Documentation
¶
Overview ¶
Package revlag provides a revision-lag "caught up" watermark for async, write-driven, lagging pipelines fed by a NATS KV watch (ADR-066).
A component that watches a KV bucket and processes each entry through a pool (with optional coalescing and inline deletes) needs to answer "have I caught up to the latest committed write?" honestly — not "have I started?". The Watermark tracks that as revision lag: Observe every delivered revision, Complete it when its processing reaches a terminal outcome, and read Indexed() — the highest revision such that every delivered revision <= it is done and nothing <= it is still in flight.
It also answers "how OLD is the view I am serving?" (ADR-083): IndexedAt pairs the watermark with the KV COMMIT timestamp of the newest observed revision it covers, so a consumer can bound staleness in wall time instead of in revisions (a revision count is load- and coalesce-dependent; see gh#590).
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Watermark ¶
type Watermark struct {
// contains filtered or unexported fields
}
Watermark is the low-water-of-pending "caught up" tracker. It answers "every DELIVERED revision <= Indexed() has completed and nothing <= it is still in flight." It is deliberately defined only over DELIVERED revisions, so it is correct even when the delivered revision set is SPARSE — as it is for a NATS KV bucket at History=1, where WatchAll delivers via OrderedConsumer + DeliverLastPerSubject and superseded revisions are purged and never delivered. The rejected "advance from Indexed+1 past every contiguous completion" would stall forever on the first purged gap.
Correctness rests on ONE property, which OrderedConsumer guarantees at the nats-server storage layer for both the bootstrap replay and live updates: delivery is monotonic ascending in the revision passed to Observe. Any un-observed revision below observedHigh is therefore purged (permanently absent) and correctly skipped. That same property makes Indexed() monotonic non-decreasing, which is what lets the covered COMMIT TIME be carried forward as a single scalar (see coveredRev/coveredAt below) instead of a growing map.
All methods are safe for concurrent use: the watch goroutine Observes; N pool workers, a coalescer callback, and/or an inline delete handler Complete; a status handler reads Indexed()/IndexedAt().
func New ¶
func New() *Watermark
New returns an empty Watermark (Indexed() == 0 until the first Observe).
func (*Watermark) Complete ¶
Complete drains every in-flight revision for key with revision <= rev — the single key-scoped completion rule. A pool worker calls it with the entry it processed, which drains any coalescer-collapsed lower revisions of that key that no worker sees individually; an inline delete handler calls it with the tombstone's revision, which drains an earlier pending update the delete supersedes. Key-scoped (never global-<=-rev) so one key's completion cannot drop a different key's pending revision.
func (*Watermark) Indexed ¶
Indexed returns the low-water-of-pending watermark: observedHigh when nothing is in flight, else minPending-1. Every delivered revision <= the result is complete and nothing <= it is still pending.
func (*Watermark) IndexedAt ¶
IndexedAt returns the Indexed() floor together with the COMMIT time of the newest observed revision that floor covers — the "the view reflects the world as of T" input to the ADR-083 staleness_ms field.
The two are read under one lock so a caller cannot pair a floor with a timestamp from a different instant. commitAt is the ZERO time when the floor covers no observed revision (cold start, or a floor that sits below every delivered revision because the lowest one is still in flight) — callers MUST treat zero as "staleness not computable", never as "zero staleness". commitAt may be older than the floor's true coverage but never newer (see coveredRev).
func (*Watermark) Observe ¶
Observe records revision r (for key, committed at commitAt) as delivered-and-in-flight. Call it from the watch goroutine for EVERY delivered entry — updates AND deletes — before dispatch, passing the KV entry's COMMIT timestamp (entry.Created()), never the local arrival time: commit time is what makes staleness mean "the view reflects the world as of T" and correctly includes server-side delivery backlog once an entry lands. A zero commitAt is allowed (it simply leaves the covered timestamp unknown, which reads as "staleness not computable" downstream) so tests and legacy paths need not fabricate one.
observedHigh is monotonic; ascending delivery means r exceeds any prior, but max() is kept for defensiveness. Revision 0 ("no revision") is ignored — KV stream sequences start at 1.
func (*Watermark) Observed ¶
Observed returns the highest revision ever DELIVERED, regardless of whether it has been applied. It is the enumeration-time target a bootstrap latch needs: when the initial-sync sentinel fires, every pre-existing entity has been delivered, so their revisions are all <= this value, and the initial build is complete once Indexed() reaches it.
Distinct from Indexed(), which is the low-water floor of what is APPLIED. Comparing against a freshly-read stream LastSeq instead would make a bootstrap latch unreachable under continuous write, because that target advances as fast as the index does.
Source Files