fix(filer): bound aggregated metadata reads by peer watermarks (#10803)

* fix(filer): watermark-bound aggregated metadata subscription against multi-source merge races The aggregated metadata subscription (SubscribeMetadata) merges per-filer sources that become readable at independent paces, but tracks its progress with a single scalar cursor. Once the cursor passes a timestamp T, anything a source materializes below T afterwards is silently skipped: a peer recovering from a stall re-inserts its backlog late (late ring merge), and a source's flush can land a log file, or a later chunk of the same file, after a subscriber's disk pass listed the files (late persisted-log landing). This is the residual documented in #10501. Bound the subscriber's two read paths by what every source has provably made visible, each with its own watermark: - Delivery low-watermark -> in-memory reads. The meta aggregator tracks, per subscribed peer (self included), the newest timestamp received on that peer's stream - real events, or idle heartbeats (peer streams now opt into ClientSupportsIdleHeartbeat). The aggregated ring is complete up to the minimum across peers; in-memory reads hold at it. - Flush low-watermark -> persisted-log reads. Each filer reports its local log-buffer flush watermark on its stream: a new flushed_ts_ns response field, carried on idle heartbeats and on periodic flush reports (gated on ClientSupportsIdleHeartbeat). Disk passes freeze the minimum across peers before listing the log files and hold at it; the day-boundary cursor jump and the metadata-chunks ref listing are bounded the same way, the latter at minute-file granularity. - Held reads keep the cursor at the last entry actually delivered and retry; the retry re-lists the log files, which is what picks up a late-landing file. Both watermarks are relaxed by the settled horizon (2 x LogFlushInterval) as a liveness escape, so a peer stalled beyond it delays subscribers by at most the horizon instead of forever - any loss that escape allows was unconditional before. With reads held at the flush watermark, a disk advance below it is proven complete on every peer's disk, so the unproven-crossing counter now only counts crossings the horizon escape allowed past a stalled peer. Live delivery on the aggregated stream may lag by up to the idle-heartbeat interval when some peers are quiet; SubscribeLocalMetadata consumers are unaffected. * fix(filer): resume evicted aggregated readers from an original-space disk anchor The aggregated ring rewrites out-of-order peer arrivals to its head, so a subscriber tailing it advances its cursor in bumped (arrival) timestamps, while persisted logs keep original timestamps. When a slow reader's unread window is evicted (e.g. a peer backlog flooding in after a stall) and the reader falls back to disk, resuming from the bumped cursor skips every original-space entry below it that memory never delivered - reproduced as a ~66% silent loss on a 3-filer cluster with one peer's stream frozen for ~70s while the subscriber lagged. Track a disk anchor: the newest original-space position the stream is proven complete through. Disk passes advance it directly; contiguous memory reads advance it to the peers' delivery low-watermark observed before the read (per-peer streams are ordered, so everything with an original timestamp at or below that watermark had already arrived and was delivered). A reader kicked off the ring resumes the disk pass from the anchor instead of the bumped cursor - redelivering what memory already sent is within the subscription's at-least-once contract, skipping what it never sent is not. * fix(filer): close review findings on the peer-watermark subscription bounds Four correctness holes found in review, one generated-file cleanup: - The flush-through claim could assert durability for events still on their way into the buffer: an event is timestamped before notification work that can block, and only then appended. Track stamped-but-unappended events on the Filer (the stamp shares a lock with the reader, and appends are bumped monotonically past the buffer head), and cap the reported flush watermark just below the oldest in-flight stamp. - Removing a peer deleted its watermark entries while its stream kept running: its next signal recreated the deleted entry, which then pinned the low-watermark forever once the stream died. Watermarks now advance only for tracked peers, and peer removal cancels the subscription context so the stream stops feeding the aggregated buffer promptly. - The pipelined sender folded flush reports (TsNs 0 reads as far behind) into batch Events tails, where the aggregator's nil-notification guard dropped them - a busy backlog replay could starve the flush watermark until the settled-horizon escape opened a loss window. Control messages are now unbatchable on the sender, and the receiver also reads watermark state off nested batch entries as belt and braces. - A give-up skip's cursor was not anchored, so the next eviction rewind undid the counted decision and re-entered the same park forever when the evicted window carried bumped timestamps. The anchor now follows give-up skips; an anchored cursor makes the rewind a no-op and keeps the gap machinery's re-arm onto the retained window reachable. - Regenerated-file churn from a different protoc-gen-go-vtproto version is dropped: the vtproto file is upstream's, plus only the flushed_ts_ns marshal/size/unmarshal cases in the same generator style. New tests pin the in-flight floor, the no-resurrection rule for removed peers, and that control messages are never nested in batches. * fix(filer): keep a removed peer's watermarks through a grace period Deleting a peer's watermark entries the moment the master removes it reopened the loss the watermarks exist to prevent: a filer frozen or partitioned long enough to miss master heartbeats is removed from the cluster, its unflushed events still exist, and with its entries gone the low-watermarks snap forward to the healthy peers - subscribers advance past the absent peer's window and its late-landing log files are silently skipped. Reproduced on a 3-filer cluster: freezing two filers for ~70s got them removed ~28s in, and a catching-up subscriber lost their entire overlapping window. Removal now only marks the peer; its watermarks keep participating in the low-watermarks for a grace period (2 x LogFlushInterval, matching the subscribe loops' settled horizon, which already bounds a stale watermark's influence meanwhile). A re-added peer clears the mark and continues its values monotonically - the flap case costs nothing. A peer that stays gone is dropped when the grace expires, so a decommission cannot pin the low-watermarks, and a dropped peer's straggling signals cannot resurrect its entry. * fix(filer): cap delivery heartbeats by the in-flight floor; harden stamps Second review pass on the watermark bounds: - Idle heartbeats on the local stream claimed delivery-completeness through "now" while an event could still sit stamped-but-unappended behind blocking notification work. A peer aggregator turns that claim into its delivery low-watermark, so it could advance (and anchor credits with it) past an event that had not been streamed yet. The heartbeat timestamp is now capped just below the oldest in-flight stamp, like the flush claim already was. - In-flight stamps are forced monotonic against the registry's own history, so a wall-clock step backwards cannot slip a new stamp under an already-sampled floor. The cross-goroutine ordering still shares the meta log's global forward-clock assumption; the comments now say so instead of overclaiming. - Duplicate removal notifications no longer refresh a removed peer's grace deadline: the first removal time wins, so a decommissioned peer cannot sit in the watermark sets forever on repeated updates. - A failed buffer append clears the event's in-flight stamp on purpose: the event is dropped from the change stream entirely (a pre-existing defect of the append path, loudly logged), and a watermark waiting for it would pin this filer's claims forever. The comments now state the decision instead of implying the failure cannot happen. * docs(filer): tighten the watermark comments Comment-only: compress the narrative comments added on this branch down to their load-bearing invariants, and fix one stale sentence (peer removal no longer deletes the watermark entries immediately). No code changes. * fix(filer): subscribe to the local filer before remote peers Self's events reach the aggregated buffer only through the aggregator's own subscription to it, but bootstrap only seeded the peers the master already listed - and self's master registration races that listing, so the watermark set could hold remote peers without self. Once the remotes signalled, the low-watermarks would claim completeness for a stream that was still missing a merge source, letting aggregated subscribers advance past the local filer's events before its subscription started. Seed self first, unconditionally: before that the watermark set is empty (a documented safe state - reads hold at the settled horizon), and after it the set can never be remotes-only. The later master update for self, or a duplicate in the listed peers, is a no-op via the already-followed check in OnPeerUpdate. * fix(filer): fence watermark claims against wall-clock regression Record issued heartbeat/flush claims in the in-flight registry and stamp later events above them, so a backward clock step cannot land an event under a watermark a peer has already advanced to. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(filer): re-check the buffer head after fencing heartbeat claims An event appended between the caught-up check and the delivery claim was covered by the claim but not yet sent on the stream. The claims fence later stamps, so re-checking the head after them proves every covered event was already sent before the heartbeat. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(filer): cross the aggregated ring's pre-subscription range only on proof The eviction gate and the gap proofs read "nothing evicted yet" as "memory holds everything after the cursor". That is false for the merge-fed aggregated ring, which is born empty while every peer's history sits on disk: before the ring's first real eviction, a subscriber whose cursor was still below the bounded chunk pass's listing stop was served the ring's earliest entry inclusively, silently skipping the withheld pre-restart files - and the idle-wait callback credited the delivery low-watermark to the disk anchor in the same disconnected state. Mark everything at or below the subscriptions' start as evicted when the aggregator is built, credit the anchor only once the run is connected to the ring, and give the aggregated gap pass a real proof to cross the marked boundary with: each disk pass's proven coverage (the peer flush low-watermark capped by the pass's listing bound). An empty pass whose proof reaches the eviction watermark crosses to it silently - no park, no loss counter - so the mark costs a bounded catch-up delay instead of the 15-minute give-up. * fix(filer): keep shipped chunk tails at or below the hold point A log file spans past its named minute (window start plus up to a flush interval), and chunk-mode clients apply a shipped file whole - so a file tail past the hold point can become a persisted client checkpoint beyond what every peer has proven, and a crash inside that window resumes past another peer's late-but-in-contract flush. Stop the ref listing a minute plus a flush interval below the hold; the withheld band is served by the memory pass (ring retention far exceeds it) or by later passes as the hold advances, so freshness is unchanged. A frozen peer flushing one window that spans its whole freeze can still overshoot; that residual is bounded by the freeze and needs a crash inside it. * docs(filer): trim the review-fix comments --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Chris Lu <chris.lu@gmail.com>
2026-08-20 14:17:07 +00:00 · 2026-08-19 18:38:46 -07:00
parent 804111745a
commit 5d5fcdf07b
13 changed files with 1332 additions and 57 deletions
@@ -67,6 +67,7 @@ type Filer struct {
 	EmptyFolderCleaner      *empty_folder_cleanup.EmptyFolderCleaner
 	EmptyFolderCleanupDelay time.Duration
 	persistedLogCache       *persistedLogCache
+	metaLogInflight         metaLogInflight
 }

 func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerHost pb.ServerAddress, filerGroup string, collection string, replication string, dataCenter string, maxFilenameLength uint32, notifyFn func()) *Filer {
@@ -142,6 +143,9 @@ func (f *Filer) AggregateFromPeers(self pb.ServerAddress, existingNodes []*maste
 	f.EmptyFolderCleaner = empty_folder_cleanup.NewEmptyFolderCleaner(f, f.Dlm.LockRing, self, f.DirBucketsPath, f.EmptyFolderCleanupDelay)

 	f.MetaAggregator = NewMetaAggregator(f, self, f.GrpcDialOption)
+	// The ring starts empty while peer history sits on disk: mark the pre-startFrom
+	// range evicted so a cursor there reads disk, not the ring's earliest entry.
+	f.MetaAggregator.MetaLogBuffer.MarkEvictedThrough(startFrom.UnixNano())
 	f.MasterClient.SetOnPeerUpdateFn(func(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
 		if update.NodeType != cluster.FilerType {
 			return
@@ -159,6 +163,16 @@ func (f *Filer) AggregateFromPeers(self pb.ServerAddress, existingNodes []*maste
 		f.Dlm.LockRing.SetSnapshot(servers, update.Version)
 	})

+	// Subscribe to the local filer first: its events reach the aggregated
+	// buffer only through this subscription, and the peer watermarks must
+	// account for it before any remote peer - a remotes-only watermark set
+	// would claim completeness without self. existingNodes can omit self
+	// (master registration races this bootstrap); duplicate adds are no-ops.
+	f.MetaAggregator.OnPeerUpdate(&master_pb.ClusterNodeUpdate{
+		NodeType: cluster.FilerType,
+		Address:  string(self),
+		IsAdd:    true,
+	}, startFrom)
 	for _, peerUpdate := range existingNodes {
 		f.MetaAggregator.OnPeerUpdate(peerUpdate, startFrom)
 	}
@@ -9,6 +9,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
@@ -56,6 +57,9 @@ func (f *Filer) notifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry
 	}

 	event := f.newMetadataEvent(oldEntry, newEntry, deleteChunks, isFromOtherCluster, signatures)
+	// Clear the stamp after the buffer append below - deliberately also on
+	// append failure (see the metaLogInflight comment).
+	defer f.metaLogInflight.done(event.TsNs)
 	eventNotification := event.EventNotification

 	if notification.Queue != nil {
@@ -76,6 +80,119 @@ func (f *Filer) notifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry
 	return event
 }

+// metaLogInflight tracks events stamped but not yet appended to the local
+// log buffer - the two are separated by notification work that can block,
+// and a claim ignoring that window would assert durability or delivery for
+// timestamps still on their way in. Stamping shares the reader's lock, so an
+// event is always visible here or (bumped monotonically) in the buffer.
+//
+// An event whose append fails also clears its stamp: it is dropped from the
+// change stream entirely (loudly logged there), and a watermark waiting for
+// it would pin this filer's claims forever.
+type metaLogInflight struct {
+	sync.Mutex
+	stamped     map[int64]int
+	lastStampNs int64
+	lastClaimNs int64
+}
+
+// stamp assigns the event timestamp and registers it as in flight. Stamps
+// are monotonic against the registry's own history and every issued claim,
+// so a wall-clock step backwards cannot slip a new stamp under a floor or
+// watermark already handed out.
+func (t *metaLogInflight) stamp() int64 {
+	t.Lock()
+	defer t.Unlock()
+	floor := t.lastStampNs
+	if t.lastClaimNs > floor {
+		floor = t.lastClaimNs
+	}
+	tsNs := time.Now().UnixNano()
+	if tsNs <= floor {
+		tsNs = floor + 1
+	}
+	t.lastStampNs = tsNs
+	if t.stamped == nil {
+		t.stamped = make(map[int64]int)
+	}
+	t.stamped[tsNs]++
+	return tsNs
+}
+
+// done removes a stamp once the event has been appended to the buffer.
+func (t *metaLogInflight) done(tsNs int64) {
+	t.Lock()
+	defer t.Unlock()
+	if t.stamped[tsNs] <= 1 {
+		delete(t.stamped, tsNs)
+	} else {
+		t.stamped[tsNs]--
+	}
+}
+
+// minTsNs returns the oldest in-flight stamp, or 0 when nothing is in flight.
+func (t *metaLogInflight) minTsNs() int64 {
+	t.Lock()
+	defer t.Unlock()
+	var min int64
+	for tsNs := range t.stamped {
+		if min == 0 || tsNs < min {
+			min = tsNs
+		}
+	}
+	return min
+}
+
+// claimThrough caps a completeness claim by the oldest in-flight stamp and
+// fences it: later stamps always land above the returned claim, so a wall
+// clock stepping backwards cannot slide a new event under a watermark a
+// peer has already advanced to.
+func (t *metaLogInflight) claimThrough(nowNs int64) int64 {
+	t.Lock()
+	defer t.Unlock()
+	claim := nowNs
+	for tsNs := range t.stamped {
+		if tsNs-1 < claim {
+			claim = tsNs - 1
+		}
+	}
+	if claim > t.lastClaimNs {
+		t.lastClaimNs = claim
+	}
+	return claim
+}
+
+// LocalFlushedThroughTsNs reports the timestamp through which the local meta
+// log is durably on disk: everything at or below it is appended and flushed,
+// and nothing can land at or below it later. The registry is consulted before
+// the buffer: an event already appended is visible to the buffer claim, one
+// still in flight caps the claim, and one stamped later is fenced above it.
+func (f *Filer) LocalFlushedThroughTsNs(nowNs int64) int64 {
+	claim := f.metaLogInflight.claimThrough(nowNs)
+	if buffered := f.LocalMetaLogBuffer.FlushedThroughTsNs(nowNs); buffered < claim {
+		claim = buffered
+	}
+	return claim
+}
+
+// LocalDeliveredThroughTsNs caps a delivery-freshness claim (an idle
+// heartbeat's timestamp) by the oldest in-flight stamp: a stamped-but-
+// unappended event has not been streamed to anyone, and a peer aggregator
+// turns the claim into its delivery low-watermark.
+func (f *Filer) LocalDeliveredThroughTsNs(nowNs int64) int64 {
+	return f.metaLogInflight.claimThrough(nowNs)
+}
+
+// StampMetaLogInflightForTest and DoneMetaLogInflightForTest let tests in
+// other packages exercise the claim caps. Not for production use.
+func (f *Filer) StampMetaLogInflightForTest() int64 {
+	return f.metaLogInflight.stamp()
+}
+
+func (f *Filer) DoneMetaLogInflightForTest(tsNs int64) {
+	f.metaLogInflight.done(tsNs)
+}
+
 func (f *Filer) newMetadataEvent(oldEntry, newEntry *Entry, deleteChunks, isFromOtherCluster bool, signatures []int32) *filer_pb.SubscribeMetadataResponse {
 	if oldEntry == nil && newEntry == nil {
 		return nil
@@ -102,7 +219,8 @@ func (f *Filer) newMetadataEvent(oldEntry, newEntry *Entry, deleteChunks, isFrom
 			IsFromOtherCluster: isFromOtherCluster,
 			Signatures:         signatures,
 		},
-		TsNs: time.Now().UnixNano(),
+		// In flight until appended to the local log buffer (see metaLogInflight).
+		TsNs: f.metaLogInflight.stamp(),
 	}
 }

@@ -0,0 +1,134 @@
+package filer
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
+)
+
+// TestMetaLogInflightFloor pins the in-flight stamp bookkeeping backing
+// Filer.LocalFlushedThroughTsNs: the floor is the oldest outstanding stamp,
+// duplicate stamps are reference-counted, and a cleared registry reports zero.
+func TestMetaLogInflightFloor(t *testing.T) {
+	var inflight metaLogInflight
+	if got := inflight.minTsNs(); got != 0 {
+		t.Fatalf("empty registry: floor=%d want 0", got)
+	}
+	ts1 := inflight.stamp()
+	ts2 := inflight.stamp()
+	if ts2 < ts1 {
+		t.Fatalf("stamps not monotonic: %d then %d", ts1, ts2)
+	}
+	if got := inflight.minTsNs(); got != ts1 {
+		t.Fatalf("floor=%d want oldest stamp %d", got, ts1)
+	}
+	inflight.done(ts1)
+	if got := inflight.minTsNs(); got != ts2 {
+		t.Fatalf("after done(ts1): floor=%d want %d", got, ts2)
+	}
+	inflight.done(ts2)
+	if got := inflight.minTsNs(); got != 0 {
+		t.Fatalf("cleared registry: floor=%d want 0", got)
+	}
+}
+
+// TestLocalFlushedThroughTsNsBoundsInflight pins the flush-watermark claim: a
+// drained buffer claims "now", but an event stamped and not yet appended caps
+// the claim just below its timestamp - a peer bounding disk reads by the
+// claim must not advance past an event still on its way into the buffer.
+func TestLocalFlushedThroughTsNsBoundsInflight(t *testing.T) {
+	f := &Filer{
+		LocalMetaLogBuffer: log_buffer.NewLogBuffer("inflight-test", time.Minute, nil, nil, nil),
+	}
+	defer f.LocalMetaLogBuffer.ShutdownLogBuffer()
+
+	now := time.Now().UnixNano()
+	if got := f.LocalFlushedThroughTsNs(now); got != now {
+		t.Fatalf("drained, nothing in flight: claim=%d want now=%d", got, now)
+	}
+
+	ts := f.metaLogInflight.stamp()
+	if got := f.LocalFlushedThroughTsNs(time.Now().UnixNano()); got != ts-1 {
+		t.Fatalf("with in-flight stamp %d: claim=%d want %d", ts, got, ts-1)
+	}
+
+	f.metaLogInflight.done(ts)
+	now = time.Now().UnixNano()
+	if got := f.LocalFlushedThroughTsNs(now); got != now {
+		t.Fatalf("stamp cleared: claim=%d want now=%d", got, now)
+	}
+}
+
+// TestLocalDeliveredThroughTsNsBoundsInflight pins the delivery-freshness
+// cap: an idle heartbeat must not claim delivery-completeness past an event
+// that is stamped but not yet appended - the peer aggregator turns that claim
+// into its delivery low-watermark.
+func TestLocalDeliveredThroughTsNsBoundsInflight(t *testing.T) {
+	f := &Filer{
+		LocalMetaLogBuffer: log_buffer.NewLogBuffer("delivered-test", time.Minute, nil, nil, nil),
+	}
+	defer f.LocalMetaLogBuffer.ShutdownLogBuffer()
+
+	now := time.Now().UnixNano()
+	if got := f.LocalDeliveredThroughTsNs(now); got != now {
+		t.Fatalf("nothing in flight: claim=%d want now=%d", got, now)
+	}
+	ts := f.metaLogInflight.stamp()
+	if got := f.LocalDeliveredThroughTsNs(time.Now().UnixNano()); got != ts-1 {
+		t.Fatalf("with in-flight stamp %d: claim=%d want %d", ts, got, ts-1)
+	}
+	f.metaLogInflight.done(ts)
+}
+
+// TestClaimFencesFutureStamps pins the claim fence: once a claim is issued, a
+// wall-clock step backwards must not let a later stamp land at or below it -
+// the peer has already advanced its watermark to the claim.
+func TestClaimFencesFutureStamps(t *testing.T) {
+	f := &Filer{
+		LocalMetaLogBuffer: log_buffer.NewLogBuffer("claim-fence-test", time.Minute, nil, nil, nil),
+	}
+	defer f.LocalMetaLogBuffer.ShutdownLogBuffer()
+
+	// Claim with a sampled clock one hour ahead: to a stamp taken at the real
+	// wall clock this is exactly a backward step after the claim was issued.
+	aheadNs := time.Now().Add(time.Hour).UnixNano()
+	if got := f.LocalDeliveredThroughTsNs(aheadNs); got != aheadNs {
+		t.Fatalf("nothing in flight: claim=%d want %d", got, aheadNs)
+	}
+	ts := f.metaLogInflight.stamp()
+	if ts <= aheadNs {
+		t.Fatalf("stamp %d not fenced above issued delivery claim %d", ts, aheadNs)
+	}
+	f.metaLogInflight.done(ts)
+
+	// The flush claim fences the same way.
+	aheadNs += int64(time.Hour)
+	if got := f.LocalFlushedThroughTsNs(aheadNs); got != aheadNs {
+		t.Fatalf("drained buffer: claim=%d want %d", got, aheadNs)
+	}
+	ts = f.metaLogInflight.stamp()
+	if ts <= aheadNs {
+		t.Fatalf("stamp %d not fenced above issued flush claim %d", ts, aheadNs)
+	}
+	f.metaLogInflight.done(ts)
+}
+
+// TestMetaLogInflightStampMonotonic pins the registry-local monotonicity: a
+// wall clock stepping backwards must not let a new stamp slip under an
+// already-sampled floor.
+func TestMetaLogInflightStampMonotonic(t *testing.T) {
+	var inflight metaLogInflight
+	ts1 := inflight.stamp()
+	// Simulate a wall-clock step backwards: force the registry's history
+	// ahead of the clock; the next stamp must still move forward.
+	inflight.Lock()
+	inflight.lastStampNs = ts1 + int64(time.Hour)
+	inflight.Unlock()
+	ts2 := inflight.stamp()
+	if ts2 <= ts1+int64(time.Hour) {
+		t.Fatalf("stamp regressed: %d after forcing history to %d", ts2, ts1+int64(time.Hour))
+	}
+	inflight.done(ts1)
+	inflight.done(ts2)
+}
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -31,16 +32,38 @@ type MetaAggregator struct {
 	MetaLogBuffer  *log_buffer.LogBuffer
 	peerChans      map[pb.ServerAddress]chan struct{}
 	peerChansLock  sync.Mutex
+	// peerWatermarks tracks, per subscribed peer (self included), the newest
+	// timestamp received on that peer's stream (event or idle heartbeat).
+	// The minimum is a delivery low-watermark: MetaLogBuffer is complete up
+	// to it, so a subscriber held at or below it cannot miss a late-merged
+	// peer event. An unsignalled peer pins it at zero.
+	peerWatermarks map[pb.ServerAddress]int64
+	// peerFlushWatermarks tracks each peer's reported flush watermark:
+	// everything at or below it is on that peer's disk. The minimum bounds
+	// persisted-log reads — beyond it a peer may still land a log file (or
+	// chunk) whose events a passed cursor would skip.
+	peerFlushWatermarks map[pb.ServerAddress]int64
+	// peerRemovedAtNs marks peers the master removed, with the removal time.
+	// A removed peer keeps participating in the low-watermarks for a grace
+	// period: removal usually means a flap (frozen or partitioned filer)
+	// whose unflushed events still exist, and de-accounting it at once would
+	// let subscribers advance past them. A re-add clears the mark; a peer
+	// gone past the grace is dropped so it cannot pin the low-watermarks.
+	peerRemovedAtNs    map[pb.ServerAddress]int64
+	peerWatermarksLock sync.Mutex
 }

 // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk.
 // The old data comes from what each LocalMetadata persisted on disk.
 func NewMetaAggregator(filer *Filer, self pb.ServerAddress, grpcDialOption grpc.DialOption) *MetaAggregator {
 	t := &MetaAggregator{
-		filer:          filer,
-		self:           self,
-		grpcDialOption: grpcDialOption,
-		peerChans:      make(map[pb.ServerAddress]chan struct{}),
+		filer:               filer,
+		self:                self,
+		grpcDialOption:      grpcDialOption,
+		peerChans:           make(map[pb.ServerAddress]chan struct{}),
+		peerWatermarks:      make(map[pb.ServerAddress]int64),
+		peerFlushWatermarks: make(map[pb.ServerAddress]int64),
+		peerRemovedAtNs:     make(map[pb.ServerAddress]int64),
 	}
 	// nil notifyFn: aggregated subscribers wake through the buffer's
 	// subscriber channels, not a cond.
@@ -61,15 +84,128 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star
 		}
 		stopChan := make(chan struct{})
 		ma.peerChans[address] = stopChan
+		// 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)
 	} else {
 		if prevChan, found := ma.peerChans[address]; found {
 			close(prevChan)
 			delete(ma.peerChans, address)
 		}
+		// Only mark: dropping the watermarks at once would let subscribers
+		// advance past a flapping peer's unflushed events (see peerRemovedAtNs).
+		ma.markPeerWatermarkRemoved(address)
 	}
 }

+// peerWatermarkRemovalGrace is how long a removed peer keeps participating
+// in the low-watermarks. Matches the subscribe loops' settled horizon, which
+// already bounds a stale watermark's influence within it.
+const peerWatermarkRemovalGrace = 2 * LogFlushInterval
+
+// initPeerWatermark adds the peer with a zero (unknown) watermark; a re-added
+// peer keeps its prior values and stops being considered removed.
+func (ma *MetaAggregator) initPeerWatermark(peer pb.ServerAddress) {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	if _, found := ma.peerWatermarks[peer]; !found {
+		ma.peerWatermarks[peer] = 0
+	}
+	if _, found := ma.peerFlushWatermarks[peer]; !found {
+		ma.peerFlushWatermarks[peer] = 0
+	}
+	delete(ma.peerRemovedAtNs, peer)
+}
+
+func (ma *MetaAggregator) markPeerWatermarkRemoved(peer pb.ServerAddress) {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	if _, found := ma.peerWatermarks[peer]; !found {
+		return
+	}
+	// First removal time wins: duplicate removals must not refresh the grace.
+	if _, marked := ma.peerRemovedAtNs[peer]; !marked {
+		ma.peerRemovedAtNs[peer] = time.Now().UnixNano()
+	}
+}
+
+// dropExpiredRemovedPeersLocked drops peers whose removal outlived the grace.
+// Caller must hold peerWatermarksLock.
+func (ma *MetaAggregator) dropExpiredRemovedPeersLocked() {
+	if len(ma.peerRemovedAtNs) == 0 {
+		return
+	}
+	cutoff := time.Now().UnixNano() - int64(peerWatermarkRemovalGrace)
+	for peer, removedAt := range ma.peerRemovedAtNs {
+		if removedAt < cutoff {
+			delete(ma.peerRemovedAtNs, peer)
+			delete(ma.peerWatermarks, peer)
+			delete(ma.peerFlushWatermarks, peer)
+		}
+	}
+}
+
+// advancePeerWatermark records the peer as received-through tsNs. Monotonic,
+// and only for tracked peers: a dropped peer's straggler must not recreate
+// its entry and pin the low-watermark.
+func (ma *MetaAggregator) advancePeerWatermark(peer pb.ServerAddress, tsNs int64) {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	if cur, found := ma.peerWatermarks[peer]; found && tsNs > cur {
+		ma.peerWatermarks[peer] = tsNs
+	}
+}
+
+// advancePeerFlushWatermark records the peer's reported flush watermark.
+// Monotonic, and only for tracked peers (see advancePeerWatermark).
+func (ma *MetaAggregator) advancePeerFlushWatermark(peer pb.ServerAddress, tsNs int64) {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	if cur, found := ma.peerFlushWatermarks[peer]; found && tsNs > cur {
+		ma.peerFlushWatermarks[peer] = tsNs
+	}
+}
+
+// PeerLowFlushWatermarkTsNs returns the minimum reported flush watermark
+// across all current peers: every peer's events at or below this time are on
+// disk. Returns 0 when any peer has not reported yet (completeness unknown).
+func (ma *MetaAggregator) PeerLowFlushWatermarkTsNs() int64 {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	ma.dropExpiredRemovedPeersLocked()
+	if len(ma.peerFlushWatermarks) == 0 {
+		return 0
+	}
+	var low int64 = math.MaxInt64
+	for _, tsNs := range ma.peerFlushWatermarks {
+		if tsNs < low {
+			low = tsNs
+		}
+	}
+	return low
+}
+
+// PeerLowWatermarkTsNs returns the minimum received-through timestamp across
+// all current peers (self included): MetaLogBuffer is complete up to this
+// time. Returns 0 when any peer has not signalled yet (or none are tracked),
+// i.e. completeness is unknown.
+func (ma *MetaAggregator) PeerLowWatermarkTsNs() int64 {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	ma.dropExpiredRemovedPeersLocked()
+	if len(ma.peerWatermarks) == 0 {
+		return 0
+	}
+	var low int64 = math.MaxInt64
+	for _, tsNs := range ma.peerWatermarks {
+		if tsNs < low {
+			low = tsNs
+		}
+	}
+	return low
+}
+
 func (ma *MetaAggregator) HasRemotePeers() bool {
 	ma.peerChansLock.Lock()
 	defer ma.peerChansLock.Unlock()
@@ -103,7 +239,7 @@ func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddres
 	lastTsNs := startFrom.UnixNano()
 	for {
 		glog.V(0).Infof("loopSubscribeToOneFiler read %s start from %v %d", peer, time.Unix(0, lastTsNs), lastTsNs)
-		nextLastTsNs, err := ma.doSubscribeToOneFiler(f, self, peer, lastTsNs)
+		nextLastTsNs, err := ma.doSubscribeToOneFiler(f, self, peer, lastTsNs, stopChan)

 		// check stopChan to see if we should stop
 		select {
@@ -127,7 +263,7 @@ func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddres
 	}
 }

-func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom int64) (int64, error) {
+func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom int64, stopChan <-chan struct{}) (int64, error) {

 	/*
 		Each filer reads the "filer.store.id", which is the store's signature when filer starts.
@@ -225,10 +361,23 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 		return nil
 	}

+	// The stream will deliver everything after lastTsNs, so the aggregated
+	// buffer is (still) complete for this peer up to that point.
+	ma.advancePeerWatermark(peer, lastTsNs)
+
 	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():
+			}
+		}()
 		atomic.AddInt32(&ma.filer.UniqueFilerEpoch, 1)
 		// Construct a log file reader that reads chunks via the peer filer's LookupVolume.
 		lookupFn := LookupFn(filerClient{client})
@@ -244,6 +393,8 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 			ClientEpoch:                  atomic.LoadInt32(&ma.filer.UniqueFilerEpoch),
 			ClientSupportsBatching:       true,
 			ClientSupportsMetadataChunks: true,
+			// Idle heartbeats keep a quiet peer from freezing the low-watermark.
+			ClientSupportsIdleHeartbeat: true,
 		})
 		if err != nil {
 			glog.V(0).Infof("SubscribeLocalMetadata %v: %v", peer, err)
@@ -257,6 +408,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 			}
 			f.onMetadataChangeEvent(event)
 			lastTsNs = event.TsNs
+			ma.advancePeerWatermark(peer, event.TsNs)
 			return nil
 		}

@@ -299,17 +451,31 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress,
 					return err
 				}
 			}
-			// Process any additional batched events. Mirror the envelope's nil
-			// guard: the server can fold a freshness signal (nil EventNotification)
-			// into the batched tail, and processOne dereferences it.
+			// Batched events, with the envelope's nil guard mirrored. A nested
+			// control message still carries watermark state - dropping it
+			// would make healthy flush progress look stalled.
 			for _, batchedEvent := range resp.Events {
+				if batchedEvent.FlushedTsNs > 0 {
+					ma.advancePeerFlushWatermark(peer, batchedEvent.FlushedTsNs)
+				}
 				if batchedEvent.EventNotification == nil {
+					if batchedEvent.TsNs > 0 {
+						ma.advancePeerWatermark(peer, batchedEvent.TsNs)
+					}
 					continue
 				}
 				if err := processOne(batchedEvent); err != nil {
 					return err
 				}
 			}
+			// Idle heartbeat: advance only the watermark, not lastTsNs, so a
+			// reconnect still resumes from the last real event.
+			if resp.EventNotification == nil && len(resp.Events) == 0 && resp.TsNs > 0 {
+				ma.advancePeerWatermark(peer, resp.TsNs)
+			}
+			if resp.FlushedTsNs > 0 {
+				ma.advancePeerFlushWatermark(peer, resp.FlushedTsNs)
+			}
 		}
 	})
 	return lastTsNs, err
@@ -0,0 +1,149 @@
+package filer
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+func newTestAggregator() *MetaAggregator {
+	return &MetaAggregator{
+		peerWatermarks:      make(map[pb.ServerAddress]int64),
+		peerFlushWatermarks: make(map[pb.ServerAddress]int64),
+		peerRemovedAtNs:     make(map[pb.ServerAddress]int64),
+	}
+}
+
+// TestPeerWatermarkBookkeeping pins the per-peer delivery watermark semantics
+// backing MetaAggregator.PeerLowWatermarkTsNs: the low-watermark is the minimum
+// received-through timestamp across tracked peers, a peer that has not
+// signalled yet holds it at zero (completeness unknown), advances are
+// monotonic, and removed peers stop participating only after the grace.
+func TestPeerWatermarkBookkeeping(t *testing.T) {
+	ma := newTestAggregator()
+	a, b := pb.ServerAddress("filer-a:8888"), pb.ServerAddress("filer-b:8888")
+
+	if got := ma.PeerLowWatermarkTsNs(); got != 0 {
+		t.Fatalf("no peers: low=%d want 0", got)
+	}
+
+	// A tracked-but-silent peer pins the low-watermark at zero.
+	ma.initPeerWatermark(a)
+	ma.initPeerWatermark(b)
+	ma.advancePeerWatermark(a, 100)
+	if got := ma.PeerLowWatermarkTsNs(); got != 0 {
+		t.Fatalf("silent peer: low=%d want 0", got)
+	}
+
+	// Both signalled: low is the minimum.
+	ma.advancePeerWatermark(b, 50)
+	if got := ma.PeerLowWatermarkTsNs(); got != 50 {
+		t.Fatalf("low=%d want 50", got)
+	}
+
+	// Advances are monotonic: a stale (lower) signal cannot regress.
+	ma.advancePeerWatermark(b, 40)
+	if got := ma.PeerLowWatermarkTsNs(); got != 50 {
+		t.Fatalf("after stale signal: low=%d want 50", got)
+	}
+
+	// Reconnect keeps the prior value (init does not reset).
+	ma.initPeerWatermark(b)
+	if got := ma.PeerLowWatermarkTsNs(); got != 50 {
+		t.Fatalf("after re-init: low=%d want 50", got)
+	}
+}
+
+// TestPeerWatermarkRemovalGrace pins the removal semantics: a removed peer is
+// usually a flap (frozen or partitioned filer), so its watermarks keep
+// holding the low-watermarks for the grace period - de-accounting it at once
+// would let subscribers advance past its still-unflushed events. A re-add
+// within the grace continues the values; a peer gone past the grace is
+// dropped, so a decommission cannot pin the low-watermark, and its straggling
+// signals cannot resurrect the entry.
+func TestPeerWatermarkRemovalGrace(t *testing.T) {
+	ma := newTestAggregator()
+	a, b := pb.ServerAddress("filer-a:8888"), pb.ServerAddress("filer-b:8888")
+	ma.initPeerWatermark(a)
+	ma.initPeerWatermark(b)
+	ma.advancePeerWatermark(a, 100)
+	ma.advancePeerWatermark(b, 50)
+	ma.advancePeerFlushWatermark(a, 100)
+	ma.advancePeerFlushWatermark(b, 50)
+
+	// Freshly removed: still participates (the flap case that loses data if
+	// dropped at once).
+	ma.markPeerWatermarkRemoved(b)
+	if got := ma.PeerLowWatermarkTsNs(); got != 50 {
+		t.Fatalf("within grace: low=%d want 50", got)
+	}
+	if got := ma.PeerLowFlushWatermarkTsNs(); got != 50 {
+		t.Fatalf("within grace: flush low=%d want 50", got)
+	}
+	// Its draining stream may still advance it while marked.
+	ma.advancePeerWatermark(b, 60)
+	if got := ma.PeerLowWatermarkTsNs(); got != 60 {
+		t.Fatalf("marked peer advance: low=%d want 60", got)
+	}
+
+	// Re-add within the grace: mark cleared, values continue.
+	ma.initPeerWatermark(b)
+	if _, marked := ma.peerRemovedAtNs[b]; marked {
+		t.Fatalf("re-added peer still marked removed")
+	}
+	if got := ma.PeerLowWatermarkTsNs(); got != 60 {
+		t.Fatalf("after re-add: low=%d want 60", got)
+	}
+
+	// Removal past the grace: dropped from both watermark sets. A duplicate
+	// removal notification must not refresh the deadline (first mark wins).
+	ma.markPeerWatermarkRemoved(b)
+	ma.peerWatermarksLock.Lock()
+	ma.peerRemovedAtNs[b] = time.Now().UnixNano() - int64(peerWatermarkRemovalGrace) - int64(time.Second)
+	ma.peerWatermarksLock.Unlock()
+	ma.markPeerWatermarkRemoved(b) // duplicate removal: must not reset the clock
+	if got := ma.PeerLowWatermarkTsNs(); got != 100 {
+		t.Fatalf("past grace: low=%d want 100", got)
+	}
+	if got := ma.PeerLowFlushWatermarkTsNs(); got != 100 {
+		t.Fatalf("past grace: flush low=%d want 100", got)
+	}
+
+	// A straggling signal after the drop must not resurrect the entry.
+	ma.advancePeerWatermark(b, 999)
+	ma.advancePeerFlushWatermark(b, 999)
+	if got := ma.PeerLowWatermarkTsNs(); got != 100 {
+		t.Fatalf("after straggler: low=%d want 100", got)
+	}
+	if _, found := ma.peerWatermarks[b]; found {
+		t.Fatalf("dropped peer resurrected in delivery watermark set")
+	}
+	if _, found := ma.peerFlushWatermarks[b]; found {
+		t.Fatalf("dropped peer resurrected in flush watermark set")
+	}
+}
+
+// TestPeerFlushWatermarkBookkeeping mirrors the delivery-watermark semantics
+// for the flush watermark that bounds persisted-log reads: min across peers,
+// zero until every peer has reported, monotonic advances.
+func TestPeerFlushWatermarkBookkeeping(t *testing.T) {
+	ma := newTestAggregator()
+	a, b := pb.ServerAddress("filer-a:8888"), pb.ServerAddress("filer-b:8888")
+
+	ma.initPeerWatermark(a)
+	ma.initPeerWatermark(b)
+	ma.advancePeerFlushWatermark(a, 200)
+	if got := ma.PeerLowFlushWatermarkTsNs(); got != 0 {
+		t.Fatalf("unreported peer: low=%d want 0", got)
+	}
+	ma.advancePeerFlushWatermark(b, 150)
+	if got := ma.PeerLowFlushWatermarkTsNs(); got != 150 {
+		t.Fatalf("low=%d want 150", got)
+	}
+	ma.advancePeerFlushWatermark(b, 120) // stale report cannot regress
+	if got := ma.PeerLowFlushWatermarkTsNs(); got != 150 {
+		t.Fatalf("after stale: low=%d want 150", got)
+	}
+	_ = a
+}
@@ -662,6 +662,7 @@ message SubscribeMetadataResponse {
    int64 ts_ns = 3;
    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
 }
 message ListMetadataSubscribersRequest {
    repeated string client_types = 1; // optional filter by client type, e.g. "mount"; empty = all
@@ -4327,8 +4327,9 @@ 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
+	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
 	unknownFields     protoimpl.UnknownFields
 	sizeCache         protoimpl.SizeCache
 }
@@ -4398,6 +4399,13 @@ func (x *SubscribeMetadataResponse) GetLogFileRefs() []*LogFileChunkRef {
 	return nil
 }

+func (x *SubscribeMetadataResponse) GetFlushedTsNs() int64 {
+	if x != nil {
+		return x.FlushedTsNs
+	}
+	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
@@ -7355,13 +7363,14 @@ 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\"\x96\x02\n" +
+	"\x1eclient_supports_idle_heartbeat\x18\r \x01(\bR\x1bclientSupportsIdleHeartbeat\"\xba\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\"C\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" +
 	"\x1eListMetadataSubscribersRequest\x12!\n" +
 	"\fclient_types\x18\x01 \x03(\tR\vclientTypes\"a\n" +
 	"\x1fListMetadataSubscribersResponse\x12>\n" +
@@ -3944,6 +3944,11 @@ func (m *SubscribeMetadataResponse) MarshalToSizedBufferVT(dAtA []byte) (int, er
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.FlushedTsNs != 0 {
+		i = protohelpers.EncodeVarint(dAtA, i, uint64(m.FlushedTsNs))
+		i--
+		dAtA[i] = 0x30
+	}
 	if len(m.LogFileRefs) > 0 {
 		for iNdEx := len(m.LogFileRefs) - 1; iNdEx >= 0; iNdEx-- {
 			size, err := m.LogFileRefs[iNdEx].MarshalToSizedBufferVT(dAtA[:i])
@@ -7816,6 +7821,9 @@ func (m *SubscribeMetadataResponse) SizeVT() (n int) {
 			n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
 		}
 	}
+	if m.FlushedTsNs != 0 {
+		n += 1 + protohelpers.SizeOfVarint(uint64(m.FlushedTsNs))
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -19463,6 +19471,25 @@ func (m *SubscribeMetadataResponse) UnmarshalVT(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 6:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field FlushedTsNs", wireType)
+			}
+			m.FlushedTsNs = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.FlushedTsNs |= int64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])
@@ -34,6 +34,11 @@ var (
 	// and logged: a dead peer makes the wait permanent, and failing the stream
 	// only moves the loop into a client that reconnects to the same wall.
 	maxGapStall = 15 * time.Minute
+
+	// metadataGapSettledHorizon is the liveness escape for the peer-watermark
+	// holds: a watermark stalled further than this stops holding reads back.
+	// Twice the flush interval so a healthy peer's flush always lands within.
+	metadataGapSettledHorizon = 2 * filer.LogFlushInterval
 )

 const (
@@ -95,7 +100,10 @@ func (s *pipelinedSender) sendLoop(stream metadataStreamSender) {
 		// envelope would drop its Events tail and refs inside Events would be
 		// applied as an (empty) event. Their TsNs is 0, which the batch
 		// heuristic would misread as far behind. Always send them solo.
-		shouldBatch := s.canBatch && len(msg.LogFileRefs) == 0 &&
+		// Control messages (nil EventNotification: heartbeats, flush reports)
+		// are unbatchable too - nested in an Events tail their watermark
+		// state is invisible to receivers.
+		shouldBatch := s.canBatch && len(msg.LogFileRefs) == 0 && msg.EventNotification != nil &&
 			time.Now().UnixNano()-msg.TsNs > int64(batchBehindThreshold)

 		if !shouldBatch {
@@ -112,7 +120,7 @@ func (s *pipelinedSender) sendLoop(stream metadataStreamSender) {
 		// go in the Events slice. Old clients ignore the Events field.
 		batch := make([]*filer_pb.SubscribeMetadataResponse, 0, maxBatchSize)
 		batch = append(batch, msg)
-		var trailingRefs *filer_pb.SubscribeMetadataResponse
+		var trailingSolo *filer_pb.SubscribeMetadataResponse
 	drain:
 		for len(batch) < maxBatchSize {
 			select {
@@ -120,9 +128,9 @@ func (s *pipelinedSender) sendLoop(stream metadataStreamSender) {
 				if !ok {
 					break drain
 				}
-				if len(next.LogFileRefs) > 0 {
+				if len(next.LogFileRefs) > 0 || next.EventNotification == nil {
 					// already consumed; send it solo right after the batch
-					trailingRefs = next
+					trailingSolo = next
 					break drain
 				}
 				batch = append(batch, next)
@@ -146,8 +154,8 @@ func (s *pipelinedSender) sendLoop(stream metadataStreamSender) {
 		if toSend.Events != nil {
 			toSend.Events = nil
 		}
-		if trailingRefs != nil {
-			if err := stream.Send(trailingRefs); err != nil {
+		if trailingSolo != nil {
+			if err := stream.Send(trailingSolo); err != nil {
 				s.reportErr(err)
 				return
 			}
@@ -192,15 +200,18 @@ func (s *pipelinedSender) Close() error {
 	}
 }

-// reportUnprovenAggregatedCrossing records the residual hole: a disk read that
-// crosses the eviction watermark may have advanced on one peer's log while a
-// lagging peer still holds unflushed events in the crossed range. Locally
-// undecidable (log files carry random filer ids, peers are tracked by address);
-// closing it needs each peer's flush watermark on the subscribe stream.
-func reportUnprovenAggregatedCrossing(cursorBeforeTsNs, cursorAfterTsNs, evictedTsNs int64, clientName, pathPrefix string) {
+// reportUnprovenAggregatedCrossing counts a disk read crossing the eviction
+// watermark without proof. A crossing at or below provenThroughTsNs (the
+// flush low-watermark frozen before the pass listed files) is proven and not
+// reported, so what remains is exactly what the settled-horizon escape
+// allowed past a stalled peer.
+func reportUnprovenAggregatedCrossing(cursorBeforeTsNs, cursorAfterTsNs, evictedTsNs, provenThroughTsNs int64, clientName, pathPrefix string) {
 	if evictedTsNs == 0 || cursorBeforeTsNs >= evictedTsNs || cursorAfterTsNs < evictedTsNs {
 		return
 	}
+	if provenThroughTsNs >= evictedTsNs {
+		return
+	}
 	stats.FilerSubscribeUnprovenGapCrossings.WithLabelValues("aggregated").Inc()
 	glog.Warningf("aggregated subscriber %s %s crossed an evicted range (%v..%v] on peer disk reads; a peer that flushes into it later will not be re-read",
 		clientName, pathPrefix, time.Unix(0, cursorBeforeTsNs), time.Unix(0, evictedTsNs))
@@ -232,6 +243,44 @@ func memoryHoldsGap(currentTsNs, lastEvictedTsNs int64) bool {
 	return currentTsNs >= lastEvictedTsNs
 }

+// errHeldByPeerWatermark aborts a read at an entry beyond the hold point; the
+// caller rewinds to the last delivered entry, waits, and re-reads (the
+// re-listing is what picks up a late-landing log file).
+var errHeldByPeerWatermark = errors.New("held by aggregated peer watermark")
+
+// resolveAggReadHoldTsNs bounds how far an aggregated subscriber may read: a
+// cursor that passes T before every source has provably made T visible loses
+// whatever arrives late. The hold is the peers' low-watermark (delivery for
+// memory reads, flush for persisted reads), relaxed by the settled horizon so
+// a stalled peer delays subscribers by at most the horizon.
+func resolveAggReadHoldTsNs(peerLowWatermarkTsNs, nowTsNs int64, settledHorizon time.Duration) int64 {
+	horizonTsNs := nowTsNs - int64(settledHorizon)
+	if peerLowWatermarkTsNs > horizonTsNs {
+		return peerLowWatermarkTsNs
+	}
+	return horizonTsNs
+}
+
+// previousMinuteEndTsNs returns the last nanosecond of the minute before
+// tsNs: log files are named per minute, so a ref listing bounded here cannot
+// include a file whose window crosses tsNs.
+func previousMinuteEndTsNs(tsNs int64) int64 {
+	return tsNs - tsNs%int64(time.Minute) - 1
+}
+
+// chunkRefsStopTsNs bounds the ref listing so no shipped file holds an entry
+// past the hold: clients apply shipped files whole and may checkpoint from a
+// tail. A file's window starts in its name's minute and spans up to a flush
+// interval, hence the double back-off; a frozen peer's freeze-spanning window
+// can still overshoot by its freeze.
+func chunkRefsStopTsNs(holdTsNs, untilNs int64) int64 {
+	stopTsNs := previousMinuteEndTsNs(holdTsNs - int64(filer.LogFlushInterval))
+	if untilNs != 0 && untilNs < stopTsNs {
+		stopTsNs = untilNs
+	}
+	return stopTsNs
+}
+
 // gapStallReporter makes a parked subscriber visible: a flush that never lands
 // stalls the stream for good, and filer.sync and mount followers just stop
 // advancing with no error on either side.
@@ -395,8 +444,9 @@ func (fs *FilerServer) parkOnGap(ctx context.Context, req *filer_pb.SubscribeMet
 // so memory still holds the whole gap; or the flush watermark observed before
 // the read had already passed the earliest in-memory timestamp, so every event
 // in the gap would have been on disk when the read ran and the miss is
-// authoritative. The aggregated ring never flushes - peers persist their own
-// logs - so it passes flushedTsNs 0 and only the eviction proof can hold.
+// authoritative. The aggregated loop passes its peers' proven-covered
+// watermark as flushedTsNs (everything at or below it was flushed and inside
+// the pass's listing), the local loop its own flush watermark.
 func resolveGapResume(currentTsNs, currentOffset, earliestMemTsNs, flushedTsNs, lastEvictedTsNs int64) (advanceToTsNs int64, advance bool) {
 	// No in-memory data (zero time → negative UnixNano), or memory not ahead of us.
 	if earliestMemTsNs <= 0 || earliestMemTsNs <= currentTsNs {
@@ -436,7 +486,7 @@ type gapPass struct {
 	gapStall  *gapStallReporter
 	earliest  func() time.Time
 	evicted   func() int64 // gap-proof watermark; aggregated uses the received-ts space
-	flushed   func() int64 // flush watermark the last disk read observed; aggregated: 0
+	flushed   func() int64 // what the last disk read proved covered: flushed AND inside its listing
 	gapChan   <-chan struct{}
 	dataChan  <-chan struct{}
 	gapReason func(earliest time.Time, evictedTsNs int64) string
@@ -471,6 +521,16 @@ func (p *gapPass) resolve(ctx context.Context, cursor *log_buffer.MessagePositio
 			*latch = nil
 			return gapProceed
 		}
+		// The last empty disk read proved coverage through the eviction
+		// watermark itself: the rest of the gap holds nothing - cross without
+		// parking or counting. The ring's pre-subscription mark is never
+		// proven by rotation, so this is its only exit.
+		if p.flushed() >= evictedTsNs {
+			p.gapStall.resumed()
+			*cursor = log_buffer.NewMessagePosition(evictedTsNs, gapResumeCursorOffset)
+			*latch = nil
+			return gapContinue
+		}
 		return p.park(ctx, cursor, latch, p.gapChan, p.gapReason(earliest, evictedTsNs))
 	}
 	if !diskAdvanced && errors.Is(*latch, log_buffer.ResumeFromDiskError) {
@@ -521,6 +581,17 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 	lastReadTime := log_buffer.NewMessagePosition(req.SinceNs, gapResumeCursorOffset)
 	glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)

+	// diskAnchorTsNs is the newest ORIGINAL-timestamp position this stream is
+	// proven complete through. Memory reads advance the cursor in the ring's
+	// bumped (arrival) space while persisted logs keep original timestamps,
+	// so a reader that falls off the ring must resume the disk pass here, not
+	// at its bumped cursor - that would skip original-space entries memory
+	// never delivered. Disk passes advance the anchor directly; contiguous
+	// memory reads advance it to the delivery low-watermark observed before
+	// the read (per-peer streams are ordered, so everything at or below it
+	// had already arrived and been delivered).
+	diskAnchorTsNs := req.SinceNs
+
 	sender := newPipelinedSender(stream, 1024, req.ClientSupportsBatching)
 	defer sender.Close()

@@ -548,9 +619,55 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 	var lastSeenTsNs int64
 	var lastHeartbeatNs int64
 	baseEachLogEntryFn := eachLogEntryFn(req, sender, eachEventNotificationFn, &unsyncedEvents)
-	eachLogEntryFn := func(logEntry *filer_pb.LogEntry) (bool, error) {
-		lastSeenTsNs = logEntry.TsNs
-		return baseEachLogEntryFn(logEntry)
+	// heldAtTsNs remembers the entry a read was held at (for the log line);
+	// the rewind target is the last entry actually delivered.
+	var heldAtTsNs int64
+	// Each read path holds at its own watermark: persisted logs are complete
+	// only up to every peer's flush watermark, the ring only up to every
+	// peer's delivery watermark.
+	holdMemTsNs := func() int64 {
+		return resolveAggReadHoldTsNs(fs.filer.MetaAggregator.PeerLowWatermarkTsNs(), time.Now().UnixNano(), metadataGapSettledHorizon)
+	}
+	// deliveredUpToTsNs tracks the newest position actually handed to the
+	// sender (or intentionally skipped by the gap machinery), so a held read
+	// can rewind to a position that skips nothing.
+	var deliveredUpToTsNs int64
+	guardedEachLogEntryFn := func(holdFn func() int64) log_buffer.EachLogEntryFuncType {
+		return func(logEntry *filer_pb.LogEntry) (bool, error) {
+			if logEntry.TsNs > holdFn() {
+				heldAtTsNs = logEntry.TsNs
+				return false, errHeldByPeerWatermark
+			}
+			lastSeenTsNs = logEntry.TsNs
+			deliveredUpToTsNs = logEntry.TsNs
+			return baseEachLogEntryFn(logEntry)
+		}
+	}
+	// Frozen BEFORE each pass lists the log files: per-source flushes are
+	// ts-ordered, so everything at or below the frozen value is in the
+	// listing; a live value could rise mid-pass and admit entries past files
+	// this pass cannot see.
+	var diskPassFlushLowTsNs int64
+	var diskPassHoldTsNs int64
+	// What the last disk pass proved covered: flushed on every peer AND inside
+	// the pass's listing, so an empty pass proves (cursor, proven] empty.
+	var diskPassProvenTsNs int64
+	diskEachLogEntryFn := guardedEachLogEntryFn(func() int64 { return diskPassHoldTsNs })
+	memEachLogEntryFn := guardedEachLogEntryFn(holdMemTsNs)
+	// waitHeld pauses a held read until new data or the retry interval (holds
+	// also release on heartbeats, which do not notify). False: context ended.
+	waitHeld := func() bool {
+		glog.V(3).Infof("held at %v (deliveredUpTo %v, flushLow %v, deliveryLow %v) for %v",
+			time.Unix(0, heldAtTsNs), time.Unix(0, deliveredUpToTsNs),
+			time.Unix(0, fs.filer.MetaAggregator.PeerLowFlushWatermarkTsNs()),
+			time.Unix(0, fs.filer.MetaAggregator.PeerLowWatermarkTsNs()), clientName)
+		select {
+		case <-aggNotifyChan:
+		case <-ctx.Done():
+			return false
+		case <-time.After(unflushedGapRetryInterval):
+		}
+		return true
 	}

 	var processedTsNs int64
@@ -566,8 +683,8 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 		gapStall: gapStall,
 		earliest: aggBuffer.GetEarliestTime,
 		evicted:  aggBuffer.GetLastEvictedOriginalTsNs,
-		flushed:  func() int64 { return 0 }, // the aggregated ring never flushes
-		gapChan:  nil,                       // nothing local signals a peer's flush; the timer paces it
+		flushed:  func() int64 { return diskPassProvenTsNs },
+		gapChan:  nil, // nothing local signals a peer's flush; the timer paces it
 		dataChan: aggNotifyChan,
 		gapReason: func(earliest time.Time, evictedTsNs int64) string {
 			return fmt.Sprintf("gap evicted through %v is not on a peer's disk yet (earliest memory %v)",
@@ -581,10 +698,42 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,

 		cursorBeforeDiskTsNs := lastReadTime.Time.UnixNano()

+		// Observe the flush low-watermark before the pass lists files (see
+		// diskPassHoldTsNs above).
+		diskPassFlushLowTsNs = fs.filer.MetaAggregator.PeerLowFlushWatermarkTsNs()
+		diskPassHoldTsNs = resolveAggReadHoldTsNs(diskPassFlushLowTsNs, time.Now().UnixNano(), metadataGapSettledHorizon)
+		diskPassProvenTsNs = diskPassFlushLowTsNs
+
 		if req.ClientSupportsMetadataChunks {
-			processedTsNs, isDone, readPersistedLogErr = fs.chunkDiskPass(ctx, sender, lastReadTime, req.UntilNs, sentRefs)
+			refsStopTsNs := chunkRefsStopTsNs(diskPassHoldTsNs, req.UntilNs)
+			// Nothing above the listing bound is proven by this pass.
+			if refsStopTsNs < diskPassProvenTsNs {
+				diskPassProvenTsNs = refsStopTsNs
+			}
+			if refsStopTsNs > lastReadTime.Time.UnixNano() {
+				processedTsNs, isDone, readPersistedLogErr = fs.chunkDiskPass(ctx, sender, lastReadTime, refsStopTsNs, sentRefs)
+			} else {
+				processedTsNs, isDone, readPersistedLogErr = 0, false, nil
+			}
 		} else {
-			processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, eachLogEntryFn)
+			processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, diskEachLogEntryFn)
+		}
+		if errors.Is(readPersistedLogErr, errHeldByPeerWatermark) {
+			// Stay at the last delivered entry; the held entry is re-read (and
+			// re-checked) by the next pass.
+			if processedTsNs > 0 {
+				lastReadTime = log_buffer.NewMessagePosition(processedTsNs, gapResumeCursorOffset)
+				if processedTsNs > diskAnchorTsNs {
+					diskAnchorTsNs = processedTsNs
+				}
+			}
+			// A hold is not a gap: clear any stale ResumeFromDiskError so the
+			// next pass's disk-miss handling cannot skip past the held entry.
+			readInMemoryLogErr = nil
+			if !waitHeld() {
+				return nil
+			}
+			continue
 		}
 		if readPersistedLogErr != nil {
 			return fmt.Errorf("reading from persisted logs: %w", readPersistedLogErr)
@@ -602,33 +751,59 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 		lastEvictedTsNs := fs.filer.MetaAggregator.MetaLogBuffer.GetLastEvictedOriginalTsNs()
 		if diskAdvanced {
 			gapStall.resumed()
-			reportUnprovenAggregatedCrossing(cursorBeforeDiskTsNs, processedTsNs, lastEvictedTsNs, clientName, req.PathPrefix)
+			reportUnprovenAggregatedCrossing(cursorBeforeDiskTsNs, processedTsNs, lastEvictedTsNs, diskPassFlushLowTsNs, clientName, req.PathPrefix)
 			lastReadTime = log_buffer.NewMessagePosition(processedTsNs, gapResumeCursorOffset)
+			if processedTsNs > diskAnchorTsNs {
+				diskAnchorTsNs = processedTsNs
+			}
 		} else if readInMemoryLogErr == nil {
 			// Nothing on disk and memory never spoke: scan forward for the next
 			// day that has logs.
 			nextDayTs := util.GetNextDayTsNano(lastReadTime.Time.UnixNano())
-			position := log_buffer.NewMessagePosition(nextDayTs, gapResumeCursorOffset)
-			found, err := fs.filer.HasPersistedLogFiles(position)
-			if err != nil {
-				return fmt.Errorf("checking persisted log files: %w", err)
-			}
-			if found {
-				gapStall.resumed()
-				reportUnprovenAggregatedCrossing(cursorBeforeDiskTsNs, nextDayTs, lastEvictedTsNs, clientName, req.PathPrefix)
-				lastReadTime = position
+			// The day jump delivers nothing; stay put until the hold point
+			// covers the skipped range.
+			if nextDayTs <= diskPassHoldTsNs {
+				position := log_buffer.NewMessagePosition(nextDayTs, gapResumeCursorOffset)
+				found, err := fs.filer.HasPersistedLogFiles(position)
+				if err != nil {
+					return fmt.Errorf("checking persisted log files: %w", err)
+				}
+				if found {
+					gapStall.resumed()
+					reportUnprovenAggregatedCrossing(cursorBeforeDiskTsNs, nextDayTs, lastEvictedTsNs, diskPassFlushLowTsNs, clientName, req.PathPrefix)
+					lastReadTime = position
+					if nextDayTs > diskAnchorTsNs {
+						diskAnchorTsNs = nextDayTs
+					}
+				}
 			}
 		}

+		cursorBeforeResolveTsNs := lastReadTime.Time.UnixNano()
 		switch gaps.resolve(ctx, &lastReadTime, &readInMemoryLogErr, diskAdvanced) {
 		case gapDone:
 			return nil
 		case gapContinue:
+			// A cursor move here is a give-up or proven-empty skip (original space);
+			// anchor it so a later eviction rewind cannot undo the decision.
+			if ts := lastReadTime.Time.UnixNano(); ts != cursorBeforeResolveTsNs && ts > diskAnchorTsNs {
+				diskAnchorTsNs = ts
+			}
 			continue
 		}

+		// A held rewind must not re-park below the gap machinery's
+		// intentional skips.
+		if lastReadTime.Time.UnixNano() > deliveredUpToTsNs {
+			deliveredUpToTsNs = lastReadTime.Time.UnixNano()
+		}
+
 		glog.V(4).Infof("read in memory %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)

+		// Sampled before the read: a contiguous (unrefused) read delivers
+		// every event whose original timestamp is at or below this.
+		preMemDeliveryLowTsNs := fs.filer.MetaAggregator.PeerLowWatermarkTsNs()
+
 		lastReadTime, isDone, readInMemoryLogErr = fs.filer.MetaAggregator.MetaLogBuffer.LoopProcessLogData(aggReaderName, lastReadTime, req.UntilNs, func() bool {
 			select {
 			case <-ctx.Done():
@@ -638,13 +813,41 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 			if !fs.hasClient(req.ClientId, req.ClientEpoch) {
 				return false
 			}
+			// Contiguous and caught up: advance the anchor to the delivery
+			// low-watermark so long live tails keep eviction rewinds short.
+			// Only once the run is connected to the ring - the empty-ring
+			// wait lands here too, with disk files still unshipped below the
+			// cursor.
+			if memoryHoldsGap(lastReadTime.Time.UnixNano(), aggBuffer.GetLastEvictedOriginalTsNs()) {
+				if dl := fs.filer.MetaAggregator.PeerLowWatermarkTsNs(); dl > diskAnchorTsNs {
+					diskAnchorTsNs = dl
+				}
+			}
 			lastHeartbeatNs = fs.maybeSendIdleHeartbeat(req, sender, fs.filer.MetaAggregator.MetaLogBuffer, lastReadTime.Time.UnixNano(), lastSeenTsNs, lastHeartbeatNs)
 			return true
-		}, eachLogEntryFn)
+		}, memEachLogEntryFn)
 		if readInMemoryLogErr != nil {
+			if errors.Is(readInMemoryLogErr, errHeldByPeerWatermark) {
+				// The read was contiguous up to the hold: credit the anchor.
+				if preMemDeliveryLowTsNs > diskAnchorTsNs {
+					diskAnchorTsNs = preMemDeliveryLowTsNs
+				}
+				// The cursor already advanced onto the held entry; rewind to
+				// the last delivered entry so nothing in between is skipped.
+				lastReadTime = log_buffer.NewMessagePosition(deliveredUpToTsNs, gapResumeCursorOffset)
+				readInMemoryLogErr = nil
+				if !waitHeld() {
+					return nil
+				}
+				continue
+			}
 			if errors.Is(readInMemoryLogErr, log_buffer.ResumeFromDiskError) {
-				// Fell behind the ring: back to the disk pass, and from there to
-				// the gap resolution above if the disk has nothing either.
+				// Fell off the ring: resume the disk pass from the anchor, not
+				// the (possibly bumped) cursor - redelivery is within the
+				// at-least-once contract, skipping is not. For an anchored
+				// cursor this is a no-op, so the gap machinery's re-arm onto
+				// the retained window stays reachable.
+				lastReadTime = log_buffer.NewMessagePosition(diskAnchorTsNs, gapResumeCursorOffset)
 				continue
 			}
 			glog.Errorf("processed to %v: %v", lastReadTime, readInMemoryLogErr)
@@ -723,6 +926,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 	// written from this single goroutine, so no synchronization is needed.
 	var lastSeenTsNs int64
 	var lastHeartbeatNs int64
+	var lastFlushReportNs int64
 	baseEachLogEntryFn := eachLogEntryFn(req, sender, eachEventNotificationFn, &unsyncedEvents)
 	eachLogEntryFn := func(logEntry *filer_pb.LogEntry) (bool, error) {
 		lastSeenTsNs = logEntry.TsNs
@@ -822,6 +1026,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 				return false
 			}
 			lastHeartbeatNs = fs.maybeSendIdleHeartbeat(req, sender, fs.filer.LocalMetaLogBuffer, lastReadTime.Time.UnixNano(), lastSeenTsNs, lastHeartbeatNs)
+			lastFlushReportNs = fs.maybeSendFlushReport(req, sender, lastFlushReportNs)
 			return true
 		}, eachLogEntryFn)
 		if readInMemoryLogErr != nil {
@@ -887,6 +1092,25 @@ func eachLogEntryFn(req *filer_pb.SubscribeMetadataRequest, sender metadataStrea
 	}
 }

+// maybeSendFlushReport reports the local flush watermark to a subscriber that
+// opted into idle heartbeats (peer aggregators). Unlike heartbeats it is sent
+// regardless of catch-up state, and its TsNs stays zero so it never advances
+// delivery freshness.
+func (fs *FilerServer) maybeSendFlushReport(req *filer_pb.SubscribeMetadataRequest, sender metadataStreamSender, lastFlushReportNs int64) int64 {
+	if !req.ClientSupportsIdleHeartbeat || fs.filer == nil {
+		return lastFlushReportNs
+	}
+	now := time.Now().UnixNano()
+	if now-lastFlushReportNs < int64(idleHeartbeatInterval) {
+		return lastFlushReportNs
+	}
+	if err := sender.Send(&filer_pb.SubscribeMetadataResponse{FlushedTsNs: fs.filer.LocalFlushedThroughTsNs(now)}); err != nil {
+		glog.V(0).Infof("=> flush report to %s: %v", req.ClientName, err)
+		return lastFlushReportNs
+	}
+	return now
+}
+
 // maybeSendIdleHeartbeat emits an empty response carrying the current time when
 // the subscriber has consumed everything up to the buffer head. The client uses
 // it to advance freshness signals (e.g. filer.sync's sync_offset) without moving
@@ -919,7 +1143,23 @@ func (fs *FilerServer) maybeSendIdleHeartbeat(req *filer_pb.SubscribeMetadataReq
 	if now-lastHeartbeatNs < int64(idleHeartbeatInterval) {
 		return lastHeartbeatNs
 	}
-	if err := sender.Send(&filer_pb.SubscribeMetadataResponse{TsNs: now}); err != nil {
+	// On the local stream the heartbeat is a delivery claim to a peer
+	// aggregator and piggybacks the flush watermark; the aggregated ring
+	// never flushes, so its heartbeats carry neither. The claims are capped
+	// by the in-flight floor and fence later stamps above themselves, so
+	// re-checking the buffer head afterwards closes the append race: an
+	// event at or below the claim was in flight (capping it), stamped later
+	// (fenced above it), or already appended here - and then the head check
+	// proves this stream has sent it before the heartbeat.
+	heartbeat := &filer_pb.SubscribeMetadataResponse{TsNs: now}
+	if fs.filer != nil && logBuffer == fs.filer.LocalMetaLogBuffer {
+		heartbeat.TsNs = fs.filer.LocalDeliveredThroughTsNs(now)
+		heartbeat.FlushedTsNs = fs.filer.LocalFlushedThroughTsNs(now)
+		if logBuffer.LastTsNs.Load() > floorTsNs {
+			return lastHeartbeatNs
+		}
+	}
+	if err := sender.Send(heartbeat); err != nil {
 		glog.V(0).Infof("=> idle heartbeat to %s: %v", req.ClientName, err)
 		return lastHeartbeatNs
 	}
@@ -7,6 +7,7 @@ import (

 	dto "github.com/prometheus/client_model/go"

+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
@@ -15,8 +16,8 @@ import (
 // TestResolveGapResume pins the one decision both subscribe paths share: a gap
 // the disk read found empty may be skipped only when it is provably so - the
 // ring never evicted past the cursor, or the flush watermark observed before
-// the read had already passed the earliest in-memory time. The aggregated ring
-// never flushes, so it is the flushedTsNs=0 column of this table.
+// the read had already passed the earliest in-memory time. The flushedTsNs=0
+// column models an aggregated pass whose peers have not proven anything yet.
 func TestResolveGapResume(t *testing.T) {
 	now := time.Date(2026, 6, 29, 12, 0, 0, 0, time.UTC).UnixNano()
 	ago := func(d time.Duration) int64 { return now - int64(d) }
@@ -160,6 +161,81 @@ func TestResolveGapResume(t *testing.T) {
 	}
 }

+// TestGapPassProvenEmptyCrossing pins the crossing that needs no memory to
+// land in: an empty disk pass proven through the eviction watermark crosses to
+// it silently - no park, no loss counter. This is the only exit past the
+// aggregated ring's pre-subscription mark, which no rotation ever proves.
+func TestGapPassProvenEmptyCrossing(t *testing.T) {
+	now := time.Date(2026, 6, 29, 12, 0, 0, 0, time.UTC).UnixNano()
+	evicted := now - int64(time.Minute) // the ring's startFrom mark
+	cursorTs := evicted - int64(2*time.Minute)
+
+	crossings := func() float64 {
+		m := &dto.Metric{}
+		if err := stats.FilerSubscribeUnprovenGapCrossings.WithLabelValues("aggregated").Write(m); err != nil {
+			t.Fatal(err)
+		}
+		return m.GetCounter().GetValue()
+	}
+
+	proven := evicted
+	p := &gapPass{
+		gapStall: &gapStallReporter{scope: "aggregated", clientName: "c", pathPrefix: "/"},
+		earliest: func() time.Time { return time.Time{} }, // ring still empty
+		evicted:  func() int64 { return evicted },
+		flushed:  func() int64 { return proven },
+	}
+
+	cursor := log_buffer.NewMessagePosition(cursorTs, gapResumeCursorOffset)
+	latch := error(log_buffer.ResumeFromDiskError) // stale latch must not survive the move
+	start := crossings()
+	if got := p.resolve(context.Background(), &cursor, &latch, false); got != gapContinue {
+		t.Fatalf("outcome = %v, want gapContinue", got)
+	}
+	if got := cursor.Time.UnixNano(); got != evicted {
+		t.Fatalf("cursor = %v, want the proven watermark %v", time.Unix(0, got), time.Unix(0, evicted))
+	}
+	if latch != nil {
+		t.Fatalf("latch not cleared: %v", latch)
+	}
+	if got := crossings(); got != start {
+		t.Fatalf("proven crossing moved the loss counter by %v", got-start)
+	}
+
+	// diskAdvanced defers everything: the disk may hold more of the gap.
+	cursor = log_buffer.NewMessagePosition(cursorTs, gapResumeCursorOffset)
+	if got := p.resolve(context.Background(), &cursor, &latch, true); got != gapContinue {
+		t.Fatalf("diskAdvanced: outcome = %v, want gapContinue", got)
+	}
+	if got := cursor.Time.UnixNano(); got != cursorTs {
+		t.Fatalf("diskAdvanced: cursor moved to %v", time.Unix(0, got))
+	}
+}
+
+// TestChunkRefsStopTsNs pins the chunk listing bound: the newest admitted file
+// name sits a minute plus a flush interval below the hold; UntilNs caps it.
+func TestChunkRefsStopTsNs(t *testing.T) {
+	hold := time.Date(2026, 6, 29, 12, 10, 45, 500, time.UTC).UnixNano()
+
+	stop := chunkRefsStopTsNs(hold, 0)
+	newestFileTsNs := stop - stop%int64(time.Minute)
+	if newestFileTsNs+int64(time.Minute)+int64(filer.LogFlushInterval) > hold {
+		t.Fatalf("file named %v may hold entries past the hold %v",
+			time.Unix(0, newestFileTsNs), time.Unix(0, hold))
+	}
+	want := time.Date(2026, 6, 29, 12, 8, 59, 999999999, time.UTC).UnixNano()
+	if stop != want {
+		t.Fatalf("stop = %v, want %v", time.Unix(0, stop), time.Unix(0, want))
+	}
+
+	if got := chunkRefsStopTsNs(hold, want-5); got != want-5 {
+		t.Fatalf("UntilNs cap: got %v, want %v", time.Unix(0, got), time.Unix(0, want-5))
+	}
+	if got := chunkRefsStopTsNs(hold, want+5); got != want {
+		t.Fatalf("UntilNs above the bound must not widen it: got %v", time.Unix(0, got))
+	}
+}
+
 // TestInclusiveDiskCursorOnWatermarkStillAdvances pins the disk-to-memory
 // handoff. A cursor built from a disk position stays inclusive, because memory
 // above the eviction watermark may hold a different entry sharing that
@@ -361,17 +437,21 @@ func TestReportUnprovenAggregatedCrossing(t *testing.T) {

 	start := crossings()
 	// Nothing evicted: no range to cross.
-	reportUnprovenAggregatedCrossing(before, after, 0, "c", "/")
+	reportUnprovenAggregatedCrossing(before, after, 0, 0, "c", "/")
 	// Cursor already past the watermark: the evicted range was behind it.
-	reportUnprovenAggregatedCrossing(evicted, after, evicted, "c", "/")
+	reportUnprovenAggregatedCrossing(evicted, after, evicted, 0, "c", "/")
 	// Cursor still short of the watermark: the gap is open, not crossed.
-	reportUnprovenAggregatedCrossing(before, evicted-1, evicted, "c", "/")
+	reportUnprovenAggregatedCrossing(before, evicted-1, evicted, 0, "c", "/")
+	// Crossed, but every peer's flush watermark already passed the eviction
+	// boundary: the crossed range was fully on peer disks — proven, not counted.
+	reportUnprovenAggregatedCrossing(before, after, evicted, evicted, "c", "/")
 	if got := crossings(); got != start {
 		t.Fatalf("counter moved by %v on advances that cross nothing", got-start)
 	}

-	// From below the watermark to above it: unproven.
-	reportUnprovenAggregatedCrossing(before, after, evicted, "c", "/")
+	// From below the watermark to above it, flush watermark short of the
+	// boundary: unproven.
+	reportUnprovenAggregatedCrossing(before, after, evicted, evicted-1, "c", "/")
 	if got := crossings(); got != start+1 {
 		t.Fatalf("counter = %v, want %v after one unproven crossing", got, start+1)
 	}
@@ -0,0 +1,271 @@
+package weed_server
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
+)
+
+// sentRecord snapshots what a Send delivered at call time - the sender clears
+// the envelope's Events after a batched Send, so asserting on the retained
+// pointer would miss the nesting.
+type sentRecord struct {
+	hasNotification bool
+	tsNs            int64
+	flushedTsNs     int64
+	nested          int
+	nestedControl   int
+}
+
+// gatedRecordingStream blocks its first Send until the gate opens, letting a
+// test queue several messages into the pipelined sender deterministically.
+type gatedRecordingStream struct {
+	gate     chan struct{}
+	gateOnce sync.Once
+	records  []sentRecord
+}
+
+func (s *gatedRecordingStream) Send(msg *filer_pb.SubscribeMetadataResponse) error {
+	rec := sentRecord{
+		hasNotification: msg.EventNotification != nil,
+		tsNs:            msg.TsNs,
+		flushedTsNs:     msg.FlushedTsNs,
+		nested:          len(msg.Events),
+	}
+	for _, nested := range msg.Events {
+		if nested.EventNotification == nil || nested.FlushedTsNs != 0 {
+			rec.nestedControl++
+		}
+	}
+	s.records = append(s.records, rec)
+	s.gateOnce.Do(func() { <-s.gate })
+	return nil
+}
+
+// TestPipelinedSenderControlMessagesNeverNested pins the batching rule flush
+// reports rely on: a control message (nil EventNotification - a flush report
+// or an idle heartbeat) must never ride in a batch's Events tail, where the
+// peer aggregator's nil-guard would drop its watermark state. A starved flush
+// watermark looks stalled, and the settled-horizon escape would then allow
+// reads past it.
+func TestPipelinedSenderControlMessagesNeverNested(t *testing.T) {
+	stream := &gatedRecordingStream{gate: make(chan struct{})}
+	sender := newPipelinedSender(stream, 16, true)
+
+	oldTs := time.Now().Add(-time.Hour).UnixNano()
+	if err := sender.Send(makeEvent("/d", "e1", oldTs)); err != nil {
+		t.Fatalf("send e1: %v", err)
+	}
+	// While the stream is blocked on e1, queue a batchable backlog event, a
+	// flush report, and another event - the drain loop sees them together.
+	if err := sender.Send(makeEvent("/d", "e2", oldTs+1)); err != nil {
+		t.Fatalf("send e2: %v", err)
+	}
+	if err := sender.Send(&filer_pb.SubscribeMetadataResponse{FlushedTsNs: 123}); err != nil {
+		t.Fatalf("send flush report: %v", err)
+	}
+	if err := sender.Send(makeEvent("/d", "e3", oldTs+2)); err != nil {
+		t.Fatalf("send e3: %v", err)
+	}
+	close(stream.gate)
+	if err := sender.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	events, flushReports := 0, 0
+	for _, rec := range stream.records {
+		if rec.nestedControl > 0 {
+			t.Fatalf("control message nested in a batch: %+v", rec)
+		}
+		if rec.hasNotification {
+			events += 1 + rec.nested
+		}
+		if rec.flushedTsNs != 0 {
+			flushReports++
+			if rec.hasNotification || rec.nested != 0 {
+				t.Fatalf("flush report not sent solo: %+v", rec)
+			}
+			if rec.flushedTsNs != 123 {
+				t.Fatalf("flush report watermark = %d, want 123", rec.flushedTsNs)
+			}
+		}
+	}
+	if events != 3 {
+		t.Fatalf("delivered %d events, want 3", events)
+	}
+	if flushReports != 1 {
+		t.Fatalf("delivered %d flush reports, want 1", flushReports)
+	}
+}
+
+// TestIdleHeartbeatCappedByInflightStamp pins the delivery-claim cap: an idle
+// heartbeat on the LOCAL stream is a delivery-completeness claim the peer
+// aggregator turns into its delivery low-watermark, so it must not claim past
+// an event that is stamped but not yet appended to the local buffer.
+func TestIdleHeartbeatCappedByInflightStamp(t *testing.T) {
+	lb := log_buffer.NewLogBuffer("hb-cap-test", time.Minute, nil, nil, nil)
+	defer lb.ShutdownLogBuffer()
+	f := &filer.Filer{LocalMetaLogBuffer: lb}
+	fs := &FilerServer{filer: f}
+	req := &filer_pb.SubscribeMetadataRequest{ClientSupportsIdleHeartbeat: true}
+
+	// Caught up, nothing in flight: heartbeat carries a current timestamp.
+	s := &collectingStream{}
+	got := fs.maybeSendIdleHeartbeat(req, s, lb, 0, 0, 0)
+	if len(s.messages) != 1 || s.messages[0].TsNs <= 0 {
+		t.Fatalf("baseline heartbeat missing: msgs=%d", len(s.messages))
+	}
+	if got != s.messages[0].TsNs {
+		// pacing returns the wall time it sent at; the claim may be lower
+		// only when something is in flight, which it is not here
+		t.Fatalf("baseline: returned %d, sent claim %d", got, s.messages[0].TsNs)
+	}
+
+	// An in-flight stamp caps the claim just below the stamp.
+	stampTsNs := f.StampMetaLogInflightForTest()
+	s = &collectingStream{}
+	fs.maybeSendIdleHeartbeat(req, s, lb, 0, 0, 0)
+	if len(s.messages) != 1 {
+		t.Fatalf("capped heartbeat missing: msgs=%d", len(s.messages))
+	}
+	if s.messages[0].TsNs != stampTsNs-1 {
+		t.Fatalf("heartbeat claim=%d, want in-flight floor %d", s.messages[0].TsNs, stampTsNs-1)
+	}
+	if s.messages[0].FlushedTsNs != stampTsNs-1 {
+		t.Fatalf("flush claim=%d, want in-flight floor %d", s.messages[0].FlushedTsNs, stampTsNs-1)
+	}
+	f.DoneMetaLogInflightForTest(stampTsNs)
+}
+
+// TestResolveAggReadHoldTsNs pins the aggregated delivery hold point: a
+// subscriber may not read past the peers' delivery low-watermark (a source
+// that is still catching up may merge older events in late), except that a
+// watermark stalled beyond the settled horizon stops holding delivery back.
+func TestResolveAggReadHoldTsNs(t *testing.T) {
+	now := time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC).UnixNano()
+	horizon := metadataGapSettledHorizon
+	ago := func(d time.Duration) int64 { return now - int64(d) }
+
+	cases := []struct {
+		name        string
+		watermarkNs int64
+		wantHoldNs  int64
+	}{
+		{
+			// Healthy: all peers signalled recently → deliver up to the watermark.
+			name:        "healthy watermark ahead of horizon",
+			watermarkNs: ago(2 * time.Second),
+			wantHoldNs:  ago(2 * time.Second),
+		},
+		{
+			// A peer stalled long ago: liveness escape takes over at the horizon.
+			name:        "stalled watermark falls back to horizon",
+			watermarkNs: ago(30 * time.Minute),
+			wantHoldNs:  ago(horizon),
+		},
+		{
+			// No signal from some peer yet (fresh cluster join): completeness
+			// unknown → horizon bounds the hold.
+			name:        "unknown watermark falls back to horizon",
+			watermarkNs: 0,
+			wantHoldNs:  ago(horizon),
+		},
+		{
+			// Watermark exactly at the horizon: either bound gives the same
+			// answer; pin the equality behavior.
+			name:        "watermark exactly at horizon",
+			watermarkNs: ago(horizon),
+			wantHoldNs:  ago(horizon),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := resolveAggReadHoldTsNs(tc.watermarkNs, now, horizon)
+			if got != tc.wantHoldNs {
+				t.Fatalf("hold=%d want %d", got, tc.wantHoldNs)
+			}
+		})
+	}
+}
+
+func TestPreviousMinuteEndTsNs(t *testing.T) {
+	ts := time.Date(2026, 7, 29, 12, 34, 56, 789, time.UTC)
+	want := time.Date(2026, 7, 29, 12, 33, 59, 999999999, time.UTC).UnixNano()
+	if got := previousMinuteEndTsNs(ts.UnixNano()); got != want {
+		t.Fatalf("got %d want %d", got, want)
+	}
+	// A timestamp exactly on a minute boundary bounds to the end of the
+	// preceding minute.
+	onBoundary := time.Date(2026, 7, 29, 12, 34, 0, 0, time.UTC)
+	wantBoundary := time.Date(2026, 7, 29, 12, 33, 59, 999999999, time.UTC).UnixNano()
+	if got := previousMinuteEndTsNs(onBoundary.UnixNano()); got != wantBoundary {
+		t.Fatalf("boundary: got %d want %d", got, wantBoundary)
+	}
+}
+
+// TestAggWatermarkHoldRewindRedelivers pins the rewind invariant SubscribeMetadata
+// relies on: when a read is held at an entry (errHeldByPeerWatermark), resuming
+// from heldAtTsNs-1 re-delivers exactly the held entry and nothing before it —
+// positions are exclusive, so the -1 keeps the held entry ahead of the cursor.
+func TestAggWatermarkHoldRewindRedelivers(t *testing.T) {
+	lb := log_buffer.NewLogBuffer("agg-hold", 10*time.Minute,
+		func(logBuffer *log_buffer.LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {
+		},
+		func(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (log_buffer.MessagePosition, bool, error) {
+			return startPosition, false, nil
+		},
+		func() {})
+	defer lb.ShutdownLogBuffer()
+
+	base := time.Now().Add(-time.Second)
+	ts1 := base.UnixNano()
+	ts2 := base.Add(100 * time.Millisecond).UnixNano()
+	for i, ts := range []int64{ts1, ts2} {
+		if err := lb.AddLogEntryToBuffer(&filer_pb.LogEntry{TsNs: ts, Data: []byte{byte(i)}, Key: []byte("k"), Offset: int64(i)}); err != nil {
+			t.Fatalf("add %d: %v", i, err)
+		}
+	}
+
+	holdTsNs := ts1 // ts2 is beyond the hold point
+	var heldAtTsNs int64
+	var delivered []int64
+	guarded := func(logEntry *filer_pb.LogEntry) (bool, error) {
+		if logEntry.TsNs > holdTsNs {
+			heldAtTsNs = logEntry.TsNs
+			return false, errHeldByPeerWatermark
+		}
+		delivered = append(delivered, logEntry.TsNs)
+		return false, nil
+	}
+
+	start := log_buffer.NewMessagePosition(ts1-1, -2)
+	_, _, err := lb.LoopProcessLogData("hold-test", start, 0, func() bool { return false }, guarded)
+	if !errors.Is(err, errHeldByPeerWatermark) {
+		t.Fatalf("want held error, got %v", err)
+	}
+	if len(delivered) != 1 || delivered[0] != ts1 {
+		t.Fatalf("before hold: delivered %v, want [%d]", delivered, ts1)
+	}
+	if heldAtTsNs != ts2 {
+		t.Fatalf("heldAt=%d want %d", heldAtTsNs, ts2)
+	}
+
+	// Release the hold and resume from just below the held entry: it must be
+	// re-delivered exactly once, without re-delivering ts1.
+	holdTsNs = ts2
+	delivered = nil
+	resume := log_buffer.NewMessagePosition(heldAtTsNs-1, -2)
+	_, _, err = lb.LoopProcessLogData("hold-test-resume", resume, 0, func() bool { return false }, guarded)
+	if err != nil && !errors.Is(err, log_buffer.ResumeFromDiskError) {
+		t.Fatalf("resume: %v", err)
+	}
+	if len(delivered) != 1 || delivered[0] != ts2 {
+		t.Fatalf("after release: delivered %v, want [%d]", delivered, ts2)
+	}
+}
@@ -941,6 +941,20 @@ func (logBuffer *LogBuffer) GetEarliestPosition() MessagePosition {
 	}
 }

+// FlushedThroughTsNs reports the timestamp through which this buffer's data
+// is durably on disk: "now" when nothing is pending a flush (a later append
+// is bumped past the head), otherwise the last flushed window's stop time.
+// Covers only entries that reached the buffer - callers stamping timestamps
+// before the append must also bound by their in-flight floor (see
+// Filer.LocalFlushedThroughTsNs).
+func (logBuffer *LogBuffer) FlushedThroughTsNs(nowNs int64) int64 {
+	flushed := logBuffer.lastFlushTsNs.Load()
+	if logBuffer.LastTsNs.Load() <= flushed {
+		return nowNs
+	}
+	return flushed
+}
+
 // GetLastFlushTsNs returns the latest flushed timestamp in Unix nanoseconds.
 // Returns 0 if nothing has been flushed yet.
 func (logBuffer *LogBuffer) GetLastFlushTsNs() int64 {
@@ -961,6 +975,18 @@ func (logBuffer *LogBuffer) GetLastEvictedTsNs() int64 {
 	return logBuffer.lastEvictedTsNs.Load()
 }

+// MarkEvictedThrough treats entries at or below tsNs as evicted even though
+// nothing was rotated out yet: a merge-fed buffer is born empty while its
+// sources hold history it must not skip. Call before the first append.
+func (logBuffer *LogBuffer) MarkEvictedThrough(tsNs int64) {
+	if tsNs > logBuffer.lastEvictedTsNs.Load() {
+		logBuffer.lastEvictedTsNs.Store(tsNs)
+	}
+	if tsNs > logBuffer.lastEvictedOriginalTsNs.Load() {
+		logBuffer.lastEvictedOriginalTsNs.Store(tsNs)
+	}
+}
+
 func (logBuffer *LogBuffer) SetLastFlushTsNs(ts int64) {
 	logBuffer.lastFlushTsNs.Store(ts)
 }
@@ -273,6 +273,46 @@ func TestEvictionGatedCursor(t *testing.T) {
 	}
 }

+// TestMarkEvictedThroughGatesYoungRing pins the young-ring gate: with the mark
+// set, a below-window gated cursor goes to disk before the first real eviction
+// instead of being served the earliest retained entry.
+func TestMarkEvictedThroughGatesYoungRing(t *testing.T) {
+	lb := NewLogBuffer("young-ring", time.Minute, nil, nil, nil)
+	defer lb.ShutdownLogBuffer()
+
+	seed := time.Now().Add(-time.Minute).UnixNano()
+	lb.MarkEvictedThrough(seed)
+	if got := lb.GetLastEvictedTsNs(); got != seed {
+		t.Fatalf("evicted through %v, want the mark %v", time.Unix(0, got), time.Unix(0, seed))
+	}
+	if got := lb.GetLastEvictedOriginalTsNs(); got != seed {
+		t.Fatalf("original watermark %v, want the mark %v", time.Unix(0, got), time.Unix(0, seed))
+	}
+	// A lower mark must not regress an established boundary.
+	lb.MarkEvictedThrough(seed - int64(time.Hour))
+	if got := lb.GetLastEvictedTsNs(); got != seed {
+		t.Fatalf("lower mark regressed the watermark to %v", time.Unix(0, got))
+	}
+
+	// One post-mark arrival: the ring's window starts well above the mark.
+	if err := lb.AddLogEntryToBuffer(&filer_pb.LogEntry{
+		TsNs: time.Now().UnixNano(), Data: []byte("x"), Key: []byte("k"),
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Below the mark: gated goes to disk, -2 keeps serving (MQ contract).
+	if _, _, _, err := lb.ReadFromBuffer(NewMessagePosition(seed-1, EvictionGatedOffset)); err != ResumeFromDiskError {
+		t.Fatalf("gated below the mark: want ResumeFromDiskError, got %v", err)
+	}
+	if buf, _, _, err := lb.ReadFromBuffer(NewMessagePosition(seed-1, -2)); err != nil || buf == nil {
+		t.Fatalf("plain sentinel below the mark: buf=%v err=%v", buf != nil, err)
+	}
+	if buf, _, _, err := lb.ReadFromBuffer(NewMessagePosition(seed, EvictionGatedOffset)); err != nil || buf == nil {
+		t.Fatalf("gated at the mark: buf=%v err=%v", buf != nil, err)
+	}
+}
+
 // TestEvictionOriginalWatermark pins the second timestamp space. The ring bumps
 // an out-of-order arrival past its head, so a bump-heavy interval (peer history
 // replay) leaves stopTimes above anything on any peer's disk; a gap gate