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