filer: converge peer metadata replication on restart (#11698)

* filer: skip replayed peer events older than the local entry

Replay deleted and re-inserted every received event unconditionally, so
a restarted filer that rewound its subscription offset re-applied stale
peer writes over newer local state (values reverted, deleted entries
came back and pointed at already-vacuumed chunks). Compare the event's
TsNs against the local entry's Mtime and skip per-path, so a move event
can still remove its source while keeping a newer destination.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: add replay last-writer-wins regression tests

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: stop peer subscriptions and flush offsets on shutdown

Filer.Shutdown closed the store while subscription goroutines were still
mid-stream; their deferred offset save was never guaranteed to run, so a
restart rewound each peer's stream back to the last periodic flush and
replayed stale events. MetaAggregator.Shutdown now closes every peer's
stop channel and waits for the loops so the saved offset covers what was
replayed, and the reconnect sleep is interruptible by the stop channel.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: test that shutdown stops and waits for peer loops

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: order replay by persisted per-path commit timestamps

Review feedback found the mtime comparison still lets stale replay win:
a local delete removes the entry it would have been compared against, an
entry mtime is client-controllable, and the check ran unlocked so a local
write could land between it and the store mutation.

- Record the event timestamp per mutated path in the store KV on every
  emitted mutation, plus a delete record for sources of deletes and
  renames. The record survives the entry, so a replayed create older than
  a local delete no longer resurrects it; ancestor delete records cover
  children swept without per-child events.
- Replay compares the event timestamp against that record under the same
  per-path entry locks local RPC mutations hold, falling back to entry
  mtime only for entries written before records existed. Lookup errors
  now abort the event instead of being treated as an absent entry.
- MetaAggregator.Shutdown marks the aggregator stopped under peerChansLock
  so a master update cannot start a subscription that outlives the wait,
  and waits for the loops without the five-second cap now that every
  blocking stage — signature RPC, bootstrap traversal, retry sleeps, and
  per-event processing — cancels on the stop channel.

* filer: pin peer offset at failed replay, fix commit-record context and key size

A replayed event that exhausted its retries (or was interrupted by
shutdown) was still followed by advancing the peer watermark, so the
persisted offset skipped the unapplied event forever. replicateMetadataChange
now reports the outcome: errPeerSubscriptionStopped aborts the subscription
so the deferred save keeps the last completed event, and an exhausted retry
records a barrier the persisted offset and reconnect resume point cannot
pass while the stream keeps flowing.

Commit records also failed on two fronts: NotifyUpdateEvent runs after
CommitTransaction with the same context, so SQL stores wrote MpTs/MpDl
through an already-committed *sql.Tx; and metaPathKey embedded the whole
path, overflowing the 766-char name column SQL KV stores use, which the
KvPut fallback update then silently dropped. The writes now use a fresh
context and a sha256-hashed fixed-size key.

* filer: lock HTTP delete handler and replay against ancestor paths

The HTTP DeleteHandler called DeleteEntryMetaAndData without the shared
entry lock, so a peer replay could still interleave with a local delete and
its commit record. It now holds the same lock the gRPC DeleteEntry handler
does.

Replay also locked only the exact event paths, but its ordering check
consults ancestor delete records: a local recursive delete holds only the
ancestor's lock, so a stale child create could check before the tombstone
landed and then reinsert. lockReplayPaths now locks every ancestor of both
event paths in one lexicographic order — parents sort before children, as
the rename handler's two-path order already does — while never locking the
root, which cannot be deleted and would serialize all replays.

* filer: record replay commit before the replacement insert

If the same-path update DeleteEntry succeeded but InsertEntry failed, the path sat deleted with no commit record until the retry arrived, and a stale peer create could resurrect it in the gap. The commit record now lands immediately after the delete — without a delete record, since an update is not a delete — and again after the successful insert.

* filer: lock replay ancestors shared, not exclusive

Exclusive locks on every ancestor serialized all replays under a shared
parent — unrelated objects in the same bucket could no longer apply in
parallel. Ancestors only need to exclude an exclusive holder (a local
delete or rename of the ancestor), so they now take shared locks; a path
the event changes stays exclusive even when it is also the other path's
ancestor.

* filer: order peer bootstrap on the sender's commit timestamp

traversePeerMetadata compared localCommitTsNs — a commit-clock value —
against the entry's mtime, which a client can set freely: a peer change
committed recently with an old mtime was skipped, and a stale entry
mtime-stamped in the future overwrote a newer local commit.

TraverseBfsMetadataResponse now carries the sender's per-path commit
timestamp (recorded MpTs record, entry mtime as the unrecorded fallback).
The bootstrap applies an entry only when its commit is newer than the
local one and records the applied timestamp, so replayed events order
against it — closing the gap between bootstrap content and the event
stream. A peer without the field sends 0 and keeps the mtime fallback,
preserving rolling-upgrade behavior.

* filer: keep failed commit records pending until they land

notePathChange ran after the mutation committed and only logged a KvPut
failure: a delete whose MpTs/MpDl record never landed left no tombstone,
so a stale replayed create could resurrect the path while the store was
unreachable.

Failed records now stay in metaTsPending and a repair pass retries them
until they land. localCommitTsNs, deletedAfterTsNs and CommitTsNs
consult the pending record, so ordering holds in the window between the
committed mutation and the durable record. A per-path flag bounds the
repair to one goroutine, and respawns cover merges landing between its
last pass and the flag release.

* filer: keep only unwritten work in the pending commit record

The pending record merged every obligation into one timestamp and flag,
and a later successful write left it untouched: a repair pass could then
write an older timestamp over a newer one, and a merged delete inherited
the newer change's time, which the ancestor check reads as a newer
delete. The record now holds a change and a delete timestamp of its own,
both discharged under the record's lock alongside the store writes that
cover them, so a repair writes only what is still owed.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer: fail traversal on a commit-record read error

CommitTsNs answered a store read failure with the entry mtime, a value
clients can set arbitrarily far ahead; a peer bootstrapping on it would
record the minted time and reject real updates below it. The lookup now
returns the error so the traversal restarts instead.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/filer/filer.go

@@ -82,6 +82,11 @@ type Filer struct {
 	persistedLogCache             *persistedLogCache
 	metaLogInflight               metaLogInflight
 	remoteTombstones              *remoteDeletionTombstones
+	// EntryLockTable serializes mutations per entry path. The filer server
+	// holds it across local read-modify-write mutations and the meta
+	// aggregator across a replayed peer change's check-and-apply, so the two
+	// cannot interleave on the same path.
+	EntryLockTable *util.LockTable[util.FullPath]
 	// remoteTombstonesDone, when non-nil, is closed once the startup tombstone
 	// rebuild finishes; lazy remote reads wait on it so a pending delete
 	// cannot resurrect in the gap.
@@ -110,6 +115,12 @@ type Filer struct {
 	// overwriting the unread ledger with a partial set.
 	deletionLedgerBlocked atomic.Bool
 	deletionLedgerFlush   chan struct{}
+	// metaTsPending holds per-path commit records whose KV write failed,
+	// keyed by path; ordering reads consult it until the retry lands them
+	// (see meta_path_ts.go). metaTsRepairing marks paths with a live
+	// repair goroutine so failures neither strand a record nor pile on.
+	metaTsPending   sync.Map
+	metaTsRepairing sync.Map
 	// flushCtx is the context metadata-log flushes append under. It stays
 	// live for the process lifetime and is cancelled once during Shutdown,
 	// bounding every retry - in-flight and queued alike - to one deadline.
@@ -133,6 +144,7 @@ func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerH
 		DeletionRetryQueue:  NewDeletionRetryQueue(),
 		persistedLogCache:   newPersistedLogCache(persistedLogCacheMaxBytes),
 		remoteTombstones:    newRemoteDeletionTombstones(),
+		EntryLockTable:      util.NewLockTable[util.FullPath](),
 	}
 	f.flushCtx, f.flushCancel = context.WithCancel(context.Background())
 	if f.UniqueFilerId < 0 {
@@ -913,6 +925,12 @@ func (f *Filer) Shutdown() {
 	f.LocalMetaLogBuffer.ShutdownLogBuffer()
 	// The final metadata-log flush still needs the store to append its entry.
 	f.LocalMetaLogBuffer.WaitForShutdown()
+	// Stop peer subscriptions and let each persist its last applied offset;
+	// the replay on the next start otherwise resumes from a stale offset and
+	// applies those old events over newer local state.
+	if f.MetaAggregator != nil {
+		f.MetaAggregator.Shutdown()
+	}
 	// Persist the deletion ledger one last time before the store closes, so a
 	// clean shutdown leaves the recovery set exactly consistent with reality.
 	f.snapshotDeletionLedger()

weed/filer/filer_lazy_remote_test.go

@@ -35,6 +35,7 @@ type stubFilerStore struct {
 	kv              map[string][]byte
 	insertErr       error
 	findErr         error
+	kvPutErr        error
 	kvGetErr        error
 	kvDeleteErr     error
 	kvGetHook       func(key []byte)
@@ -49,6 +50,12 @@ func newStubFilerStore() *stubFilerStore {
 	}
 }
 
+func (s *stubFilerStore) setKvPutErr(err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kvPutErr = err
+}
+
 func (s *stubFilerStore) GetName() string                             { return "stub" }
 func (s *stubFilerStore) Initialize(util.Configuration, string) error { return nil }
 func (s *stubFilerStore) Shutdown()                                   {}
@@ -60,6 +67,9 @@ func (s *stubFilerStore) RollbackTransaction(context.Context) error { return nil
 func (s *stubFilerStore) KvPut(_ context.Context, key []byte, value []byte) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if s.kvPutErr != nil {
+		return s.kvPutErr
+	}
 	s.kv[string(key)] = append([]byte(nil), value...)
 	return nil
 }

weed/filer/filer_notify.go

@@ -71,6 +71,10 @@ func (f *Filer) notifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry
 	}
 
 	f.logMetaEvent(ctx, event)
+	// Record the commit timestamp so a peer replay ordered by the entry lock
+	// this mutation may still be under sees the local change, including as a
+	// tombstone after the entry itself is gone.
+	f.noteCommittedEvent(event)
 	if sink := metadataEventSinkFromContext(ctx); sink != nil {
 		sink.Record(event)
 	}

weed/filer/meta_aggregator.go

@@ -63,6 +63,11 @@ type MetaAggregator struct {
 	flushAdvanced         chan struct{}
 	remotePeerArrived     chan struct{}
 	peerWatermarksLock    sync.Mutex
+	peerWg                sync.WaitGroup
+	// stopped is set under peerChansLock by Shutdown; OnPeerUpdate then
+	// rejects adds so no subscription can outlive the shutdown wait and run
+	// against a store Filer.Shutdown has already closed.
+	stopped bool
 }
 
 // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk.
@@ -87,6 +92,13 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star
 	ma.peerChansLock.Lock()
 	defer ma.peerChansLock.Unlock()
 
+	// After Shutdown the master connection may still deliver an update; a
+	// subscription started now would have no stop signal and could run past
+	// the store close.
+	if ma.stopped {
+		return
+	}
+
 	address := pb.ServerAddress(update.Address)
 	if update.IsAdd {
 		// the peer is already followed, restarting would only lose the events
@@ -102,7 +114,11 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star
 		// Account for the peer before its stream signals; keep prior values
 		// on reconnect.
 		ma.initPeerWatermark(address)
-		go ma.loopSubscribeToOneFiler(ma.filer, ma.self, address, startFrom, stopChan)
+		ma.peerWg.Add(1)
+		go func() {
+			defer ma.peerWg.Done()
+			ma.loopSubscribeToOneFiler(ma.filer, ma.self, address, startFrom, stopChan)
+		}()
 	} else {
 		if prevChan, found := ma.peerChans[address]; found {
 			close(prevChan)
@@ -344,6 +360,42 @@ func (ma *MetaAggregator) HasPeer(address pb.ServerAddress) bool {
 	return false
 }
 
+// Shutdown stops every peer subscription and waits for each to return,
+// letting the deferred offset save persist the last applied position before
+// the store closes. Without it a graceful restart resumes from the last
+// periodically-saved offset — up to a minute stale — and replays those
+// events over newer local state.
+func (ma *MetaAggregator) Shutdown() {
+	ma.peerChansLock.Lock()
+	ma.stopped = true
+	chans := ma.peerChans
+	ma.peerChans = make(map[pb.ServerAddress]chan struct{})
+	for _, stopChan := range chans {
+		close(stopChan)
+	}
+	ma.peerChansLock.Unlock()
+
+	// Wait without a cap: every blocking point in the subscribe loop is
+	// stop-aware, so a peer still running here is genuinely wedged and
+	// returning anyway would let Filer.Shutdown close the store under it and
+	// skip its deferred offset save.
+	done := make(chan struct{})
+	go func() {
+		ma.peerWg.Wait()
+		close(done)
+	}()
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-done:
+			return
+		case <-ticker.C:
+			glog.Warningf("meta aggregator shutdown: peer subscriptions still stopping")
+		}
+	}
+}
+
 func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom time.Time, stopChan chan struct{}) {
 	lastTsNs := startFrom.UnixNano()
 	for {
@@ -368,7 +420,12 @@ func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddres
 		if lastTsNs < nextLastTsNs {
 			lastTsNs = nextLastTsNs
 		}
-		time.Sleep(1733 * time.Millisecond)
+		select {
+		case <-stopChan:
+			glog.V(0).Infof("stop subscribing peer %s meta change", peer)
+			return
+		case <-time.After(1733 * time.Millisecond):
+		}
 	}
 }
 
@@ -384,11 +441,34 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 
 	*/
 
-	var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse)
+	var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse) error
 	lastPersistTime := time.Now()
 	lastTsNs := startFrom
+	// replayFloorTsNs is the earliest event this session gave up replaying.
+	// The persisted offset and the reconnect resume point must stay behind
+	// it so the event is redelivered on the next subscribe instead of being
+	// skipped forever.
+	var replayFloorTsNs int64
+	persistableTsNs := func(tsNs int64) int64 {
+		if replayFloorTsNs > 0 && replayFloorTsNs-1 < tsNs {
+			return replayFloorTsNs - 1
+		}
+		return tsNs
+	}
 
-	peerSignature, err := ma.readFilerStoreSignature(peer)
+	// Everything below derives from ctx, cancelled when the peer is removed
+	// or the aggregator shuts down, so no stage can block past its stop.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		select {
+		case <-stopChan:
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+
+	peerSignature, err := ma.readFilerStoreSignature(ctx, peer)
 	if err != nil {
 		return lastTsNs, fmt.Errorf("connecting to peer filer %s: %v", peer, err)
 	}
@@ -408,7 +488,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 			// different concept and can be set to arbitrary values.
 			preTraverseTime := time.Now()
 			glog.V(0).Infof("no previous offset for peer %s, starting full metadata sync", peer)
-			if traverseErr := ma.traversePeerMetadata(f, peer); traverseErr != nil {
+			if traverseErr := ma.traversePeerMetadata(ctx, f, peer); traverseErr != nil {
 				return lastTsNs, fmt.Errorf("initial metadata sync from %s: %v", peer, traverseErr)
 			}
 			lastTsNs = preTraverseTime.Add(-time.Minute).UnixNano()
@@ -420,11 +500,12 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 			return lastTsNs, fmt.Errorf("read offset for peer %s: %w", peer, err)
 		}
 		defer func(prevTsNs int64) {
-			if lastTsNs != prevTsNs && lastTsNs != lastPersistTime.UnixNano() {
-				if err := ma.updateOffset(f, peer, peerSignature, lastTsNs); err == nil {
-					glog.V(0).Infof("last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
+			saveTsNs := persistableTsNs(lastTsNs)
+			if saveTsNs != prevTsNs && lastTsNs != lastPersistTime.UnixNano() {
+				if err := ma.updateOffset(f, peer, peerSignature, saveTsNs); err == nil {
+					glog.V(0).Infof("last sync time with %s at %v (%d)", peer, time.Unix(0, saveTsNs), saveTsNs)
 				} else {
-					glog.Errorf("failed to save last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
+					glog.Errorf("failed to save last sync time with %s at %v (%d)", peer, time.Unix(0, saveTsNs), saveTsNs)
 				}
 			}
 		}(lastTsNs)
@@ -432,11 +513,23 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 		glog.V(0).Infof("follow peer: %v, last %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs)
 		var counter int64
 		var synced bool
-		maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) {
-			replicateMetadataChange(f.Store, peer, event)
+		maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) error {
+			if err := replicateMetadataChange(f, peer, event, stopChan); err != nil {
+				if errors.Is(err, errPeerSubscriptionStopped) {
+					return err
+				}
+				// A failed event becomes a barrier the persisted offset and
+				// the reconnect resume point cannot pass, while the stream
+				// keeps flowing: blocking on an event that can never replay
+				// would stall every later event from this peer.
+				if replayFloorTsNs == 0 || event.TsNs < replayFloorTsNs {
+					replayFloorTsNs = event.TsNs
+				}
+				return nil
+			}
 			counter++
 			if lastPersistTime.Add(time.Minute).Before(time.Now()) {
-				if err := ma.updateOffset(f, peer, peerSignature, event.TsNs); err == nil {
+				if err := ma.updateOffset(f, peer, peerSignature, persistableTsNs(event.TsNs)); err == nil {
 					if event.TsNs < time.Now().Add(-2*time.Minute).UnixNano() {
 						glog.V(0).Infof("sync with %s progressed to: %v %0.2f/sec", peer, time.Unix(0, event.TsNs), float64(counter)/60.0)
 					} else if !synced {
@@ -449,6 +542,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 					glog.V(0).Infof("failed to update offset for %v: %v", peer, err)
 				}
 			}
+			return nil
 		}
 	}
 
@@ -465,7 +559,9 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 			return err
 		}
 		if maybeReplicateMetadataChange != nil {
-			maybeReplicateMetadataChange(event)
+			if err := maybeReplicateMetadataChange(event); err != nil {
+				return err
+			}
 		}
 		return nil
 	}
@@ -476,17 +572,8 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 
 	glog.V(0).Infof("subscribing remote %s meta change: %v, clientId:%d", peer, time.Unix(0, lastTsNs), ma.filer.UniqueFilerId)
 	err = pb.WithFilerClient(true, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-		// Stop the stream promptly on removal; the blocking Recv below would
-		// otherwise only notice when the stream errors on its own.
-		go func() {
-			select {
-			case <-stopChan:
-				cancel()
-			case <-ctx.Done():
-			}
-		}()
+		// ctx is cancelled on stopChan by the watch above; the blocking Recv
+		// below stops promptly on removal or shutdown.
 		atomic.AddInt32(&ma.filer.UniqueFilerEpoch, 1)
 		// Construct a log file reader that reads chunks via the peer filer's LookupVolume.
 		lookupFn := LookupFn(filerClient{client})
@@ -511,6 +598,14 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 		}
 
 		processOne := func(event *filer_pb.SubscribeMetadataResponse) error {
+			// Bail promptly on stop; a cancelled stream may still deliver
+			// queued events while unwinding, and replaying them now only
+			// delays the offset save the shutdown is waiting on.
+			select {
+			case <-stopChan:
+				return errPeerSubscriptionStopped
+			default:
+			}
 			if err := processEventFn(event); err != nil {
 				glog.V(0).Infof("SubscribeLocalMetadata process %v: %v", event, err)
 				return fmt.Errorf("process %v: %w", event, err)
@@ -587,19 +682,37 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 			}
 		}
 	})
-	return lastTsNs, err
+	return persistableTsNs(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
+// backoff and reports the outcome. errPeerSubscriptionStopped is returned
+// when the subscription stops mid-retry so the caller aborts and the offset
+// save keeps the last completed event. A failure that outlives the retry
+// budget is counted and logged, then returned: the caller pins the peer's
+// offset at the event so a restart replays it rather than skipping it.
+func replicateMetadataChange(f *Filer, peer pb.ServerAddress, event *filer_pb.SubscribeMetadataResponse, stopChan <-chan struct{}) error {
+	waitTime := time.Second
+	var err error
+	for waitTime < util.RetryWaitTime {
+		select {
+		case <-stopChan:
+			return errPeerSubscriptionStopped
+		default:
+		}
+		if err = f.Replay(event); err == nil {
+			return nil
+		}
+		if !util.IsTransientError(err) {
+			break
+		}
+		glog.V(0).Infof("retry replicate metadata change from %s: %v", peer, err)
+		select {
+		case <-stopChan:
+			return errPeerSubscriptionStopped
+		case <-time.After(waitTime):
+		}
+		waitTime += waitTime / 2
 	}
 	stats.FilerMetaAggregatorReplayFailures.WithLabelValues(string(peer)).Inc()
 	name := event.GetEventNotification().GetNewEntry().GetName()
@@ -607,15 +720,20 @@ func replicateMetadataChange(store FilerStore, peer pb.ServerAddress, event *fil
 		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)
+	return err
 }
 
+// errPeerSubscriptionStopped ends event processing once the subscription's
+// stop channel closed.
+var errPeerSubscriptionStopped = errors.New("peer subscription stopped")
+
 // 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.
-func (ma *MetaAggregator) traversePeerMetadata(f *Filer, peer pb.ServerAddress) error {
+// The ctx cancels on the peer's stop channel so a shutdown does not have to
+// out-wait a long traversal.
+func (ma *MetaAggregator) traversePeerMetadata(ctx context.Context, f *Filer, peer pb.ServerAddress) error {
 	return pb.WithFilerClient(true, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
 		stream, err := client.TraverseBfsMetadata(ctx, &filer_pb.TraverseBfsMetadataRequest{
 			Directory:        "/",
 			ExcludedPrefixes: []string{SystemLogDir},
@@ -635,22 +753,17 @@ func (ma *MetaAggregator) traversePeerMetadata(f *Filer, peer pb.ServerAddress)
 			if resp.Entry == nil {
 				continue
 			}
-			fullpath := util.Join(resp.Directory, resp.Entry.Name)
 			entry := FromPbEntry(resp.Directory, resp.Entry)
-			if insertErr := f.Store.InsertEntry(context.Background(), entry); insertErr != nil {
-				// Entry may already exist (root dir, or partial previous bootstrap).
-				existing, findErr := f.Store.FindEntry(context.Background(), entry.FullPath)
-				if findErr != nil {
-					return fmt.Errorf("insert entry %s: %w", fullpath, insertErr)
-				}
-				// Only overwrite if the peer's entry is newer.
-				if entry.Attr.Mtime.After(existing.Attr.Mtime) {
-					if updateErr := f.Store.UpdateEntry(context.Background(), entry); updateErr != nil {
-						return fmt.Errorf("update entry %s: %w", fullpath, updateErr)
-					}
-				} else {
-					glog.V(1).Infof("skip older peer entry %s (peer mtime %v <= local mtime %v)", fullpath, entry.Attr.Mtime, existing.Attr.Mtime)
-				}
+			// Order on the sender's per-path commit timestamp, not the
+			// entry's mtime — mtime is client-settable and does not track
+			// when the peer committed the change. Peers that predate the
+			// field send 0 and fall back to mtime.
+			peerTsNs := resp.EntryTsNs
+			if peerTsNs == 0 {
+				peerTsNs = entry.Attr.Mtime.UnixNano()
+			}
+			if applyErr := f.applyTraversedEntry(ctx, entry, peerTsNs); applyErr != nil {
+				return applyErr
 			}
 			count++
 			if count%10000 == 0 {
@@ -662,9 +775,38 @@ func (ma *MetaAggregator) traversePeerMetadata(f *Filer, peer pb.ServerAddress)
 	})
 }
 
-func (ma *MetaAggregator) readFilerStoreSignature(peer pb.ServerAddress) (sig int32, err error) {
+// applyTraversedEntry merges one bootstrap entry under the path's replay
+// locks, keeping local state when its commit is newer than the sender's —
+// most importantly a delete, whose tombstone must not let the bootstrap
+// resurrect the path. An applied entry records the sender's commit time so a
+// later peer replay orders against it rather than the entry's mtime.
+func (f *Filer) applyTraversedEntry(ctx context.Context, entry *Entry, peerTsNs int64) error {
+	unlock := f.lockReplayPaths(entry.FullPath, "")
+	defer unlock()
+	localTsNs, err := f.localCommitTsNs(ctx, entry.FullPath)
+	if err != nil {
+		return fmt.Errorf("check local state for %s: %w", entry.FullPath, err)
+	}
+	if localTsNs >= peerTsNs {
+		return nil
+	}
+	if insertErr := f.Store.InsertEntry(ctx, entry); insertErr != nil {
+		// Entry already exists (root dir, or partial previous bootstrap);
+		// the lock kept the check above current, so the peer commit is newer.
+		if _, findErr := f.Store.FindEntry(ctx, entry.FullPath); findErr != nil {
+			return fmt.Errorf("insert entry %s: %w", entry.FullPath, insertErr)
+		}
+		if updateErr := f.Store.UpdateEntry(ctx, entry); updateErr != nil {
+			return fmt.Errorf("update entry %s: %w", entry.FullPath, updateErr)
+		}
+	}
+	f.notePathChange(entry.FullPath, peerTsNs, false)
+	return nil
+}
+
+func (ma *MetaAggregator) readFilerStoreSignature(ctx context.Context, peer pb.ServerAddress) (sig int32, err error) {
 	err = pb.WithFilerClient(false, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
-		resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
+		resp, err := client.GetFilerConfiguration(ctx, &filer_pb.GetFilerConfigurationRequest{})
 		if err != nil {
 			return err
 		}

weed/filer/meta_aggregator_replay_test.go

@@ -15,6 +15,14 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
+// newReplayTestFiler builds the minimum Filer Replay needs over a stub store.
+func newReplayTestFiler(store FilerStore) *Filer {
+	return &Filer{
+		Store:          NewFilerStoreWrapper(store),
+		EntryLockTable: util.NewLockTable[util.FullPath](),
+	}
+}
+
 // flakyReplayStore fails InsertEntry for its first failUntil calls, then
 // delegates to the embedded stubFilerStore.
 type flakyReplayStore struct {
@@ -73,8 +81,11 @@ func TestReplicateMetadataChangeRetriesTransientFailure(t *testing.T) {
 		err:            errors.New("i/o timeout talking to store"),
 	}
 	peer := pb.ServerAddress("peer-transient:1")
+	f := newReplayTestFiler(store)
 
-	replicateMetadataChange(store, peer, quotaChangeEvent("my-bucket", wantQuota))
+	if err := replicateMetadataChange(f, peer, quotaChangeEvent("my-bucket", wantQuota), nil); err != nil {
+		t.Fatalf("replicateMetadataChange: %v", err)
+	}
 
 	inserted, err := store.FindEntry(context.Background(), util.NewFullPath("/buckets", "my-bucket"))
 	if err != nil {
@@ -97,10 +108,14 @@ func TestReplicateMetadataChangeGivesUpLoudlyOnPermanentFailure(t *testing.T) {
 		err:            errors.New("entry checksum mismatch"),
 	}
 	peer := pb.ServerAddress("peer-permanent:1")
+	f := newReplayTestFiler(store)
 
 	before := replayFailureCount(t, peer)
 
-	replicateMetadataChange(store, peer, quotaChangeEvent("poison-bucket", 65536<<20))
+	err := replicateMetadataChange(f, peer, quotaChangeEvent("poison-bucket", 65536<<20), nil)
+	if err == nil {
+		t.Fatal("expected the exhausted replay error so the caller can pin the peer offset at the failed event")
+	}
 
 	if got := store.attemptCount(); got != 1 {
 		t.Fatalf("attempts = %d, want exactly 1: a non-transient error must fail fast, not retry", got)
@@ -109,3 +124,587 @@ func TestReplicateMetadataChangeGivesUpLoudlyOnPermanentFailure(t *testing.T) {
 		t.Fatalf("FilerMetaAggregatorReplayFailures[%s] = %v, want %v", peer, got, before+1)
 	}
 }
+
+func overwriteEvent(dir, name string, tsNs int64, content string) *filer_pb.SubscribeMetadataResponse {
+	return &filer_pb.SubscribeMetadataResponse{
+		Directory: dir,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: name},
+			NewEntry: &filer_pb.Entry{Name: name, Content: []byte(content)},
+		},
+		TsNs: tsNs,
+	}
+}
+
+func deleteEvent(dir, name string, tsNs int64) *filer_pb.SubscribeMetadataResponse {
+	return &filer_pb.SubscribeMetadataResponse{
+		Directory: dir,
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: name},
+		},
+		TsNs: tsNs,
+	}
+}
+
+func TestReplaySkipsEventOlderThanLocalEntry(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+	store.entries[string(path)] = &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now},
+		Content:  []byte("newer local write"),
+	}
+
+	if err := f.Replay(overwriteEvent("/dir", "file", now.Add(-time.Minute).UnixNano(), "old replayed version")); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	entry, err := store.FindEntry(context.Background(), path)
+	if err != nil {
+		t.Fatalf("FindEntry: %v", err)
+	}
+	if string(entry.Content) != "newer local write" {
+		t.Fatalf("content = %q, want local write kept", entry.Content)
+	}
+}
+
+func TestReplayAppliesEventNewerThanLocalEntry(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+	store.entries[string(path)] = &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now.Add(-time.Minute)},
+		Content:  []byte("old local write"),
+	}
+
+	if err := f.Replay(overwriteEvent("/dir", "file", now.UnixNano(), "newer peer write")); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	entry, err := store.FindEntry(context.Background(), path)
+	if err != nil {
+		t.Fatalf("FindEntry: %v", err)
+	}
+	if string(entry.Content) != "newer peer write" {
+		t.Fatalf("content = %q, want peer write applied", entry.Content)
+	}
+}
+
+func TestReplaySkipsDeleteOlderThanLocalEntry(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+	store.entries[string(path)] = &Entry{FullPath: path, Attr: Attr{Mtime: now}}
+
+	if err := f.Replay(deleteEvent("/dir", "file", now.Add(-time.Minute).UnixNano())); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), path); err != nil {
+		t.Fatalf("newer local entry must survive a stale delete: %v", err)
+	}
+}
+
+func TestReplayAppliesNewerDelete(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+	store.entries[string(path)] = &Entry{FullPath: path, Attr: Attr{Mtime: now.Add(-time.Minute)}}
+
+	if err := f.Replay(deleteEvent("/dir", "file", now.UnixNano())); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), path); err == nil {
+		t.Fatal("newer delete must remove the local entry")
+	}
+}
+
+// A move event touches two paths; the local check applies to each side
+// independently, so a newer destination is kept while the source is removed.
+func TestReplayMoveChecksEachPathSeparately(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	oldPath := util.NewFullPath("/a", "file")
+	newPath := util.NewFullPath("/b", "file")
+	store.entries[string(oldPath)] = &Entry{FullPath: oldPath, Attr: Attr{Mtime: now.Add(-time.Minute)}}
+	store.entries[string(newPath)] = &Entry{FullPath: newPath, Attr: Attr{Mtime: now}, Content: []byte("kept")}
+
+	event := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/a",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      &filer_pb.Entry{Name: "file"},
+			NewEntry:      &filer_pb.Entry{Name: "file"},
+			NewParentPath: "/b",
+		},
+		TsNs: now.Add(-time.Minute).UnixNano(),
+	}
+	if err := f.Replay(event); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), oldPath); err == nil {
+		t.Fatal("old source entry should be deleted")
+	}
+	entry, err := store.FindEntry(context.Background(), newPath)
+	if err != nil {
+		t.Fatalf("FindEntry: %v", err)
+	}
+	if string(entry.Content) != "kept" {
+		t.Fatalf("content = %q, want newer destination kept", entry.Content)
+	}
+}
+
+// A local delete leaves a recorded tombstone, so a replayed create older than
+// the delete must not resurrect the path even though no entry remains to
+// compare against.
+func TestReplaySkipsCreateOlderThanLocalDelete(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+
+	// The delete removed the entry; its commit record survives.
+	f.notePathChange(path, now.UnixNano(), true)
+
+	createEvent := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "file", Content: []byte("stale resurrect")},
+		},
+		TsNs: now.Add(-time.Minute).UnixNano(),
+	}
+	if err := f.Replay(createEvent); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), path); err == nil {
+		t.Fatal("stale replayed create resurrected a locally deleted entry")
+	}
+}
+
+// A deleted directory covers its children: a bulk drop emits no per-child
+// events, so the tombstone on the ancestor is what a replayed child create
+// must check.
+func TestReplaySkipsCreateUnderDeletedAncestor(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+
+	f.notePathChange("/dir", now.UnixNano(), true)
+
+	createEvent := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "child", Content: []byte("stale")},
+		},
+		TsNs: now.Add(-time.Minute).UnixNano(),
+	}
+	if err := f.Replay(createEvent); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), "/dir/child"); err == nil {
+		t.Fatal("stale replayed create resurrected a child of a deleted directory")
+	}
+}
+
+// A recorded commit timestamp orders replay, not the entry's mtime: a rename
+// keeps the source's mtime on the destination, and a mounted client can set
+// mtime arbitrarily, so an entry whose mtime postdates the event must still
+// be overwritten when its commit record is older.
+func TestReplayAppliesWhenCommitRecordIsOlderThanFutureMtime(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/b", "x")
+	store.entries[string(path)] = &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now.Add(time.Hour)}, // future mtime, e.g. client-set or kept from a rename source
+		Content:  []byte("old destination"),
+	}
+	// The last committed change predates the event.
+	f.notePathChange(path, now.Add(-time.Minute).UnixNano(), false)
+
+	event := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/a",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      &filer_pb.Entry{Name: "x"},
+			NewEntry:      &filer_pb.Entry{Name: "x", Content: []byte("renamed in")},
+			NewParentPath: "/b",
+		},
+		TsNs: now.UnixNano(),
+	}
+	if err := f.Replay(event); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	entry, err := store.FindEntry(context.Background(), path)
+	if err != nil {
+		t.Fatalf("FindEntry: %v", err)
+	}
+	if string(entry.Content) != "renamed in" {
+		t.Fatalf("content = %q, want the rename destination applied", entry.Content)
+	}
+}
+
+// A store read failure is not a not-found: the replayed change must not be
+// applied on the guess that no newer local entry exists.
+func TestReplayPropagatesLookupError(t *testing.T) {
+	store := newStubFilerStore()
+	store.findErr = errors.New("store read failed")
+	f := newReplayTestFiler(store)
+
+	err := f.Replay(overwriteEvent("/dir", "file", time.Now().UnixNano(), "v"))
+	if err == nil {
+		t.Fatal("expected Replay to propagate the FindEntry failure")
+	}
+}
+
+// The check and the mutation share the per-path entry lock, so a local write
+// holding that lock cannot be split by a replayed change.
+func TestReplaySerializesWithEntryLock(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+	store.entries[string(path)] = &Entry{FullPath: path, Attr: Attr{Mtime: now}}
+
+	// A local mutation holding the path lock; a replay started behind it
+	// sees the recorded commit and skips.
+	held := f.EntryLockTable.AcquireLock("test local write", path, util.ExclusiveLock)
+	f.notePathChange(path, now.UnixNano(), false)
+
+	done := make(chan error, 1)
+	go func() {
+		done <- f.Replay(overwriteEvent("/dir", "file", now.Add(-time.Minute).UnixNano(), "stale"))
+	}()
+	select {
+	case <-done:
+		t.Fatal("replay finished while the path lock was still held")
+	case <-time.After(50 * time.Millisecond):
+	}
+	f.EntryLockTable.ReleaseLock(path, held)
+	if err := <-done; err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	entry, err := store.FindEntry(context.Background(), path)
+	if err != nil {
+		t.Fatalf("FindEntry: %v", err)
+	}
+	if entry.Content != nil {
+		t.Fatalf("content = %q, want local entry kept", entry.Content)
+	}
+}
+
+// A closed stop channel must abort the retry loop with
+// errPeerSubscriptionStopped so the subscription tears down and the offset
+// save keeps the last completed event instead of skipping this one.
+func TestReplicateMetadataChangeAbortsOnStop(t *testing.T) {
+	store := &flakyReplayStore{
+		stubFilerStore: newStubFilerStore(),
+		failUntil:      1 << 30,
+		err:            errors.New("i/o timeout talking to store"),
+	}
+	peer := pb.ServerAddress("peer-stopped:1")
+	f := newReplayTestFiler(store)
+
+	stopChan := make(chan struct{})
+	close(stopChan)
+	err := replicateMetadataChange(f, peer, quotaChangeEvent("stuck-bucket", 1<<20), stopChan)
+	if !errors.Is(err, errPeerSubscriptionStopped) {
+		t.Fatalf("err = %v, want errPeerSubscriptionStopped", err)
+	}
+	if got := store.attemptCount(); got != 0 {
+		t.Fatalf("attempts = %d, want 0: a stopped subscription must not replay", got)
+	}
+}
+
+// Commit-record keys must stay small: SQL KV stores keep the key suffix in a
+// VARCHAR(766) column, so a raw path over ~578 bytes silently drops the
+// record (the insert fails and the fallback update touches no row).
+func TestMetaPathKeyFixedSizeForLongPaths(t *testing.T) {
+	path := util.NewFullPath("/", string(make([]rune, 600)))
+	for _, p := range []util.FullPath{"/a", path} {
+		if got := len(metaPathKey(metaPathTsPrefix, p)); got != len(metaPathTsPrefix)+16 {
+			t.Fatalf("metaPathKey(%d-byte path) = %d bytes, want %d", len(p), got, len(metaPathTsPrefix)+16)
+		}
+	}
+}
+
+// Exact-path locking alone lets a replayed child create slip between a local
+// ancestor delete's tombstone check and the sweep: replay locks every
+// ancestor too, so it waits for the delete's lock and then sees the
+// tombstone.
+func TestReplaySerializesWithAncestorLock(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+
+	now := time.Now()
+	child := util.NewFullPath("/dir", "file")
+	held := f.EntryLockTable.AcquireLock("test local delete", "/dir", util.ExclusiveLock)
+	f.notePathChange("/dir", now.UnixNano(), true)
+
+	createEvent := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "file", Content: []byte("stale resurrect")},
+		},
+		TsNs: now.Add(-time.Minute).UnixNano(),
+	}
+	done := make(chan error, 1)
+	go func() {
+		done <- f.Replay(createEvent)
+	}()
+	select {
+	case <-done:
+		t.Fatal("replay finished while the ancestor lock was still held")
+	case <-time.After(50 * time.Millisecond):
+	}
+	f.EntryLockTable.ReleaseLock("/dir", held)
+	if err := <-done; err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), child); err == nil {
+		t.Fatal("stale replayed create resurrected a child of a locally deleted directory")
+	}
+}
+
+// Bootstrap ordering uses the sender's commit timestamp: a local commit
+// newer than the traversed entry keeps its state even when the peer entry's
+// mtime is far in the future.
+func TestBootstrapSkipsPeerEntryBehindLocalDelete(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+
+	f.notePathChange(path, now.UnixNano(), true)
+
+	entry := &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now.Add(time.Hour)}, // client-settable, must not win
+		Content:  []byte("stale peer entry"),
+	}
+	if err := f.applyTraversedEntry(context.Background(), entry, now.Add(-time.Minute).UnixNano()); err != nil {
+		t.Fatalf("applyTraversedEntry: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), path); err == nil {
+		t.Fatal("bootstrap resurrected a path newer local delete covered")
+	}
+}
+
+// The bug this guards: a peer entry committed recently but stamped with an
+// old client mtime must still apply over an absent-or-older local commit.
+func TestBootstrapAppliesPeerCommitNewerThanLocal(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+
+	entry := &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now.Add(-time.Hour)}, // old client-set mtime
+		Content:  []byte("fresh peer write"),
+	}
+	if err := f.applyTraversedEntry(context.Background(), entry, now.UnixNano()); err != nil {
+		t.Fatalf("applyTraversedEntry: %v", err)
+	}
+	got, err := store.FindEntry(context.Background(), path)
+	if err != nil {
+		t.Fatalf("peer entry with newer commit must bootstrap: %v", err)
+	}
+	if string(got.Content) != "fresh peer write" {
+		t.Fatalf("content = %q, want peer entry applied", got.Content)
+	}
+}
+
+// A bootstrap-applied entry records the sender's commit timestamp, so a
+// replayed event older than it — but newer than the entry's mtime — is
+// still skipped.
+func TestBootstrapAppliedCommitBlocksStaleReplay(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+
+	entry := &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now.Add(-time.Hour)},
+		Content:  []byte("fresh peer write"),
+	}
+	if err := f.applyTraversedEntry(context.Background(), entry, now.UnixNano()); err != nil {
+		t.Fatalf("applyTraversedEntry: %v", err)
+	}
+	if err := f.Replay(overwriteEvent("/dir", "file", now.Add(-time.Minute).UnixNano(), "stale replay")); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	got, _ := store.FindEntry(context.Background(), path)
+	if string(got.Content) != "fresh peer write" {
+		t.Fatalf("content = %q, stale replay must not overwrite the bootstrap entry", got.Content)
+	}
+}
+
+// An existing local entry is replaced only when the peer commit is newer —
+// not when its mtime happens to be.
+func TestBootstrapUpdatesExistingByCommitTime(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+
+	store.entries[string(path)] = &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now.Add(time.Hour)}, // future mtime, e.g. client-set
+		Content:  []byte("old local write"),
+	}
+	f.notePathChange(path, now.Add(-time.Minute).UnixNano(), false)
+
+	entry := &Entry{
+		FullPath: path,
+		Attr:     Attr{Mtime: now.Add(-time.Hour)},
+		Content:  []byte("newer peer commit"),
+	}
+	if err := f.applyTraversedEntry(context.Background(), entry, now.UnixNano()); err != nil {
+		t.Fatalf("applyTraversedEntry: %v", err)
+	}
+	got, _ := store.FindEntry(context.Background(), path)
+	if string(got.Content) != "newer peer commit" {
+		t.Fatalf("content = %q, newer peer commit must update despite the future local mtime", got.Content)
+	}
+}
+
+// A delete whose commit record failed to persist still blocks a stale
+// replayed create: the pending record stands in until the retry lands it.
+func TestPendingDeleteRecordBlocksStaleReplay(t *testing.T) {
+	store := newStubFilerStore()
+	store.setKvPutErr(errors.New("kv store unreachable"))
+	f := newReplayTestFiler(store)
+	f.flushCtx, f.flushCancel = context.WithCancel(context.Background())
+	defer f.flushCancel()
+	now := time.Now()
+	path := util.NewFullPath("/dir", "file")
+
+	// The delete already committed; only its ordering record failed.
+	f.notePathChange(path, now.UnixNano(), true)
+
+	createEvent := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/dir",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "file", Content: []byte("stale resurrect")},
+		},
+		TsNs: now.Add(-time.Minute).UnixNano(),
+	}
+	if err := f.Replay(createEvent); err != nil {
+		t.Fatalf("Replay: %v", err)
+	}
+	if _, err := store.FindEntry(context.Background(), path); err == nil {
+		t.Fatal("stale create resurrected a path whose delete record is still pending")
+	}
+}
+
+// The failed record write retries until it lands, then leaves the pending set.
+func TestPendingCommitRecordRepairs(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	f.flushCtx, f.flushCancel = context.WithCancel(context.Background())
+	defer f.flushCancel()
+	path := util.NewFullPath("/dir", "file")
+
+	store.setKvPutErr(errors.New("kv store unreachable"))
+	f.notePathChange(path, 100, true)
+	if _, loaded := f.metaTsPending.Load(string(path)); !loaded {
+		t.Fatal("failed record write must stay pending")
+	}
+	store.setKvPutErr(nil)
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, loaded := f.metaTsPending.Load(string(path)); !loaded {
+			ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathDelPrefix, path))
+			if err != nil || !found || ts != 100 {
+				t.Fatalf("delete record after repair: ts=%v found=%v err=%v", ts, found, err)
+			}
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("pending record was not repaired")
+}
+
+// CommitTsNs reports the recorded commit time over the entry mtime, and a
+// still-pending record over both.
+func TestCommitTsNsPrefersRecordAndPending(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	path := util.NewFullPath("/dir", "file")
+
+	if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 7 {
+		t.Fatalf("no record: CommitTsNs = %d, %v; want fallback 7", got, err)
+	}
+	f.notePathChange(path, 50, false)
+	if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 50 {
+		t.Fatalf("recorded: CommitTsNs = %d, %v; want 50", got, err)
+	}
+	pending := &metaTsPendingRec{}
+	pending.changeTsNs.Store(80)
+	f.metaTsPending.Store(string(path), pending)
+	if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 80 {
+		t.Fatalf("pending: CommitTsNs = %d, %v; want 80", got, err)
+	}
+}
+
+// A successful write discharges what the pending record owed before it, so a
+// repair pass can never put an older timestamp back over it.
+func TestPendingCommitRecordDischargedByNewerWrite(t *testing.T) {
+	store := newStubFilerStore()
+	f := newReplayTestFiler(store)
+	f.flushCtx, f.flushCancel = context.WithCancel(context.Background())
+	defer f.flushCancel()
+	path := util.NewFullPath("/dir", "file")
+
+	store.setKvPutErr(errors.New("kv store unreachable"))
+	f.notePathChange(path, 100, false)
+	store.setKvPutErr(nil)
+	f.notePathChange(path, 200, false)
+
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, loaded := f.metaTsPending.Load(string(path)); !loaded {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathTsPrefix, path))
+	if err != nil || !found || ts != 200 {
+		t.Fatalf("MpTs after repair = %v, %v, %v; want 200", ts, found, err)
+	}
+}
+
+// A failed delete record keeps its own time even when a later failed change
+// merges into the pending record: the repair must not stamp the delete at
+// the newer change's time.
+func TestPendingDeleteRecordKeepsItsTime(t *testing.T) {
+	store := newStubFilerStore()
+	store.setKvPutErr(errors.New("kv store unreachable"))
+	f := newReplayTestFiler(store)
+	f.flushCtx, f.flushCancel = context.WithCancel(context.Background())
+	defer f.flushCancel()
+	path := util.NewFullPath("/dir", "file")
+
+	f.notePathChange(path, 100, true)
+	f.notePathChange(path, 200, false)
+	store.setKvPutErr(nil)
+
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, loaded := f.metaTsPending.Load(string(path)); !loaded {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathTsPrefix, path)); err != nil || !found || ts != 200 {
+		t.Fatalf("MpTs = %v, %v, %v; want 200", ts, found, err)
+	}
+	if ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathDelPrefix, path)); err != nil || !found || ts != 100 {
+		t.Fatalf("MpDl = %v, %v, %v; want 100", ts, found, err)
+	}
+}

weed/filer/meta_aggregator_test.go

@@ -34,3 +34,55 @@ func TestOnPeerUpdateRepeatedAdd(t *testing.T) {
 		t.Fatal("expecting the subscription to be removed")
 	}
 }
+
+// Shutdown must close every peer's stopChan and wait for the loops, which is
+// what lets each loop's deferred offset save run before the store closes.
+func TestMetaAggregatorShutdownStopsPeerLoops(t *testing.T) {
+	ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil)
+
+	stopChan := make(chan struct{})
+	peer := pb.ServerAddress("127.0.0.1:1")
+	ma.peerChans[peer] = stopChan
+	ma.peerWg.Add(1)
+	exited := make(chan struct{})
+	go func() {
+		defer ma.peerWg.Done()
+		<-stopChan
+		close(exited)
+	}()
+
+	done := make(chan struct{})
+	go func() {
+		ma.Shutdown()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Shutdown did not return")
+	}
+	select {
+	case <-exited:
+	default:
+		t.Fatal("peer loop was not stopped")
+	}
+}
+
+// After Shutdown begins, a peer-add update arriving on the still-connected
+// master watch must not start a subscription: it would have no stop signal
+// and could run past the store close.
+func TestOnPeerUpdateAfterShutdownRejected(t *testing.T) {
+	ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil)
+	ma.Shutdown()
+
+	peer := pb.ServerAddress("127.0.0.1:1")
+	ma.OnPeerUpdate(&master_pb.ClusterNodeUpdate{Address: string(peer), IsAdd: true}, time.Now())
+
+	ma.peerChansLock.Lock()
+	_, found := ma.peerChans[peer]
+	ma.peerChansLock.Unlock()
+	if found {
+		t.Fatal("peer subscription started after shutdown")
+	}
+}

weed/filer/meta_path_ts.go

@@ -0,0 +1,324 @@
+package filer
+
+import (
+	"context"
+	"crypto/sha256"
+	"errors"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+const (
+	// metaPathTsPrefix keys per-path commit timestamps in the filer store KV:
+	// the newest event timestamp this filer committed or applied for the
+	// path. Written for every emitted mutation event and kept after the entry
+	// is deleted, so a replayed peer event can still be ordered against a
+	// change that left no entry behind.
+	metaPathTsPrefix = "MpTs"
+	// metaPathDelPrefix keys delete-only timestamps. A replayed event checks
+	// the path's ancestors too, so a directory delete also covers children
+	// swept without per-child events (a dropped bucket's whole-store delete,
+	// or a walk aborted partway).
+	metaPathDelPrefix = "MpDl"
+)
+
+// metaPathKey hashes the path into a fixed-size key. SQL-backed KV stores
+// split the key at byte 8 and store the remainder base64-encoded in a
+// VARCHAR(766) name column: a raw path over ~578 bytes overflows it, the
+// KvPut insert fails, and the fallback update touches no row yet returns no
+// error — the commit record would never land and stale replay could restore
+// the path.
+func metaPathKey(prefix string, path util.FullPath) []byte {
+	sum := sha256.Sum256([]byte(path))
+	key := make([]byte, len(prefix)+16)
+	copy(key, prefix)
+	copy(key[len(prefix):], sum[:16])
+	return key
+}
+
+// kvReadTsNs returns the timestamp recorded for key, or found=false when the
+// key is absent or malformed. Any other store read failure is an error: a
+// caller deciding whether a replayed change is stale must not guess.
+func (f *Filer) kvReadTsNs(ctx context.Context, key []byte) (tsNs int64, found bool, err error) {
+	value, err := f.Store.KvGet(ctx, key)
+	if err != nil {
+		if errors.Is(err, ErrKvNotFound) {
+			return 0, false, nil
+		}
+		return 0, false, err
+	}
+	if len(value) < 8 {
+		return 0, false, nil
+	}
+	return int64(util.BytesToUint64(value)), true, nil
+}
+
+// notePathChange persists tsNs as path's newest committed change. Local event
+// timestamps are stamped monotonically (metaLogInflight.stamp), so a plain
+// put cannot regress an earlier local record; a higher peer record it may
+// overwrite was already newer under last-writer-wins ordering anyway. A
+// delete additionally writes a delete record the ancestor lookup consults.
+//
+// The write uses a fresh context: callers notify after CommitTransaction,
+// and SQL stores keep the already-committed *sql.Tx in the request context —
+// a record written through it would fail and the path would lose ordering.
+//
+// A failed write is kept in metaTsPending and retried in the background: the
+// mutation the record orders has already committed, so dropping it would let
+// a stale replayed change overwrite or resurrect the path while the store is
+// unreachable. Ordering reads consult the pending record until the retry
+// lands it.
+func (f *Filer) notePathChange(path util.FullPath, tsNs int64, deleted bool) {
+	if f.Store == nil {
+		return
+	}
+	ctx := context.Background()
+	key := string(path)
+	rec := f.pendingRec(key)
+	defer rec.mu.Unlock()
+
+	value := make([]byte, 8)
+	util.Uint64toBytes(value, uint64(tsNs))
+	if err := f.Store.KvPut(ctx, metaPathKey(metaPathTsPrefix, path), value); err != nil {
+		glog.WarningfCtx(ctx, "record meta change time %s: %v", path, err)
+		if tsNs > rec.changeTsNs.Load() {
+			rec.changeTsNs.Store(tsNs)
+		}
+	} else if rec.changeTsNs.Load() <= tsNs {
+		// This write covers every change record owed so far.
+		rec.changeTsNs.Store(0)
+	}
+	if deleted {
+		if err := f.Store.KvPut(ctx, metaPathKey(metaPathDelPrefix, path), value); err != nil {
+			glog.WarningfCtx(ctx, "record meta delete time %s: %v", path, err)
+			if tsNs > rec.deleteTsNs.Load() {
+				rec.deleteTsNs.Store(tsNs)
+			}
+		} else if rec.deleteTsNs.Load() <= tsNs {
+			rec.deleteTsNs.Store(0)
+		}
+	}
+	if rec.changeTsNs.Load() == 0 && rec.deleteTsNs.Load() == 0 {
+		f.metaTsPending.CompareAndDelete(key, rec)
+		return
+	}
+	if _, running := f.metaTsRepairing.LoadOrStore(key, struct{}{}); !running {
+		go f.repairMetaPathTs(path)
+	}
+}
+
+// metaTsPendingRec is a commit record still on its way to the store, holding
+// only work not yet written: changeTsNs is the newest committed change the
+// MpTs record lacks and deleteTsNs the newest delete the MpDl record lacks —
+// the two keep their own times so a later update cannot masquerade as the
+// delete. Store writes and the bookkeeping they discharge share mu, so a
+// repair pass can never write an older record over a newer one that already
+// landed; readers take the atomic fields directly.
+type metaTsPendingRec struct {
+	mu         sync.Mutex
+	changeTsNs atomic.Int64
+	deleteTsNs atomic.Int64
+}
+
+// pendingRec returns the live pending record for key, locked for mutation,
+// recreating it when a repair already dropped the previous one.
+func (f *Filer) pendingRec(key string) *metaTsPendingRec {
+	for {
+		cur, _ := f.metaTsPending.LoadOrStore(key, &metaTsPendingRec{})
+		rec := cur.(*metaTsPendingRec)
+		rec.mu.Lock()
+		if live, ok := f.metaTsPending.Load(key); ok && live == cur {
+			return rec
+		}
+		rec.mu.Unlock()
+	}
+}
+
+// repairMetaPathTs retries a pending commit record until it lands. Each pass
+// writes only what the record still owes, so a merge by a later failure is
+// picked up without another goroutine and a landed write is never rewritten.
+// The retry is unbounded — the mutation is already committed and only the
+// record preserves its ordering.
+func (f *Filer) repairMetaPathTs(path util.FullPath) {
+	key := string(path)
+	ctx := f.flushCtx
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	defer func() {
+		f.metaTsRepairing.Delete(key)
+		// A failure landing between the last pass and the flag release
+		// still needs a repair pass — unless the retry context is done.
+		if ctx.Err() != nil {
+			return
+		}
+		if _, ok := f.metaTsPending.Load(key); ok {
+			if _, running := f.metaTsRepairing.LoadOrStore(key, struct{}{}); !running {
+				go f.repairMetaPathTs(path)
+			}
+		}
+	}()
+	for {
+		if _, ok := f.metaTsPending.Load(key); !ok {
+			return
+		}
+		rec := f.pendingRec(key)
+		var tsErr, delErr error
+		if tsNs := rec.changeTsNs.Load(); tsNs > 0 {
+			value := make([]byte, 8)
+			util.Uint64toBytes(value, uint64(tsNs))
+			if tsErr = f.Store.KvPut(ctx, metaPathKey(metaPathTsPrefix, path), value); tsErr == nil {
+				rec.changeTsNs.Store(0)
+			}
+		}
+		if tsNs := rec.deleteTsNs.Load(); tsNs > 0 {
+			value := make([]byte, 8)
+			util.Uint64toBytes(value, uint64(tsNs))
+			if delErr = f.Store.KvPut(ctx, metaPathKey(metaPathDelPrefix, path), value); delErr == nil {
+				rec.deleteTsNs.Store(0)
+			}
+		}
+		done := rec.changeTsNs.Load() == 0 && rec.deleteTsNs.Load() == 0
+		if done {
+			f.metaTsPending.CompareAndDelete(key, rec)
+		}
+		rec.mu.Unlock()
+		if !done {
+			glog.V(0).InfofCtx(ctx, "retry record meta change time %s: %v %v", path, tsErr, delErr)
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(time.Second):
+			}
+		}
+	}
+}
+
+// pendingPathTsNs returns the newest pending commit timestamp for path and
+// the newest pending delete timestamp.
+func (f *Filer) pendingPathTsNs(path util.FullPath) (changeTsNs, deleteTsNs int64) {
+	cur, ok := f.metaTsPending.Load(string(path))
+	if !ok {
+		return 0, 0
+	}
+	rec := cur.(*metaTsPendingRec)
+	return rec.changeTsNs.Load(), rec.deleteTsNs.Load()
+}
+
+// CommitTsNs returns the newest commit timestamp known for path — the
+// recorded event time, or a still-pending one — and fallbackTsNs when none
+// exists. Peers stamp traversal entries with it so a bootstrap subscriber
+// orders each entry on the sender's commit clock rather than the entry's
+// client-settable mtime. A store read failure is an error, not a fallback:
+// the mtime it would mint could outdate real commits.
+func (f *Filer) CommitTsNs(ctx context.Context, path util.FullPath, fallbackTsNs int64) (int64, error) {
+	if f.Store == nil {
+		return fallbackTsNs, nil
+	}
+	tsNs := fallbackTsNs
+	recTs, found, err := f.kvReadTsNs(ctx, metaPathKey(metaPathTsPrefix, path))
+	if err != nil {
+		return 0, err
+	}
+	if found {
+		tsNs = recTs
+	}
+	if pendTs, _ := f.pendingPathTsNs(path); pendTs > tsNs {
+		tsNs = pendTs
+	}
+	return tsNs, nil
+}
+
+// noteCommittedEvent records the event's commit timestamp for every path it
+// touches: the source of a delete or rename, and the target of a create,
+// update, or rename. It runs after the local mutation commits while the
+// caller still holds the path's entry lock (for filer-server mutations), so
+// a peer replay ordered by the same lock sees either the whole local change
+// or none of it.
+func (f *Filer) noteCommittedEvent(event *filer_pb.SubscribeMetadataResponse) {
+	message := event.EventNotification
+	if message == nil {
+		return
+	}
+	var source, target util.FullPath
+	if message.OldEntry != nil {
+		source = util.FullPath(filer_pb.MetadataEventSourceFullPath(event))
+	}
+	if message.NewEntry != nil {
+		target = util.FullPath(filer_pb.MetadataEventTargetFullPath(event))
+	}
+	// A delete or a rename removes the source path; an update reports the
+	// same path on both sides and counts once.
+	if source != "" && source != target {
+		f.notePathChange(source, event.TsNs, true)
+	}
+	if target != "" {
+		f.notePathChange(target, event.TsNs, false)
+	}
+}
+
+// deletedAfterTsNs returns the newest recorded delete timestamp among path
+// and its ancestors.
+func (f *Filer) deletedAfterTsNs(ctx context.Context, path util.FullPath) (int64, error) {
+	var tsNs int64
+	for p := string(path); p != "" && p != "/"; {
+		ts, found, err := f.kvReadTsNs(ctx, metaPathKey(metaPathDelPrefix, util.FullPath(p)))
+		if err != nil {
+			return 0, err
+		}
+		if found && ts > tsNs {
+			tsNs = ts
+		}
+		if _, pendDel := f.pendingPathTsNs(util.FullPath(p)); pendDel > tsNs {
+			tsNs = pendDel
+		}
+		i := strings.LastIndexByte(p, '/')
+		if i <= 0 {
+			break
+		}
+		p = p[:i]
+	}
+	return tsNs, nil
+}
+
+// localCommitTsNs returns the newest timestamp this filer is known to have
+// committed for path: the recorded event time when one exists — it is not
+// client-controllable the way an entry's mtime is — else the entry's mtime
+// for entries written before recording existed, plus any recorded delete of
+// the path or an ancestor.
+func (f *Filer) localCommitTsNs(ctx context.Context, path util.FullPath) (int64, error) {
+	if f.Store == nil {
+		return 0, nil
+	}
+	var tsNs int64
+	recTs, found, err := f.kvReadTsNs(ctx, metaPathKey(metaPathTsPrefix, path))
+	if err != nil {
+		return 0, err
+	}
+	if found {
+		tsNs = recTs
+	} else if entry, findErr := f.Store.FindEntry(ctx, path); findErr != nil {
+		if !errors.Is(findErr, filer_pb.ErrNotFound) {
+			return 0, findErr
+		}
+	} else if entry != nil {
+		tsNs = entry.Mtime.UnixNano()
+	}
+	if pendTs, _ := f.pendingPathTsNs(path); pendTs > tsNs {
+		tsNs = pendTs
+	}
+	delTs, err := f.deletedAfterTsNs(ctx, path)
+	if err != nil {
+		return 0, err
+	}
+	if delTs > tsNs {
+		tsNs = delTs
+	}
+	return tsNs, nil
+}

weed/filer/meta_replay.go

@@ -2,6 +2,8 @@ package filer
 
 import (
 	"context"
+	"slices"
+	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -9,31 +11,166 @@ import (
 )
 
 // 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 {
+// store calls, so retrying it after a partial failure is not atomic; both
+// operations are idempotent and the timestamp check tolerates re-delivery.
+//
+// A replayed event can be older than local state: a peer resuming from a
+// stale offset replays everything since the offset, and two writes through
+// different filers race within the replication lag. Apply each path's change
+// only when no newer local change is recorded, so a late event cannot revert
+// or resurrect entries.
+//
+// Ordering uses the per-path commit timestamp recorded for every emitted
+// mutation (meta_path_ts.go), which survives deletes as a tombstone and is
+// not client-controllable the way entry mtime is — a mounted client can set
+// mtime into the future, and a rename keeps the source file's mtime on the
+// destination entry. Entries written before recording existed fall back to
+// their mtime.
+//
+// The check and the mutation run under the same per-path entry lock the
+// filer server's local writes hold, so they cannot be split by a concurrent
+// local update or by another peer's replay.
+func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
 	message := resp.EventNotification
-	var oldPath util.FullPath
-	var newEntry *Entry
+	dir := resp.Directory
+	if message.NewParentPath != "" {
+		dir = message.NewParentPath
+	}
+	var oldPath, newPath util.FullPath
 	if message.OldEntry != nil {
 		oldPath = util.NewFullPath(resp.Directory, message.OldEntry.Name)
-		glog.V(4).Infof("deleting %v", oldPath)
-		if err := filerStore.DeleteEntry(context.Background(), oldPath); err != nil {
+	}
+	if message.NewEntry != nil {
+		newPath = util.NewFullPath(dir, message.NewEntry.Name)
+	}
+	ctx := context.Background()
+
+	unlock := f.lockReplayPaths(oldPath, newPath)
+	defer unlock()
+
+	samePath := oldPath != "" && oldPath == newPath
+	if samePath {
+		newer, err := f.replayBlockedByLocal(ctx, oldPath, resp.TsNs)
+		if err != nil || newer {
 			return err
 		}
+		glog.V(4).Infof("deleting %v", oldPath)
+		if err := f.Store.DeleteEntry(ctx, oldPath); err != nil {
+			return err
+		}
+		// Record the commit before the insert too: if the insert fails and
+		// the event is retried later, the path sits deleted in between —
+		// without the record a stale peer create could resurrect it in the
+		// gap. The entry still exists afterward, so no delete record.
+		f.notePathChange(oldPath, resp.TsNs, false)
+		glog.V(4).Infof("creating %v", newPath)
+		newEntry := FromPbEntry(dir, message.NewEntry)
+		if err := f.Store.InsertEntry(ctx, newEntry); err != nil {
+			return err
+		}
+		f.notePathChange(oldPath, resp.TsNs, false)
+		return nil
+	}
+
+	if message.OldEntry != nil {
+		newer, err := f.replayBlockedByLocal(ctx, oldPath, resp.TsNs)
+		if err != nil {
+			return err
+		}
+		if !newer {
+			glog.V(4).Infof("deleting %v", oldPath)
+			if err := f.Store.DeleteEntry(ctx, oldPath); err != nil {
+				return err
+			}
+			f.notePathChange(oldPath, resp.TsNs, true)
+		}
 	}
 
 	if message.NewEntry != nil {
-		dir := resp.Directory
-		if message.NewParentPath != "" {
-			dir = message.NewParentPath
-		}
-		key := util.NewFullPath(dir, message.NewEntry.Name)
-		glog.V(4).Infof("creating %v", key)
-		newEntry = FromPbEntry(dir, message.NewEntry)
-		if err := filerStore.InsertEntry(context.Background(), newEntry); err != nil {
+		newer, err := f.replayBlockedByLocal(ctx, newPath, resp.TsNs)
+		if err != nil {
 			return err
 		}
+		if !newer {
+			glog.V(4).Infof("creating %v", newPath)
+			newEntry := FromPbEntry(dir, message.NewEntry)
+			if err := f.Store.InsertEntry(ctx, newEntry); err != nil {
+				return err
+			}
+			f.notePathChange(newPath, resp.TsNs, false)
+		}
 	}
 
 	return nil
 }
+
+// replayBlockedByLocal reports whether the local store already holds a
+// version of path committed after the event's timestamp. Lookup failures are
+// propagated: a replayed change must not be applied on the guess that the
+// path was simply absent, or a transient read error could let a stale event
+// overwrite a newer entry past the point of no return.
+func (f *Filer) replayBlockedByLocal(ctx context.Context, path util.FullPath, tsNs int64) (bool, error) {
+	if tsNs <= 0 {
+		return false, nil
+	}
+	localTsNs, err := f.localCommitTsNs(ctx, path)
+	if err != nil {
+		return false, err
+	}
+	return localTsNs > tsNs, nil
+}
+
+// lockReplayPaths holds the shared entry locks for every path the event
+// touches, in the same sorted order the rename handler takes them.
+// lockReplayPaths takes the per-path entry locks for the event's paths and
+// every ancestor of them, in lexicographic order — parents sort before
+// children, the same ordering the local rename handler's two-path lock uses,
+// so lock acquisition is deadlock-free across operations. Ancestors are
+// included because the replay check consults ancestor delete records: a
+// local recursive delete or directory rename holds only the ancestor's lock,
+// and without contending on it a replay could reinsert a stale child after
+// its check. Ancestors are locked shared — they only need to exclude an
+// exclusive holder, not each other, or every replay under a busy parent
+// would serialize — while a path the event actually changes stays exclusive
+// even when it is also the other path's ancestor. The root is never locked:
+// it cannot be deleted, and a shared root lock would still be excluded by
+// every top-level delete.
+func (f *Filer) lockReplayPaths(oldPath, newPath util.FullPath) func() {
+	lt := f.EntryLockTable
+	if lt == nil {
+		return func() {}
+	}
+	var paths []util.FullPath
+	addAncestors := func(p util.FullPath) {
+		for i := strings.LastIndexByte(string(p), '/'); i > 0; i = strings.LastIndexByte(string(p), '/') {
+			p = p[:i]
+			paths = append(paths, p)
+		}
+	}
+	addAncestors(oldPath)
+	addAncestors(newPath)
+	if oldPath != "" {
+		paths = append(paths, oldPath)
+	}
+	if newPath != "" {
+		paths = append(paths, newPath)
+	}
+	if len(paths) == 0 {
+		return func() {}
+	}
+	slices.Sort(paths)
+	paths = slices.Compact(paths)
+	locks := make([]*util.ActiveLock, len(paths))
+	for i, p := range paths {
+		lockType := util.SharedLock
+		if p == oldPath || p == newPath {
+			lockType = util.ExclusiveLock
+		}
+		locks[i] = lt.AcquireLock("MetaReplay", p, lockType)
+	}
+	return func() {
+		for i := len(paths) - 1; i >= 0; i-- {
+			lt.ReleaseLock(paths[i], locks[i])
+		}
+	}
+}

weed/pb/filer.proto

@@ -714,6 +714,10 @@ message TraverseBfsMetadataRequest {
 message TraverseBfsMetadataResponse {
     string directory = 1;
     Entry entry = 2;
+    // The sender's per-path commit timestamp (see meta_path_ts.go): the newest
+    // event time this filer committed for the path, 0 when unrecorded so
+    // subscribers fall back to the entry's mtime.
+    int64 entry_ts_ns = 3;
 }
 
 message LogEntry {

weed/pb/filer_pb/filer.pb.go

@@ -4777,9 +4777,13 @@ func (x *TraverseBfsMetadataRequest) GetExcludedPrefixes() []string {
 }
 
 type TraverseBfsMetadataResponse struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	Directory     string                 `protobuf:"bytes,1,opt,name=directory,proto3" json:"directory,omitempty"`
-	Entry         *Entry                 `protobuf:"bytes,2,opt,name=entry,proto3" json:"entry,omitempty"`
+	state     protoimpl.MessageState `protogen:"open.v1"`
+	Directory string                 `protobuf:"bytes,1,opt,name=directory,proto3" json:"directory,omitempty"`
+	Entry     *Entry                 `protobuf:"bytes,2,opt,name=entry,proto3" json:"entry,omitempty"`
+	// The sender's per-path commit timestamp (see meta_path_ts.go): the newest
+	// event time this filer committed for the path, 0 when unrecorded so
+	// subscribers fall back to the entry's mtime.
+	EntryTsNs     int64 `protobuf:"varint,3,opt,name=entry_ts_ns,json=entryTsNs,proto3" json:"entry_ts_ns,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -4828,6 +4832,13 @@ func (x *TraverseBfsMetadataResponse) GetEntry() *Entry {
 	return nil
 }
 
+func (x *TraverseBfsMetadataResponse) GetEntryTsNs() int64 {
+	if x != nil {
+		return x.EntryTsNs
+	}
+	return 0
+}
+
 type LogEntry struct {
 	state            protoimpl.MessageState `protogen:"open.v1"`
 	TsNs             int64                  `protobuf:"varint,1,opt,name=ts_ns,json=tsNs,proto3" json:"ts_ns,omitempty"`
@@ -7688,10 +7699,11 @@ const file_filer_proto_rawDesc = "" +
 	"\bfiler_id\x18\x03 \x01(\tR\afilerId\"g\n" +
 	"\x1aTraverseBfsMetadataRequest\x12\x1c\n" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12+\n" +
-	"\x11excluded_prefixes\x18\x02 \x03(\tR\x10excludedPrefixes\"b\n" +
+	"\x11excluded_prefixes\x18\x02 \x03(\tR\x10excludedPrefixes\"\x82\x01\n" +
 	"\x1bTraverseBfsMetadataResponse\x12\x1c\n" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12%\n" +
-	"\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\"\x8b\x01\n" +
+	"\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12\x1e\n" +
+	"\ventry_ts_ns\x18\x03 \x01(\x03R\tentryTsNs\"\x8b\x01\n" +
 	"\bLogEntry\x12\x13\n" +
 	"\x05ts_ns\x18\x01 \x01(\x03R\x04tsNs\x12,\n" +
 	"\x12partition_key_hash\x18\x02 \x01(\x05R\x10partitionKeyHash\x12\x12\n" +

weed/pb/filer_pb/filer_vtproto.pb.go

@@ -4379,6 +4379,11 @@ func (m *TraverseBfsMetadataResponse) MarshalToSizedBufferVT(dAtA []byte) (int,
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.EntryTsNs != 0 {
+		i = protohelpers.EncodeVarint(dAtA, i, uint64(m.EntryTsNs))
+		i--
+		dAtA[i] = 0x18
+	}
 	if m.Entry != nil {
 		size, err := m.Entry.MarshalToSizedBufferVT(dAtA[:i])
 		if err != nil {
@@ -8267,6 +8272,9 @@ func (m *TraverseBfsMetadataResponse) SizeVT() (n int) {
 		l = m.Entry.SizeVT()
 		n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
 	}
+	if m.EntryTsNs != 0 {
+		n += 1 + protohelpers.SizeOfVarint(uint64(m.EntryTsNs))
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -20865,6 +20873,25 @@ func (m *TraverseBfsMetadataResponse) UnmarshalVT(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field EntryTsNs", wireType)
+			}
+			m.EntryTsNs = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.EntryTsNs |= int64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])

weed/server/filer_grpc_server_traverse_meta.go

@@ -42,9 +42,14 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ
 	// the filer on large trees (e.g. during a peer's first-time bootstrap).
 	sendEntry := func(entry *filer.Entry) error {
 		parent, _ := entry.FullPath.DirAndName()
+		entryTsNs, err := fs.filer.CommitTsNs(ctx, entry.FullPath, entry.Mtime.UnixNano())
+		if err != nil {
+			return fmt.Errorf("commit timestamp of %s: %w", entry.FullPath, err)
+		}
 		if err := stream.Send(&filer_pb.TraverseBfsMetadataResponse{
 			Directory: parent,
 			Entry:     entry.ToProtoEntry(),
+			EntryTsNs: entryTsNs,
 		}); err != nil {
 			return fmt.Errorf("send traverse bfs metadata response: %w", err)
 		}

weed/server/filer_grpc_server_traverse_meta_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/stretchr/testify/assert"
@@ -37,8 +38,9 @@ func TestPtrie(t *testing.T) {
 
 type captureTraverseStream struct {
 	grpc.ServerStream
-	ctx     context.Context
-	visited []string
+	ctx       context.Context
+	visited   []string
+	entryTsNs map[string]int64
 }
 
 func (s *captureTraverseStream) Context() context.Context { return s.ctx }
@@ -48,7 +50,12 @@ func (s *captureTraverseStream) Send(resp *filer_pb.TraverseBfsMetadataResponse)
 	if dir != "/" {
 		dir += "/"
 	}
-	s.visited = append(s.visited, dir+resp.Entry.Name)
+	path := dir + resp.Entry.Name
+	s.visited = append(s.visited, path)
+	if s.entryTsNs == nil {
+		s.entryTsNs = make(map[string]int64)
+	}
+	s.entryTsNs[path] = resp.EntryTsNs
 	return nil
 }
 
@@ -89,6 +96,16 @@ func TestTraverseBfsMetadata(t *testing.T) {
 		}
 		seen[path] = true
 	}
+
+	// Every response carries a commit timestamp for ordering; with no
+	// recorded commit it falls back to the entry's mtime.
+	for path, tsNs := range stream.entryTsNs {
+		if path == "/" {
+			assert.Equal(t, filer.Root.Attr.Mtime.UnixNano(), tsNs)
+			continue
+		}
+		assert.Equalf(t, store.entries[path].Attr.Mtime.UnixNano(), tsNs, "entry_ts_ns for %s", path)
+	}
 }
 
 var _ filer_pb.SeaweedFiler_TraverseBfsMetadataServer = (*captureTraverseStream)(nil)

weed/server/filer_server.go

@@ -268,6 +268,9 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	maxFilenameLength := v.GetUint32("filer.options.max_file_name_length")
 	glog.V(0).Infof("max_file_name_length %d", maxFilenameLength)
 	fs.filer = filer.NewFiler(*option.Masters, fs.grpcDialOption, option.Host, option.FilerGroup, option.Collection, option.DefaultReplication, option.DataCenter, maxFilenameLength, nil)
+	// Peer metadata replay takes the same per-path locks, so a replayed change
+	// cannot split a local mutation's read-modify-write.
+	fs.entryLockTable = fs.filer.EntryLockTable
 	fs.filer.Cipher = option.Cipher
 	fs.filer.DefaultDiskType = option.DiskType
 	fs.filer.BuildGuardedRemoteClient = BuildGuardedRemoteStorageClient

weed/server/filer_server_handlers_write.go

@@ -255,7 +255,15 @@ func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	err = fs.filer.DeleteEntryMetaAndData(context.Background(), util.FullPath(objectPath), isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil, 0)
+	// The delete and the commit record it emits must serialize with peer
+	// replay of this path, same as the gRPC DeleteEntry handler.
+	fullPath := util.FullPath(objectPath)
+	if fs.entryLockTable != nil {
+		pathLock := fs.entryLockTable.AcquireLock("DeleteHandler", fullPath, util.ExclusiveLock)
+		defer fs.entryLockTable.ReleaseLock(fullPath, pathLock)
+	}
+
+	err = fs.filer.DeleteEntryMetaAndData(context.Background(), fullPath, isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil, 0)
 	if err != nil && err != filer_pb.ErrNotFound {
 		glog.V(1).Infoln("deleting", objectPath, ":", err.Error())
 		writeJsonError(w, r, http.StatusInternalServerError, err)