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 <bucketsPath>/<collection> 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 <noreply@anthropic.com>

* test: volume.fsck keeps chunks referenced outside the volume's bucket

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: user.email <chrislusf@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-10-26 07:58:43 -07:00
1 parent 937633a77d
commit b5af82409f
2 files changed
+119 -108

No files matched your search

+4 -77
View File
@@ -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, "<expert only!> 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, "<expert only!> 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))
}
+115 -31
View File
@@ -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.