diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 069db1970..89629e588 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -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) } diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index 25430b901..730a46226 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -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(), } } diff --git a/weed/filer/filer_notify_inflight_test.go b/weed/filer/filer_notify_inflight_test.go new file mode 100644 index 000000000..91c2f34ac --- /dev/null +++ b/weed/filer/filer_notify_inflight_test.go @@ -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) +} diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 6c1e0d1d1..9f2b7b63f 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -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 diff --git a/weed/filer/meta_aggregator_watermark_test.go b/weed/filer/meta_aggregator_watermark_test.go new file mode 100644 index 000000000..bbe10ae9d --- /dev/null +++ b/weed/filer/meta_aggregator_watermark_test.go @@ -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 +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index b8beefd71..c6f00af3f 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -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 diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index a00719891..6e7f65a72 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -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" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index efcfa0819..1f6b2ea0f 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -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:]) diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index f57b3dba6..427c94be9 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -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 } diff --git a/weed/server/filer_grpc_server_sub_meta_gap_test.go b/weed/server/filer_grpc_server_sub_meta_gap_test.go index 8d5d1286d..6505b0828 100644 --- a/weed/server/filer_grpc_server_sub_meta_gap_test.go +++ b/weed/server/filer_grpc_server_sub_meta_gap_test.go @@ -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) } diff --git a/weed/server/filer_grpc_server_sub_meta_watermark_test.go b/weed/server/filer_grpc_server_sub_meta_watermark_test.go new file mode 100644 index 000000000..b482d089a --- /dev/null +++ b/weed/server/filer_grpc_server_sub_meta_watermark_test.go @@ -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) + } +} diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index 279c343b6..01d6100c9 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -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) } diff --git a/weed/util/log_buffer/log_buffer_eviction_gate_test.go b/weed/util/log_buffer/log_buffer_eviction_gate_test.go index 927f13952..5997324f4 100644 --- a/weed/util/log_buffer/log_buffer_eviction_gate_test.go +++ b/weed/util/log_buffer/log_buffer_eviction_gate_test.go @@ -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