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
- func EnsureBucket(ctx context.Context, client *natsclient.Client) (jetstream.KeyValue, error)
- func FreshnessWindow(heartbeat time.Duration) time.Duration
- type BucketSource
- type Dump
- type GaugeOption
- type Gauges
- type Option
- type ProducerNames
- type Publisher
- type Reading
- type Set
- type StatusWriter
- type Verdict
- type Watcher
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. // // Re-exported from graph.BucketGraphStatus (the single source of truth) so // this name and the framework-owned-bucket set that write-protects it against // a generic rule update_kv (framework-owned-bucket-guards F3) can never drift. // Producers and consumers keep referring to readiness.BucketGraphStatus. BucketGraphStatus = graph.BucketGraphStatus // KeyGraphIndex is graph-index's status key (one key per producer). KeyGraphIndex = "graph-index" // KeyGraphEmbedding is graph-embedding's status key. KeyGraphEmbedding = "graph-embedding" // KeyGraphIngest is graph-ingest's status key. Its envelope is a BACKLOG // envelope (graph.ComputeBacklogStatus): Lag is in MESSAGES, and the revision // fields are absent because graph-ingest consumes multiple streams whose // sequence spaces are independent. KeyGraphIngest = "graph-ingest" // KeyRule is the rule processor's status key, reporting bootstrap-replay // completion for its entity watchers. KeyRule = "rule" )
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 acquires the GRAPH_STATUS bucket through the catalog owner seam 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; the seam's create-or-open resolves the race, and an adopted bucket is RECONCILED to the catalog declaration (History, no-lifecycle retention) rather than adopted config-unseen.
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 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 Dump ¶
type Dump struct {
Key string `json:"key"`
Known bool `json:"known"`
Fresh bool `json:"fresh"`
AgeMs int64 `json:"age_ms"`
// State, Ready, Lag and BootstrapComplete are echoed from the last envelope so an
// operator can see WHY a consumer is deferring without a second round trip.
State string `json:"state,omitempty"`
Ready bool `json:"ready"`
Lag uint64 `json:"lag,omitempty"`
BootstrapComplete bool `json:"bootstrap_complete"`
BootstrapScope uint64 `json:"bootstrap_scope,omitempty"`
// Error is the last transport/decode failure, if any. A dead feed and a
// not-ready producer are different operator problems.
Error string `json:"error,omitempty"`
}
Dump is a read-only projection of every declared key's local state, for an operator endpoint. It is NOT a verdict: baking a verdict into a framework surface would bake the key list into the framework, which is exactly what the client-side-fold rule forbids.
type GaugeOption ¶
type GaugeOption func(*Gauges)
GaugeOption configures the set at construction.
func WithRevisionGauges ¶
func WithRevisionGauges() GaugeOption
WithRevisionGauges adds indexed_revision and target_revision. Only revision-lag producers (graph-index, graph-embedding) may pass it: their Lag is in ENTITY_STATES revisions and the two fields are meaningful. A backlog producer must not.
type Gauges ¶
type Gauges struct {
// contains filtered or unexported fields
}
Gauges is the readiness envelope's Prometheus projection, owned ONCE for every producer.
It exists because the duplication it replaces did not merely permit a bug, it produced one twice: graph-index and graph-embedding each hand-rolled this set, and both independently omitted bootstrap_complete — the field the KV envelope treats as load-bearing and EvaluateReadinessGate consumes. A second producer shape (backlog) would have made four copies with the same drift mode.
Set is the single projection site, so a field added to IndexStatusResponse cannot be wired in three producers and forgotten in a fourth.
func NewGauges ¶
func NewGauges(names ProducerNames, opts ...GaugeOption) *Gauges
NewGauges builds the readiness gauge set for one producer.
The emitted metric NAMES are an external contract — dashboards and operator alerts consume them, and the graph-index-readiness spec explicitly protects graph-index's published output. They are therefore reproduced here exactly as the two hand-rolled sets emitted them; MetricNames pins that, and a test compares it against the live registry.
func (*Gauges) MetricNames ¶
MetricNames returns the registry metric names this set emits, in registration order. Exported so a producer's test can pin its external contract without reaching into unexported fields.
func (*Gauges) RecordPublishFailure ¶
func (g *Gauges) RecordPublishFailure()
RecordPublishFailure counts one failed GRAPH_STATUS heartbeat write (ADR-083).
func (*Gauges) Register ¶
func (g *Gauges) Register(registry *metric.MetricsRegistry)
Register wires every gauge into the metrics registry, falling back to the default Prometheus registerer when the registry is nil — the fallback each producer previously hand-rolled for tests.
func (*Gauges) Set ¶
func (g *Gauges) Set(resp graph.IndexStatusResponse)
Set projects one readiness envelope onto the gauges.
THE SINGLE PROJECTION SITE. Every producer funnels through here, so adding a field to IndexStatusResponse means adding it once rather than remembering four places — which is exactly the omission that left bootstrap_complete unexposed on two independently-written producers.
The state one-hot iterates graph.AllIndexStates rather than the states this producer happens to emit, so a new readiness state renders as an explicit 0 instead of a silently absent series.
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 ProducerNames ¶
type ProducerNames struct {
// Service is the metrics-registry key, hyphenated: "graph-index".
Service string
// Subsystem is the Prometheus subsystem, underscored: "graph_index".
Subsystem string
}
ProducerNames carries the two naming vocabularies this repo already uses for the same component.
A struct rather than two string parameters: they differ only in punctuation ("graph-index" vs "graph_index"), so positional arguments are a silent-swap footgun whose failure mode is renaming EVERY metric the producer emits — with no error raised anywhere and dashboards simply going dark. That is the same reasoning graph.IndexStatusInputs carries, and the same reason this package refuses to DERIVE one spelling from the other: guessing wrong fails silently, so the caller states both.
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 Set ¶
type Set struct {
// contains filtered or unexported fields
}
Set folds several producers' readiness envelopes into one verdict for a consumer that depends on all of them.
AGGREGATION IS CLIENT-SIDE AND DELIBERATELY NOT PUBLISHED. A published aggregate would become the one envelope that is DERIVED rather than observed, so its staleness would report the aggregator's liveness rather than the producers'; on any defer the consumer needs per-producer detail anyway; and the producer set is deployment-dependent (graph-ingest appears in 18 shipped component instances, graph-index in 8, rule in 8, graph-embedding in 4), so a framework-published aggregate would have to guess its own membership.
The KEY LIST IS THE CONSUMER'S. There is no framework-declared "all producers" list and no optional-key flag — an optional key is one you did not declare. Declaring a key you do not depend on makes you defer on someone else's outage; omitting one you do depend on is the bug this whole change exists to fix, and it is visible in your own call site rather than buried in a framework default.
func NewSet ¶
func NewSet(src BucketSource, keys []string, opts ...Option) *Set
NewSet builds a Set over the given producer keys. Duplicate keys collapse; an empty key list yields a Set that always proceeds, which is correct — a consumer that declared no dependencies has nothing to wait for.
func (*Set) FullyCovered ¶
FullyCovered reports the STRICTER predicate a snapshot caller needs: every declared producer is not merely healthy but has zero outstanding work.
IT IS DELIBERATELY SEPARATE FROM Evaluate AND MUST NOT GATE ANY READ PATH. ADR-085 banned coverage as admission control FOR READS — lag is a property of the answer, not a fault, and gating reads on it makes a healthy system flap unavailable under ordinary write load. That ADR explicitly defers the NON-read case to "that consumer's evidence", and gh#712 is that evidence: a parity snapshot compared two systems while one was still ingesting, and no health signal could have caught it because nothing was unhealthy.
So: use this to decide WHEN TO TAKE A SNAPSHOT, never to decide whether to answer a query. It implies Evaluate's proceed — an unhealthy or unknown producer can never be covered.
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 Verdict ¶
type Verdict struct {
// OK reports that every declared key passed.
OK bool
// Key names the first key (in sorted order) that did not pass. Empty when OK.
Key string
// Reason is that key's typed defer reason. It is DeferNone when OK, and also when
// a key is healthy but merely not drained — FullyCovered's extra condition is not
// a health fault and invents no reason for one.
Reason graph.DeferReason
}
Verdict is one fold's answer. The three values are correlated — Key and Reason are only meaningful together, and only when !OK — so they travel as a named type rather than a positional tuple that drifts and misbinds at call sites.
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