filer: stop aggregated metadata subscribers from spinning on a peer watermark hold (#10863)

* fix(filer): stop logging a held aggregated read as an error

An aggregated subscriber may not read past the peers' low-watermark, and
it stops at the first entry beyond it by returning a sentinel from the
read callback. LoopProcessLogData logs every callback error, so on a
cluster that keeps writing - where there is almost always an entry newer
than the watermark - every read wrote an ERROR line naming the entry it
stopped at, thousands per minute per filer.

Mark the stop as control flow: an error wrapping StopReadingError is
handed back to the caller unlogged, and the held-read sentinel wraps it.

* fix(filer): release an aggregated watermark hold on peer progress

A held read waited on the aggregated buffer's data channel, which the
next write signalled - but a write cannot release a hold, only a peer
reporting further progress can. On a cluster that keeps writing the loop
therefore re-ran a whole pass per arriving event, log file listing and
all, and held again on the same entry every time.

Signal held readers from the meta aggregator instead, whenever a
low-watermark rises: a peer reporting, or one dropped past its removal
grace. The retry interval stays as the backstop for what no watermark
covers. Count the holds so a parked subscriber stays visible.

* fix(filer): floor how often an aggregated watermark hold releases

Peers advance their delivery watermark on every event they stream, so
releasing a hold on every advance is the same pass-per-event storm as
releasing on every write, just without the log lines - and each pass
lists a day of log files.

Floor the release at 20ms. Advances inside the floor collapse into one
release, which then delivers everything they covered.

* fix(filer): pace a peer's delivery claim by what its subscribers hold at

A filer's local metadata stream carries an idle heartbeat to its peer
aggregators, and each peer turns it into that filer's delivery
low-watermark. Aggregated subscribers hold at the minimum across peers,
so a filer quiet enough to fall back on the heartbeat parked every
subscriber in the cluster up to a keepalive interval - 5 seconds -
behind live writes. With nine filers, most of them quiet at any moment,
the minimum sat there permanently.

Pace that heartbeat at 200ms once the filer has peers. It stays a
keepalive, at the keepalive interval, for a filer with none.

* fix(filer): wake each aggregated hold on its own watermark

A persisted-log read is held by what the peers have flushed, an
in-memory read by what they have delivered, but both parked on one
channel closed whenever either minimum rose. Peers advance their
delivery watermark on every event they stream, so a flush-held reader
woke at the coalescing floor to re-list a day of log files and park
again on the same entry - the storm this set out to fix, in the one
place asymmetric peer progress still reached.

Signal the two separately and park each read on the one that bounds it.

weed/filer/meta_aggregator.go

@@ -49,8 +49,19 @@ type MetaAggregator struct {
 	// 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
+	peerRemovedAtNs map[pb.ServerAddress]int64
+	// lowWatermarkTsNs and lowFlushWatermarkTsNs are the last minima signalled
+	// on deliveryAdvanced and flushAdvanced, each closed and replaced when its
+	// own minimum rises. Held readers park on the one that bounds them:
+	// arriving data cannot release a watermark hold, only a peer reporting
+	// the progress that hold waits on. The two are kept apart because a
+	// delivery advance - one per event a peer streams - cannot release a read
+	// held at the flush watermark, and waking it costs a whole pass.
+	lowWatermarkTsNs      int64
+	lowFlushWatermarkTsNs int64
+	deliveryAdvanced      chan struct{}
+	flushAdvanced         chan struct{}
+	peerWatermarksLock    sync.Mutex
 }
 
 // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk.
@@ -116,6 +127,7 @@ func (ma *MetaAggregator) initPeerWatermark(peer pb.ServerAddress) {
 		ma.peerFlushWatermarks[peer] = 0
 	}
 	delete(ma.peerRemovedAtNs, peer)
+	ma.noteLowWatermarksLocked()
 }
 
 func (ma *MetaAggregator) markPeerWatermarkRemoved(peer pb.ServerAddress) {
@@ -144,6 +156,8 @@ func (ma *MetaAggregator) dropExpiredRemovedPeersLocked() {
 			delete(ma.peerFlushWatermarks, peer)
 		}
 	}
+	// Dropping the peer that pinned a minimum raises it, same as it reporting.
+	ma.noteLowWatermarksLocked()
 }
 
 // advancePeerWatermark records the peer as received-through tsNs. Monotonic,
@@ -154,6 +168,7 @@ func (ma *MetaAggregator) advancePeerWatermark(peer pb.ServerAddress, tsNs int64
 	defer ma.peerWatermarksLock.Unlock()
 	if cur, found := ma.peerWatermarks[peer]; found && tsNs > cur {
 		ma.peerWatermarks[peer] = tsNs
+		ma.noteLowWatermarksLocked()
 	}
 }
 
@@ -164,6 +179,7 @@ func (ma *MetaAggregator) advancePeerFlushWatermark(peer pb.ServerAddress, tsNs
 	defer ma.peerWatermarksLock.Unlock()
 	if cur, found := ma.peerFlushWatermarks[peer]; found && tsNs > cur {
 		ma.peerFlushWatermarks[peer] = tsNs
+		ma.noteLowWatermarksLocked()
 	}
 }
 
@@ -174,16 +190,7 @@ 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
+	return lowWatermarkOf(ma.peerFlushWatermarks)
 }
 
 // PeerLowWatermarkTsNs returns the minimum received-through timestamp across
@@ -194,11 +201,67 @@ func (ma *MetaAggregator) PeerLowWatermarkTsNs() int64 {
 	ma.peerWatermarksLock.Lock()
 	defer ma.peerWatermarksLock.Unlock()
 	ma.dropExpiredRemovedPeersLocked()
-	if len(ma.peerWatermarks) == 0 {
+	return lowWatermarkOf(ma.peerWatermarks)
+}
+
+// DeliveryWatermarkAdvancedChan returns a channel closed the next time the
+// delivery low-watermark rises, which is what releases an in-memory read held
+// at it. PeerLowFlushWatermarkTsNs has FlushWatermarkAdvancedChan. Callers
+// must take the channel before reading the watermark they hold at, so a rise
+// in between wakes them instead of being missed.
+func (ma *MetaAggregator) DeliveryWatermarkAdvancedChan() <-chan struct{} {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	if ma.deliveryAdvanced == nil {
+		ma.deliveryAdvanced = make(chan struct{})
+	}
+	return ma.deliveryAdvanced
+}
+
+// FlushWatermarkAdvancedChan returns a channel closed the next time the flush
+// low-watermark rises, which is what releases a persisted-log read held at it.
+func (ma *MetaAggregator) FlushWatermarkAdvancedChan() <-chan struct{} {
+	ma.peerWatermarksLock.Lock()
+	defer ma.peerWatermarksLock.Unlock()
+	if ma.flushAdvanced == nil {
+		ma.flushAdvanced = make(chan struct{})
+	}
+	return ma.flushAdvanced
+}
+
+// noteLowWatermarksLocked recomputes both minima and wakes the readers parked
+// on each one that rose. Caller must hold peerWatermarksLock.
+func (ma *MetaAggregator) noteLowWatermarksLocked() {
+	low, flushLow := lowWatermarkOf(ma.peerWatermarks), lowWatermarkOf(ma.peerFlushWatermarks)
+	// Falls are recorded too (a joining peer sits at 0 until it signals), so
+	// the climb back out of one is seen as a rise.
+	if low > ma.lowWatermarkTsNs {
+		ma.deliveryAdvanced = closeWatermarkChan(ma.deliveryAdvanced)
+	}
+	if flushLow > ma.lowFlushWatermarkTsNs {
+		ma.flushAdvanced = closeWatermarkChan(ma.flushAdvanced)
+	}
+	ma.lowWatermarkTsNs, ma.lowFlushWatermarkTsNs = low, flushLow
+}
+
+// closeWatermarkChan wakes everyone parked on ch and clears it, so the next
+// caller parks on a fresh one.
+func closeWatermarkChan(ch chan struct{}) chan struct{} {
+	if ch != nil {
+		close(ch)
+	}
+	return nil
+}
+
+// lowWatermarkOf returns the minimum across the tracked peers, or 0 when none
+// are tracked: a peer that has not signalled sits at 0 and pins the minimum
+// there, which is what makes completeness unknown.
+func lowWatermarkOf(watermarks map[pb.ServerAddress]int64) int64 {
+	if len(watermarks) == 0 {
 		return 0
 	}
 	var low int64 = math.MaxInt64
-	for _, tsNs := range ma.peerWatermarks {
+	for _, tsNs := range watermarks {
 		if tsNs < low {
 			low = tsNs
 		}

weed/filer/meta_aggregator_testing.go

@@ -0,0 +1,23 @@
+package filer
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// TrackPeerForTesting registers a peer as if the master had announced it, but
+// without starting its subscription goroutine, so loop tests can drive the
+// low-watermarks by hand. Test support only.
+func (ma *MetaAggregator) TrackPeerForTesting(peer pb.ServerAddress) {
+	ma.peerChansLock.Lock()
+	ma.peerChans[peer] = make(chan struct{})
+	ma.peerChansLock.Unlock()
+	ma.initPeerWatermark(peer)
+}
+
+// ReportPeerWatermarksForTesting stands in for what a peer's stream reports:
+// its delivery watermark (events and idle heartbeats) and its flush
+// watermark. Test support only.
+func (ma *MetaAggregator) ReportPeerWatermarksForTesting(peer pb.ServerAddress, deliveredTsNs, flushedTsNs int64) {
+	ma.advancePeerWatermark(peer, deliveredTsNs)
+	ma.advancePeerFlushWatermark(peer, flushedTsNs)
+}

weed/filer/meta_aggregator_watermark_test.go

@@ -147,3 +147,67 @@ func TestPeerFlushWatermarkBookkeeping(t *testing.T) {
 	}
 	_ = a
 }
+
+// closed reports whether ch has been signalled, without blocking.
+func closed(ch <-chan struct{}) bool {
+	select {
+	case <-ch:
+		return true
+	default:
+		return false
+	}
+}
+
+// TestWatermarkAdvancedChansAreScoped pins what each wake channel promises. A
+// held read re-runs a whole pass - a persisted-log listing in it - so waking
+// one whose bound did not move is pure waste, and peers advance their delivery
+// watermark on every event they stream.
+func TestWatermarkAdvancedChansAreScoped(t *testing.T) {
+	ma := newTestAggregator()
+	a, b := pb.ServerAddress("filer-a:8888"), pb.ServerAddress("filer-b:8888")
+	ma.initPeerWatermark(a)
+	ma.initPeerWatermark(b)
+
+	delivery, flush := ma.DeliveryWatermarkAdvancedChan(), ma.FlushWatermarkAdvancedChan()
+
+	// One peer alone does not move either minimum: the other still pins both.
+	ma.advancePeerWatermark(a, 100)
+	ma.advancePeerFlushWatermark(a, 100)
+	if closed(delivery) || closed(flush) {
+		t.Fatalf("a single peer moved a minimum: delivery=%v flush=%v", closed(delivery), closed(flush))
+	}
+
+	// Delivery progress on every peer wakes the in-memory holds only.
+	ma.advancePeerWatermark(b, 50)
+	if !closed(delivery) {
+		t.Fatal("delivery low-watermark rose without waking the in-memory holds")
+	}
+	if closed(flush) {
+		t.Fatal("delivery progress woke the persisted-log holds, which it cannot release")
+	}
+
+	// And flush progress the persisted-log holds only.
+	delivery = ma.DeliveryWatermarkAdvancedChan()
+	ma.advancePeerFlushWatermark(b, 50)
+	if !closed(flush) {
+		t.Fatal("flush low-watermark rose without waking the persisted-log holds")
+	}
+	if closed(delivery) {
+		t.Fatal("flush progress woke the in-memory holds, which it cannot release")
+	}
+
+	// A fresh channel is handed out after each signal, so the next park is on
+	// the next rise rather than on one already consumed.
+	flush = ma.FlushWatermarkAdvancedChan()
+	if closed(flush) {
+		t.Fatal("re-handed a signalled channel")
+	}
+	ma.advancePeerFlushWatermark(a, 200)
+	if closed(flush) {
+		t.Fatal("one peer above the minimum signalled a rise")
+	}
+	ma.advancePeerFlushWatermark(b, 200)
+	if !closed(flush) {
+		t.Fatal("the trailing peer catching up did not signal a rise")
+	}
+}

weed/server/filer_grpc_server_sub_meta.go

@@ -50,6 +50,21 @@ const (
 	// newer. It keeps freshness signals such as filer.sync's sync_offset metric
 	// from looking stuck during read-only periods on the source.
 	idleHeartbeatInterval = 5 * time.Second
+
+	// peerDeliveryClaimInterval paces that heartbeat on the local stream of a
+	// filer with peers, where it is not a keepalive but a delivery claim: the
+	// peer aggregator turns it into its delivery low-watermark, and every
+	// aggregated subscriber in the cluster holds at the minimum across peers.
+	// So this, not idleHeartbeatInterval, is how far behind live writes a
+	// quiet filer leaves them. Rounded up to the reader's own poll interval.
+	peerDeliveryClaimInterval = 200 * time.Millisecond
+
+	// heldWakeFloor coalesces hold releases. Peers advance their watermarks
+	// per event they stream, and each release costs a whole pass - a log file
+	// listing included - so releasing on every advance turns a busy cluster
+	// into a listing storm. It is added to delivery latency, so it stays well
+	// under the claim interval that already bounds it.
+	heldWakeFloor = 20 * time.Millisecond
 )
 
 // metadataStreamSender is satisfied by both gRPC stream types and pipelinedSender.
@@ -245,8 +260,9 @@ func memoryHoldsGap(currentTsNs, lastEvictedTsNs int64) bool {
 
 // 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")
+// re-listing is what picks up a late-landing log file). Holding is the normal
+// state on a cluster that keeps writing, hence the quiet stop.
+var errHeldByPeerWatermark = fmt.Errorf("held by aggregated peer watermark: %w", log_buffer.StopReadingError)
 
 // resolveAggReadHoldTsNs bounds how far an aggregated subscriber may read: a
 // cursor that passes T before every source has provably made T visible loses
@@ -654,15 +670,28 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 	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 {
+	// waitHeld pauses a held read until a peer reports further progress, or
+	// the retry interval elapses (a peer dropped past its grace, or a log file
+	// landing that no watermark covers). Arriving data is deliberately not a
+	// wake-up: on a cluster that keeps writing there is always an entry past
+	// the hold, so waking on it spins the loop without ever releasing the
+	// hold. The channel is the one for this read's own watermark - a delivery
+	// advance cannot release a flush-held read - and was taken before the pass
+	// sampled it, so a rise in between wakes us here instead of being missed.
+	// False: context ended.
+	waitHeld := func(scope string, watermarkChan <-chan struct{}) bool {
+		stats.FilerSubscribeWatermarkHolds.WithLabelValues(scope).Inc()
 		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(heldWakeFloor):
+		}
+		select {
+		case <-watermarkChan:
 		case <-ctx.Done():
 			return false
 		case <-time.After(unflushedGapRetryInterval):
@@ -696,6 +725,10 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 
 		glog.V(4).Infof("read on disk %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
 
+		// Taken before either read samples its watermark, so a rise mid-pass
+		// cannot land between the sample and the park below.
+		flushChan := fs.filer.MetaAggregator.FlushWatermarkAdvancedChan()
+		deliveryChan := fs.filer.MetaAggregator.DeliveryWatermarkAdvancedChan()
 		cursorBeforeDiskTsNs := lastReadTime.Time.UnixNano()
 
 		// Observe the flush low-watermark before the pass lists files (see
@@ -730,7 +763,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 			// 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() {
+			if !waitHeld("disk", flushChan) {
 				return nil
 			}
 			continue
@@ -836,7 +869,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 				// the last delivered entry so nothing in between is skipped.
 				lastReadTime = log_buffer.NewMessagePosition(deliveredUpToTsNs, gapResumeCursorOffset)
 				readInMemoryLogErr = nil
-				if !waitHeld() {
+				if !waitHeld("memory", deliveryChan) {
 					return nil
 				}
 				continue
@@ -1139,8 +1172,13 @@ func (fs *FilerServer) maybeSendIdleHeartbeat(req *filer_pb.SubscribeMetadataReq
 		// the buffer holds data the subscriber has not reached yet
 		return lastHeartbeatNs
 	}
+	isLocalStream := fs.filer != nil && logBuffer == fs.filer.LocalMetaLogBuffer
+	interval := idleHeartbeatInterval
+	if isLocalStream && fs.filer.MetaAggregator != nil && fs.filer.MetaAggregator.HasRemotePeers() {
+		interval = peerDeliveryClaimInterval
+	}
 	now := time.Now().UnixNano()
-	if now-lastHeartbeatNs < int64(idleHeartbeatInterval) {
+	if now-lastHeartbeatNs < int64(interval) {
 		return lastHeartbeatNs
 	}
 	// On the local stream the heartbeat is a delivery claim to a peer
@@ -1152,7 +1190,7 @@ func (fs *FilerServer) maybeSendIdleHeartbeat(req *filer_pb.SubscribeMetadataReq
 	// (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 {
+	if isLocalStream {
 		heartbeat.TsNs = fs.filer.LocalDeliveredThroughTsNs(now)
 		heartbeat.FlushedTsNs = fs.filer.LocalFlushedThroughTsNs(now)
 		if logBuffer.LastTsNs.Load() > floorTsNs {

weed/server/filer_subscribe_agg_test.go

@@ -0,0 +1,318 @@
+package weed_server
+
+// End-to-end tests for the aggregated (SubscribeMetadata) loop's peer
+// watermark holds. Peers report their progress at their own pace, so on a
+// cluster that keeps writing there is almost always an entry newer than the
+// low-watermark: a hold is the normal state, and has to be quiet and paced
+// rather than logged and retried per arriving event.
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"google.golang.org/grpc/peer"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+)
+
+const (
+	testSelfAddress = pb.ServerAddress("self:8888")
+	testPeerAddress = pb.ServerAddress("peer:8888")
+)
+
+// startAggregator gives the harness a meta aggregator with one remote peer, so
+// SubscribeMetadata takes the aggregated path instead of delegating to
+// SubscribeLocalMetadata.
+func (h *subscribeHarness) startAggregator() *filer.MetaAggregator {
+	ma := filer.NewMetaAggregator(h.f, testSelfAddress, nil)
+	ma.TrackPeerForTesting(testSelfAddress)
+	ma.TrackPeerForTesting(testPeerAddress)
+	h.f.MetaAggregator = ma
+	h.t.Cleanup(ma.MetaLogBuffer.ShutdownLogBuffer)
+	return ma
+}
+
+// reportPeers stands in for both peers' streams reporting through tsNs.
+func reportPeers(ma *filer.MetaAggregator, tsNs int64) {
+	reportPeersAt(ma, tsNs, tsNs)
+}
+
+// reportPeersAt is reportPeers with the two watermarks apart: a peer streams
+// an event well before it flushes it, so they advance independently.
+func reportPeersAt(ma *filer.MetaAggregator, deliveredTsNs, flushedTsNs int64) {
+	ma.ReportPeerWatermarksForTesting(testSelfAddress, deliveredTsNs, flushedTsNs)
+	ma.ReportPeerWatermarksForTesting(testPeerAddress, deliveredTsNs, flushedTsNs)
+}
+
+func (h *subscribeHarness) appendAggregated(tsNs int64) {
+	if err := h.f.MetaAggregator.MetaLogBuffer.AddLogEntryToBuffer(testEvent(tsNs, fmt.Sprintf("a-%d", tsNs))); err != nil {
+		h.t.Fatalf("append aggregated: %v", err)
+	}
+}
+
+func (h *subscribeHarness) subscribeAggregated(sinceNs int64) *runningSubscribe {
+	// A peer in the context keeps findClientAddress quiet, so an ERROR count
+	// across the run sees only what the subscribe loop itself writes.
+	ctx, cancel := context.WithCancel(peer.NewContext(context.Background(),
+		&peer.Peer{Addr: &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 12345}}))
+	stream := &fakeSubscribeStream{ctx: ctx}
+	req := &filer_pb.SubscribeMetadataRequest{
+		ClientName:  "agg-loop-test",
+		ClientId:    11,
+		ClientEpoch: 1,
+		SinceNs:     sinceNs,
+	}
+	r := &runningSubscribe{stream: stream, cancel: cancel, done: make(chan error, 1), finished: make(chan struct{})}
+	go func() {
+		r.done <- h.fs.SubscribeMetadata(req, stream)
+		close(r.finished)
+	}()
+	h.t.Cleanup(func() {
+		cancel()
+		select {
+		case <-r.finished:
+		case <-time.After(5 * time.Second):
+			h.t.Error("aggregated subscribe loop did not exit on cancel")
+		}
+	})
+	return r
+}
+
+// heldSubscriber starts a subscriber whose peers reported a moment ago and
+// then went quiet - the hold sits just behind live writes, exactly where a
+// quiet peer's idle heartbeat leaves it - and returns it with the write
+// timestamp the peers last reported.
+func (h *subscribeHarness) heldSubscriber() (*filer.MetaAggregator, *runningSubscribe, int64) {
+	ma := h.startAggregator()
+	frozen := time.Now().Add(-500 * time.Millisecond).UnixNano()
+	reportPeers(ma, frozen)
+	return ma, h.subscribeAggregated(frozen), frozen
+}
+
+// writeFor appends an entry every 2ms for d, reporting the peers through the
+// PREVIOUS entry each time when report is set: the live-cluster shape, where
+// the low-watermark keeps moving but always trails the newest entry, so every
+// read ends held.
+func (h *subscribeHarness) writeFor(ma *filer.MetaAggregator, d time.Duration, report bool) (written []int64, elapsed time.Duration) {
+	start := time.Now()
+	for stop := time.After(d); ; {
+		select {
+		case <-stop:
+			if report && len(written) > 0 {
+				reportPeers(ma, written[len(written)-1])
+			}
+			return written, time.Since(start)
+		default:
+		}
+		ts := time.Now().UnixNano()
+		h.appendAggregated(ts)
+		if report && len(written) > 0 {
+			reportPeers(ma, written[len(written)-1])
+		}
+		written = append(written, ts)
+		time.Sleep(2 * time.Millisecond)
+	}
+}
+
+// heldReads reads back the hold counter the loop keeps in place of the log
+// line, which is also how an operator sees a held subscriber now.
+func heldReads(scopes ...string) int {
+	if len(scopes) == 0 {
+		scopes = []string{"memory", "disk"}
+	}
+	var total int
+	for _, scope := range scopes {
+		total += int(testutil.ToFloat64(stats.FilerSubscribeWatermarkHolds.WithLabelValues(scope)))
+	}
+	return total
+}
+
+// waitForEventsAtLeastOnce asserts every want arrives, in order. The cursor a
+// hold rewinds to is inclusive of the last delivered entry, so each cycle
+// repeats it - at-least-once, which is the contract; a skip is not.
+func waitForEventsAtLeastOnce(t *testing.T, r *runningSubscribe, want []int64, timeout time.Duration) {
+	t.Helper()
+	last := want[len(want)-1]
+	deadline := time.Now().Add(timeout)
+	var got []int64
+	for time.Now().Before(deadline) {
+		got = eventTimestamps(r.stream.snapshot())
+		if len(got) > 0 && got[len(got)-1] >= last {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	seen := make(map[int64]bool, len(got))
+	var prev int64
+	for _, ts := range got {
+		if ts < prev {
+			t.Fatalf("delivered %v after %v: out of order", time.Unix(0, ts), time.Unix(0, prev))
+		}
+		prev, seen[ts] = ts, true
+	}
+	for i, ts := range want {
+		if !seen[ts] {
+			t.Fatalf("event %d of %d (%v) never delivered", i+1, len(want), time.Unix(0, ts))
+		}
+	}
+}
+
+// TestSubscribeLoop_AggregatedHoldIsQuiet: every held read used to log an
+// ERROR, so a busy cluster wrote one line per arriving event.
+func TestSubscribeLoop_AggregatedHoldIsQuiet(t *testing.T) {
+	h := newSubscribeHarness(t)
+	ma, r, _ := h.heldSubscriber()
+
+	errorsBefore := glog.Stats.Error.Lines()
+	written, elapsed := h.writeFor(ma, time.Second, false)
+	errorLines := glog.Stats.Error.Lines() - errorsBefore
+
+	t.Logf("%d writes over %v produced %d ERROR lines", len(written), elapsed, errorLines)
+	if errorLines > 0 {
+		t.Fatalf("a held read logged %d ERROR lines over %d writes; holds are control flow", errorLines, len(written))
+	}
+	if got := eventTimestamps(r.stream.snapshot()); len(got) > 0 {
+		t.Fatalf("delivered %d events past the peers' watermark", len(got))
+	}
+}
+
+// TestSubscribeLoop_AggregatedHoldIsPaced is the other half: a held read used
+// to wait on the buffer's data channel, which the very next write signalled,
+// so the loop re-ran a whole pass - a log file listing in it - per event.
+// Arriving data cannot release a hold; only peer progress can.
+func TestSubscribeLoop_AggregatedHoldIsPaced(t *testing.T) {
+	h := newSubscribeHarness(t)
+	// Long enough that a hold paced by the retry interval alone is far below
+	// one hold per write, short enough to keep the test quick.
+	prevRetry := unflushedGapRetryInterval
+	unflushedGapRetryInterval = 200 * time.Millisecond
+	t.Cleanup(func() { unflushedGapRetryInterval = prevRetry })
+	ma, _, _ := h.heldSubscriber()
+
+	holdsBefore := heldReads()
+	written, elapsed := h.writeFor(ma, time.Second, false)
+	holds := heldReads() - holdsBefore
+
+	t.Logf("%d writes over %v produced %d holds", len(written), elapsed, holds)
+	// The watermarks never moved, so the retry interval alone paces the holds.
+	if maxHolds := int(elapsed/unflushedGapRetryInterval) + 2; holds > maxHolds {
+		t.Fatalf("held %d times in %v, want at most %d: a hold must wait for peer progress, not for the next write",
+			holds, elapsed, maxHolds)
+	}
+}
+
+// TestSubscribeLoop_AggregatedHoldReleasesOnPeerProgress pins what pacing the
+// hold may not cost: once the peers report through the held entry it must be
+// delivered, without waiting out the retry interval.
+func TestSubscribeLoop_AggregatedHoldReleasesOnPeerProgress(t *testing.T) {
+	h := newSubscribeHarness(t)
+	// Far longer than this test runs: only the watermark signal can deliver.
+	prevRetry := unflushedGapRetryInterval
+	unflushedGapRetryInterval = 30 * time.Second
+	t.Cleanup(func() { unflushedGapRetryInterval = prevRetry })
+	ma, r, _ := h.heldSubscriber()
+
+	ts := time.Now().UnixNano()
+	h.appendAggregated(ts)
+	assertNoEventsFor(t, r, 200*time.Millisecond)
+
+	reportPeers(ma, ts)
+	waitForEvents(t, r, []int64{ts}, 3*time.Second)
+}
+
+// TestSubscribeLoop_AggregatedHoldCoalescesPeerProgress covers the rest of the
+// cost: peers advance their watermark on every event they stream, so releasing
+// a hold per advance is the same pass-per-event storm as releasing per write,
+// just without the log lines.
+func TestSubscribeLoop_AggregatedHoldCoalescesPeerProgress(t *testing.T) {
+	h := newSubscribeHarness(t)
+	ma, r, _ := h.heldSubscriber()
+
+	holdsBefore := heldReads()
+	written, elapsed := h.writeFor(ma, time.Second, true)
+	holds := heldReads() - holdsBefore
+
+	t.Logf("%d writes over %v produced %d holds", len(written), elapsed, holds)
+	if maxHolds := int(elapsed/heldWakeFloor) + 2; holds > maxHolds {
+		t.Fatalf("held %d times in %v, want at most %d: releases must coalesce", holds, elapsed, maxHolds)
+	}
+	// Coalescing may not lose anything the peers reported through.
+	waitForEventsAtLeastOnce(t, r, written, 3*time.Second)
+}
+
+// TestPeerDeliveryClaimPacing pins what a filer with peers owes them: the idle
+// heartbeat on its local stream is a delivery claim every aggregated
+// subscriber in the cluster holds at, so it refreshes at the claim interval,
+// not at the keepalive interval that used to park them ~5s behind live writes.
+func TestPeerDeliveryClaimPacing(t *testing.T) {
+	h := newSubscribeHarness(t)
+	req := &filer_pb.SubscribeMetadataRequest{ClientSupportsIdleHeartbeat: true}
+	lb := h.f.LocalMetaLogBuffer
+	// Claimed a second ago: stale for a delivery claim, fresh for a keepalive.
+	claimedAtNs := time.Now().Add(-time.Second).UnixNano()
+
+	s := &collectingStream{}
+	if got := h.fs.maybeSendIdleHeartbeat(req, s, lb, 0, 0, claimedAtNs); got != claimedAtNs || len(s.messages) != 0 {
+		t.Fatalf("standalone filer sent %d heartbeats early", len(s.messages))
+	}
+
+	h.startAggregator()
+	s = &collectingStream{}
+	if got := h.fs.maybeSendIdleHeartbeat(req, s, lb, 0, 0, claimedAtNs); got == claimedAtNs || len(s.messages) != 1 {
+		t.Fatalf("filer with peers sent %d delivery claims, want 1", len(s.messages))
+	}
+	if s.messages[0].TsNs <= claimedAtNs {
+		t.Fatalf("claim ts %d did not advance past %d", s.messages[0].TsNs, claimedAtNs)
+	}
+}
+
+// TestSubscribeLoop_AggregatedDiskHoldIgnoresDeliveryProgress pins that each
+// hold parks on its own watermark. A persisted-log read is held by what the
+// peers have FLUSHED, and peers advance their DELIVERY watermark on every
+// event they stream, so waking it on delivery progress would re-list a day of
+// log files per event and re-park on the same entry.
+func TestSubscribeLoop_AggregatedDiskHoldIgnoresDeliveryProgress(t *testing.T) {
+	h := newSubscribeHarness(t)
+	prevRetry := unflushedGapRetryInterval
+	unflushedGapRetryInterval = 500 * time.Millisecond
+	t.Cleanup(func() { unflushedGapRetryInterval = prevRetry })
+
+	// Recent enough that the settled horizon does not take the hold over.
+	onDisk := time.Now().Add(-30 * time.Second).UnixNano()
+	h.append(onDisk)
+	h.f.LocalMetaLogBuffer.ForceFlush()
+	waitForFlushedFiles(t, h, onDisk)
+
+	ma := h.startAggregator()
+	// Flushed below the entry on disk, delivered well past it: only the disk
+	// pass is held.
+	reportPeersAt(ma, time.Now().UnixNano(), onDisk-int64(time.Second))
+	r := h.subscribeAggregated(onDisk - int64(5*time.Second))
+	assertNoEventsFor(t, r, 200*time.Millisecond)
+
+	holdsBefore := heldReads("disk")
+	start := time.Now()
+	for deadline := start.Add(time.Second); time.Now().Before(deadline); {
+		reportPeersAt(ma, time.Now().UnixNano(), 0)
+		time.Sleep(2 * time.Millisecond)
+	}
+	holds, elapsed := heldReads("disk")-holdsBefore, time.Since(start)
+
+	t.Logf("%v of delivery-only progress produced %d disk holds", elapsed, holds)
+	if maxHolds := int(elapsed/unflushedGapRetryInterval) + 2; holds > maxHolds {
+		t.Fatalf("held on disk %d times in %v, want at most %d: delivery progress cannot release a flush hold",
+			holds, elapsed, maxHolds)
+	}
+
+	// The flush watermark reaching the entry is what delivers it.
+	reportPeersAt(ma, 0, onDisk)
+	waitForEvents(t, r, []int64{onDisk}, 3*time.Second)
+}

weed/stats/metrics.go

@@ -242,6 +242,14 @@ var (
 			Help:      "Times a metadata subscriber moved past a log range without proof it was persisted: scope=aggregated means a peer may not have flushed it, scope=local means this filer's own log flush was wedged past the give-up bound.",
 		}, []string{"scope"})
 
+	FilerSubscribeWatermarkHolds = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: subsystemFiler,
+			Name:      "subscribe_watermark_holds",
+			Help:      "Times an aggregated metadata read stopped at an entry newer than the peers' low-watermark and waited for a peer to report further progress: scope=memory held at the delivery watermark, scope=disk at the flush watermark.",
+		}, []string{"scope"})
+
 	FilerSubscribeGapStalledGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: Namespace,
@@ -907,6 +915,7 @@ func init() {
 	Gather.MustRegister(FilerServerLastSendTsOfSubscribeGauge)
 	Gather.MustRegister(FilerSubscribeGapStalledGauge)
 	Gather.MustRegister(FilerSubscribeUnprovenGapCrossings)
+	Gather.MustRegister(FilerSubscribeWatermarkHolds)
 	Gather.MustRegister(FilerMetaAggregatorReplayFailures)
 	Gather.MustRegister(FilerObjectSizeBytesHistogram)
 	Gather.MustRegister(collectors.NewGoCollector())

weed/util/log_buffer/log_read.go

@@ -14,6 +14,10 @@ import (
 var (
 	ResumeError         = fmt.Errorf("resume")
 	ResumeFromDiskError = fmt.Errorf("resumeFromDisk")
+	// StopReadingError, wrapped by an eachLogDataFn's error, marks a read the
+	// callback ended on purpose - control flow, not a failure - so the loop
+	// hands it back to the caller without logging it.
+	StopReadingError = fmt.Errorf("stopReading")
 )
 
 // notificationHealthCheckInterval bounds how long an idle subscriber blocks
@@ -297,7 +301,9 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 			lastReadPosition = NewMessagePosition(logEntry.TsNs, batchIndex)
 
 			if isDone, err = eachLogDataFn(logEntry); err != nil {
-				glog.Errorf("LoopProcessLogData: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err)
+				if !errors.Is(err, StopReadingError) {
+					glog.Errorf("LoopProcessLogData: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err)
+				}
 				return
 			}
 			if isDone {
@@ -570,7 +576,9 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 
 			glog.V(4).Infof("Calling eachLogDataFn for entry at offset %d, next position will be %d", logEntry.Offset, logEntry.Offset+1)
 			if isDone, err = eachLogDataFn(logEntry, logEntry.Offset); err != nil {
-				glog.Errorf("LoopProcessLogDataWithOffset: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err)
+				if !errors.Is(err, StopReadingError) {
+					glog.Errorf("LoopProcessLogDataWithOffset: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err)
+				}
 				return
 			}
 			if isDone {