Documentation
¶
Overview ¶
Package readiness owns BOTH sides of the ADR-083 readiness distribution contract. Producers (publisher.go) create the GRAPH_STATUS KV bucket and write their envelope to their own key on every heartbeat tick; consumers (watcher.go) watch that key, hold the last-known ADR-066 envelope, and answer "(status, fresh | unknown)" with no per-decision NATS round-trip.
It exists because per-decision request/reply died under exactly the load that makes readiness interesting: in a single-binary deployment the status request travels the same connection as the ENTITY_STATES firehose, so an in-process requester pays the saturated stream twice per round-trip, times out, and fails closed — indistinguishable in the logs from a genuine not-ready (gh#590). The KV twofer inverts it: the producer pays one write per heartbeat and N consumers hold state.
Freshness is judged by CONSUMER-LOCAL arrival time. No producer timestamp is ever compared against the consumer's clock (ADR-083 D2) — arrival-time freshness needs no clock agreement between processes, and cross-process skew would buy nothing.
This package holds the bucket/key identifiers for BOTH sides of the contract, so producers and consumers cannot drift onto different names. It deliberately does not import natsclient: BucketSource is the narrow method set *natsclient.Client already satisfies, which keeps the dependency one-way and the unit tests free of a live NATS.
Index ¶
Constants ¶
const ( // BucketGraphStatus is the dedicated readiness bucket. It is operational // component status, NOT domain entity state: it never routes through // graph-ingest, carries no ADR-055 semantic envelope, and is kept out of // ENTITY_STATES so the sole-writer invariant and the graph itself stay clean. BucketGraphStatus = "GRAPH_STATUS" // BucketHistory is the KV history depth for the bucket: enough replay to see // the last few transitions after an incident, not enough to hoard. BucketHistory = 3 // KeyGraphIndex is graph-index's status key (one key per producer). KeyGraphIndex = "graph-index" // KeyGraphEmbedding is graph-embedding's status key. KeyGraphEmbedding = "graph-embedding" )
The GRAPH_STATUS identifiers, shared by producers and consumers.
const ( // DefaultHeartbeat is the producers' status tick: they publish the envelope // every tick unconditionally, so the write doubles as a liveness heartbeat. DefaultHeartbeat = 5 * time.Second // FreshnessMultiplier is how many heartbeats a held status stays fresh. It is a // CONSTANT, not config (owner decision): 3 tolerates a lost write and a slow // delivery without letting a dead feed masquerade as live, and one fewer knob // is one fewer way to configure a fail-open. // // "Freshness" here is about the ENVELOPE, not the view it describes — can this // consumer still vouch for the status reading it is holding, or has the producer // gone quiet? That is a transport-liveness question, and it is what // StatusReading.Fresh answers and what the health gate fails closed on. It is NOT // the retired view-age concept (how far behind ENTITY_STATES the index itself is); // view age is reported on IndexStatusResponse.StalenessMs and gates nothing. FreshnessMultiplier = 3 )
Variables ¶
This section is empty.
Functions ¶
func EnsureBucket ¶
EnsureBucket creates-or-opens the GRAPH_STATUS bucket and returns a handle. Every producer calls it at Start, EAGERLY — before any consumer could bind — because a consumer that binds a watch to a not-yet-existent bucket reads permanently unknown (fail-closed) until it happens to rebind, and the whole point of this contract is that unknown means something is wrong.
It is IDEMPOTENT across producers and restarts: graph-index and graph-embedding both run in cmd/semstreams and both call it, in either order and possibly concurrently. Idempotency comes from natsclient's CreateKeyValueBucket, which gets an existing bucket before attempting a create and treats a concurrent create as success.
That get-first behavior also means an existing bucket is adopted WITHOUT comparing its config, so a pre-existing GRAPH_STATUS with a different History is used as-is rather than rejected. Nothing depends on that today precisely because every caller routes through THIS function and therefore asks for one shape — which is the reason to keep it that way rather than hand-rolling a second bucket config elsewhere.
The bucket carries no TTL and no size-based eviction: staleness is judged consumer-side from arrival time (D2), so expiring the key server-side would only destroy the last-known state a consumer needs to diagnose a dead producer.
func FreshnessWindow ¶
FreshnessWindow is how long a status value stays fresh for a producer publishing on the given heartbeat. Exported so consumers and their tests express the window in one vocabulary instead of re-deriving 3x from the two constants.
Types ¶
type BucketCreator ¶
type BucketCreator interface {
CreateKeyValueBucket(ctx context.Context, cfg jetstream.KeyValueConfig) (jetstream.KeyValue, error)
}
BucketCreator creates-or-opens a KV bucket. *natsclient.Client satisfies it via CreateKeyValueBucket — declared here as the narrow method set with the EXACT signature (a divergent embedded signature silently fails the structural match and no-ops the capability) rather than importing natsclient, which would make this package non-leaf and drag a live NATS into its unit tests.
type BucketSource ¶
type BucketSource interface {
GetKeyValueBucket(ctx context.Context, name string) (jetstream.KeyValue, error)
}
BucketSource opens a KV bucket by name. *natsclient.Client satisfies it via GetKeyValueBucket — declared here as the narrow method set (with the EXACT signature, or the structural match silently breaks) rather than importing natsclient, so this package stays a leaf and tests need no live NATS.
type Option ¶
type Option func(*Watcher)
Option configures a Watcher.
func WithHeartbeat ¶
WithHeartbeat declares the producer's publish interval, which sets the freshness window (FreshnessMultiplier x heartbeat). Defaults to DefaultHeartbeat.
func WithLogger ¶
WithLogger sets the logger used for bind/decode failures.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher writes one producer's readiness envelope to its GRAPH_STATUS key. One instance per producer; Publish is called from the status tick and is safe to call from any single goroutine.
func NewPublisher ¶
func NewPublisher(bucket StatusWriter, key string) *Publisher
NewPublisher builds a publisher for one producer key (KeyGraphIndex, KeyGraphEmbedding). It returns nil when the wiring is incomplete, and a nil *Publisher's Publish is a safe no-op: readiness reporting must never be the thing that panics the component whose health it reports.
func (*Publisher) Key ¶
Key reports the producer key this publisher writes, for the caller's failure log.
func (*Publisher) Publish ¶
Publish writes the envelope as the KV value for this producer's key.
The value is PLAIN graph.IndexStatusResponse JSON — the same struct fusion decodes, with no BaseMessage wrapper. The payload registry governs polymorphic message publishes on subjects, where a receiver must discriminate a type it did not choose; this is a KV value on a fixed key whose type is fixed by the contract, and wrapping it would break every consumer's plain decode (graph/readiness, pkg/fusion) for nothing.
Callers publish on EVERY tick, unconditionally, without comparing to the last value: the write is the liveness heartbeat that lets a consumer tell "not ready" from "the producer is gone", and skipping unchanged values would make a healthy steady state indistinguishable from a dead one.
The error is returned, never swallowed, so the tick loop can count and log it; the loop must keep ticking regardless, because the next heartbeat is the recovery.
type Reading ¶
type Reading struct {
// Status is the last envelope received on the key.
Status graph.IndexStatusResponse
// Raw is the untouched wire value the Status field was decoded from, for a caller
// that must decode into its OWN field-identical struct (pkg/fusion.IndexStatus).
// Handing such a caller only the decoded graph.IndexStatusResponse would force a
// hand-copied field remap, which is exactly the bug that silently dropped
// IndexedRevision/Lag once before and read downstream as a false caught-up. Nil
// when !Known.
Raw []byte
// Fresh reports that the last update arrived within FreshnessMultiplier
// heartbeats. False means UNKNOWN — fail closed.
Fresh bool
// Known reports that an envelope has been received at least once. It separates
// "never saw the producer" (absent bucket/key, standalone deployment) from "the
// feed went quiet", which are different operator problems.
Known bool
// Age is how long ago the last update arrived, consumer-local. Zero when
// !Known.
Age time.Duration
// Err is the last bucket/watch/decode failure, if any, for the structured defer
// log. It is cleared by the next good update.
Err error
}
Reading is the consumer-side answer: the last-known envelope plus everything needed to attribute a defer without correlating log lines.
Status is authoritative ONLY when Fresh. When Fresh is false the envelope is last-known-and-aged (or the zero value if none ever arrived) and is carried for DIAGNOSTICS only — a caller must fail closed, never read the stale bits as index state. That distinction is the whole point: gh#590 cost three investigation cycles because a transport failure wore not-ready's log line.
type StatusWriter ¶
StatusWriter writes one status value. jetstream.KeyValue satisfies it; taking the one method the publisher needs keeps the seam narrow and the unit tests honest.
type Watcher ¶
type Watcher struct {
// contains filtered or unexported fields
}
Watcher holds the last-known readiness envelope for ONE producer key. One instance serves every gate decision in a consumer: Read is safe from any goroutine, while Start and Stop belong to the owning component's lifecycle goroutine (Start once, Stop once).
func NewWatcher ¶
func NewWatcher(src BucketSource, key string, opts ...Option) *Watcher
NewWatcher builds a watcher for one producer key (KeyGraphIndex, KeyGraphEmbedding) in the GRAPH_STATUS bucket. It performs no I/O; call Start.
func (*Watcher) Read ¶
Read returns the current reading. It never blocks on I/O — the answer comes from held state, which is the point of watching rather than polling.
func (*Watcher) Start ¶
Start begins watching in the background and returns immediately. It does NOT fail when the bucket or key is absent: a deployment running a consumer without its producer is a legitimate configuration whose readiness is simply UNKNOWN, and Read reports it as such (fail closed) while the watcher keeps trying to bind. The caller's own escape hatch (e.g. allow_ungated_reads) is what covers that case — never a fabricated envelope.
Start is idempotent; subsequent calls are no-ops.
func (*Watcher) Stop ¶
func (w *Watcher) Stop()
Stop cancels the watch and waits for the background goroutine to exit. It is safe to call without Start and safe to call more than once.
func (*Watcher) WaitForFirst ¶
WaitForFirst blocks until an envelope has been received on the key, or ctx ends. It returns nil once something has arrived (fresh or not — Read is what judges that) and ctx.Err() otherwise.
It exists for the LAZILY-BOUND consumers (the graph/query client, fusion): they bind the watch on their first gate decision, and without a bounded wait that first decision would fail closed purely because the watch had not delivered yet — a regression against the request/reply this replaces, which always got an answer or an error. Waiting once at bind, bounded by the same timeout the request used, keeps the worst-case latency of the first gated call unchanged while every later call becomes a local read. Callers must wait AT MOST once: with no producer deployed nothing will ever arrive, and a per-call wait would turn the standalone deployment into a per-call stall.
Source Files