From 7c3c5ed2a455b00c9aa1363821d77ecb0cbcb8d4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 26 Jun 2026 10:27:18 -0700 Subject: [PATCH] fix(filer.sync.verify): sort listings client-side before merge (#10117) * fix(filer.sync.verify): sort listings client-side before merge The merge walks both filers' directory listings in lockstep and needs them in the same byte order. A filer before 4.32 with a locale SQL collation lists case-insensitively while a 4.32+ peer lists byte-ordered, so comparing two such clusters returns the same names in a different order and the merge desyncs into spurious MISSING / ONLY_IN_B. Buffer and sort each directory client-side so both sides agree on order regardless of filer version or store backend. Trades the streaming source's O(buffer) memory for O(directory) per side, fine for a one-shot verify CLI; both sides still load concurrently. Claude-Session: https://claude.ai/code/session_01BKsBdKYFNCEjeHLjJfumPF * fix(filer.sync.verify): surface listing errors before merging A listing that fails mid-stream leaves a partial, unsorted buffer. Now that both sides are fully buffered anyway, check each side's error right after the loads finish and before the merge, so partial entries can't emit spurious MISSING / ONLY_IN_B before the error aborts the run. Claude-Session: https://claude.ai/code/session_01BKsBdKYFNCEjeHLjJfumPF --- weed/command/filer_sync_verify.go | 123 ++++++++++--------- weed/command/filer_sync_verify_test.go | 158 +++++++++++++++++++++++-- 2 files changed, 219 insertions(+), 62 deletions(-) diff --git a/weed/command/filer_sync_verify.go b/weed/command/filer_sync_verify.go index 366f9e4fe..0031eb0e3 100644 --- a/weed/command/filer_sync_verify.go +++ b/weed/command/filer_sync_verify.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "path" + "sort" "sync" "sync/atomic" "time" @@ -261,59 +262,73 @@ func runVerifySync(filerA, filerB pb.ServerAddress, aPath, bPath string, return nil } -// entryStream is a sorted, streaming view of a single directory's entries. -// A background goroutine pages through the directory via ReadDirAllEntries -// and forwards each entry to a buffered channel; the caller consumes entries -// one at a time through peek/advance. Memory usage is O(channel buffer) — -// independent of directory size — rather than O(total entries). -type entryStream struct { - ch <-chan *filer_pb.Entry - head *filer_pb.Entry - done bool - err error // written before ch is closed; safe to read once done==true +// dirEntries holds one directory's entries buffered and sorted in Go byte +// order, which the merge in compareDirectory requires (it pairs equal names and +// uses eA.Name < eB.Name as the tie-break). The filer's own list order is not +// trusted: since PR #9824 (4.32) a filer forces byte-lexicographic listing +// regardless of SQL collation, while an older filer whose name column uses a +// locale collation lists case-insensitively. Comparing two such clusters yields +// the same name set in a different order, desyncing the merge into spurious +// MISSING / ONLY_IN_B. Sorting both sides client-side makes the order identical +// regardless of filer version or store backend. +// +// The load runs in a background goroutine so both sides of a compareDirectory +// list concurrently; peek/advance block on `ready` until it completes. Memory +// is O(directory size) per side — acceptable for a one-shot verify CLI. +type dirEntries struct { + ready chan struct{} // closed once entries and err are set + entries []*filer_pb.Entry + idx int + err error } -// newEntryStream starts the background goroutine. It exits when listing -// completes, an error occurs, or ctx is cancelled; the channel is always -// closed before exit so consumers do not block indefinitely. -func newEntryStream(ctx context.Context, client filer_pb.FilerClient, dir string) *entryStream { - ch := make(chan *filer_pb.Entry, 64) - s := &entryStream{ch: ch} +// newDirEntries starts loading the whole directory and sorting it by Name, then +// returns immediately. The goroutine exits when listing completes, errors, or +// ctx is cancelled; it always closes `ready` so consumers never block forever. +func newDirEntries(ctx context.Context, client filer_pb.FilerClient, dir string) *dirEntries { + s := &dirEntries{ready: make(chan struct{})} go func() { - defer close(ch) + defer close(s.ready) s.err = filer_pb.ReadDirAllEntries(ctx, client, util.FullPath(dir), "", func(entry *filer_pb.Entry, isLast bool) error { select { - case ch <- entry: - return nil case <-ctx.Done(): return ctx.Err() + default: } + s.entries = append(s.entries, entry) + return nil }) + if s.err == nil { + sort.Slice(s.entries, func(i, j int) bool { + return s.entries[i].Name < s.entries[j].Name + }) + } }() return s } +// wait blocks until the directory has finished loading and sorting. Its close +// of `ready` happens-before the return, so entries and err are then safe to read +// without additional synchronisation. +func (s *dirEntries) wait() { <-s.ready } + // peek returns the next entry without consuming it, or nil at end-of-stream. -func (s *entryStream) peek() *filer_pb.Entry { - if s.done { +// It blocks until the directory has finished loading and sorting. +func (s *dirEntries) peek() *filer_pb.Entry { + s.wait() + if s.idx >= len(s.entries) { return nil } - if s.head == nil { - e, ok := <-s.ch - if !ok { - s.done = true - return nil - } - s.head = e - } - return s.head + return s.entries[s.idx] } // advance consumes and returns the next entry. -func (s *entryStream) advance() *filer_pb.Entry { +func (s *dirEntries) advance() *filer_pb.Entry { e := s.peek() - s.head = nil + if e != nil { + s.idx++ + } return e } @@ -340,26 +355,38 @@ func compareDirectory(ctx context.Context, result.dirCount.Add(1) - // A child context ensures that stream goroutines are cancelled and their - // channels are closed if compareDirectory returns early (e.g. on error). + // A child context cancels the load goroutines (and their listing RPCs) if + // compareDirectory returns early, e.g. on error. Both sides load concurrently. mergeCtx, cancelMerge := context.WithCancel(ctx) defer cancelMerge() - streamA := newEntryStream(mergeCtx, clientA, dirA) - streamB := newEntryStream(mergeCtx, clientB, dirB) + entriesA := newDirEntries(mergeCtx, clientA, dirA) + entriesB := newDirEntries(mergeCtx, clientB, dirB) + + // Both sides are fully buffered before merging; surface any listing error + // here so a failed or cancelled listing can't emit bogus diffs from partial + // data. wait() makes entries/err visible without extra synchronisation. + entriesA.wait() + entriesB.wait() + if err := entriesA.err; err != nil && err != context.Canceled { + return fmt.Errorf("list %s on filer A: %v", dirA, err) + } + if err := entriesB.err; err != nil && err != context.Canceled { + return fmt.Errorf("list %s on filer B: %v", dirB, err) + } // collect subdirectories for recursive comparison type dirPair struct{ a, b string } var subDirs []dirPair - for streamA.peek() != nil || streamB.peek() != nil { - eA := streamA.peek() - eB := streamB.peek() + for entriesA.peek() != nil || entriesB.peek() != nil { + eA := entriesA.peek() + eB := entriesB.peek() switch { case eA != nil && (eB == nil || eA.Name < eB.Name): // entry only in A - entryA := streamA.advance() + entryA := entriesA.advance() if entryA.IsDirectory { // Always recurse for missing-in-B directories: a recent // child write can bump the parent's mtime even though @@ -375,7 +402,7 @@ func compareDirectory(ctx context.Context, case eB != nil && (eA == nil || eB.Name < eA.Name): // entry only in B - entryB := streamB.advance() + entryB := entriesB.advance() if !isActivePassive { if isTooRecent(entryB, cutoffTime) { result.skippedRecent.Add(1) @@ -386,8 +413,8 @@ func compareDirectory(ctx context.Context, default: // same name in both - entryA := streamA.advance() - entryB := streamB.advance() + entryA := entriesA.advance() + entryB := entriesB.advance() if entryA.IsDirectory && entryB.IsDirectory { subDirs = append(subDirs, dirPair{ @@ -411,15 +438,6 @@ func compareDirectory(ctx context.Context, } } - // Both channels are closed: close happens-before the receive of the zero - // value, so stream.err is visible here without additional synchronisation. - if err := streamA.err; err != nil && err != context.Canceled { - return fmt.Errorf("list %s on filer A: %v", dirA, err) - } - if err := streamB.err; err != nil && err != context.Canceled { - return fmt.Errorf("list %s on filer B: %v", dirB, err) - } - // Release our slot before recursing so children can acquire it. Holding // it across wg.Wait would deadlock once depth exceeds verifySyncConcurrency. releaseSlot() @@ -465,7 +483,6 @@ func compareDirectory(ctx context.Context, return nil } - func compareEntries(dir string, entryA, entryB *filer_pb.Entry, result *VerifyResult) { result.fileCount.Add(1) diff --git a/weed/command/filer_sync_verify_test.go b/weed/command/filer_sync_verify_test.go index 7a8faaec4..1fc6a6b24 100644 --- a/weed/command/filer_sync_verify_test.go +++ b/weed/command/filer_sync_verify_test.go @@ -18,10 +18,14 @@ import ( type verifyTestStream struct { entries []*filer_pb.Entry idx int + recvErr error // returned after all entries instead of io.EOF, if set } func (s *verifyTestStream) Recv() (*filer_pb.ListEntriesResponse, error) { if s.idx >= len(s.entries) { + if s.recvErr != nil { + return nil, s.recvErr + } return nil, io.EOF } resp := &filer_pb.ListEntriesResponse{Entry: s.entries[s.idx]} @@ -39,11 +43,12 @@ func (s *verifyTestStream) RecvMsg(_ any) error { return nil } // verifyTestInnerClient is the SeaweedFilerClient passed to fn inside WithFilerClient. type verifyTestInnerClient struct { filer_pb.SeaweedFilerClient // embed for unimplemented RPCs - entriesByDir map[string][]*filer_pb.Entry + entriesByDir map[string][]*filer_pb.Entry + recvErr error // injected listing error, if set } func (c *verifyTestInnerClient) ListEntries(_ context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { - return &verifyTestStream{entries: c.entriesByDir[in.Directory]}, nil + return &verifyTestStream{entries: c.entriesByDir[in.Directory], recvErr: c.recvErr}, nil } // verifyTestFilerClient implements filer_pb.FilerClient and tracks concurrent @@ -57,6 +62,8 @@ type verifyTestFilerClient struct { inFlight atomic.Int64 peakFlight atomic.Int64 delay time.Duration + onList func() // called at the start of each listing, if set + recvErr error // injected listing error, surfaced after entries } func (c *verifyTestFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { @@ -69,10 +76,13 @@ func (c *verifyTestFilerClient) WithFilerClient(_ bool, fn func(filer_pb.Seaweed break } } + if c.onList != nil { + c.onList() + } if c.delay > 0 { time.Sleep(c.delay) } - return fn(&verifyTestInnerClient{entriesByDir: c.entriesByDir}) + return fn(&verifyTestInnerClient{entriesByDir: c.entriesByDir, recvErr: c.recvErr}) } func (c *verifyTestFilerClient) AdjustedUrl(_ *filer_pb.Location) string { return "" } @@ -422,8 +432,8 @@ func TestVerifySyncMissingDirRecursesEvenWithRecentMtime(t *testing.T) { clientA := &verifyTestFilerClient{ entriesByDir: map[string][]*filer_pb.Entry{ - "/": {recentDir}, - "/subdir": {oldChild}, + "/": {recentDir}, + "/subdir": {oldChild}, }, } clientB := &verifyTestFilerClient{ @@ -454,14 +464,14 @@ func TestVerifySyncMissingDirRecursesEvenWithRecentMtime(t *testing.T) { func TestVerifySyncRootPath(t *testing.T) { clientA := &verifyTestFilerClient{ entriesByDir: map[string][]*filer_pb.Entry{ - "/": {verifyDirEntry("data")}, - "/data": {verifyFileEntry("file.txt", 42)}, + "/": {verifyDirEntry("data")}, + "/data": {verifyFileEntry("file.txt", 42)}, }, } clientB := &verifyTestFilerClient{ entriesByDir: map[string][]*filer_pb.Entry{ - "/": {verifyDirEntry("data")}, - "/data": {verifyFileEntry("file.txt", 42)}, + "/": {verifyDirEntry("data")}, + "/data": {verifyFileEntry("file.txt", 42)}, }, } @@ -531,3 +541,133 @@ func TestVerifySyncNoDeadlockDeepTree(t *testing.T) { t.Fatal("compareDirectory did not complete within 10s — possible deadlock") } } + +// TestVerifySyncByteOrderSkew covers a collation skew between two filer +// versions: both return the SAME name set in DIFFERENT order. Filer A (older, +// pre-4.32) lists in locale collation (case-insensitive: lowercase mixes in +// among uppercase); filer B (4.32+, with PR #9824) lists in byte order +// (uppercase before lowercase). The mock returns each side's slice verbatim. +// +// Because compareDirectory sorts both sides client-side before merging, the two +// orders converge and no spurious diffs are reported. Were the sort removed, the +// streaming merge would desync and count 3 false MISSING + 3 false ONLY_IN_B. +func TestVerifySyncByteOrderSkew(t *testing.T) { + // locale order (case-insensitive): lowercase mixes in among uppercase. + localeOrder := []*filer_pb.Entry{ + verifyFileEntry("sk-4", 10), + verifyFileEntry("sk-8", 10), + verifyFileEntry("sk-mmsJ", 10), + verifyFileEntry("sk-nFGE", 10), + verifyFileEntry("sk-RH0Z", 10), + verifyFileEntry("sk-Xp", 10), + verifyFileEntry("sk-Z06", 10), + } + // byte order: uppercase (R,X,Z) sort before lowercase (m,n). + byteOrder := []*filer_pb.Entry{ + verifyFileEntry("sk-4", 10), + verifyFileEntry("sk-8", 10), + verifyFileEntry("sk-RH0Z", 10), + verifyFileEntry("sk-Xp", 10), + verifyFileEntry("sk-Z06", 10), + verifyFileEntry("sk-mmsJ", 10), + verifyFileEntry("sk-nFGE", 10), + } + + clientA := &verifyTestFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{"/root": localeOrder}, + } + clientB := &verifyTestFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{"/root": byteOrder}, + } + + result := &VerifyResult{} + sem := make(chan struct{}, verifySyncConcurrency) + if err := compareDirectory(context.Background(), clientA, clientB, + "/root", "/root", false, time.Time{}, sem, result); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := result.missingCount.Load(); got != 0 { + t.Errorf("missingCount = %d, want 0", got) + } + if got := result.onlyInB.Load(); got != 0 { + t.Errorf("onlyInB = %d, want 0", got) + } +} + +// TestVerifySyncSourcesListConcurrently guards against re-serializing the two +// directory listings: both sides buffer + sort, and their loads must run +// concurrently — not A fully then B. A shared gate releases only once both +// listings are in-flight at the same time; if they were sequential the first +// would block on the gate and time out. +func TestVerifySyncSourcesListConcurrently(t *testing.T) { + var started atomic.Int32 + var timedOut atomic.Bool + release := make(chan struct{}) + gate := func() { + if started.Add(1) == 2 { + close(release) // both listings reached the gate → proceed + } + select { + case <-release: + case <-time.After(3 * time.Second): + timedOut.Store(true) // only one listing ever in-flight → serialized + } + } + + entries := []*filer_pb.Entry{verifyFileEntry("a", 1), verifyFileEntry("b", 1)} + clientA := &verifyTestFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{"/root": entries}, + onList: gate, + } + clientB := &verifyTestFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{"/root": entries}, + onList: gate, + } + + result := &VerifyResult{} + sem := make(chan struct{}, verifySyncConcurrency) + if err := compareDirectory(context.Background(), clientA, clientB, + "/root", "/root", false, time.Time{}, sem, result); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if timedOut.Load() { + t.Fatal("sources listed sequentially: both listings were not in-flight at once") + } +} + +// TestVerifySyncListErrorNoBogusDiffs verifies that a listing failure aborts +// with the error and reports no differences. A side whose listing errors keeps +// only a partial, unsorted buffer; the error must be surfaced before the merge +// so those partial entries never produce spurious MISSING / ONLY_IN_B. +func TestVerifySyncListErrorNoBogusDiffs(t *testing.T) { + entries := []*filer_pb.Entry{ + verifyFileEntry("a", 1), + verifyFileEntry("b", 1), + verifyFileEntry("c", 1), + } + // A errors mid-listing (partial buffer); B lists cleanly. + clientA := &verifyTestFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{"/root": entries}, + recvErr: fmt.Errorf("injected list failure"), + } + clientB := &verifyTestFilerClient{ + entriesByDir: map[string][]*filer_pb.Entry{"/root": entries}, + } + + result := &VerifyResult{} + sem := make(chan struct{}, verifySyncConcurrency) + err := compareDirectory(context.Background(), clientA, clientB, + "/root", "/root", false, time.Time{}, sem, result) + if err == nil { + t.Fatal("expected a listing error, got nil") + } + if got := result.missingCount.Load(); got != 0 { + t.Errorf("missingCount = %d, want 0 (no bogus diffs on list error)", got) + } + if got := result.onlyInB.Load(); got != 0 { + t.Errorf("onlyInB = %d, want 0 (no bogus diffs on list error)", got) + } + if got := result.fileCount.Load(); got != 0 { + t.Errorf("fileCount = %d, want 0 (merge must not run on failed listing)", got) + } +}