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
2026-08-18 21:26:56 +00:00 · 2026-06-26 10:27:18 -07:00
parent 378f9a64ff
commit 7c3c5ed2a4
2 changed files with 219 additions and 62 deletions
@@ -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)

@@ -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)
+	}
+}