filer/etcd: serve volume.fsck from one etcd revision (#12555)

* filer/etcd: scan entries from one revision

volume.fsck walked etcd one directory at a time, so a file moved from a
directory not yet listed into one already listed was seen under neither
name and its chunks were reported as orphans. The etcd store now
implements filer.EntryScanner: every page is read at the revision of the
first read, and a compacted revision fails the scan instead of falling
back to the walk.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/etcd: read hard links at the scan revision

Hard-link records resolved during a scan were read at the current
revision, so a link updated meanwhile mixed two instants. Reads made with
the scan's context now use its revision.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* shell: list etcd among the stores volume.fsck reads from a snapshot

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/etcd: skip hard-link records in a scan

Hard-link records live beside the entries under their random ids, so a
root scan could read one as an entry with a path made of random bytes,
which the traversal stream cannot send. Skip hard-link ids and keys whose
directory is not valid UTF-8.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/etcd: recognize a hard-link record by the id it stores

A hard-link id can be valid UTF-8, so the key shape alone let some
records through a root scan as phantom files. The record's value is the
shared entry, whose HardLinkId is the record's own key; no entry key
can equal its own hard-link id.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/etcd: scan entries under an invalid-UTF-8 directory

The filer sanitizes names but stores directories as given, so a path
written over HTTP can leave entries the listing never reaches. Skipping
them let volume.fsck count their chunks as orphans. Send them under the
sanitized directory instead, the substitution the filer uses for names.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer/etcd: fail a scan whose sanitized path names another entry

An entry under an invalid-UTF-8 directory is reported under the sanitized
directory. When that path is also another stored entry at the scan revision,
or another sanitized entry in the same scan, acting on the reported path
(volume.fsck deleting it) would reach the other file, so the scan fails and
names the stored path instead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

go.mod

@@ -84,6 +84,7 @@ require (
 	github.com/xdg-go/scram v1.2.0
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
+	go.etcd.io/etcd/api/v3 v3.7.1
 	go.etcd.io/etcd/client/v3 v3.7.1
 	go.mongodb.org/mongo-driver v1.17.9
 	go.opencensus.io v0.24.0 // indirect
@@ -479,7 +480,6 @@ require (
 	github.com/zeebo/blake3 v0.2.4 // indirect
 	github.com/zeebo/errs v1.4.0 // indirect
 	go.etcd.io/bbolt v1.5.0 // indirect
-	go.etcd.io/etcd/api/v3 v3.7.1 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.46.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.71.0 // indirect

weed/filer/etcd/etcd_store.go

@@ -138,7 +138,7 @@ func (store *EtcdStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (er
 func (store *EtcdStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
 	key := genKey(fullpath.DirAndName())
 
-	resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key))
+	resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...)
 	if err != nil {
 		return nil, fmt.Errorf("get %s : %v", fullpath, err)
 	}

weed/filer/etcd/etcd_store_kv.go

@@ -20,7 +20,7 @@ func (store *EtcdStore) KvPut(ctx context.Context, key []byte, value []byte) (er
 
 func (store *EtcdStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
 
-	resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key))
+	resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...)
 
 	if err != nil {
 		return nil, fmt.Errorf("kv get: %w", err)

weed/filer/etcd/etcd_store_scan.go

@@ -0,0 +1,131 @@
+package etcd
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+
+	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	weed_util "github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+var _ filer.EntryScanner = (*EtcdStore)(nil)
+
+var scanPageSize int64 = 1000
+
+type scanRevisionKey struct{}
+
+// atScanRevision pins a read made with a ScanEntries context to the scan's
+// revision, so hard-link records resolve as of the same instant.
+func atScanRevision(ctx context.Context, opts ...clientv3.OpOption) []clientv3.OpOption {
+	if rev, ok := ctx.Value(scanRevisionKey{}).(int64); ok {
+		opts = append(opts, clientv3.WithRev(rev))
+	}
+	return opts
+}
+
+// ScanEntries implements filer.EntryScanner. Every page is read at the
+// revision of the first read, so a rename committed meanwhile shows up under
+// exactly one of its names. If etcd compacts that revision before the scan
+// ends, the scan fails rather than mixing revisions.
+func (store *EtcdStore) ScanEntries(ctx context.Context, dirPath weed_util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error {
+	return scanEntries(ctx, store.client, store.etcdKeyPrefix, dirPath, fn)
+}
+
+func scanEntries(ctx context.Context, kv clientv3.KV, keyPrefix string, dirPath weed_util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error {
+	resp, err := kv.Get(ctx, keyPrefix+string(genKey(dirPath.DirAndName())))
+	if err != nil {
+		return fmt.Errorf("scan %s: %v", dirPath, err)
+	}
+	rev := resp.Header.Revision
+	scanCtx := context.WithValue(ctx, scanRevisionKey{}, rev)
+	if dirPath != "/" && len(resp.Kvs) > 0 {
+		root := &filer.Entry{FullPath: dirPath}
+		if err := root.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(resp.Kvs[0].Value)); err != nil {
+			return fmt.Errorf("scan decode %s: %v", dirPath, err)
+		}
+		if err := fn(scanCtx, root); err != nil {
+			return err
+		}
+	}
+
+	// Children of dirPath, then everything deeper; "/" covers both.
+	prefixes := []string{string(dirPath)}
+	if dirPath != "/" {
+		prefixes = []string{string(genDirectoryKeyPrefix(dirPath, "")), string(dirPath) + "/"}
+	}
+	sanitized := map[weed_util.FullPath]struct{}{}
+	for _, prefix := range prefixes {
+		if err := scanPrefix(scanCtx, kv, keyPrefix, prefix, dirPath, rev, sanitized, fn); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func scanPrefix(ctx context.Context, kv clientv3.KV, keyPrefix, prefix string, dirPath weed_util.FullPath, rev int64, sanitized map[weed_util.FullPath]struct{}, fn func(ctx context.Context, entry *filer.Entry) error) error {
+	end := clientv3.GetPrefixRangeEnd(keyPrefix + prefix)
+	for start := keyPrefix + prefix; ; {
+		resp, err := kv.Get(ctx, start, clientv3.WithRange(end), clientv3.WithLimit(scanPageSize), clientv3.WithRev(rev))
+		if errors.Is(err, rpctypes.ErrCompacted) {
+			return fmt.Errorf("scan %s: etcd compacted revision %d before the scan finished; rerun it, or keep more revisions (etcd --auto-compaction-retention): %w", dirPath, rev, err)
+		}
+		if err != nil {
+			return fmt.Errorf("scan %s: %w", dirPath, err)
+		}
+		for _, item := range resp.Kvs {
+			key := item.Key[len(keyPrefix):]
+			name := getNameFromKey(key)
+			if name == "" || len(name) == len(key) {
+				continue
+			}
+			// Names are sanitized when stored, directories are not; the stream
+			// needs valid UTF-8, and fsck must still count the chunks.
+			rawDir := string(key[:len(key)-len(name)-1])
+			dir := weed_util.SanitizeUTF8Name(rawDir)
+			entry := &filer.Entry{FullPath: weed_util.NewFullPath(dir, name)}
+			if err := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(item.Value)); err != nil {
+				return fmt.Errorf("scan decode %s: %v", entry.FullPath, err)
+			}
+			// A hard-link record is stored under its own id (filer.NewHardLinkId).
+			if bytes.Equal(entry.HardLinkId, key) {
+				continue
+			}
+			if dir != rawDir {
+				if err := claimSanitizedPath(ctx, kv, keyPrefix, rev, entry.FullPath, weed_util.NewFullPath(rawDir, name), sanitized); err != nil {
+					return err
+				}
+			}
+			if err := fn(ctx, entry); err != nil {
+				return err
+			}
+		}
+		if !resp.More || len(resp.Kvs) == 0 {
+			return nil
+		}
+		start = string(resp.Kvs[len(resp.Kvs)-1].Key) + "\x00"
+	}
+}
+
+// claimSanitizedPath fails the scan when the path an entry is reported under
+// names a different entry, so acting on it (fsck deleting it) cannot reach
+// another file.
+func claimSanitizedPath(ctx context.Context, kv clientv3.KV, keyPrefix string, rev int64, p, rawPath weed_util.FullPath, sanitized map[weed_util.FullPath]struct{}) error {
+	_, taken := sanitized[p]
+	if !taken {
+		resp, err := kv.Get(ctx, keyPrefix+string(genKey(p.DirAndName())), clientv3.WithRev(rev), clientv3.WithKeysOnly())
+		if err != nil {
+			return fmt.Errorf("scan lookup %s: %w", p, err)
+		}
+		taken = len(resp.Kvs) > 0
+	}
+	if taken {
+		return fmt.Errorf("scan: entry %q has a directory that is not valid UTF-8 and would be reported as %s, which names another entry; rename or delete one of them", string(rawPath), p)
+	}
+	sanitized[p] = struct{}{}
+	return nil
+}

weed/filer/etcd/etcd_store_scan_test.go

@@ -0,0 +1,290 @@
+package etcd
+
+import (
+	"context"
+	"errors"
+	"os"
+	"sort"
+	"testing"
+
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+type fakeVersion struct {
+	rev   int64
+	value []byte // nil marks a delete
+}
+
+// fakeMVCC keeps every revision of every key, as etcd does until compaction.
+type fakeMVCC struct {
+	clientv3.KV
+	rev       int64
+	compacted int64
+	keys      map[string][]fakeVersion
+	gets      int
+	beforeGet func(gets int)
+}
+
+func newFakeMVCC() *fakeMVCC {
+	return &fakeMVCC{keys: map[string][]fakeVersion{}}
+}
+
+func (m *fakeMVCC) put(key string, value []byte) {
+	m.rev++
+	m.keys[key] = append(m.keys[key], fakeVersion{m.rev, value})
+}
+
+func (m *fakeMVCC) del(key string) {
+	m.rev++
+	m.keys[key] = append(m.keys[key], fakeVersion{m.rev, nil})
+}
+
+func (m *fakeMVCC) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
+	m.gets++
+	if m.beforeGet != nil {
+		m.beforeGet(m.gets)
+	}
+	op := clientv3.OpGet(key, opts...)
+	rev := op.Rev()
+	if rev == 0 {
+		rev = m.rev
+	}
+	if rev < m.compacted {
+		return nil, rpctypes.ErrCompacted
+	}
+	end := string(op.RangeBytes())
+	var names []string
+	for k := range m.keys {
+		if k == key || (end != "" && k > key && k < end) {
+			names = append(names, k)
+		}
+	}
+	sort.Strings(names)
+	resp := &clientv3.GetResponse{Header: &etcdserverpb.ResponseHeader{Revision: rev}}
+	for _, k := range names {
+		var value []byte
+		for _, v := range m.keys[k] {
+			if v.rev <= rev {
+				value = v.value
+			}
+		}
+		if value == nil {
+			continue
+		}
+		if limit := op.Limit(); limit > 0 && int64(len(resp.Kvs)) == limit {
+			resp.More = true
+			break
+		}
+		resp.Kvs = append(resp.Kvs, &mvccpb.KeyValue{Key: []byte(k), Value: value})
+	}
+	return resp, nil
+}
+
+const testKeyPrefix = "pfx"
+
+func (m *fakeMVCC) putEntry(t *testing.T, p util.FullPath, isDir bool) {
+	t.Helper()
+	entry := &filer.Entry{FullPath: p}
+	if isDir {
+		entry.Attr.Mode = os.ModeDir | 0755
+	} else {
+		entry.Chunks = []*filer_pb.FileChunk{{FileId: "3,01" + p.Name(), Size: 1}}
+	}
+	meta, err := entry.EncodeAttributesAndChunks()
+	if err != nil {
+		t.Fatal(err)
+	}
+	m.put(entryKey(p), meta)
+}
+
+func entryKey(p util.FullPath) string {
+	return testKeyPrefix + string(genKey(p.DirAndName()))
+}
+
+func scanPaths(t *testing.T, m *fakeMVCC, dirPath util.FullPath) map[util.FullPath]int {
+	t.Helper()
+	seen := map[util.FullPath]int{}
+	if err := scanEntries(context.Background(), m, testKeyPrefix, dirPath, func(_ context.Context, entry *filer.Entry) error {
+		seen[entry.FullPath]++
+		return nil
+	}); err != nil {
+		t.Fatalf("scan %s: %v", dirPath, err)
+	}
+	return seen
+}
+
+func withScanPageSize(t *testing.T, n int64) {
+	prev := scanPageSize
+	scanPageSize = n
+	t.Cleanup(func() { scanPageSize = prev })
+}
+
+func TestScanEntriesSeesFileMovedDuringScan(t *testing.T) {
+	withScanPageSize(t, 2)
+	m := newFakeMVCC()
+	for _, d := range []util.FullPath{"/buckets", "/buckets/b", "/buckets/b/aaa", "/buckets/b/pad", "/buckets/b/zzz", "/buckets/b/zzz/late", "/buckets/bc"} {
+		m.putEntry(t, d, true)
+	}
+	for _, f := range []util.FullPath{"/buckets/b/pad/p0", "/buckets/b/pad/p1", "/buckets/b/pad/p2", "/buckets/b/zzz/late/f1", "/buckets/b/zzz/late/f2", "/buckets/bc/x"} {
+		m.putEntry(t, f, false)
+	}
+	m.put(testKeyPrefix+"/buckets/b/hardlink-without-separator", []byte("not an entry"))
+
+	// Pages run root, children of /buckets/b, then deeper paths in key order:
+	// the move lands after aaa's subtree was read and before late's.
+	m.beforeGet = func(gets int) {
+		if gets == 5 {
+			from, to := util.FullPath("/buckets/b/zzz/late/f1"), util.FullPath("/buckets/b/aaa/f1")
+			value := m.keys[entryKey(from)][len(m.keys[entryKey(from)])-1].value
+			m.del(entryKey(from))
+			m.put(entryKey(to), value)
+		}
+	}
+
+	seen := scanPaths(t, m, "/buckets/b")
+	want := []util.FullPath{"/buckets/b", "/buckets/b/aaa", "/buckets/b/pad", "/buckets/b/zzz", "/buckets/b/zzz/late",
+		"/buckets/b/pad/p0", "/buckets/b/pad/p1", "/buckets/b/pad/p2", "/buckets/b/zzz/late/f1", "/buckets/b/zzz/late/f2"}
+	for _, p := range want {
+		if seen[p] != 1 {
+			t.Errorf("%s seen %d times, want once", p, seen[p])
+		}
+	}
+	if len(seen) != len(want) {
+		t.Errorf("seen %v, want exactly %v", seen, want)
+	}
+	if m.gets < 5 {
+		t.Fatalf("the move never ran: %d reads", m.gets)
+	}
+	if _, found := m.keys[entryKey("/buckets/b/aaa/f1")]; !found {
+		t.Fatal("the move did not land")
+	}
+}
+
+func TestScanEntriesFromRoot(t *testing.T) {
+	withScanPageSize(t, 3)
+	m := newFakeMVCC()
+	m.putEntry(t, "/a", true)
+	m.putEntry(t, "/a/x", false)
+	m.putEntry(t, "/top", false)
+	m.put(testKeyPrefix+"filer.store.id", []byte{1, 2, 3, 4})
+	m.put(testKeyPrefix+"/\x00", []byte("root marker"))
+	for _, id := range []string{"/\x00\xff\xfe\x80abcdefghijk\x01", "/\x00abcdefghijklmn\x01"} {
+		link, err := (&filer.Entry{FullPath: "/top", HardLinkId: filer.HardLinkId(id), HardLinkCounter: 2}).EncodeAttributesAndChunks()
+		if err != nil {
+			t.Fatal(err)
+		}
+		m.put(testKeyPrefix+id, link)
+	}
+	m.putEntry(t, "/\xc3\x28xyz/name", false)
+
+	seen := scanPaths(t, m, "/")
+	if len(seen) != 4 || seen["/a"] != 1 || seen["/a/x"] != 1 || seen["/top"] != 1 || seen["/_(xyz/name"] != 1 {
+		t.Fatalf("seen %v", seen)
+	}
+}
+
+func TestScanEntriesReadsHardLinkAtScanRevision(t *testing.T) {
+	m := newFakeMVCC()
+	m.putEntry(t, "/d", true)
+	m.putEntry(t, "/d/f", false)
+	m.put(testKeyPrefix+"link", []byte("v1"))
+	m.beforeGet = func(gets int) {
+		if gets == 2 {
+			m.put(testKeyPrefix+"link", []byte("v2"))
+		}
+	}
+	var got []string
+	err := scanEntries(context.Background(), m, testKeyPrefix, "/d", func(ctx context.Context, entry *filer.Entry) error {
+		resp, err := m.Get(ctx, testKeyPrefix+"link", atScanRevision(ctx)...)
+		if err != nil {
+			return err
+		}
+		got = append(got, string(resp.Kvs[0].Value))
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, v := range got {
+		if v != "v1" {
+			t.Fatalf("hard link read %q inside the scan, want the scan revision's v1", got)
+		}
+	}
+	if len(got) != 2 {
+		t.Fatalf("visited %d entries, want 2", len(got))
+	}
+}
+
+func TestScanEntriesFailsWhenRevisionCompacted(t *testing.T) {
+	withScanPageSize(t, 1)
+	m := newFakeMVCC()
+	m.putEntry(t, "/d", true)
+	m.putEntry(t, "/d/a", false)
+	m.putEntry(t, "/d/b", false)
+	m.beforeGet = func(gets int) {
+		if gets == 3 {
+			m.putEntry(t, "/d/c", false)
+			m.compacted = m.rev
+		}
+	}
+	err := scanEntries(context.Background(), m, testKeyPrefix, "/d", func(context.Context, *filer.Entry) error { return nil })
+	if !errors.Is(err, rpctypes.ErrCompacted) {
+		t.Fatalf("got %v, want a compaction error", err)
+	}
+	if errors.Is(err, filer.ErrUnsupportedEntryScan) {
+		t.Fatalf("a compacted scan must fail, not fall back to the walk: %v", err)
+	}
+}
+
+func TestScanEntriesRejectsSanitizedPathOfAnotherEntry(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		paths   []util.FullPath
+		lateDup bool
+		wantErr bool
+	}{
+		{name: "free", paths: []util.FullPath{"/\xc3\x28xyz/name"}},
+		{name: "stored entry", paths: []util.FullPath{"/\xc3\x28xyz/name", "/_(xyz/name"}, wantErr: true},
+		{name: "two invalid directories", paths: []util.FullPath{"/a\xff/b", "/a\xfe/b"}, wantErr: true},
+		{name: "created after the scan revision", paths: []util.FullPath{"/\xc3\x28xyz/name"}, lateDup: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			withScanPageSize(t, 1)
+			m := newFakeMVCC()
+			for _, p := range tc.paths {
+				m.putEntry(t, p, false)
+			}
+			if tc.lateDup {
+				m.beforeGet = func(gets int) {
+					if gets == 2 {
+						m.putEntry(t, "/_(xyz/name", false)
+					}
+				}
+			}
+			var seen []util.FullPath
+			err := scanEntries(context.Background(), m, testKeyPrefix, "/", func(_ context.Context, entry *filer.Entry) error {
+				seen = append(seen, entry.FullPath)
+				return nil
+			})
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("scan succeeded with %v, want a collision error", seen)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(seen) != 1 || seen[0] != "/_(xyz/name" {
+				t.Fatalf("seen %v", seen)
+			}
+		})
+	}
+}

weed/shell/command_volume_fsck.go

@@ -100,10 +100,10 @@ func (c *commandVolumeFsck) Help() string {
 		The default 5h cutoff provides sufficient buffer for metadata commits.
 
 	Step 2 reads the filer from one consistent snapshot when its store supports
-	that (the SQL stores do), so a file renamed while the filer is being read is
-	still counted and its chunks are not mistaken for orphans. Other stores are
-	walked directory by directory; there, avoid running with -reallyDeleteFromVolume
-	while files are being renamed or moved.
+	that (the SQL stores and etcd do), so a file renamed while the filer is being
+	read is still counted and its chunks are not mistaken for orphans. Other
+	stores are walked directory by directory; there, avoid running with
+	-reallyDeleteFromVolume while files are being renamed or moved.
 
 	With -findMissingChunksInFiler, -volumeId may name a volume the master no
 	longer lists, to find the filer entries that still reference it. A volume is