filer: deliver meta-log events under their stream position, not their stamp (#12084)

* filer: deliver meta-log events under their stream position, not their stamp

A mutation's event timestamp is stamped before the filer store write so
the entry and its event share a commit time, while the log-buffer append
happens after the write. A write delayed behind later-stamped mutations
keeps its early stamp in the payload while AddDataToBuffer bumps
LogEntry.TsNs to keep the stream monotone - so subscribers could
receive events whose payload TsNs arrives out of order.

Consumers treat the delivered TsNs as a stream position (cursors,
watermarks, replay windows). The s3 lifecycle daily-run drains events in
order and pins its per-shard cursor behind the first not-yet-due match;
an event delivered with a stale stamp after the cursor already passed it
is filtered out forever and its pending match silently never fires -
the flaky test_lifecyclev2_expiration signature of exactly one object of
a same-rule pair surviving, with errors=0 on every pass.

Normalize event.TsNs to logEntry.TsNs at delivery (and for the
skeleton-filtered heartbeat) so delivered timestamps are monotone by
construction. The commit-time stamp remains available on the entries
themselves via CommitTsNs.

* pb,filer: stamp metadata events with a mutation timestamp

TsNs on the wire now carries the delivery stream position, so the
mutation's commit stamp needs its own field. Add mutation_ts_ns to
SubscribeMetadataResponse and set it when the event is created.

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

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

* filer: carry the mutation stamp through event delivery

Delivery rewrites TsNs to the stream position; the payload stamp survives
in MutationTsNs, backfilled from TsNs for payloads written before the
field existed. The notification callback forwards it to the wire message.

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

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

* pb: normalize chunk-replay timestamps like eachLogEntryFn

ReadLogFileRefs bypasses the server decode path, so a delayed append
still arrived under its stale payload stamp and could hide behind a
subscriber cursor. Deliver the stream position and preserve the stamp in
MutationTsNs.

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

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

* filer: order replay and tombstones by the mutation stamp

Replay treated the delivered TsNs as the commit time, so a peer write
delayed in its own log could overwrite a newer local entry. Compare and
stamp with MutationTsNs, falling back to TsNs for older peers.

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

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

* remote.sync: carry the mutation stamp through reshaped renames

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

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

* filer: test replay ordering on the mutation stamp

* filer: tombstone remote deletes by delivery position, order by commit stamp

* filer: keep the peer commit stamp when restamping aggregated events

* pb: keep the log cursor on the entry position past callback restamps

---------

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

weed/command/filer_remote_sync_dir.go

@@ -193,6 +193,7 @@ func reshapeCrossMountRename(mountedDir string, resp *filer_pb.SubscribeMetadata
 			Directory:         message.NewParentPath,
 			EventNotification: &filer_pb.EventNotification{NewEntry: message.NewEntry, NewParentPath: message.NewParentPath, Signatures: message.Signatures},
 			TsNs:              resp.TsNs,
+			MutationTsNs:      resp.MutationTsNs,
 		}, true
 	case oldInMount:
 		if isRemoteOnly(message.OldEntry) {
@@ -204,6 +205,7 @@ func reshapeCrossMountRename(mountedDir string, resp *filer_pb.SubscribeMetadata
 			Directory:         resp.Directory,
 			EventNotification: &filer_pb.EventNotification{OldEntry: message.OldEntry, Signatures: message.Signatures},
 			TsNs:              resp.TsNs,
+			MutationTsNs:      resp.MutationTsNs,
 		}, true
 	default:
 		return nil, false

weed/filer/filer_notify.go

@@ -282,7 +282,8 @@ func (f *Filer) newMetadataEvent(oldEntry, newEntry *Entry, deleteChunks, isFrom
 			Signatures:         signatures,
 		},
 		// In flight until appended to the local log buffer (see metaLogInflight).
-		TsNs: tsNs,
+		TsNs:         tsNs,
+		MutationTsNs: tsNs,
 	}
 }
 
@@ -293,9 +294,14 @@ func (f *Filer) logMetaEvent(ctx context.Context, event *filer_pb.SubscribeMetad
 		return
 	}
 
-	if err := f.LocalMetaLogBuffer.AddDataToBuffer([]byte(event.Directory), data, event.TsNs); err != nil {
+	tsNs, err := f.LocalMetaLogBuffer.AddDataToBufferAndGetTs([]byte(event.Directory), data, event.TsNs)
+	if err != nil {
 		glog.Errorf("failed to add data to log buffer for %s: %v", event.Directory, err)
+		return
 	}
+	// The in-memory copy now carries the position the stream assigns it, the
+	// same value subscribers will see delivered.
+	event.TsNs = tsNs
 
 }
 

weed/filer/filer_remote_tombstone.go

@@ -36,17 +36,26 @@ const (
 // timestamps, which only the event itself knows, so a pending tombstone can
 // only be lifted by the event confirming it or by TTL. Once the event stamps
 // the real timestamp the tombstone is releasable by the offset.
+// remoteDeletionTombstone carries the delete's commit stamp — the basis for
+// ordering it against other mutations — and, once the event lands, the local
+// stream position it was delivered under, which the write-back offset must
+// pass before the remote delete counts as consumed.
+type remoteDeletionTombstone struct {
+	commitTsNs int64
+	streamTsNs int64
+}
+
 type remoteDeletionTombstones struct {
 	mu      sync.Mutex
-	files   map[string]int64 // file path -> delete event TsNs
-	dirs    map[string]int64 // deleted directory path -> event TsNs; covers its subtree
-	pending map[string]bool  // tombstone path recorded ahead of its event
+	files   map[string]remoteDeletionTombstone // file path -> delete event
+	dirs    map[string]remoteDeletionTombstone // deleted directory path -> event; covers its subtree
+	pending map[string]bool                   // tombstone path recorded ahead of its event
 }
 
 func newRemoteDeletionTombstones() *remoteDeletionTombstones {
 	return &remoteDeletionTombstones{
-		files:   make(map[string]int64),
-		dirs:    make(map[string]int64),
+		files:   make(map[string]remoteDeletionTombstone),
+		dirs:    make(map[string]remoteDeletionTombstone),
 		pending: make(map[string]bool),
 	}
 }
@@ -54,19 +63,24 @@ func newRemoteDeletionTombstones() *remoteDeletionTombstones {
 // add records a tombstone ahead of its metadata event — the timestamp is the
 // local delete time, a lower bound the sync offset cannot order against.
 func (t *remoteDeletionTombstones) add(path string, isDir bool, tsNs int64) {
-	t.upsert(path, isDir, tsNs, false)
+	t.upsert(path, isDir, tsNs, 0, false)
 }
 
-// addFromEvent records a tombstone stamped by the delete event itself, so the
-// mount's sync offset can release it once the daemon passes that event.
-func (t *remoteDeletionTombstones) addFromEvent(path string, isDir bool, tsNs int64) {
-	t.upsert(path, isDir, tsNs, true)
+// addFromEvent records a tombstone stamped by the delete event itself:
+// commitTsNs orders it against other mutations, streamTsNs is the delivery
+// position the mount's sync offset must pass to release it.
+func (t *remoteDeletionTombstones) addFromEvent(path string, isDir bool, commitTsNs, streamTsNs int64) {
+	t.upsert(path, isDir, commitTsNs, streamTsNs, true)
 }
 
-func (t *remoteDeletionTombstones) upsert(path string, isDir bool, tsNs int64, fromEvent bool) {
+func (t *remoteDeletionTombstones) upsert(path string, isDir bool, commitTsNs, streamTsNs int64, fromEvent bool) {
 	if t == nil {
 		return
 	}
+	if streamTsNs == 0 {
+		// No delivery position yet — the offset can never vouch for it.
+		streamTsNs = math.MaxInt64
+	}
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	// An ancestor directory tombstone at least as new already covers the
@@ -77,7 +91,7 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, tsNs int64, f
 			break
 		}
 		p = p[:i]
-		if ancestorTs, ok := t.dirs[p]; ok && ancestorTs >= tsNs {
+		if ancestor, ok := t.dirs[p]; ok && ancestor.commitTsNs >= commitTsNs {
 			return
 		}
 	}
@@ -86,16 +100,16 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, tsNs int64, f
 		m = t.dirs
 	}
 	if cur, ok := m[path]; ok {
-		if tsNs > cur {
-			m[path] = tsNs
-			if !fromEvent {
-				// a newer local delete restamps the tombstone ahead of its
-				// event — the offset cannot vouch for it until the event lands
-				t.pending[path] = true
-			}
+		if commitTsNs < cur.commitTsNs {
+			return
 		}
-		if fromEvent && m[path] == tsNs {
+		m[path] = remoteDeletionTombstone{commitTsNs: commitTsNs, streamTsNs: streamTsNs}
+		if fromEvent {
 			delete(t.pending, path)
+		} else {
+			// a newer local delete restamps the tombstone ahead of its
+			// event — the offset cannot vouch for it until the event lands
+			t.pending[path] = true
 		}
 		return
 	}
@@ -106,14 +120,12 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, tsNs int64, f
 			return
 		}
 	}
-	m[path] = tsNs
-	if fromEvent {
-		delete(t.pending, path)
-	} else {
+	m[path] = remoteDeletionTombstone{commitTsNs: commitTsNs, streamTsNs: streamTsNs}
+	if !fromEvent {
 		t.pending[path] = true
 	}
 	if isDir {
-		t.dropCoveredLocked(path, tsNs)
+		t.dropCoveredLocked(path, commitTsNs)
 	}
 }
 
@@ -123,14 +135,14 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, tsNs int64, f
 // Caller must hold t.mu.
 func (t *remoteDeletionTombstones) dropCoveredLocked(dirPath string, tsNs int64) {
 	prefix := dirPath + "/"
-	for p, ts := range t.files {
-		if ts <= tsNs && strings.HasPrefix(p, prefix) {
+	for p, cur := range t.files {
+		if cur.commitTsNs <= tsNs && strings.HasPrefix(p, prefix) {
 			delete(t.files, p)
 			delete(t.pending, p)
 		}
 	}
-	for p, ts := range t.dirs {
-		if ts <= tsNs && strings.HasPrefix(p, prefix) {
+	for p, cur := range t.dirs {
+		if cur.commitTsNs <= tsNs && strings.HasPrefix(p, prefix) {
 			delete(t.dirs, p)
 			delete(t.pending, p)
 		}
@@ -145,11 +157,11 @@ func (t *remoteDeletionTombstones) drop(path string, tsNs int64) {
 	}
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if cur, ok := t.dirs[path]; ok && cur <= tsNs {
+	if cur, ok := t.dirs[path]; ok && cur.commitTsNs <= tsNs {
 		delete(t.dirs, path)
 		delete(t.pending, path)
 	}
-	if cur, ok := t.files[path]; ok && cur <= tsNs {
+	if cur, ok := t.files[path]; ok && cur.commitTsNs <= tsNs {
 		delete(t.files, path)
 		delete(t.pending, path)
 	}
@@ -163,29 +175,27 @@ func (t *remoteDeletionTombstones) clear(path string, tsNs int64) {
 	}
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if cur, ok := t.files[path]; ok && tsNs >= cur {
+	if cur, ok := t.files[path]; ok && tsNs >= cur.commitTsNs {
 		delete(t.files, path)
 		delete(t.pending, path)
 	}
 }
 
-// blockedSince returns the newest delete timestamp governing path — its own
-// file tombstone or one from a deleted ancestor directory — and whether that
-// tombstone is still waiting for its event. 0 means clear.
-func (t *remoteDeletionTombstones) blockedSince(path string) (tsNs int64, pending bool) {
+// blockedSince returns the tombstone governing path — its own file tombstone
+// or the newest one from a deleted ancestor directory — and whether that
+// tombstone is still waiting for its event. commitTsNs == 0 means clear.
+func (t *remoteDeletionTombstones) blockedSince(path string) (streamTsNs, commitTsNs int64, pending bool) {
 	if t == nil {
-		return 0, false
+		return 0, 0, false
 	}
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if ts, ok := t.files[path]; ok {
-		tsNs = ts
-		pending = t.pending[path]
+	if cur, ok := t.files[path]; ok {
+		streamTsNs, commitTsNs, pending = cur.streamTsNs, cur.commitTsNs, t.pending[path]
 	}
 	for p := path; ; {
-		if ts, ok := t.dirs[p]; ok && ts > tsNs {
-			tsNs = ts
-			pending = t.pending[p]
+		if cur, ok := t.dirs[p]; ok && cur.commitTsNs > commitTsNs {
+			streamTsNs, commitTsNs, pending = cur.streamTsNs, cur.commitTsNs, t.pending[p]
 		}
 		i := strings.LastIndexByte(p, '/')
 		if i <= 0 {
@@ -193,23 +203,49 @@ func (t *remoteDeletionTombstones) blockedSince(path string) (tsNs int64, pendin
 		}
 		p = p[:i]
 	}
-	return tsNs, pending
+	return streamTsNs, commitTsNs, pending
 }
 
-// releaseThrough drops the tombstones governing path that are no newer than
-// tsNs, once their remote deletes are confirmed consumed.
+// releaseThrough drops the tombstones governing path whose remote delete the
+// write-back offset has passed, i.e. recorded at or below streamTsNs.
+// Pending tombstones carry no delivery position yet and are never dropped.
 func (t *remoteDeletionTombstones) releaseThrough(path string, tsNs int64) {
 	if t == nil {
 		return
 	}
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if cur, ok := t.files[path]; ok && cur <= tsNs {
+	if cur, ok := t.files[path]; ok && cur.streamTsNs <= tsNs {
 		delete(t.files, path)
 		delete(t.pending, path)
 	}
 	for p := path; ; {
-		if cur, ok := t.dirs[p]; ok && cur <= tsNs {
+		if cur, ok := t.dirs[p]; ok && cur.streamTsNs <= tsNs {
+			delete(t.dirs, p)
+			delete(t.pending, p)
+		}
+		i := strings.LastIndexByte(p, '/')
+		if i <= 0 {
+			break
+		}
+		p = p[:i]
+	}
+}
+
+// releaseExpired drops the tombstones governing path that outlived the TTL —
+// the only release a tombstone still waiting for its event can get.
+func (t *remoteDeletionTombstones) releaseExpired(path string, nowNs int64) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if cur, ok := t.files[path]; ok && nowNs-cur.commitTsNs >= int64(remoteDeletionTombstoneTTL) {
+		delete(t.files, path)
+		delete(t.pending, path)
+	}
+	for p := path; ; {
+		if cur, ok := t.dirs[p]; ok && nowNs-cur.commitTsNs >= int64(remoteDeletionTombstoneTTL) {
 			delete(t.dirs, p)
 			delete(t.pending, p)
 		}
@@ -222,14 +258,14 @@ func (t *remoteDeletionTombstones) releaseThrough(path string, tsNs int64) {
 }
 
 func (t *remoteDeletionTombstones) evictExpiredLocked(nowNs int64) {
-	for p, ts := range t.files {
-		if nowNs-ts >= int64(remoteDeletionTombstoneTTL) {
+	for p, cur := range t.files {
+		if nowNs-cur.commitTsNs >= int64(remoteDeletionTombstoneTTL) {
 			delete(t.files, p)
 			delete(t.pending, p)
 		}
 	}
-	for p, ts := range t.dirs {
-		if nowNs-ts >= int64(remoteDeletionTombstoneTTL) {
+	for p, cur := range t.dirs {
+		if nowNs-cur.commitTsNs >= int64(remoteDeletionTombstoneTTL) {
 			delete(t.dirs, p)
 			delete(t.pending, p)
 		}
@@ -264,21 +300,21 @@ func (f *Filer) isRemoteDeletionPending(ctx context.Context, p util.FullPath, mo
 	if f.remoteTombstones == nil {
 		return false
 	}
-	tsNs, pending := f.remoteTombstones.blockedSince(string(p))
-	if tsNs == 0 {
+	streamTsNs, commitTsNs, pending := f.remoteTombstones.blockedSince(string(p))
+	if commitTsNs == 0 {
 		return false
 	}
 	if pending {
 		// Recorded ahead of its delete event — the write-back offset cannot
 		// vouch for it yet; only the TTL lifts it.
-		if f.remoteDeletionExpired(tsNs) {
-			f.remoteTombstones.releaseThrough(string(p), tsNs)
+		if f.remoteDeletionExpired(commitTsNs) {
+			f.remoteTombstones.releaseExpired(string(p), time.Now().UnixNano())
 			return false
 		}
 		return true
 	}
-	if f.remoteDeletionConsumed(ctx, mountDir, tsNs) {
-		f.remoteTombstones.releaseThrough(string(p), tsNs)
+	if f.remoteDeletionConsumed(ctx, mountDir, streamTsNs) {
+		f.remoteTombstones.releaseThrough(string(p), streamTsNs)
 		return false
 	}
 	return true
@@ -352,6 +388,12 @@ func (f *Filer) rebuildRemoteDeletionTombstones(ctx context.Context, mounts []ut
 				if err := proto.Unmarshal(logEntry.Data, event); err != nil {
 					return false, nil
 				}
+				// Same normalization as delivery: the payload ts is the
+				// commit stamp, the wrapper ts the stream position.
+				if event.MutationTsNs == 0 {
+					event.MutationTsNs = event.TsNs
+				}
+				event.TsNs = logEntry.TsNs
 				f.onRemoteDeletionEvents(event)
 				return false, nil
 			})
@@ -407,17 +449,20 @@ func (f *Filer) onRemoteDeletionEvents(event *filer_pb.SubscribeMetadataResponse
 	if message == nil {
 		return
 	}
+	// The commit stamp orders the delete against other mutations; event.TsNs
+	// is the delivery position the write-back offset must pass.
+	commitTsNs := eventMutationTsNs(event)
 	if message.OldEntry != nil && !IsMetadataOnlyDelete(message.OldEntry) {
 		sourcePath := filer_pb.MetadataEventSourceFullPath(event)
 		if message.NewEntry == nil || sourcePath != filer_pb.MetadataEventTargetFullPath(event) {
 			if f.RemoteStorage != nil && f.remoteTombstones != nil {
 				if _, remoteLoc := f.RemoteStorage.FindMountDirectory(util.FullPath(sourcePath)); remoteLoc != nil {
-					f.remoteTombstones.addFromEvent(sourcePath, message.OldEntry.IsDirectory, event.TsNs)
+					f.remoteTombstones.addFromEvent(sourcePath, message.OldEntry.IsDirectory, commitTsNs, event.TsNs)
 				}
 			}
 		}
 	}
 	if message.NewEntry != nil {
-		f.remoteTombstones.clear(filer_pb.MetadataEventTargetFullPath(event), event.TsNs)
+		f.remoteTombstones.clear(filer_pb.MetadataEventTargetFullPath(event), commitTsNs)
 	}
 }

weed/filer/filer_remote_tombstone_test.go

@@ -41,34 +41,34 @@ func putRemoteSyncOffset(t *testing.T, store *stubFilerStore, dir string, offset
 func TestRemoteDeletionTombstones_BlocksAndReleases(t *testing.T) {
 	tombs := newRemoteDeletionTombstones()
 
-	tombs.addFromEvent("/m/a.txt", false, 100)
-	ts, _ := tombs.blockedSince("/m/a.txt")
+	tombs.addFromEvent("/m/a.txt", false, 100, 100)
+	_, ts, _ := tombs.blockedSince("/m/a.txt")
 	assert.Equal(t, int64(100), ts)
-	ts, _ = tombs.blockedSince("/m/b.txt")
+	_, ts, _ = tombs.blockedSince("/m/b.txt")
 	assert.Zero(t, ts)
 
 	// a delete of a directory blocks its whole subtree
-	tombs.addFromEvent("/m/dir", true, 200)
-	ts, _ = tombs.blockedSince("/m/dir")
+	tombs.addFromEvent("/m/dir", true, 200, 200)
+	_, ts, _ = tombs.blockedSince("/m/dir")
 	assert.Equal(t, int64(200), ts)
-	ts, _ = tombs.blockedSince("/m/dir/deep/x.txt")
+	_, ts, _ = tombs.blockedSince("/m/dir/deep/x.txt")
 	assert.Equal(t, int64(200), ts)
-	ts, _ = tombs.blockedSince("/m/dirx/y.txt")
+	_, ts, _ = tombs.blockedSince("/m/dirx/y.txt")
 	assert.Zero(t, ts)
 
 	// a rewrite at or after the delete lifts only that file's tombstone
 	tombs.clear("/m/a.txt", 100)
-	ts, _ = tombs.blockedSince("/m/a.txt")
+	_, ts, _ = tombs.blockedSince("/m/a.txt")
 	assert.Zero(t, ts)
-	tombs.addFromEvent("/m/a.txt", false, 300)
+	tombs.addFromEvent("/m/a.txt", false, 300, 300)
 	tombs.clear("/m/a.txt", 250)
-	ts, _ = tombs.blockedSince("/m/a.txt")
+	_, ts, _ = tombs.blockedSince("/m/a.txt")
 	assert.Equal(t, int64(300), ts, "older create must not lift newer delete")
 
 	tombs.releaseThrough("/m/dir/deep/x.txt", 200)
-	ts, _ = tombs.blockedSince("/m/dir/deep/x.txt")
+	_, ts, _ = tombs.blockedSince("/m/dir/deep/x.txt")
 	assert.Zero(t, ts)
-	ts, _ = tombs.blockedSince("/m/dir")
+	_, ts, _ = tombs.blockedSince("/m/dir")
 	assert.Zero(t, ts)
 }
 
@@ -77,22 +77,22 @@ func TestRemoteDeletionTombstones_AncestorSubsumesAndCovers(t *testing.T) {
 
 	// a child tombstone recorded before its ancestor is dropped once the
 	// ancestor's newer delete covers the whole subtree
-	tombs.addFromEvent("/m/dir/a.txt", false, 100)
-	tombs.addFromEvent("/m/dir", true, 200)
-	ts, _ := tombs.blockedSince("/m/dir")
+	tombs.addFromEvent("/m/dir/a.txt", false, 100, 100)
+	tombs.addFromEvent("/m/dir", true, 200, 200)
+	_, ts, _ := tombs.blockedSince("/m/dir")
 	assert.Equal(t, int64(200), ts)
-	ts, _ = tombs.blockedSince("/m/dir/a.txt")
+	_, ts, _ = tombs.blockedSince("/m/dir/a.txt")
 	assert.Equal(t, int64(200), ts)
 	_, exists := tombs.files["/m/dir/a.txt"]
 	assert.False(t, exists, "descendant tombstone is subsumed by the ancestor")
 
 	// adds under the covered subtree are skipped while the ancestor stands
-	tombs.addFromEvent("/m/dir/b.txt", false, 150)
+	tombs.addFromEvent("/m/dir/b.txt", false, 150, 150)
 	_, exists = tombs.files["/m/dir/b.txt"]
 	assert.False(t, exists)
 	// ...but a child deleted after the ancestor still records its own tombstone
-	tombs.addFromEvent("/m/dir/c.txt", false, 300)
-	ts, _ = tombs.blockedSince("/m/dir/c.txt")
+	tombs.addFromEvent("/m/dir/c.txt", false, 300, 300)
+	_, ts, _ = tombs.blockedSince("/m/dir/c.txt")
 	assert.Equal(t, int64(300), ts)
 }
 
@@ -119,22 +119,49 @@ func TestRemoteDeletionTombstones_PendingIgnoresOffset(t *testing.T) {
 	assert.False(t, f.isRemoteDeletionPending(context.Background(), util.FullPath(filePath), "/buckets/mybucket"))
 }
 
+// A delayed delete is recorded at its commit stamp but delivered under a
+// bumped stream position. The write-back offset moves on stream positions,
+// so the tombstone must hold until the offset passes the delivery position,
+// not the commit stamp — otherwise a lazy read resurrects the remote object.
+func TestRemoteDeletionTombstones_OffsetMustPassDeliveryPosition(t *testing.T) {
+	f, store := newMountedTestFiler(t, "stub_tomb_stream_pos", nil, 0)
+
+	base := time.Now().UnixNano()
+	f.onMetadataChangeEvent(&filer_pb.SubscribeMetadataResponse{
+		Directory:    "/buckets/mybucket/dir",
+		TsNs:         base + 2, // delivered stream position
+		MutationTsNs: base,     // commit stamp before a delayed append
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry: &filer_pb.Entry{Name: "a.txt"},
+		},
+	})
+	filePath := util.FullPath("/buckets/mybucket/dir/a.txt")
+
+	putRemoteSyncOffset(t, store, "/buckets/mybucket", base+1)
+	assert.True(t, f.isRemoteDeletionPending(context.Background(), filePath, "/buckets/mybucket"),
+		"offset past the commit stamp must not release a delete delivered later")
+
+	putRemoteSyncOffset(t, store, "/buckets/mybucket", base+2)
+	assert.False(t, f.isRemoteDeletionPending(context.Background(), filePath, "/buckets/mybucket"),
+		"offset at the delivery position releases the tombstone")
+}
+
 func TestRemoteDeletionTombstones_RaisedLocalTombstoneIsPendingAgain(t *testing.T) {
 	tombs := newRemoteDeletionTombstones()
 
-	tombs.addFromEvent("/m/a.txt", false, 100)
-	_, pending := tombs.blockedSince("/m/a.txt")
+	tombs.addFromEvent("/m/a.txt", false, 100, 100)
+	_, _, pending := tombs.blockedSince("/m/a.txt")
 	assert.False(t, pending)
 
 	// a newer local delete restamps the tombstone before its own event
 	// lands, so the write-back offset cannot vouch for it yet
 	tombs.add("/m/a.txt", false, 200)
-	_, pending = tombs.blockedSince("/m/a.txt")
+	_, _, pending = tombs.blockedSince("/m/a.txt")
 	assert.True(t, pending)
 
 	// once that delete's event confirms the new timestamp it releases normally
-	tombs.addFromEvent("/m/a.txt", false, 200)
-	_, pending = tombs.blockedSince("/m/a.txt")
+	tombs.addFromEvent("/m/a.txt", false, 200, 200)
+	_, _, pending = tombs.blockedSince("/m/a.txt")
 	assert.False(t, pending)
 }
 
@@ -436,7 +463,7 @@ func TestRebuildRemoteDeletionTombstones_EmptyLogReleasesGate(t *testing.T) {
 		t.Fatal("rebuild never released the gate")
 	}
 	assert.Nil(t, f.remoteTombstonesDone.Load())
-	ts, _ := f.remoteTombstones.blockedSince("/buckets/mybucket/a.txt")
+	_, ts, _ := f.remoteTombstones.blockedSince("/buckets/mybucket/a.txt")
 	assert.Zero(t, ts)
 }
 

weed/filer/meta_aggregator.go

@@ -441,7 +441,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 
 	*/
 
-	var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse) error
+	var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse, int64) error
 	lastPersistTime := time.Now()
 	lastTsNs := startFrom
 	// replayFloorTsNs is the earliest event this session gave up replaying.
@@ -513,7 +513,7 @@ 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) error {
+		maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse, peerTsNs int64) error {
 			if err := replicateMetadataChange(f, peer, event, stopChan); err != nil {
 				if errors.Is(err, errPeerSubscriptionStopped) {
 					return err
@@ -522,16 +522,16 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 				// 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
+				if replayFloorTsNs == 0 || peerTsNs < replayFloorTsNs {
+					replayFloorTsNs = peerTsNs
 				}
 				return nil
 			}
 			counter++
 			if lastPersistTime.Add(time.Minute).Before(time.Now()) {
-				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)
+				if err := ma.updateOffset(f, peer, peerSignature, persistableTsNs(peerTsNs)); err == nil {
+					if peerTsNs < time.Now().Add(-2*time.Minute).UnixNano() {
+						glog.V(0).Infof("sync with %s progressed to: %v %0.2f/sec", peer, time.Unix(0, peerTsNs), float64(counter)/60.0)
 					} else if !synced {
 						synced = true
 						glog.V(0).Infof("synced with %s", peer)
@@ -554,12 +554,22 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 		}
 		dir := event.Directory
 		// println("received meta change", dir, "size", len(data))
-		if err := ma.MetaLogBuffer.AddDataToBuffer([]byte(dir), data, event.TsNs); err != nil {
+		peerTsNs := event.TsNs
+		if event.MutationTsNs == 0 {
+			// Older peers predate the field; their TsNs was the commit stamp,
+			// so Replay must keep ordering by it after TsNs is restamped.
+			event.MutationTsNs = peerTsNs
+		}
+		tsNs, err := ma.MetaLogBuffer.AddDataToBufferAndGetTs([]byte(dir), data, event.TsNs)
+		if err != nil {
 			glog.Errorf("failed to add data to log buffer for %s: %v", dir, err)
 			return err
 		}
+		// The event now sits at tsNs in the local stream; offset tracking
+		// still needs the peer's own position, so it is passed separately.
+		event.TsNs = tsNs
 		if maybeReplicateMetadataChange != nil {
-			if err := maybeReplicateMetadataChange(event); err != nil {
+			if err := maybeReplicateMetadataChange(event, peerTsNs); err != nil {
 				return err
 			}
 		}
@@ -606,13 +616,14 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 				return errPeerSubscriptionStopped
 			default:
 			}
+			peerTsNs := event.TsNs
 			if err := processEventFn(event); err != nil {
 				glog.V(0).Infof("SubscribeLocalMetadata process %s -> %s ts_ns=%d: %v", filer_pb.MetadataEventSourceFullPath(event), filer_pb.MetadataEventTargetFullPath(event), event.TsNs, err)
 				return fmt.Errorf("process %s -> %s ts_ns=%d: %w", filer_pb.MetadataEventSourceFullPath(event), filer_pb.MetadataEventTargetFullPath(event), event.TsNs, err)
 			}
 			f.onMetadataChangeEvent(event)
-			lastTsNs = event.TsNs
-			ma.advancePeerWatermark(peer, event.TsNs)
+			lastTsNs = peerTsNs
+			ma.advancePeerWatermark(peer, peerTsNs)
 			return nil
 		}
 

weed/filer/meta_aggregator_replay_test.go

@@ -195,6 +195,35 @@ func TestReplayAppliesEventNewerThanLocalEntry(t *testing.T) {
 	}
 }
 
+// A delayed peer write is delivered under its bumped stream position, which
+// sorts after a newer local commit; Replay must order on MutationTsNs so the
+// older mutation cannot overwrite the newer local entry.
+func TestReplayOrdersByMutationStampNotStreamPosition(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"),
+		CommitTsNs: now.UnixNano(),
+	}
+
+	event := overwriteEvent("/dir", "file", now.Add(time.Minute).UnixNano(), "stale peer write")
+	event.MutationTsNs = now.Add(-time.Minute).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) != "newer local write" {
+		t.Fatalf("content = %q, want local write kept", entry.Content)
+	}
+}
+
 func TestReplaySkipsDeleteOlderThanLocalEntry(t *testing.T) {
 	store := newStubFilerStore()
 	f := newReplayTestFiler(store)
@@ -917,8 +946,8 @@ func TestReplayDirectoryDeleteSweepsRemoteObjects(t *testing.T) {
 	if len(stub.deleteCalls) != 1 || stub.deleteCalls[0].Path != "/d/f.txt" {
 		t.Fatalf("expected one remote delete of /d/f.txt, got %v", stub.deleteCalls)
 	}
-	if got := f.remoteTombstones.dirs["/buckets/mybucket/d"]; got != event.TsNs {
-		t.Fatalf("dir tombstone stamped %d, want the event ts %d", got, event.TsNs)
+	if got := f.remoteTombstones.dirs["/buckets/mybucket/d"]; got.commitTsNs != event.TsNs {
+		t.Fatalf("dir tombstone stamped %+v, want the event ts %d", got, event.TsNs)
 	}
 }
 
@@ -1353,7 +1382,7 @@ func TestReplaySweepKeepsRemoteOnlyOwnerOfPath(t *testing.T) {
 	}
 	sweep := &folderSweep{chunksByPath: make(map[util.FullPath][]*filer_pb.FileChunk)}
 	message := &filer_pb.EventNotification{OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}}
-	kept, err := f.deleteSweptChild(ctx, stale, message, 0, sweep, false, true)
+	kept, err := f.deleteSweptChild(ctx, stale, message, 0, 0, sweep, false, true)
 	if err != nil {
 		t.Fatalf("deleteSweptChild: %v", err)
 	}

weed/filer/meta_replay.go

@@ -47,18 +47,19 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
 		newPath = util.NewFullPath(dir, message.NewEntry.Name)
 	}
 	ctx := context.Background()
+	mutationTsNs := eventMutationTsNs(resp)
 
 	unlock := f.lockReplayPaths(oldPath, newPath)
 	defer unlock()
 
 	samePath := oldPath != "" && oldPath == newPath
 	if samePath {
-		newer, err := f.replayBlockedByLocal(ctx, oldPath, resp.TsNs)
+		newer, err := f.replayBlockedByLocal(ctx, oldPath, mutationTsNs)
 		if err != nil || newer {
 			return err
 		}
 		newEntry := FromPbEntry(dir, message.NewEntry)
-		newEntry.CommitTsNs = resp.TsNs
+		newEntry.CommitTsNs = mutationTsNs
 		if message.OldEntry.IsDirectory {
 			glog.V(4).Infof("updating %v", newPath)
 			return f.updateExistingEntry(ctx, oldPath, newEntry)
@@ -75,7 +76,7 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
 	}
 
 	if message.OldEntry != nil {
-		newer, err := f.replayBlockedByLocal(ctx, oldPath, resp.TsNs)
+		newer, err := f.replayBlockedByLocal(ctx, oldPath, mutationTsNs)
 		if err != nil {
 			return err
 		}
@@ -86,7 +87,7 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
 			keepDir := false
 			if message.OldEntry.IsDirectory {
 				var err error
-				keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message, resp.TsNs)
+				keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message, mutationTsNs, resp.TsNs)
 				if err != nil {
 					return err
 				}
@@ -101,14 +102,14 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
 	}
 
 	if message.NewEntry != nil {
-		newer, err := f.replayBlockedByLocal(ctx, newPath, resp.TsNs)
+		newer, err := f.replayBlockedByLocal(ctx, newPath, mutationTsNs)
 		if err != nil {
 			return err
 		}
 		if !newer {
 			glog.V(4).Infof("creating %v", newPath)
 			newEntry := FromPbEntry(dir, message.NewEntry)
-			newEntry.CommitTsNs = resp.TsNs
+			newEntry.CommitTsNs = mutationTsNs
 			if err := f.Store.InsertEntry(ctx, newEntry); err != nil {
 				return err
 			}
@@ -118,6 +119,16 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
 	return nil
 }
 
+// eventMutationTsNs returns the mutation's commit stamp: TsNs carries the
+// delivery stream position while MutationTsNs was stamped before the store
+// write. Events from older peers lack the field and fall back to TsNs.
+func eventMutationTsNs(event *filer_pb.SubscribeMetadataResponse) int64 {
+	if event.MutationTsNs != 0 {
+		return event.MutationTsNs
+	}
+	return event.TsNs
+}
+
 // updateExistingEntry applies a same-path update without deleting the row in
 // between: the path never sits absent, so a failed retry cannot leave it
 // unordered — a tombstone would instead read as a subtree delete that
@@ -174,9 +185,9 @@ func (f *Filer) updateExistingEntry(ctx context.Context, path util.FullPath, new
 // the chunks of removed files for deletion. Children this filer committed
 // after the event stay — a newer local write wins over a delayed delete —
 // along with the directories that hold them.
-func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, tsNs int64) (bool, error) {
+func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, commitTsNs, streamTsNs int64) (bool, error) {
 	sweep := &folderSweep{chunksByPath: make(map[util.FullPath][]*filer_pb.FileChunk)}
-	kept, err := f.sweepReplayedFolder(ctx, dir, message, tsNs, sweep)
+	kept, err := f.sweepReplayedFolder(ctx, dir, message, commitTsNs, streamTsNs, sweep)
 	// Flush what was collected even on error: the entries already deleted
 	// cannot be listed again on retry, so their chunks must be queued now.
 	sweep.flush(ctx, f)
@@ -213,7 +224,7 @@ func (s *folderSweep) flush(ctx context.Context, f *Filer) {
 // newer than the event, working subtree-first so nested entries go too.
 // It reports whether a newer descendant keeps dir alive — the caller then
 // keeps the dir entry itself.
-func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, tsNs int64, sweep *folderSweep) (kept bool, err error) {
+func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, commitTsNs, streamTsNs int64, sweep *folderSweep) (kept bool, err error) {
 	keepRemote := IsMetadataOnlyDelete(message.OldEntry)
 	deleteRemote := !keepRemote && !message.GetIsFromOtherCluster()
 	lastFileName := ""
@@ -225,15 +236,15 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, mess
 		for _, sub := range entries {
 			lastFileName = sub.Name()
 			childKept := false
-			if tsNs > 0 {
+			if commitTsNs > 0 {
 				localTsNs, tsErr := f.localCommitTsNs(ctx, sub.FullPath)
 				if tsErr != nil {
 					return false, tsErr
 				}
-				childKept = localTsNs > tsNs
+				childKept = localTsNs > commitTsNs
 			}
 			if !childKept && sub.IsDirectory() {
-				childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, message, tsNs, sweep)
+				childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, message, commitTsNs, streamTsNs, sweep)
 				if err != nil {
 					return false, err
 				}
@@ -242,7 +253,7 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, mess
 				kept = true
 				continue
 			}
-			childKept, err = f.deleteSweptChild(ctx, sub, message, tsNs, sweep, keepRemote, deleteRemote)
+			childKept, err = f.deleteSweptChild(ctx, sub, message, commitTsNs, streamTsNs, sweep, keepRemote, deleteRemote)
 			if err != nil {
 				return false, err
 			}
@@ -265,7 +276,7 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, mess
 // lazy fetch materialized from the remote — reads the object at the key as
 // its only copy: the write-back daemon's processDeleteEvent keeps such an
 // object, and the sweep does too.
-func (f *Filer) deleteSweptChild(ctx context.Context, sub *Entry, message *filer_pb.EventNotification, tsNs int64, sweep *folderSweep, keepRemote, deleteRemote bool) (bool, error) {
+func (f *Filer) deleteSweptChild(ctx context.Context, sub *Entry, message *filer_pb.EventNotification, commitTsNs, streamTsNs int64, sweep *folderSweep, keepRemote, deleteRemote bool) (bool, error) {
 	var childLock *util.ActiveLock
 	if lt := f.EntryLockTable; lt != nil {
 		// The sweep already holds locks on the event's paths; waiting for a
@@ -279,12 +290,12 @@ func (f *Filer) deleteSweptChild(ctx context.Context, sub *Entry, message *filer
 		}
 		defer lt.ReleaseLock(sub.FullPath, childLock)
 	}
-	if tsNs > 0 {
+	if commitTsNs > 0 {
 		localTsNs, tsErr := f.localCommitTsNs(ctx, sub.FullPath)
 		if tsErr != nil {
 			return false, tsErr
 		}
-		if localTsNs > tsNs {
+		if localTsNs > commitTsNs {
 			return true, nil
 		}
 	}
@@ -299,7 +310,7 @@ func (f *Filer) deleteSweptChild(ctx context.Context, sub *Entry, message *filer
 	}
 	if !keepRemote && f.RemoteStorage != nil && f.remoteTombstones != nil {
 		if _, remoteLoc := f.RemoteStorage.FindMountDirectory(sub.FullPath); remoteLoc != nil {
-			f.remoteTombstones.addFromEvent(string(sub.FullPath), sub.IsDirectory(), tsNs)
+			f.remoteTombstones.addFromEvent(string(sub.FullPath), sub.IsDirectory(), commitTsNs, streamTsNs)
 		}
 	}
 	if deleteRemote && !sub.IsDirectory() {

weed/pb/filer.proto

@@ -692,6 +692,7 @@ message SubscribeMetadataResponse {
     repeated SubscribeMetadataResponse events = 4; // batch of additional events (backlog catch-up)
     repeated LogFileChunkRef log_file_refs = 5;    // log file chunk refs for client direct-read
     int64 flushed_ts_ns = 6; // local log-buffer flush watermark: everything at or below it is on disk
+    int64 mutation_ts_ns = 7; // metadata mutation commit stamp; ts_ns carries the delivery stream position instead
 }
 message ListMetadataSubscribersRequest {
     repeated string client_types = 1; // optional filter by client type, e.g. "mount"; empty = all

weed/pb/filer_pb/filer.pb.go

@@ -4440,9 +4440,10 @@ type SubscribeMetadataResponse struct {
 	Directory         string                       `protobuf:"bytes,1,opt,name=directory,proto3" json:"directory,omitempty"`
 	EventNotification *EventNotification           `protobuf:"bytes,2,opt,name=event_notification,json=eventNotification,proto3" json:"event_notification,omitempty"`
 	TsNs              int64                        `protobuf:"varint,3,opt,name=ts_ns,json=tsNs,proto3" json:"ts_ns,omitempty"`
-	Events            []*SubscribeMetadataResponse `protobuf:"bytes,4,rep,name=events,proto3" json:"events,omitempty"`                                 // batch of additional events (backlog catch-up)
-	LogFileRefs       []*LogFileChunkRef           `protobuf:"bytes,5,rep,name=log_file_refs,json=logFileRefs,proto3" json:"log_file_refs,omitempty"`  // log file chunk refs for client direct-read
-	FlushedTsNs       int64                        `protobuf:"varint,6,opt,name=flushed_ts_ns,json=flushedTsNs,proto3" json:"flushed_ts_ns,omitempty"` // local log-buffer flush watermark: everything at or below it is on disk
+	Events            []*SubscribeMetadataResponse `protobuf:"bytes,4,rep,name=events,proto3" json:"events,omitempty"`                                    // batch of additional events (backlog catch-up)
+	LogFileRefs       []*LogFileChunkRef           `protobuf:"bytes,5,rep,name=log_file_refs,json=logFileRefs,proto3" json:"log_file_refs,omitempty"`     // log file chunk refs for client direct-read
+	FlushedTsNs       int64                        `protobuf:"varint,6,opt,name=flushed_ts_ns,json=flushedTsNs,proto3" json:"flushed_ts_ns,omitempty"`    // local log-buffer flush watermark: everything at or below it is on disk
+	MutationTsNs      int64                        `protobuf:"varint,7,opt,name=mutation_ts_ns,json=mutationTsNs,proto3" json:"mutation_ts_ns,omitempty"` // metadata mutation commit stamp; ts_ns carries the delivery stream position instead
 	unknownFields     protoimpl.UnknownFields
 	sizeCache         protoimpl.SizeCache
 }
@@ -4519,6 +4520,13 @@ func (x *SubscribeMetadataResponse) GetFlushedTsNs() int64 {
 	return 0
 }
 
+func (x *SubscribeMetadataResponse) GetMutationTsNs() int64 {
+	if x != nil {
+		return x.MutationTsNs
+	}
+	return 0
+}
+
 type ListMetadataSubscribersRequest struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	ClientTypes   []string               `protobuf:"bytes,1,rep,name=client_types,json=clientTypes,proto3" json:"client_types,omitempty"` // optional filter by client type, e.g. "mount"; empty = all
@@ -7723,14 +7731,15 @@ const file_filer_proto_rawDesc = "" +
 	" \x03(\tR\vdirectories\x128\n" +
 	"\x18client_supports_batching\x18\v \x01(\bR\x16clientSupportsBatching\x12E\n" +
 	"\x1fclient_supports_metadata_chunks\x18\f \x01(\bR\x1cclientSupportsMetadataChunks\x12C\n" +
-	"\x1eclient_supports_idle_heartbeat\x18\r \x01(\bR\x1bclientSupportsIdleHeartbeat\"\xba\x02\n" +
+	"\x1eclient_supports_idle_heartbeat\x18\r \x01(\bR\x1bclientSupportsIdleHeartbeat\"\xe0\x02\n" +
 	"\x19SubscribeMetadataResponse\x12\x1c\n" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12J\n" +
 	"\x12event_notification\x18\x02 \x01(\v2\x1b.filer_pb.EventNotificationR\x11eventNotification\x12\x13\n" +
 	"\x05ts_ns\x18\x03 \x01(\x03R\x04tsNs\x12;\n" +
 	"\x06events\x18\x04 \x03(\v2#.filer_pb.SubscribeMetadataResponseR\x06events\x12=\n" +
 	"\rlog_file_refs\x18\x05 \x03(\v2\x19.filer_pb.LogFileChunkRefR\vlogFileRefs\x12\"\n" +
-	"\rflushed_ts_ns\x18\x06 \x01(\x03R\vflushedTsNs\"C\n" +
+	"\rflushed_ts_ns\x18\x06 \x01(\x03R\vflushedTsNs\x12$\n" +
+	"\x0emutation_ts_ns\x18\a \x01(\x03R\fmutationTsNs\"C\n" +
 	"\x1eListMetadataSubscribersRequest\x12!\n" +
 	"\fclient_types\x18\x01 \x03(\tR\vclientTypes\"a\n" +
 	"\x1fListMetadataSubscribersResponse\x12>\n" +

weed/pb/filer_pb/filer_vtproto.pb.go

@@ -4053,6 +4053,11 @@ func (m *SubscribeMetadataResponse) MarshalToSizedBufferVT(dAtA []byte) (int, er
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.MutationTsNs != 0 {
+		i = protohelpers.EncodeVarint(dAtA, i, uint64(m.MutationTsNs))
+		i--
+		dAtA[i] = 0x38
+	}
 	if m.FlushedTsNs != 0 {
 		i = protohelpers.EncodeVarint(dAtA, i, uint64(m.FlushedTsNs))
 		i--
@@ -8191,6 +8196,9 @@ func (m *SubscribeMetadataResponse) SizeVT() (n int) {
 	if m.FlushedTsNs != 0 {
 		n += 1 + protohelpers.SizeOfVarint(uint64(m.FlushedTsNs))
 	}
+	if m.MutationTsNs != 0 {
+		n += 1 + protohelpers.SizeOfVarint(uint64(m.MutationTsNs))
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -20240,6 +20248,25 @@ func (m *SubscribeMetadataResponse) UnmarshalVT(dAtA []byte) error {
 					break
 				}
 			}
+		case 7:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field MutationTsNs", wireType)
+			}
+			m.MutationTsNs = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.MutationTsNs |= int64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])

weed/pb/filer_pb_direct_read.go

@@ -274,13 +274,24 @@ func processOneLogEntry(logEntry *filer_pb.LogEntry, filter PathFilter, processE
 		glog.Errorf("unmarshal log entry: %v", err)
 		return 0, nil // skip corrupt entries
 	}
+	// Same normalization the server applies in eachLogEntryFn: the payload
+	// stamp predates the log append, so deliver the stream position and keep
+	// the stamp in MutationTsNs (payloads written before that field existed
+	// still carry it in event.TsNs).
+	if event.MutationTsNs == 0 {
+		event.MutationTsNs = event.TsNs
+	}
+	event.TsNs = logEntry.TsNs
 	if !matchesFilter(event, filter) {
-		return event.TsNs, nil
+		return logEntry.TsNs, nil
 	}
+	// Return the entry's stream position, not event.TsNs: the callback may
+	// restamp the event with its own delivery position, and callers resume
+	// from the returned offset.
 	if err := processEventFn(event); err != nil {
-		return event.TsNs, fmt.Errorf("process event: %w", err)
+		return logEntry.TsNs, fmt.Errorf("process event: %w", err)
 	}
-	return event.TsNs, nil
+	return logEntry.TsNs, nil
 }
 
 // --- path filtering (mirrors server-side eachEventNotificationFn logic) ---

weed/pb/filer_pb_direct_read_test.go

@@ -589,3 +589,67 @@ func TestProcessOneLogEntrySkipsInvalidUTF8(t *testing.T) {
 		t.Fatalf("tsNs = %d, want 0 for a skipped entry", tsNs)
 	}
 }
+
+// TestProcessOneLogEntryDeliversStreamPosition verifies the same
+// stamp-vs-position normalization the server applies in eachLogEntryFn: the
+// payload TsNs predates the log append, so the entry is delivered under its
+// stream position while the mutation stamp moves to MutationTsNs.
+func TestProcessOneLogEntryDeliversStreamPosition(t *testing.T) {
+	data, err := proto.Marshal(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/b",
+		TsNs:      99,
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "foo"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got *filer_pb.SubscribeMetadataResponse
+	lastTsNs, err := processOneLogEntry(&filer_pb.LogEntry{TsNs: 101, Data: data},
+		PathFilter{PathPrefix: "/"},
+		func(resp *filer_pb.SubscribeMetadataResponse) error {
+			got = resp
+			return nil
+		})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if lastTsNs != 101 {
+		t.Fatalf("cursor = %d, want stream position 101", lastTsNs)
+	}
+	if got.TsNs != 101 {
+		t.Fatalf("delivered ts = %d, want stream position 101", got.TsNs)
+	}
+	if got.MutationTsNs != 99 {
+		t.Fatalf("mutation ts = %d, want payload stamp 99", got.MutationTsNs)
+	}
+}
+
+// A callback may restamp the delivered event (the meta aggregator moves TsNs
+// to the local buffer position). The returned cursor must stay the log
+// entry's own position, or a resume offset silently skips peer events.
+func TestProcessOneLogEntryCursorIgnoresCallbackRestamp(t *testing.T) {
+	data, err := proto.Marshal(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/b",
+		TsNs:      99,
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "foo"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	lastTsNs, err := processOneLogEntry(&filer_pb.LogEntry{TsNs: 101, Data: data},
+		PathFilter{PathPrefix: "/"},
+		func(resp *filer_pb.SubscribeMetadataResponse) error {
+			resp.TsNs = 1 << 40
+			return nil
+		})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if lastTsNs != 101 {
+		t.Fatalf("cursor = %d, want stream position 101", lastTsNs)
+	}
+}

weed/server/filer_grpc_server_sub_meta.go

@@ -1305,7 +1305,7 @@ func (fs *FilerServer) subscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 
 }
 
-func eachLogEntryFn(ctx context.Context, req *filer_pb.SubscribeMetadataRequest, sender metadataStreamSender, eachEventNotificationFn func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error, filtered *int64) log_buffer.EachLogEntryFuncType {
+func eachLogEntryFn(ctx context.Context, req *filer_pb.SubscribeMetadataRequest, sender metadataStreamSender, eachEventNotificationFn func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64, mutationTsNs int64) error, filtered *int64) log_buffer.EachLogEntryFuncType {
 	// A shallow scan of the path fields skips unmarshaling chunk-heavy events
 	// this subscriber would filter out anyway; scan surprises fall back to the
 	// full decode. Only a delivery resets the shared unsynced-events counter.
@@ -1323,7 +1323,7 @@ func eachLogEntryFn(ctx context.Context, req *filer_pb.SubscribeMetadataRequest,
 				if *filtered > MaxUnsyncedEvents {
 					if err := sender.Send(&filer_pb.SubscribeMetadataResponse{
 						EventNotification: &filer_pb.EventNotification{},
-						TsNs:              skeleton.TsNs,
+						TsNs:              logEntry.TsNs,
 					}); err != nil {
 						return false, err
 					}
@@ -1341,7 +1341,21 @@ func eachLogEntryFn(ctx context.Context, req *filer_pb.SubscribeMetadataRequest,
 			return false, fmt.Errorf("unexpected unmarshal filer_pb.SubscribeMetadataResponse: %w", err)
 		}
 
-		if err := eachEventNotificationFn(event.Directory, event.EventNotification, event.TsNs); err != nil {
+		// Deliver the stream position (logEntry.TsNs), not the payload stamp:
+		// event.TsNs was stamped before the store write while the append lands
+		// after it, so an event delayed behind later-stamped writes keeps an
+		// early stamp in Data while the buffer bumps logEntry.TsNs to stay
+		// monotone. Subscribers assume delivered timestamps are ordered; an
+		// early-stamped event delivered under its stale stamp can sit behind a
+		// cursor that already advanced past it and never be seen again.
+		// The mutation stamp survives in MutationTsNs; payloads written before
+		// that field existed still carry it in event.TsNs.
+		mutationTsNs := event.MutationTsNs
+		if mutationTsNs == 0 {
+			mutationTsNs = event.TsNs
+		}
+		event.TsNs = logEntry.TsNs
+		if err := eachEventNotificationFn(event.Directory, event.EventNotification, event.TsNs, mutationTsNs); err != nil {
 			return false, err
 		}
 
@@ -1614,8 +1628,8 @@ func deltaLogFileRefs(refs []*filer_pb.LogFileChunkRef, sent map[string]sentRefS
 	return out
 }
 
-func (fs *FilerServer) eachEventNotificationFn(req *filer_pb.SubscribeMetadataRequest, sender metadataStreamSender, clientName string, filtered *int64) func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
-	return func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
+func (fs *FilerServer) eachEventNotificationFn(req *filer_pb.SubscribeMetadataRequest, sender metadataStreamSender, clientName string, filtered *int64) func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64, mutationTsNs int64) error {
+	return func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64, mutationTsNs int64) error {
 		defer func() {
 			if *filtered > MaxUnsyncedEvents {
 				if err := sender.Send(&filer_pb.SubscribeMetadataResponse{
@@ -1660,6 +1674,7 @@ func (fs *FilerServer) eachEventNotificationFn(req *filer_pb.SubscribeMetadataRe
 			Directory:         dirPath,
 			EventNotification: eventNotification,
 			TsNs:              tsNs,
+			MutationTsNs:      mutationTsNs,
 		}
 
 		if !filer_pb.MetadataEventMatchesSubscription(message, req.PathPrefix, req.PathPrefixes, req.Directories) {

weed/server/filer_grpc_server_sub_meta_prefilter_test.go

@@ -38,7 +38,7 @@ func TestEachLogEntryFnPrefilterSkipsDecode(t *testing.T) {
 	sender := &recordingSender{}
 	var decoded int
 	var unsyncedEvents int64
-	fn := eachLogEntryFn(context.Background(), req, sender, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
+	fn := eachLogEntryFn(context.Background(), req, sender, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64, mutationTsNs int64) error {
 		decoded++
 		unsyncedEvents = 0 // emulate a delivery, like the notification fn after a send
 		return nil
@@ -84,7 +84,7 @@ func TestEachLogEntryFnNoFilterDecodesEverything(t *testing.T) {
 	req := &filer_pb.SubscribeMetadataRequest{}
 	var decoded int
 	var unsyncedEvents int64
-	fn := eachLogEntryFn(context.Background(), req, &recordingSender{}, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
+	fn := eachLogEntryFn(context.Background(), req, &recordingSender{}, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64, mutationTsNs int64) error {
 		decoded++
 		return nil
 	}, &unsyncedEvents)
@@ -95,3 +95,66 @@ func TestEachLogEntryFnNoFilterDecodesEverything(t *testing.T) {
 		t.Fatalf("unfiltered subscriber must decode every entry, got %d", decoded)
 	}
 }
+
+func TestEachLogEntryFnDeliversStreamPositionTs(t *testing.T) {
+	// The event stamp is taken before the store write while the buffer
+	// append lands after it, so a delayed append keeps an early stamp in
+	// the payload while LogEntry.TsNs is bumped to stay monotone.
+	// Subscribers must see the stream position: a cursor saved past the
+	// payload stamp would filter the event out forever.
+	req := &filer_pb.SubscribeMetadataRequest{}
+	var unsyncedEvents int64
+	var gotTsNs, gotMutationTsNs int64
+	fn := eachLogEntryFn(context.Background(), req, &recordingSender{}, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64, mutationTsNs int64) error {
+		gotTsNs = tsNs
+		gotMutationTsNs = mutationTsNs
+		return nil
+	}, &unsyncedEvents)
+
+	data, err := proto.Marshal(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/b",
+		TsNs:      99, // stamped before the store write
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "expire1/foo"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The buffer appended it after a ts=100 write already landed: bumped to 101.
+	if _, err := fn(&filer_pb.LogEntry{TsNs: 101, Data: data}); err != nil {
+		t.Fatal(err)
+	}
+	if gotTsNs != 101 {
+		t.Fatalf("delivered ts = %d, want stream position 101", gotTsNs)
+	}
+	if gotMutationTsNs != 99 {
+		t.Fatalf("mutation ts = %d, want payload stamp 99", gotMutationTsNs)
+	}
+}
+
+func TestEachLogEntryFnFilteredHeartbeatUsesStreamPosition(t *testing.T) {
+	req := &filer_pb.SubscribeMetadataRequest{PathPrefix: "/buckets/keep/"}
+	sender := &recordingSender{}
+	unsyncedEvents := int64(MaxUnsyncedEvents)
+	fn := eachLogEntryFn(context.Background(), req, sender, func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64, mutationTsNs int64) error {
+		return nil
+	}, &unsyncedEvents)
+
+	data, err := proto.Marshal(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/other",
+		TsNs:      99,
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{Name: "x"},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := fn(&filer_pb.LogEntry{TsNs: 101, Data: data}); err != nil {
+		t.Fatal(err)
+	}
+	if len(sender.sent) != 1 || sender.sent[0].TsNs != 101 {
+		t.Fatalf("heartbeat ts = %+v, want stream position 101", sender.sent)
+	}
+}

weed/server/filer_grpc_server_sub_meta_test.go

@@ -197,11 +197,12 @@ func TestEachEventNotificationFnMatchesRenameTargetsForAllWatchTypes(t *testing.
 			if len(tt.req.Directories) > 0 {
 				newDir = tt.req.Directories[0]
 			}
+			now := time.Now().UnixNano()
 			err := eachEventFn("/tmp", &filer_pb.EventNotification{
 				OldEntry:      &filer_pb.Entry{Name: "old"},
 				NewEntry:      &filer_pb.Entry{Name: "new"},
 				NewParentPath: newDir,
-			}, time.Now().UnixNano())
+			}, now, now)
 			if err != nil {
 				t.Fatalf("eachEventFn: %v", err)
 			}
@@ -523,7 +524,7 @@ func TestFilteredEventsEmitMaxUnsyncedMarker(t *testing.T) {
 		lastTsNs = base + int64(i)
 		err := eachEventFn("/other/dir", &filer_pb.EventNotification{
 			NewEntry: &filer_pb.Entry{Name: fmt.Sprintf("file%d", i)},
-		}, lastTsNs)
+		}, lastTsNs, lastTsNs)
 		if err != nil {
 			t.Fatalf("eachEventFn: %v", err)
 		}

weed/util/log_buffer/log_buffer.go

@@ -558,6 +558,14 @@ func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) err
 }
 
 func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processingTsNs int64) error {
+	_, err := logBuffer.AddDataToBufferAndGetTs(partitionKey, data, processingTsNs)
+	return err
+}
+
+// AddDataToBufferAndGetTs appends like AddDataToBuffer and returns the
+// position the entry was assigned, bumped past processingTsNs when the
+// buffer must stay monotone.
+func (logBuffer *LogBuffer) AddDataToBufferAndGetTs(partitionKey, data []byte, processingTsNs int64) (int64, error) {
 
 	// An entry this large gets a window to itself, so it will seal and copy one;
 	// wait for the queue to have room before joining the queue for the lock.
@@ -582,7 +590,7 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	}
 
 	if !logBuffer.beginWrite() {
-		return ErrBufferStopped
+		return 0, ErrBufferStopped
 	}
 	var toFlush *dataToFlush
 	var marshalErr error
@@ -653,7 +661,7 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 			if size < 0 || size > math.MaxInt-4 || size > maxBufferSize-4 {
 				marshalErr = fmt.Errorf("message size %d exceeds maximum allowed size", size)
 				glog.Errorf("%v", marshalErr)
-				return marshalErr
+				return 0, marshalErr
 			}
 			// Fit the entry exactly. Doubling left room for a second oversized
 			// record in the same window, which only doubles the flush copy and
@@ -669,7 +677,7 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	if _, err := logEntry.MarshalToSizedBufferVT(logBuffer.buf[logBuffer.pos+4 : logBuffer.pos+4+size]); err != nil {
 		marshalErr = fmt.Errorf("failed to marshal LogEntry: %w", err)
 		glog.Errorf("%v", marshalErr)
-		return marshalErr
+		return 0, marshalErr
 	}
 	logBuffer.idx = append(logBuffer.idx, logBuffer.pos)
 	util.Uint32toBytes(logBuffer.sizeBuf, uint32(size))
@@ -684,7 +692,7 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	}
 
 	logBuffer.offset++
-	return nil
+	return logEntry.TsNs, nil
 }
 
 func (logBuffer *LogBuffer) IsStopping() bool {