From 679835e4fc5959e2f054ca09a7d11f50b5b45b86 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 23 Oct 2026 18:33:51 -0700 Subject: [PATCH] volume.fsck: run all filer-entry purges through the shared worker pool (#11960) * volume.fsck: queue per-volume orphan deletes through the shared purge pool The worker pool added for -reallyDeleteFilerEntries only covered entries whose volume was missing entirely; the per-volume check that finds filer entries pointing at absent needles still issued one synchronous HTTP DELETE per orphan, which is the bulk of the work the report measured. One pool now serves both purge phases and is drained before the empty- directory sweep. * volume.fsck: keep hard link ids in the filer fid file Orphans queued by the per-volume check must hash into the pool by their hard link id just like the missing-volume path, so two paths sharing an id still serialize on one worker. The fid file now stores the id beside each path and readFilerFileIdFile hands it to the caller. * volume.fsck: cover orphan purge queuing and hard link id round-trip * volume.fsck: fail on a truncated hard link id in the fid file A .fid record cut short inside the hard link id length was emitted with a nil id, so its purge could hash to a different worker than the entries it shares the link with. Read the length strictly like the header and path. * volume.fsck: drain the save consumer before returning a root genFn error The early return left the save goroutine blocked on outputChan while the caller ran finishPurge, so a late queuePurge could send on a closed channel and panic. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * storage: skip EC shard-size encoding tests when disk space is low TestCalculateExpectedShardSizeWithRealEncoding writes a .dat file plus 14 EC shards per subtest (~1.3GB peak for the 512MB case) into the temp dir. Shared CI runners can run low on space while other test packages write to /tmp in parallel, and the encode then dies with ENOSPC mid-write, surfacing as a test failure. Check free space up front and skip the subtest instead. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/shell/command_fs_meta_save.go | 3 + weed/shell/command_volume_fsck.go | 71 +++++++---- weed/shell/command_volume_fsck_test.go | 110 ++++++++++++++++++ .../disk_location_ec_realworld_test.go | 28 ++++- 4 files changed, 183 insertions(+), 29 deletions(-) diff --git a/weed/shell/command_fs_meta_save.go b/weed/shell/command_fs_meta_save.go index b936252f62..4034655350 100644 --- a/weed/shell/command_fs_meta_save.go +++ b/weed/shell/command_fs_meta_save.go @@ -323,6 +323,9 @@ func doTraverseBfsAndSaving(filerClient filer_pb.FilerClient, writer io.Writer, firstErr = genErr hasErr.Store(true) }) + close(outputChan) + wg.Wait() + <-saveErrChan return genErr } else { if e.IsDirectory { diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index 7578c71823..e3c301992d 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -271,19 +271,33 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. if *c.findMissingChunksInFiler { // collect all filer file ids and paths - if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, *purgeAbsent, collectModifyFromAtNs, collectCutoffFromAtNs); err != nil { - return fmt.Errorf("collectFilerFileIdAndPaths: %w", err) + var queuePurge func(purgeItem) + var finishPurge func() + if *applyPurging || *purgeAbsent { + queuePurge, finishPurge = startPurgeWorkers(*c.purgeConcurrency, c.httpDelete) } - for dataNodeId, volumeIdToVInfo := range dataNodeVolumeIdToVInfo { - // for each volume, check filer file ids - if err = c.findFilerChunksMissingInVolumeServers(volumeIdToVInfo, dataNodeId, *applyPurging || *purgeAbsent); err != nil { - return fmt.Errorf("findFilerChunksMissingInVolumeServers: %w", err) + err = func() error { + if err := c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, *purgeAbsent, queuePurge, collectModifyFromAtNs, collectCutoffFromAtNs); err != nil { + return fmt.Errorf("collectFilerFileIdAndPaths: %w", err) } + for dataNodeId, volumeIdToVInfo := range dataNodeVolumeIdToVInfo { + // for each volume, check filer file ids + if err := c.findFilerChunksMissingInVolumeServers(volumeIdToVInfo, dataNodeId, queuePurge); err != nil { + return fmt.Errorf("findFilerChunksMissingInVolumeServers: %w", err) + } + } + return nil + }() + if finishPurge != nil { + finishPurge() + } + if err != nil { + return err } c.purgeEmptyDirectories() } else { // collect all filer file ids - if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, false, 0, 0); err != nil { + if err = c.collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo, false, nil, 0, 0); err != nil { return fmt.Errorf("failed to collect file ids from filer: %w", err) } // If any entry's manifest could not be resolved, our in-use fid set @@ -306,7 +320,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. return nil } -func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, collectModifyFromAtNs int64, cutoffFromAtNs int64) error { +func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, queuePurge func(purgeItem), collectModifyFromAtNs int64, cutoffFromAtNs int64) error { if *c.verbose { fmt.Fprintf(c.writer, "checking each file from filer path %s...\n", c.getCollectFilerFilePath()) } @@ -382,12 +396,6 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m }, func(outputChan chan interface{}) error { buffer := make([]byte, readbufferSize) - var queuePurge func(purgeItem) - var finishPurge func() - if purgeAbsent { - queuePurge, finishPurge = startPurgeWorkers(*c.purgeConcurrency, c.httpDelete) - defer finishPurge() - } for item := range outputChan { i := item.(*Item) if f, ok := files[i.vid]; ok { @@ -400,6 +408,13 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m if _, err := f.Write([]byte(i.path)); err != nil { return err } + util.Uint32toBytes(buffer[:4], uint32(len(i.hardLinkId))) + if _, err := f.Write(buffer[:4]); err != nil { + return err + } + if _, err := f.Write(i.hardLinkId); err != nil { + return err + } } else if *c.findMissingChunksInFiler { // check if the volume matches the filter if len(c.volumeIds) > 0 { @@ -423,10 +438,10 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m }) } -func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInfo map[uint32]VInfo, dataNodeId string, applyPurging bool) error { +func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInfo map[uint32]VInfo, dataNodeId string, queuePurge func(purgeItem)) error { for volumeId, vinfo := range volumeIdToVInfo { - checkErr := c.oneVolumeFileIdsCheckOneVolume(dataNodeId, volumeId, applyPurging) + checkErr := c.oneVolumeFileIdsCheckOneVolume(dataNodeId, volumeId, queuePurge) if checkErr != nil { return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, checkErr) } @@ -650,7 +665,7 @@ func (lw *lockedWriter) Write(p []byte) (int, error) { return lw.w.Write(p) } -func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleId types.NeedleId, itemPath util.FullPath)) error { +func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleId types.NeedleId, itemPath util.FullPath, hardLinkId []byte)) error { fp, err := os.Open(getFilerFileIdFile(c.tempFolder, volumeId)) if err != nil { return err @@ -685,13 +700,23 @@ func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleI return fmt.Errorf("read fid path size mismatch for volume %d,%x%08x: got %d want %d", volumeId, item.fileKey, item.cookie, n, pathSize) } item.path = util.FullPath(pathBytes) + item.hardLinkId = nil + if _, err := io.ReadFull(br, buffer[:4]); err != nil { + return fmt.Errorf("read fid hard link id size for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err) + } + if hardLinkIdSize := util.BytesToUint32(buffer[:4]); hardLinkIdSize > 0 { + item.hardLinkId = make([]byte, int(hardLinkIdSize)) + if _, err := io.ReadFull(br, item.hardLinkId); err != nil { + return fmt.Errorf("read fid hard link id for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err) + } + } needleId := types.NeedleId(item.fileKey) - fn(needleId, item.path) + fn(needleId, item.path, item.hardLinkId) } return nil } -func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, volumeId uint32, applyPurging bool) (err error) { +func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, volumeId uint32, queuePurge func(purgeItem)) (err error) { if *c.verbose { fmt.Fprintf(c.writer, "find missing file chunks in dataNodeId %s volume %d ...\n", dataNodeId, volumeId) } @@ -702,11 +727,11 @@ func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, vo if err = db.LoadFromIdx(getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)); err != nil { return } - if err = c.readFilerFileIdFile(volumeId, func(needleId types.NeedleId, itemPath util.FullPath) { + if err = c.readFilerFileIdFile(volumeId, func(needleId types.NeedleId, itemPath util.FullPath, hardLinkId []byte) { if _, found := db.Get(needleId); !found { fmt.Fprintf(c.writer, "%s\n", itemPath) - if applyPurging { - c.httpDelete(itemPath) + if queuePurge != nil { + queuePurge(purgeItem{path: itemPath, hardLinkId: hardLinkId}) } } }); err != nil { @@ -889,7 +914,7 @@ func (c *commandVolumeFsck) oneVolumeFileIdsSubtractFilerFileIds(dataNodeId stri return } - if err = c.readFilerFileIdFile(volumeId, func(filerNeedleId types.NeedleId, itemPath util.FullPath) { + if err = c.readFilerFileIdFile(volumeId, func(filerNeedleId types.NeedleId, itemPath util.FullPath, _ []byte) { inUseCount++ if *c.verifyNeedle && !vinfo.isEcVolume { if needleValue, ok := volumeFileIdDb.Get(filerNeedleId); ok && !needleValue.Size.IsDeleted() { diff --git a/weed/shell/command_volume_fsck_test.go b/weed/shell/command_volume_fsck_test.go index 50bdd0f8fa..7804b85224 100644 --- a/weed/shell/command_volume_fsck_test.go +++ b/weed/shell/command_volume_fsck_test.go @@ -1,12 +1,14 @@ package shell import ( + "bytes" "context" "fmt" "io" "net" "net/http" "net/http/httptest" + "os" "sort" "strings" "sync/atomic" @@ -16,6 +18,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" + "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -310,3 +314,109 @@ func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) { t.Fatalf("processed %d deletes, want 8", got) } } + +// Orphans found by the per-volume check go through the same purge queue as +// the missing-volume path, and the fid file keeps each entry's hard link id +// so same-id deletes still land on one worker. +func TestVolumeFsckCheckOneVolumeQueuesOrphansWithHardLinkId(t *testing.T) { + tempDir := t.TempDir() + var verbose bool + var buf bytes.Buffer + c := &commandVolumeFsck{tempFolder: tempDir, verbose: &verbose, writer: &buf} + + volumeId := uint32(7) + db := needle_map.NewMemDb() + if err := db.Set(types.NeedleId(11), types.ToOffset(16), types.Size(1)); err != nil { + t.Fatalf("seed volume idx: %v", err) + } + if err := db.SaveToIdx(getVolumeFileIdFile(tempDir, "dn1", volumeId)); err != nil { + t.Fatalf("write volume idx: %v", err) + } + db.Close() + + f, err := os.Create(getFilerFileIdFile(tempDir, volumeId)) + if err != nil { + t.Fatalf("create fid file: %v", err) + } + writeFid := func(fileKey uint64, cookie uint32, p string, hardLinkId []byte) { + buffer := make([]byte, readbufferSize) + util.Uint64toBytes(buffer, fileKey) + util.Uint32toBytes(buffer[8:], cookie) + util.Uint32toBytes(buffer[12:], uint32(len(p))) + if _, err := f.Write(buffer); err != nil { + t.Fatalf("write fid header: %v", err) + } + if _, err := f.Write([]byte(p)); err != nil { + t.Fatalf("write fid path: %v", err) + } + util.Uint32toBytes(buffer[:4], uint32(len(hardLinkId))) + if _, err := f.Write(buffer[:4]); err != nil { + t.Fatalf("write fid hard link id size: %v", err) + } + if _, err := f.Write(hardLinkId); err != nil { + t.Fatalf("write fid hard link id: %v", err) + } + } + writeFid(11, 1, "/keep/a", nil) + writeFid(22, 2, "/orphan/b", []byte{1, 2, 3}) + writeFid(33, 3, "/orphan/c", nil) + if err := f.Close(); err != nil { + t.Fatalf("close fid file: %v", err) + } + + var queued []purgeItem + if err := c.oneVolumeFileIdsCheckOneVolume("dn1", volumeId, func(it purgeItem) { + queued = append(queued, it) + }); err != nil { + t.Fatalf("check volume: %v", err) + } + if len(queued) != 2 { + t.Fatalf("queued %d purges, want 2 (the two orphans): %v", len(queued), queued) + } + if queued[0].path != "/orphan/b" || queued[1].path != "/orphan/c" { + t.Fatalf("queued paths %v, want /orphan/b then /orphan/c", queued) + } + if !bytes.Equal(queued[0].hardLinkId, []byte{1, 2, 3}) { + t.Fatalf("queued hardLinkId %v, want [1 2 3]", queued[0].hardLinkId) + } + if queued[1].hardLinkId != nil { + t.Fatalf("queued hardLinkId %v, want nil", queued[1].hardLinkId) + } +} + +// A .fid file cut short inside the hard link id field is corrupt input, not a +// record without a link id: the reader must fail instead of emitting the entry +// with a nil id. +func TestReadFilerFileIdFileRejectsTruncatedHardLinkId(t *testing.T) { + tempDir := t.TempDir() + var verbose bool + c := &commandVolumeFsck{tempFolder: tempDir, verbose: &verbose, writer: io.Discard} + + f, err := os.Create(getFilerFileIdFile(tempDir, 7)) + if err != nil { + t.Fatalf("create fid file: %v", err) + } + buffer := make([]byte, readbufferSize) + util.Uint64toBytes(buffer, 11) + util.Uint32toBytes(buffer[8:], 1) + util.Uint32toBytes(buffer[12:], uint32(len("/orphan/b"))) + if _, err := f.Write(buffer); err != nil { + t.Fatalf("write fid header: %v", err) + } + if _, err := f.Write([]byte("/orphan/b")); err != nil { + t.Fatalf("write fid path: %v", err) + } + if _, err := f.Write([]byte{0, 0}); err != nil { + t.Fatalf("write partial hard link id size: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("close fid file: %v", err) + } + + err = c.readFilerFileIdFile(7, func(types.NeedleId, util.FullPath, []byte) { + t.Fatal("callback fired for a truncated record") + }) + if err == nil { + t.Fatal("expected an error for a truncated hard link id size") + } +} diff --git a/weed/storage/disk_location_ec_realworld_test.go b/weed/storage/disk_location_ec_realworld_test.go index 42ce29bd78..f54682a47d 100644 --- a/weed/storage/disk_location_ec_realworld_test.go +++ b/weed/storage/disk_location_ec_realworld_test.go @@ -5,9 +5,24 @@ import ( "path/filepath" "testing" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" ) +// requireDiskSpaceForEncoding skips the subtest when dir's filesystem cannot +// hold the .dat file plus the shards EC encoding writes for it, so ENOSPC on +// space-constrained runners doesn't surface as a test failure. +func requireDiskSpaceForEncoding(t *testing.T, dir string, datFileSize int64, expectedShardSize int64) { + t.Helper() + needBytes := uint64(datFileSize) + + uint64(expectedShardSize)*erasure_coding.TotalShardsCount + + 16*erasure_coding.ErasureCodingSmallBlockSize + if free := stats.NewDiskStatus(dir).Free; free < needBytes { + t.Skipf("insufficient free space in %s: have %d bytes, need ~%d to encode a %d-byte volume", + dir, free, needBytes, datFileSize) + } +} + // TestCalculateExpectedShardSizeWithRealEncoding validates our shard size calculation // by actually running EC encoding on real files and comparing the results func TestCalculateExpectedShardSizeWithRealEncoding(t *testing.T) { @@ -52,6 +67,10 @@ func TestCalculateExpectedShardSizeWithRealEncoding(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + // Calculate expected shard size using our function + expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) + requireDiskSpaceForEncoding(t, tempDir, tt.datFileSize, expectedShardSize) + // Create a test .dat file with the specified size baseFileName := filepath.Join(tempDir, "test_volume") datFileName := baseFileName + ".dat" @@ -82,9 +101,6 @@ func TestCalculateExpectedShardSizeWithRealEncoding(t *testing.T) { } datFile.Close() - // Calculate expected shard size using our function - expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) - // Run actual EC encoding _, err = erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()) if err != nil { @@ -146,6 +162,9 @@ func TestCalculateExpectedShardSizeEdgeCases(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) + requireDiskSpaceForEncoding(t, tempDir, tt.datFileSize, expectedShardSize) + baseFileName := filepath.Join(tempDir, tt.name) datFileName := baseFileName + ".dat" @@ -163,9 +182,6 @@ func TestCalculateExpectedShardSizeEdgeCases(t *testing.T) { datFile.Write(data) datFile.Close() - // Calculate expected - expectedShardSize := calculateExpectedShardSize(tt.datFileSize, erasure_coding.DataShardsCount) - // Run actual EC encoding _, err = erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()) if err != nil {