filer: retry and surface metadata replay failures from peers (#10714)

* filer: stop silently dropping metadata replay failures from peers

When two filers do not share a store (e.g. one leveldb3 per pod), each
subscribes to its peers' metadata streams and replays their events
locally (meta_aggregator.go's maybeReplicateMetadataChange, wired into
doSubscribeToOneFiler). A failed Replay() was logged and then treated
as done anyway: processEventFn always returned nil regardless of the
replay outcome, and processOne advanced lastTsNs unconditionally. The
offset is the only record of subscription progress, so a dropped event
was gone for good - no retry, and nothing else ever observed it.

An entry that fails to replay this way diverges from its peer
permanently. This is how a bucket's quota (entry.Quota, carried on
peer events like everything else - see entry_codec.go's EqualEntry
comparing Quota, and FromPbEntry copying it in entry.go) can end up
different across filers indefinitely: one replay hiccup on one filer,
and its enforcement and any metric reading its own store diverges from
the others' with no signal anything went wrong.

Fix: replicateMetadataChange now retries a failure with util.Retry,
which already distinguishes transient errors (timeouts, connection
resets, throttling, ...) from everything else and bounds the backoff.
That covers the common case - a busy store, a blip talking to a
remote-backed backend - without changing behavior when replay
succeeds. An error that is not transient, or outlives the retry
budget, is not retried further: propagating it so the offset never
advances would stall this peer's entire stream behind one event that
may never replay, which is worse than the one entry staying stale.
Instead it is skipped, loudly - counted in a new
stats.FilerMetaAggregatorReplayFailures metric and logged at error
level - so the divergence is discoverable instead of silent.

Tested: go build ./... and go test ./weed/filer/... ./weed/server/...
Added meta_aggregator_replay_test.go: one test fails against the old
one-shot Replay call (a single transient failure is never retried, so
the store never converges) and passes with the fix; a second covers a
permanently-failing event completing quickly and being counted instead
of retried forever.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* filer: keep the test quota constant int64 for 32-bit builds

An untyped shift constant passed to t.Fatalf's ...any defaults to int and
overflows on 32-bit, failing go vet there.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* filer: name the diverged entry in the give-up log line

event.Directory is only the parent (typically /buckets), so for any
directory with more than one child the previous log line could not say
which entry failed to replay - the exact thing the change exists to
make discoverable. Name comes from NewEntry, falling back to OldEntry
for deletes; both getters are nil-safe.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* chore(filer): trim metadata replay comments to the non-obvious why

Compress the added comments on replicateMetadataChange and its tests down to
the reasoning a maintainer cannot get from the code: why a retry-exhausted
failure is skipped rather than propagated, what the old one-shot Replay body
did that the test pins, and why the quota constant is typed int64. Drops
deployment-specific narration and restatement of the code. No behaviour change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* chore: restore load-bearing clauses trimmed in the comment pass

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* filer: document and test the multi-step DeleteEntry replay hazard

CodeRabbit flagged that FilerStoreWrapper.DeleteEntry skips the delete
once FindEntry reports the path already gone, and that the redis store
families remove the primary key before parent-directory membership.
Chained together, a delete that fails between those two steps is
retried as a no-op: the stale membership is never revisited, and
replicateMetadataChange now reports overall success for it without
incrementing FilerMetaAggregatorReplayFailures, whereas before this PR
every such failure was unconditionally logged. The underlying store
inconsistency is pre-existing (a single non-retried Replay already
leaves the same stale membership behind); what retry adds is that this
one case no longer surfaces it.

Making Replay atomic or teaching every store to repair secondary
mutations on retry is out of scope here. Instead: document the hazard
at Replay, filerstore_wrapper.go's DeleteEntry, and
replicateMetadataChange, and add a test against the real
FilerStoreWrapper (not a strawman) that pins down the current,
documented behavior.

* filer: trim replay retry comments and tests

Drop the comment-only hunks documenting the pre-existing DeleteEntry
partial-failure hazard, the test that asserted that hazard still exists,
and the second hand-rolled fake store. Reuse stubFilerStore for the two
retry tests.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/filer/meta_aggregator.go

@@ -19,6 +19,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
 )
 
@@ -187,10 +188,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 		var counter int64
 		var synced bool
 		maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) {
-			if err := Replay(f.Store, event); err != nil {
-				glog.Errorf("failed to reply metadata change from %v: %v", peer, err)
-				return
-			}
+			replicateMetadataChange(f.Store, peer, event)
 			counter++
 			if lastPersistTime.Add(time.Minute).Before(time.Now()) {
 				if err := ma.updateOffset(f, peer, peerSignature, event.TsNs); err == nil {
@@ -317,6 +315,25 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 	return lastTsNs, err
 }
 
+// replicateMetadataChange retries transient Replay failures with bounded
+// backoff. A failure that outlives the retry budget is counted and logged,
+// then skipped: blocking on an event that can never replay would stall every
+// later event from this peer, which is worse than one entry staying stale.
+func replicateMetadataChange(store FilerStore, peer pb.ServerAddress, event *filer_pb.SubscribeMetadataResponse) {
+	err := util.Retry("replicate metadata change from "+string(peer), func() error {
+		return Replay(store, event)
+	})
+	if err == nil {
+		return
+	}
+	stats.FilerMetaAggregatorReplayFailures.WithLabelValues(string(peer)).Inc()
+	name := event.GetEventNotification().GetNewEntry().GetName()
+	if name == "" {
+		name = event.GetEventNotification().GetOldEntry().GetName()
+	}
+	glog.Errorf("giving up replicating metadata change from %s for %s/%s (ts=%d): %v", peer, event.Directory, name, event.TsNs, err)
+}
+
 // traversePeerMetadata does a full BFS traversal of a peer filer's metadata
 // and inserts all entries into the local store. This is used when a filer
 // connects to a peer for the first time and needs to bootstrap pre-existing data.

weed/filer/meta_aggregator_replay_test.go

@@ -0,0 +1,111 @@
+package filer
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	dto "github.com/prometheus/client_model/go"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// flakyReplayStore fails InsertEntry for its first failUntil calls, then
+// delegates to the embedded stubFilerStore.
+type flakyReplayStore struct {
+	*stubFilerStore
+	countMu   sync.Mutex
+	failUntil int
+	err       error
+	attempts  int
+}
+
+func (s *flakyReplayStore) InsertEntry(ctx context.Context, entry *Entry) error {
+	s.countMu.Lock()
+	s.attempts++
+	fail := s.attempts <= s.failUntil
+	s.countMu.Unlock()
+	if fail {
+		return s.err
+	}
+	return s.stubFilerStore.InsertEntry(ctx, entry)
+}
+
+func (s *flakyReplayStore) attemptCount() int {
+	s.countMu.Lock()
+	defer s.countMu.Unlock()
+	return s.attempts
+}
+
+func replayFailureCount(t *testing.T, peer pb.ServerAddress) float64 {
+	t.Helper()
+	var m dto.Metric
+	if err := stats.FilerMetaAggregatorReplayFailures.WithLabelValues(string(peer)).Write(&m); err != nil {
+		t.Fatalf("read counter: %v", err)
+	}
+	return m.GetCounter().GetValue()
+}
+
+func quotaChangeEvent(bucket string, quota int64) *filer_pb.SubscribeMetadataResponse {
+	return &filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: bucket, IsDirectory: true, Quota: quota},
+		},
+		TsNs: time.Now().UnixNano(),
+	}
+}
+
+// The old body logged a Replay error and returned, so one transient failure
+// permanently diverged the entry from the peer.
+func TestReplicateMetadataChangeRetriesTransientFailure(t *testing.T) {
+	// int64: untyped, the value defaults to int and overflows a 32-bit build
+	const wantQuota int64 = 131072 << 20
+
+	store := &flakyReplayStore{
+		stubFilerStore: newStubFilerStore(),
+		failUntil:      1,
+		err:            errors.New("i/o timeout talking to store"),
+	}
+	peer := pb.ServerAddress("peer-transient:1")
+
+	replicateMetadataChange(store, peer, quotaChangeEvent("my-bucket", wantQuota))
+
+	inserted, err := store.FindEntry(context.Background(), util.NewFullPath("/buckets", "my-bucket"))
+	if err != nil {
+		t.Fatal("expected the entry to be inserted once the transient failure is retried past")
+	}
+	if inserted.Quota != wantQuota {
+		t.Fatalf("quota = %d, want %d", inserted.Quota, wantQuota)
+	}
+	if got := store.attemptCount(); got < 2 {
+		t.Fatalf("attempts = %d, want at least 2", got)
+	}
+}
+
+// An event that can never replay must fail fast instead of blocking the
+// subscribe stream, and must be counted so the divergence is not silent.
+func TestReplicateMetadataChangeGivesUpLoudlyOnPermanentFailure(t *testing.T) {
+	store := &flakyReplayStore{
+		stubFilerStore: newStubFilerStore(),
+		failUntil:      1 << 30,
+		err:            errors.New("entry checksum mismatch"),
+	}
+	peer := pb.ServerAddress("peer-permanent:1")
+
+	before := replayFailureCount(t, peer)
+
+	replicateMetadataChange(store, peer, quotaChangeEvent("poison-bucket", 65536<<20))
+
+	if got := store.attemptCount(); got != 1 {
+		t.Fatalf("attempts = %d, want exactly 1: a non-transient error must fail fast, not retry", got)
+	}
+	if got := replayFailureCount(t, peer); got != before+1 {
+		t.Fatalf("FilerMetaAggregatorReplayFailures[%s] = %v, want %v", peer, got, before+1)
+	}
+}

weed/filer/meta_replay.go

@@ -8,6 +8,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
+// Replay applies the delete and the insert as separate, non-transactional
+// store calls, so retrying it after a partial failure is not atomic.
 func Replay(filerStore FilerStore, resp *filer_pb.SubscribeMetadataResponse) error {
 	message := resp.EventNotification
 	var oldPath util.FullPath

weed/stats/metrics.go

@@ -250,6 +250,14 @@ var (
 			Help:      "Number of metadata subscribers currently parked waiting to read past a gap in the metadata log.",
 		}, []string{"scope"})
 
+	FilerMetaAggregatorReplayFailures = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: subsystemFiler,
+			Name:      "meta_aggregator_replay_failures",
+			Help:      "Number of peer metadata events skipped after replay retries were exhausted, leaving that entry diverged from the peer.",
+		}, []string{"peer"})
+
 	// Sampled only on first creation, so counts track distinct objects.
 	FilerObjectSizeBytesHistogram = prometheus.NewHistogram(
 		prometheus.HistogramOpts{
@@ -899,6 +907,7 @@ func init() {
 	Gather.MustRegister(FilerServerLastSendTsOfSubscribeGauge)
 	Gather.MustRegister(FilerSubscribeGapStalledGauge)
 	Gather.MustRegister(FilerSubscribeUnprovenGapCrossings)
+	Gather.MustRegister(FilerMetaAggregatorReplayFailures)
 	Gather.MustRegister(FilerObjectSizeBytesHistogram)
 	Gather.MustRegister(collectors.NewGoCollector())
 	Gather.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))