From 8d3cb2ed2dfd2c75dac8d9ac56fdb1d59fb84209 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 31 Oct 2026 11:49:50 -0700 Subject: [PATCH] filer: resume a rejoining peer's meta subscription at its last position (#12279) * filer: resume a rejoining peer's meta subscription at its last position OnPeerUpdate starts every peer add at the update's arrival time. For filers sharing one store there is no persisted offset to correct it, so a peer that drops out and rejoins (a graceful restart) resubscribes from "now" and the events it logged while its stream was down never reach the aggregated buffer - filer.backup and other meta subscribers silently skip them. Track each peer's delivered-through position (the same timestamp domain as the stream's events and idle-heartbeat claims) and rewind the start of a rejoining subscription to it. First-seen peers keep the given start, and non-shared peers still resume from their persisted offset. * filer: keep the re-add resume point behind a replay-floored event The session's replay floor already pins the persisted offset and the reconnect resume point so an event that outlived its retry budget is redelivered on resubscribe; cap peerResumeTsNs to the same position on session exit so the new re-add resume cannot skip past it either. * filer: stop test aggregators' log-buffer workers * filer: cover peer rejoin resubscribing at the resume position * filer: drop the unreachable replay-floor cap on the resume point Only a peer with its own store can floor an event, and that peer's subscribe start always comes from its persisted offset, never from the re-add resume point. The cap also ran in the old session's defer, after the re-add had already read the value. * filer: say why a re-add may resume from an idle heartbeat The heartbeat comments promised a reconnect never resumes from one, while the re-add resume point does. Spell out the difference and the fence that makes the local-stream claim safe to resume from. * filer: forget a removed peer's resume point after an hour The map held an entry for every address ever seen; with pod IPs that is one per restart. The removal mark now outlives the watermark grace and takes the resume point with it once the retention passes. * filer: advance the resume point only for tracked peers Same rule as the watermarks: a straggler from a dropped session must not recreate the entry. * filer: set the test aggregator's subscribe loop instead of a nil fallback --- weed/filer/meta_aggregator.go | 82 +++++++++++++++++--- weed/filer/meta_aggregator_test.go | 77 ++++++++++++++++++ weed/filer/meta_aggregator_watermark_test.go | 8 +- weed/server/filer_grpc_server_sub_meta.go | 4 +- 4 files changed, 157 insertions(+), 14 deletions(-) diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 96eedd0797..367a74f678 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -49,6 +49,7 @@ 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. + // The mark itself stays until the peer's resume point is forgotten. peerRemovedAtNs map[pb.ServerAddress]int64 // lowWatermarkTsNs and lowFlushWatermarkTsNs are the last minima signalled // on deliveryAdvanced and flushAdvanced, each closed and replaced when its @@ -62,8 +63,17 @@ type MetaAggregator struct { deliveryAdvanced chan struct{} flushAdvanced chan struct{} remotePeerArrived chan struct{} - peerWatermarksLock sync.Mutex - peerWg sync.WaitGroup + // peerResumeTsNs keeps each peer's delivered-through position across + // subscription restarts; the watermark maps forget a removed peer past + // the grace, but a rejoining peer still needs it to resume. Forgotten + // peerResumeRetention after the removal so addresses that never return, + // such as restarted pods' IPs, do not accumulate. + peerResumeTsNs map[pb.ServerAddress]int64 + peerWatermarksLock sync.Mutex + peerWg sync.WaitGroup + // peerSubscribeLoop is replaceable so a test can observe the subscribe + // start without a live peer. + peerSubscribeLoop func(f *Filer, self, peer pb.ServerAddress, startFrom time.Time, stopChan chan struct{}) // stopped is set under peerChansLock by Shutdown; OnPeerUpdate then // rejects adds so no subscription can outlive the shutdown wait and run // against a store Filer.Shutdown has already closed. @@ -81,7 +91,9 @@ func NewMetaAggregator(filer *Filer, self pb.ServerAddress, grpcDialOption grpc. peerWatermarks: make(map[pb.ServerAddress]int64), peerFlushWatermarks: make(map[pb.ServerAddress]int64), peerRemovedAtNs: make(map[pb.ServerAddress]int64), + peerResumeTsNs: make(map[pb.ServerAddress]int64), } + t.peerSubscribeLoop = t.loopSubscribeToOneFiler // nil notifyFn: aggregated subscribers wake through the buffer's // subscriber channels, not a cond. t.MetaLogBuffer = log_buffer.NewLogBuffer("aggr", LogFlushInterval, nil, nil, nil) @@ -114,10 +126,11 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star // Account for the peer before its stream signals; keep prior values // on reconnect. ma.initPeerWatermark(address) + startFrom = ma.peerResumeStartFrom(address, startFrom) ma.peerWg.Add(1) go func() { defer ma.peerWg.Done() - ma.loopSubscribeToOneFiler(ma.filer, ma.self, address, startFrom, stopChan) + ma.peerSubscribeLoop(ma.filer, ma.self, address, startFrom, stopChan) }() } else { if prevChan, found := ma.peerChans[address]; found { @@ -135,6 +148,11 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star // already bounds a stale watermark's influence within it. const peerWatermarkRemovalGrace = 2 * LogFlushInterval +// peerResumeRetention is how long a removed peer's resume point outlives the +// watermark grace, so a slow restart or a partition still rejoins at its old +// position. +const peerResumeRetention = time.Hour + // 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) { @@ -162,19 +180,25 @@ func (ma *MetaAggregator) markPeerWatermarkRemoved(peer pb.ServerAddress) { } } -// dropExpiredRemovedPeersLocked drops peers whose removal outlived the grace. -// Caller must hold peerWatermarksLock. +// dropExpiredRemovedPeersLocked drops the watermarks of peers whose removal +// outlived the grace, and the mark with the resume point once it outlived +// peerResumeRetention. Caller must hold peerWatermarksLock. func (ma *MetaAggregator) dropExpiredRemovedPeersLocked() { if len(ma.peerRemovedAtNs) == 0 { return } - cutoff := time.Now().UnixNano() - int64(peerWatermarkRemovalGrace) + now := time.Now().UnixNano() + graceCutoff := now - int64(peerWatermarkRemovalGrace) + resumeCutoff := now - int64(peerResumeRetention) for peer, removedAt := range ma.peerRemovedAtNs { - if removedAt < cutoff { - delete(ma.peerRemovedAtNs, peer) + if removedAt < graceCutoff { delete(ma.peerWatermarks, peer) delete(ma.peerFlushWatermarks, peer) } + if removedAt < resumeCutoff { + delete(ma.peerRemovedAtNs, peer) + delete(ma.peerResumeTsNs, peer) + } } // Dropping the peer that pinned a minimum raises it, same as it reporting. ma.noteLowWatermarksLocked() @@ -182,14 +206,46 @@ func (ma *MetaAggregator) dropExpiredRemovedPeersLocked() { // 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. +// its entry and pin the low-watermark, nor revive a forgotten resume point. 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 { + cur, found := ma.peerWatermarks[peer] + if !found { + return + } + if tsNs > cur { ma.peerWatermarks[peer] = tsNs ma.noteLowWatermarksLocked() } + if tsNs > ma.peerResumeTsNs[peer] { + ma.peerResumeTsNs[peer] = tsNs + } +} + +// peerResumeStartFrom rewinds a (re)joining peer's subscribe start to the +// position the previous subscription was delivered through. The peer's own +// stream carries delivery claims (idle heartbeats and events alike) in this +// timestamp domain, so anything at or below it was already received; starting +// from the caller's "now" instead would drop what the peer logged while its +// stream was down. A peer seen for the first time keeps the given start. +// +// An in-session reconnect resumes from the last real event, but this may +// trust an idle heartbeat too: on the peer's local stream it is a +// delivered-through claim capped by the in-flight floor and fenced against +// later stamps, so no event can land at or below it afterwards. The claim +// came from the peer's previous process, so it holds as long as the +// restarted peer's clock does not run backwards. +// +// A replay floor never has to cap this: only a peer with its own store can +// floor an event, and such a peer starts from its persisted offset instead. +func (ma *MetaAggregator) peerResumeStartFrom(peer pb.ServerAddress, startFrom time.Time) time.Time { + ma.peerWatermarksLock.Lock() + defer ma.peerWatermarksLock.Unlock() + if prev := ma.peerResumeTsNs[peer]; prev > 0 && prev < startFrom.UnixNano() { + return time.Unix(0, prev) + } + return startFrom } // advancePeerFlushWatermark records the peer's reported flush watermark. @@ -694,8 +750,10 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, return err } } - // Idle heartbeat: advance only the watermark, not lastTsNs, so a - // reconnect still resumes from the last real event. + // Idle heartbeat: a delivery claim. It advances the watermark and + // the re-add resume point (see peerResumeStartFrom) but not + // lastTsNs, so this loop's own reconnect still resumes from the + // last real event. if resp.EventNotification == nil && len(resp.Events) == 0 && resp.TsNs > 0 { ma.advancePeerWatermark(peer, resp.TsNs) } diff --git a/weed/filer/meta_aggregator_test.go b/weed/filer/meta_aggregator_test.go index 9ea75a4f68..0655ed0037 100644 --- a/weed/filer/meta_aggregator_test.go +++ b/weed/filer/meta_aggregator_test.go @@ -70,6 +70,83 @@ func TestMetaAggregatorShutdownStopsPeerLoops(t *testing.T) { } } +// A peer removed and re-added must resume where its previous subscription +// was delivered through: subscribing from the update's arrival time drops +// whatever the peer logged while its stream was down. +func TestPeerResumeStartFromSurvivesRemoval(t *testing.T) { + peer := pb.ServerAddress("127.0.0.1:1") + ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil) + defer ma.Shutdown() + + subscribeStarts := make(chan int64, 3) + ma.peerSubscribeLoop = func(_ *Filer, _ pb.ServerAddress, _ pb.ServerAddress, startFrom time.Time, stopChan chan struct{}) { + subscribeStarts <- startFrom.UnixNano() + <-stopChan + } + nextSubscribeStart := func() int64 { + select { + case tsNs := <-subscribeStarts: + return tsNs + case <-time.After(3 * time.Second): + t.Fatal("no peer subscription started") + return 0 + } + } + + add := &master_pb.ClusterNodeUpdate{Address: string(peer), IsAdd: true} + firstStart := time.Now() + ma.OnPeerUpdate(add, firstStart) + if got := nextSubscribeStart(); got != firstStart.UnixNano() { + t.Fatalf("expecting first subscribe at %d, got %d", firstStart.UnixNano(), got) + } + + deliveredThrough := time.Now().Add(-time.Hour).UnixNano() + ma.advancePeerWatermark(peer, deliveredThrough) + if got := ma.peerResumeStartFrom(peer, time.Now()); got.UnixNano() != deliveredThrough { + t.Fatalf("expecting resume at %d, got %d", deliveredThrough, got.UnixNano()) + } + + // Removal and even the watermark-grace expiry keep the resume position. + ma.OnPeerUpdate(&master_pb.ClusterNodeUpdate{Address: string(peer)}, time.Now()) + ma.peerWatermarksLock.Lock() + ma.peerRemovedAtNs[peer] = time.Now().Add(-2 * peerWatermarkRemovalGrace).UnixNano() + ma.dropExpiredRemovedPeersLocked() + ma.peerWatermarksLock.Unlock() + + ma.OnPeerUpdate(add, time.Now()) + if got := nextSubscribeStart(); got != deliveredThrough { + t.Fatalf("expecting re-added peer to resume at %d, got %d", deliveredThrough, got) + } + + // A peer gone past the retention is forgotten and starts afresh. + ma.OnPeerUpdate(&master_pb.ClusterNodeUpdate{Address: string(peer)}, time.Now()) + ma.peerWatermarksLock.Lock() + ma.peerRemovedAtNs[peer] = time.Now().Add(-2 * peerResumeRetention).UnixNano() + ma.dropExpiredRemovedPeersLocked() + _, retained := ma.peerResumeTsNs[peer] + ma.peerWatermarksLock.Unlock() + if retained { + t.Fatal("expecting the resume point to be forgotten past the retention") + } + freshStart := time.Now() + ma.OnPeerUpdate(add, freshStart) + if got := nextSubscribeStart(); got != freshStart.UnixNano() { + t.Fatalf("expecting forgotten peer to start at %d, got %d", freshStart.UnixNano(), got) + } + + // A first-seen peer, or a caller start already behind the recorded + // position, keeps the caller's start. + other := pb.ServerAddress("127.0.0.1:3") + start := time.Now().Add(-time.Minute) + if got := ma.peerResumeStartFrom(other, start); !got.Equal(start) { + t.Fatalf("expecting caller start %v for unseen peer, got %v", start, got) + } + early := time.Unix(0, deliveredThrough).Add(-time.Second) + if got := ma.peerResumeStartFrom(peer, early); !got.Equal(early) { + t.Fatalf("expecting caller start %v behind resume, got %v", early, got) + } +} + // After Shutdown begins, a peer-add update arriving on the still-connected // master watch must not start a subscription: it would have no stop signal // and could run past the store close. diff --git a/weed/filer/meta_aggregator_watermark_test.go b/weed/filer/meta_aggregator_watermark_test.go index f81722adf3..5edd2fb1e1 100644 --- a/weed/filer/meta_aggregator_watermark_test.go +++ b/weed/filer/meta_aggregator_watermark_test.go @@ -8,11 +8,14 @@ import ( ) func newTestAggregator() *MetaAggregator { - return &MetaAggregator{ + ma := &MetaAggregator{ peerWatermarks: make(map[pb.ServerAddress]int64), peerFlushWatermarks: make(map[pb.ServerAddress]int64), peerRemovedAtNs: make(map[pb.ServerAddress]int64), + peerResumeTsNs: make(map[pb.ServerAddress]int64), } + ma.peerSubscribeLoop = ma.loopSubscribeToOneFiler + return ma } // TestPeerWatermarkBookkeeping pins the per-peer delivery watermark semantics @@ -116,6 +119,9 @@ func TestPeerWatermarkRemovalGrace(t *testing.T) { if got := ma.PeerLowWatermarkTsNs(); got != 100 { t.Fatalf("after straggler: low=%d want 100", got) } + if got := ma.peerResumeTsNs[b]; got != 60 { + t.Fatalf("after straggler: resume=%d want 60", got) + } if _, found := ma.peerWatermarks[b]; found { t.Fatalf("dropped peer resurrected in delivery watermark set") } diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index 775f1b3bf2..66af39618a 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -1385,7 +1385,9 @@ func (fs *FilerServer) maybeSendFlushReport(req *filer_pb.SubscribeMetadataReque // 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 -// its resume checkpoint, so a restart still re-reads from the last real event. +// its persisted resume checkpoint, so a restart still re-reads from the last +// real event. The fenced local-stream claim is the one exception: a peer +// aggregator resumes a re-added peer from it (MetaAggregator.peerResumeStartFrom). // // The catch-up floor is the max of two read-progress markers: // - readPositionTsNs: how far the read cursor has advanced. It starts at