From b5af82409f35ed3c79b688129a88a6af6c1c24a7 Mon Sep 17 00:00:00 2001 From: Jaehoon Kim Date: Mon, 26 Oct 2026 23:58:43 +0900 Subject: [PATCH] fix: volume.fsck -volumeId purges live chunks referenced from outside the volume's bucket (#12076) * fix: volume.fsck -volumeId purges live chunks referenced from outside the volume's bucket -volumeId narrowed the filer read to / when the selected volumes shared one bucketed collection (#9347). A filer started with -collection, or an fs.configure rule, stores files from any path in that collection, so the narrowed read misses their references and reports live chunks as orphans; with -reallyDeleteFromVolume they are deleted. #11794 already made -collection read from the root for this reason. Do the same for -volumeId: the flag keeps selecting which volumes are checked and purged, and the filer is always read from "/". Co-authored-by: Claude Fable 5.1 * test: volume.fsck keeps chunks referenced outside the volume's bucket --------- Co-authored-by: Claude Fable 5.1 Co-authored-by: user.email --- weed/shell/command_volume_fsck.go | 81 +------------- weed/shell/command_volume_fsck_test.go | 146 +++++++++++++++++++------ 2 files changed, 119 insertions(+), 108 deletions(-) diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index ff3c3fe0e8..67b0fa08e5 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -59,7 +59,6 @@ type commandVolumeFsck struct { collectionMatcher *wildcard.CollectionMatcher knownVolumeIds map[uint32]bool volumeIds map[uint32]bool - scopedFilerPath string tempFolder string verbose *bool forcePurging *bool @@ -120,7 +119,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. c.skipEcVolumes = fsckCommand.Bool("skipEcVolumes", false, "skip erasure coded volumes") c.findMissingChunksInFiler = fsckCommand.Bool("findMissingChunksInFiler", false, "see \"help volume.fsck\"") c.collection = fsckCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty or \"*\" matches all, \""+CollectionDefault+"\" the collection with no name") - volumeIds := fsckCommand.String("volumeId", "", "comma separated the volume id") + volumeIds := fsckCommand.String("volumeId", "", "comma separated the volume id; the filer is still read from its root, since a file under any path may reference these volumes") applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, " after detection, delete missing data from volumes / delete missing file entries from filer. Currently this only works with default filerGroup.") c.forcePurging = fsckCommand.Bool("forcePurging", false, "delete missing data from volumes in one replica used together with applyPurging") purgeAbsent := fsckCommand.Bool("reallyDeleteFilerEntries", false, " delete missing file entries from filer if the corresponding volume is missing for any reason, please ensure all still existing/expected volumes are connected! used together with findMissingChunksInFiler") @@ -218,11 +217,6 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. } } - c.scopedFilerPath = c.resolveScopedFilerPath(dataNodeVolumeIdToVInfo) - if *c.verbose && c.scopedFilerPath != "/" { - fmt.Fprintf(c.writer, "scoping filer walk to %s\n", c.scopedFilerPath) - } - var collectCutoffFromAtNs int64 = 0 if cutoffTimeAgo.Seconds() != 0 { collectCutoffFromAtNs = time.Now().Add(-*cutoffTimeAgo).UnixNano() @@ -322,7 +316,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. 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()) + fmt.Fprintf(c.writer, "checking each file from filer path /...\n") } files := make(map[uint32]*os.File) @@ -344,7 +338,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m } }() - return doTraverseAndSaving(c.env, c.writer, c.getCollectFilerFilePath(), false, false, + return doTraverseAndSaving(c.env, c.writer, "/", false, false, func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) { if *c.verbose && entry.Entry.IsDirectory { fmt.Fprintf(c.writer, "checking directory %s\n", util.NewFullPath(entry.Dir, entry.Entry.Name)) @@ -900,8 +894,7 @@ func (c *commandVolumeFsck) deleteEmptyDirectory(dir util.FullPath, attr *filer_ } func (c *commandVolumeFsck) canPurgeDirectory(dir util.FullPath) bool { - root := c.getCollectFilerFilePath() - if string(dir) == root || !strings.HasPrefix(string(dir), strings.TrimSuffix(root, "/")+"/") { + if string(dir) == "/" { return false } // deleting a bucket drops its whole collection @@ -1082,72 +1075,6 @@ func (c *commandVolumeFsck) purgeFileIdsForOneVolume(volumeId uint32, fileIds [] return } -func (c *commandVolumeFsck) getCollectFilerFilePath() string { - return c.scopedFilerPath -} - -func (c *commandVolumeFsck) resolveScopedFilerPath(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo) string { - if *c.collection != "" { - // fs.configure can bind a collection to any path prefix, so files - // referencing the filtered volumes may live anywhere. A bucket-dir - // walk could miss them and report their live chunks as orphans; - // only a full walk is safe for a collection filter. - return "/" - } - if len(c.volumeIds) == 0 { - return "/" - } - collections := make(map[string]struct{}) - for _, vidMap := range dataNodeVolumeIdToVInfo { - for vid, vinfo := range vidMap { - if _, ok := c.volumeIds[vid]; !ok { - continue - } - // empty collection: volume can be referenced from anywhere - if vinfo.collection == "" { - return "/" - } - collections[vinfo.collection] = struct{}{} - if len(collections) > 1 { - return "/" - } - } - } - if len(collections) != 1 { - return "/" - } - var collection string - for col := range collections { - collection = col - } - exists, err := c.bucketDirExists(collection) - if err != nil || !exists { - return "/" - } - return fmt.Sprintf("%s/%s", c.bucketsPath, collection) -} - -func (c *commandVolumeFsck) bucketDirExists(name string) (bool, error) { - var found bool - err := c.env.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{ - Directory: c.bucketsPath, - Name: name, - }) - if err != nil { - if strings.Contains(err.Error(), filer_pb.ErrNotFound.Error()) { - return nil - } - return err - } - if resp.Entry != nil && resp.Entry.IsDirectory { - found = true - } - return nil - }) - return found, err -} - func getVolumeFileIdFile(tempFolder string, dataNodeid string, vid uint32) string { return filepath.Join(tempFolder, fmt.Sprintf("%s_%d.idx", dataNodeid, vid)) } diff --git a/weed/shell/command_volume_fsck_test.go b/weed/shell/command_volume_fsck_test.go index 43bb07efea..0c742f82ab 100644 --- a/weed/shell/command_volume_fsck_test.go +++ b/weed/shell/command_volume_fsck_test.go @@ -27,25 +27,20 @@ import ( func TestVolumeFsckCanPurgeDirectory(t *testing.T) { testCases := []struct { - scopedFilerPath string - dir util.FullPath - expected bool + dir util.FullPath + expected bool }{ - {"/", "/orphan/dir/deep", true}, - {"/", "/orphan", true}, - {"/", "/", false}, - {"/", "/buckets", false}, - {"/", "/buckets/bucket1", false}, - {"/", "/buckets/bucket1/dir", true}, - {"/buckets/bucket1", "/buckets/bucket1", false}, - {"/buckets/bucket1", "/buckets/bucket1/dir", true}, - {"/buckets/bucket1", "/buckets/bucket11/dir", false}, - {"/buckets/bucket1", "/orphan/dir", false}, + {"/orphan/dir/deep", true}, + {"/orphan", true}, + {"/", false}, + {"/buckets", false}, + {"/buckets/bucket1", false}, + {"/buckets/bucket1/dir", true}, } for _, tc := range testCases { - c := &commandVolumeFsck{bucketsPath: "/buckets", scopedFilerPath: tc.scopedFilerPath} + c := &commandVolumeFsck{bucketsPath: "/buckets"} if actual := c.canPurgeDirectory(tc.dir); actual != tc.expected { - t.Errorf("scope %s: canPurgeDirectory(%s) = %v, expected %v", tc.scopedFilerPath, tc.dir, actual, tc.expected) + t.Errorf("canPurgeDirectory(%s) = %v, expected %v", tc.dir, actual, tc.expected) } } } @@ -101,6 +96,37 @@ func (s *stubFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_p return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil } +func (s *stubFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error { + var names []string + children := make(map[string]*filer_pb.Entry) + for path, entry := range s.entries { + parent, name := util.FullPath(path).DirAndName() + if parent != req.Directory || name == "" { + continue + } + if name < req.StartFromFileName || (name == req.StartFromFileName && !req.InclusiveStartFrom) { + continue + } + names = append(names, name) + children[name] = entry + } + sort.Strings(names) + for _, name := range names { + stored := children[name] + entry := &filer_pb.Entry{ + Name: name, + IsDirectory: stored.IsDirectory, + Chunks: stored.Chunks, + Attributes: stored.Attributes, + HardLinkId: stored.HardLinkId, + } + if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: entry}); err != nil { + return err + } + } + return nil +} + func (s *stubFilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { fullPath := string(util.NewFullPath(req.Directory, req.Name)) entry, found := s.entries[fullPath] @@ -164,10 +190,9 @@ func TestVolumeFsckPurgeEmptyDirectories(t *testing.T) { verbose := false c := &commandVolumeFsck{ - verbose: &verbose, - writer: io.Discard, - bucketsPath: "/buckets", - scopedFilerPath: "/", + verbose: &verbose, + writer: io.Discard, + bucketsPath: "/buckets", purgedDirs: map[util.FullPath]struct{}{ "/orphan/dir/deep": {}, "/orphan/dir/wide": {}, @@ -197,12 +222,11 @@ func TestVolumeFsckPurgeEmptyDirectoriesKeepsFreshDirectory(t *testing.T) { verbose := false c := &commandVolumeFsck{ - verbose: &verbose, - writer: io.Discard, - bucketsPath: "/buckets", - scopedFilerPath: "/", - purgedDirs: map[util.FullPath]struct{}{"/fresh/dir": {}}, - env: startStubFiler(t, stub), + verbose: &verbose, + writer: io.Discard, + bucketsPath: "/buckets", + purgedDirs: map[util.FullPath]struct{}{"/fresh/dir": {}}, + env: startStubFiler(t, stub), } c.purgeEmptyDirectories() @@ -223,12 +247,11 @@ func TestVolumeFsckPurgeEmptyDirectoriesKeepsChangedDirectory(t *testing.T) { verbose := false c := &commandVolumeFsck{ - verbose: &verbose, - writer: io.Discard, - bucketsPath: "/buckets", - scopedFilerPath: "/", - purgedDirs: map[util.FullPath]struct{}{"/race/dir": {}}, - env: startStubFiler(t, stub), + verbose: &verbose, + writer: io.Discard, + bucketsPath: "/buckets", + purgedDirs: map[util.FullPath]struct{}{"/race/dir": {}}, + env: startStubFiler(t, stub), } c.purgeEmptyDirectories() @@ -321,6 +344,67 @@ func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) { } } +// A file outside the selected volumes' bucket can still reference their +// chunks — a filer can bind a collection to any path — so the filer walk must +// reach them or their live chunks would be reported as orphans. +func TestVolumeFsckCollectFilerFileIdsReachesOutsideBucket(t *testing.T) { + tempDir := t.TempDir() + chunk := func(fileKey uint64) *filer_pb.FileChunk { + return &filer_pb.FileChunk{Size: 1, Fid: &filer_pb.FileId{VolumeId: 7, FileKey: fileKey, Cookie: 1}} + } + directory := &filer_pb.Entry{IsDirectory: true} + stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{ + "/": directory, + "/buckets": directory, + "/buckets/col1": directory, + "/buckets/col1/object": {Chunks: []*filer_pb.FileChunk{chunk(11)}}, + "/topics": directory, + "/topics/system-log": {Chunks: []*filer_pb.FileChunk{chunk(22)}}, + }} + + verbose, findMissing, verify, force := false, false, false, false + c := &commandVolumeFsck{ + tempFolder: tempDir, + verbose: &verbose, + findMissingChunksInFiler: &findMissing, + verifyNeedle: &verify, + forcePurging: &force, + writer: io.Discard, + bucketsPath: "/buckets", + env: startStubFiler(t, stub), + } + + vinfo := map[string]map[uint32]VInfo{"dn1": {7: {collection: "col1"}}} + if err := c.collectFilerFileIdAndPaths(vinfo, false, func(purgeItem) {}, 0, 0); err != nil { + t.Fatalf("collect filer file ids: %v", err) + } + + db := needle_map.NewMemDb() + for _, needleId := range []types.NeedleId{11, 22, 33} { + if err := db.Set(needleId, types.ToOffset(16), types.Size(1)); err != nil { + t.Fatalf("seed volume idx: %v", err) + } + } + if err := db.SaveToIdx(getVolumeFileIdFile(tempDir, "dn1", 7)); err != nil { + t.Fatalf("write volume idx: %v", err) + } + db.Close() + + inUseCount, orphanFileIds, _, err := c.oneVolumeFileIdsSubtractFilerFileIds("dn1", 7, &VInfo{collection: "col1"}, 0, 0) + if err != nil { + t.Fatalf("subtract filer file ids: %v", err) + } + if inUseCount != 2 { + t.Fatalf("in-use count %d, want 2 (the in-bucket and out-of-bucket references)", inUseCount) + } + if len(orphanFileIds) != 1 { + t.Fatalf("orphans %v, want only the unreferenced needle", orphanFileIds) + } + if want := types.NeedleId(33).FileId(7); orphanFileIds[0] != want { + t.Fatalf("orphan %s, want the unreferenced needle %s", orphanFileIds[0], want) + } +} + // 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.