graphview

package
v1.0.0-beta.158 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package graphview provides a shared read-side fan-out primitive over a NATS KV bucket (ADR-081): ONE WatchAll feeding one validated in-memory current-state projection, coalesced to a view-rate tick and fanned out to N local subscribers with snapshot+delta consistency, per-subscriber at-most-once backpressure, and an honest degraded-path contract.

It exists to retire the O(N x writeRate) per-client-watcher trap (gh#579): each write is serialized, decoded, and contract-validated exactly once regardless of subscriber count. The view is domain-agnostic — the bucket handle and a validating decode func are injected; the decode func receives the key, raw value, and authoritative entry metadata (revision + server write time) and decides what a servable record is ((T, keep, err): keep=false maps the key to absent, err poisons the key per ADR-079 semantics).

Coherence: why fan-out is one critical section (G1, the tick seam)

A materialized view IS a cache, so the read-after-write coherence class that bit the graph-ingest read-through cache (PR #583) applies here at two seams:

  • Attach seam: {snapshot capture + subscriber registration} is atomic with delta application at one view sequence S under the projection mutex. The snapshot reflects every applied change at <= S; the subscriber receives exactly the changes > S. Snapshot capture holds the lock (bounded map copy); snapshot delivery happens after unlock.

  • Tick seam: the prior-art coalescers fire their callbacks OUTSIDE their lock, so a detached batch holding K@R5 can be enqueued to a subscriber that attached with a snapshot already at K@R6 — stale delivery, the PR #583 shape one seam over. This package closes that seam STRUCTURALLY: batch detach, subscriber-set iteration, and per-subscriber enqueue are ONE critical section under the projection mutex (fanOut). A detached batch never exists outside the lock, so no apply and no attach can interleave between capture and enqueue — the stale-delivery interleaving is unrepresentable rather than guarded at each enqueue site (a guard every future enqueue path would have to remember).

The work under the lock is bounded map writes (changed keys x subscribers) — never a channel send, decode, or I/O. Actual delivery to subscriber channels happens in per-subscriber goroutines outside the lock, so a slow subscriber degrades to staleness (its pending deltas coalesce last-writer-wins per key, bounded by live changed-key cardinality) and never blocks the watcher, the projection apply, or its peers.

Degraded paths (G5/G6)

Readiness is caught-up, not started: attach and point reads before the initial WatchAll replay completes fail with ErrNotReady. Loss of the shared watcher fails closed — every subscriber's delta channel closes with ErrWatcherLost surfaced via Err(), and the frozen projection is never served as live. Restart re-bootstraps and reconciles ghost keys (keys absent from the fresh replay are removed) before reporting caught-up. Decode/contract failures surface as typed per-key *PoisonError signals in the delta lane and on point reads; they never launder through as upserts and never halt delivery for unrelated keys.

The view coexists with raw WatchAll: consumers needing independent filters, historical replay, per-revision delivery, or independent ack semantics keep opening real JetStream consumers.

Index

Constants

View Source 
const DefaultTickInterval = 250 * time.Millisecond

DefaultTickInterval is the fan-out coalescing window used when no WithTickInterval option is given.

Variables

View Source 
var (
	// ErrNotReady gates attach and point reads before the initial WatchAll
	// replay has completed (readiness = caught-up, not started) and after a
	// watcher loss (fail-closed is a not-ready state).
	ErrNotReady = errors.New("graphview: view not ready")

	// ErrWatcherLost marks the fail-closed state after the shared watcher
	// died (updates channel closed, unrecoverable error, or context
	// cancellation). The frozen projection is never served as live.
	ErrWatcherLost = errors.New("graphview: watcher lost")

	// ErrViewStopped is returned by every operation after Stop, and is the
	// terminal reason reported by Subscription.Err after a view shutdown.
	ErrViewStopped = errors.New("graphview: view stopped")

	// ErrKeyNotFound is returned by Get for a key absent from the projection.
	ErrKeyNotFound = errors.New("graphview: key not found")

	// ErrAlreadyStarted is returned by Start on a view that already started.
	ErrAlreadyStarted = errors.New("graphview: view already started")
)

Sentinel errors returned by view operations. Failed-state errors wrap BOTH ErrNotReady and ErrWatcherLost so callers can gate on not-ready semantics or distinguish "wait for bootstrap" from "the watcher died; restart".

Functions

This section is empty.

Types

type DecodeFunc 

type DecodeFunc[T any] func(key string, value []byte, meta EntryMeta) (decoded T, keep bool, err error)

DecodeFunc decodes and validates one stored KV value; meta carries the entry's authoritative revision and server write time. keep=false means the key is not part of the view (non-record keys, present-but-not-ready records) and maps it to absence; a non-nil err poisons the key (G6) — the value is never delivered as an upsert while other keys keep flowing. Decode runs exactly once per delivered write, on the watcher goroutine, regardless of subscriber count.

type Delta 

type Delta[T any] struct {
	// Op is the operation kind.
	Op DeltaOp
	// Key is the KV key.
	Key string
	// Value is the decoded value; set only for DeltaUpsert.
	Value T
	// Revision is the KV revision of the operation.
	Revision uint64
	// Created is the KV server timestamp of the operation (entry.Created()),
	// carried for every op including tombstones — deletes have no decoded T,
	// so the delta lane is their only channel for the write time.
	Created time.Time
	// Err is the *PoisonError; set only for DeltaPoison.
	Err error
}

Delta is one coalesced per-key operation delivered to subscribers: at most one per changed key per tick window, carrying the greatest-revision operation observed in that window. Values are shared across subscribers (for pointer-shaped T, do not mutate delivered values).

type DeltaOp 

type DeltaOp uint8

DeltaOp is the kind of a Delta: upsert, delete (tombstone), or poison. 

const (
	// DeltaUpsert carries the newest decoded value for a key.
	DeltaUpsert DeltaOp = iota
	// DeltaDelete is a tombstone: the key converged to absence.
	DeltaDelete
	// DeltaPoison is a typed per-key poison signal; Delta.Err holds the
	// *PoisonError.
	DeltaPoison
)

Delta operation kinds. Tombstones and poison signals ride the same ordered, coalesced lane as upserts — last-writer-wins by revision (G4).

func (DeltaOp) String 

func (op DeltaOp) String() string

String implements fmt.Stringer for readable test failures and logs.

type Entry 

type Entry[T any] struct {
	// Value is the decoded value (shared, not copied per subscriber).
	Value T
	// Revision is the KV revision that produced the value.
	Revision uint64
}

Entry is a projection value in a Snapshot.

type EntryMeta 

type EntryMeta struct {
	// Revision is the KV revision of the write being decoded.
	Revision uint64
	// Created is the KV server timestamp of the write (entry.Created()).
	Created time.Time
}

EntryMeta carries the authoritative KV entry metadata into the validating decode: the operation's revision and the server write timestamp (entry.Created()). Revision duplicates Delta.Revision deliberately — decoded records that embed their own write metadata should not need to reach back into the delta lane for it.

type Hooks 

type Hooks struct {
	// OnApply fires after every delivered entry has been applied (including
	// skips and tombstones), with the entry's key and KV revision.
	OnApply func(key string, revision uint64)
	// OnCaughtUp fires when the view transitions to caught-up (initial
	// bootstrap and every successful restart).
	OnCaughtUp func()
	// OnWatcherLost fires once per watcher loss with the cause.
	OnWatcherLost func(err error)
	// OnPoison fires per poisoned write with the key and *PoisonError.
	OnPoison func(key string, err error)
	// OnTick fires after every processed tick window with the number of
	// changed keys detached and the number of subscribers fanned out to.
	OnTick func(changedKeys, subscribers int)
	// OnSubscribers fires with the new subscriber count after every attach
	// and every effective detach (Unsubscribe, subscription-context cancel,
	// and the batch detaches at Stop and watcher loss, which report zero).
	// Counts are delivered outside the projection mutex, so under concurrent
	// attach/detach two callbacks may arrive out of order — treat the value
	// as an eventually-correct gauge, not an ordered event stream.
	OnSubscribers func(n int)
	// OnFanOut fires after every tick window that enqueued at least one
	// changed key to at least one subscriber, with the number of pending
	// deltas that were overwritten before delivery across all subscribers in
	// the window (the at-most-once coalescing drop on undrained buffers —
	// the slow-subscriber staleness signal) and the largest per-subscriber
	// pending-buffer size after enqueue.
	OnFanOut func(overwritten, maxPending int)
}

Hooks receives view-internal signals. All callbacks are optional, are invoked outside the projection mutex, and MUST be fast and non-blocking — they run on the watcher and ticker goroutines. This is the observability seam for metrics (ADR-081 build task 2.5); tests use it for deterministic synchronization.

type KeyedEntry 

type KeyedEntry[T any] struct {
	// Key is the KV key.
	Key string
	// Value is the decoded value.
	Value T
	// Revision is the KV revision that produced the value.
	Revision uint64
}

KeyedEntry is a projection value returned by List.

type Option 

type Option func(*options)

Option configures a View at construction.

func WithHooks 

func WithHooks(h Hooks) Option

WithHooks installs observability hooks.

func WithTickInterval 

func WithTickInterval(d time.Duration) Option

WithTickInterval sets the fan-out coalescing window (default DefaultTickInterval). Non-positive values fall back to the default.

type PoisonError 

type PoisonError struct {
	// Key is the poisoned KV key.
	Key string
	// Revision is the KV revision whose value failed decode.
	Revision uint64
	// Err is the decode/contract error from the injected DecodeFunc.
	Err error
}

PoisonError is the typed per-key poison signal (G6, ADR-079 semantics): a stored value for Key failed the injected validating decode at Revision. It is surfaced in the delta lane (Delta.Op == DeltaPoison, Delta.Err), on point reads of the poisoned key, and in Snapshot.Poisoned — never as a normal upsert. It wraps the decode/contract error so consumer poison latches can errors.As through to their contract-error types.

func (*PoisonError) Error 

func (e *PoisonError) Error() string

Error implements error.

func (*PoisonError) Unwrap 

func (e *PoisonError) Unwrap() error

Unwrap exposes the underlying decode/contract error for errors.Is/As.

type Snapshot 

type Snapshot[T any] struct {
	// Entries maps key to its current decoded value.
	Entries map[string]Entry[T]
	// Poisoned maps key to its sticky poison signal (G6) so consumers that
	// attach after a poison event can still drive their latches.
	Poisoned map[string]*PoisonError
	// Sequence is the view apply-sequence S the snapshot was captured at.
	Sequence uint64
	// Revision is the applied-revision watermark at capture.
	Revision uint64
	// AppliedAt is the KV server write time of the Revision watermark — how
	// current this snapshot was when it was captured, taken in the same
	// critical section as Entries, Sequence, and Revision so it names the same
	// instant. It is REPORTING ONLY (nothing in this package gates on it) and
	// is the zero time when currency is not computable; see View.Applied for
	// the floor, zero-value, and clock-skew contract.
	AppliedAt time.Time
}

Snapshot is the consistent-at-S projection copy returned by SnapshotAndSubscribe: every key applied at sequence <= Sequence is present (or absent if deleted), and the paired subscription delivers exactly the changes after Sequence — no gap, no duplicate, no inversion (G1).

type Subscription 

type Subscription[T any] struct {
	// contains filtered or unexported fields
}

Subscription is one subscriber's attachment to a View. Deltas() delivers coalesced batches; channel close is the explicit terminal signal and Err() reports why (nil after Unsubscribe, the subscription context's error after its cancellation, ErrViewStopped after Stop, a wrapped ErrWatcherLost after a watcher loss). Slowness never disconnects a subscription: an undrained subscriber's pending deltas coalesce last-writer-wins per key until it resumes.

func (*Subscription[T]) Deltas 

func (s *Subscription[T]) Deltas() <-chan []Delta[T]

Deltas returns the delivery channel. Each batch holds at most one delta per key (the greatest-revision operation), sorted by revision ascending. The channel closes on Unsubscribe, subscription-context cancellation, view Stop, or watcher loss — check Err() after close.

func (*Subscription[T]) Err 

func (s *Subscription[T]) Err() error

Err reports the terminal reason once Deltas has closed: nil for a clean Unsubscribe, the context error for a context detach, ErrViewStopped for view shutdown, or a wrapped ErrWatcherLost staleness signal.

func (*Subscription[T]) Unsubscribe 

func (s *Subscription[T]) Unsubscribe()

Unsubscribe detaches the subscription and releases its buffered state. Idempotent; safe to call concurrently with delivery.

type View 

type View[T any] struct {
	// contains filtered or unexported fields
}

View maintains one authoritative in-memory projection of a KV bucket from a single WatchAll and fans coalesced deltas out to local subscribers. It is explicitly constructed and owned — inject it into consumers; there is no process-global registry. All methods are safe for concurrent use.

func New 

func New[T any](source WatcherSource, decode DecodeFunc[T], opts ...Option) (*View[T], error)

New constructs a View over source with the injected validating decode. The view does not touch the bucket until Start.

func (*View[T]) Applied 

func (v *View[T]) Applied() (revision uint64, appliedAt time.Time)

Applied returns the applied-revision watermark together with the KV SERVER WRITE TIME (entry.Created()) of the newest revision the view has applied — the "this view reflects the bucket as of T" pair. It mirrors pkg/revlag.Watermark.IndexedAt, the pair graph-index projects into the ADR-083 staleness_ms field.

Both values are read in ONE critical section, so the revision and the timestamp always name the same write; reading them through two accessors could observe a torn pair.

REPORTING ONLY — no graphview API gates on the timestamp, and none may grow one. This package's gates stay exactly the two it has: ErrNotReady until the initial replay completes (still building) and fail-closed on watcher loss (broken). Age is neither, so a healthy view that is merely behind serves and the CONSUMER stamps currency on its own output, the way community detection reports staleness_at_detection_ms alongside its answer.

It is a FLOOR on currency, not an oracle. Only a DELIVERED write can age it, so a feed that stalls entirely upstream leaves the pair frozen and looks arbitrarily current by this measure alone. Detecting a total stall is the watcher-loss path's job, not this value's; graph.IndexStatusResponse.StalenessMs carries the same caveat for the same reason.

appliedAt is the ZERO time until something has been applied, and stays zero rather than being fabricated as time.Now(). A caller computing an age MUST check IsZero and treat it as "currency not computable" — the contract graph.IndexStatusInputs.IndexedAt uses, where the projection deliberately declines to make an uncomputable view look fresh. It is zero for the same reason when the newest applied write carried no server timestamp (real KV entries always do; fakes and synthetic replays need not).

Subtracting it from a local time.Now() crosses a NATS server clock against a local one and is off by the skew between them. Accepted, the same trade as ADR-083 D3: the alternative — measuring currency in revisions — is wrong by multiples under a coalescing change alone.

func (*View[T]) AppliedRevision 

func (v *View[T]) AppliedRevision() uint64

AppliedRevision returns the applied-revision watermark: the highest KV revision the view has applied. WatchAll delivery is ordered, so the watermark reaching R proves every delivered revision <= R was applied.

A caller that also wants the watermark's write time MUST use Applied instead of pairing this with a second read — the two would come from different critical sections and could name different writes.

func (*View[T]) CaughtUp 

func (v *View[T]) CaughtUp() bool

CaughtUp reports whether the view is live: initial replay complete and the shared watcher healthy (readiness = caught-up, not started).

func (*View[T]) Get 

func (v *View[T]) Get(key string) (T, uint64, error)

Get is the coherent point read: it serves the projection under the same mutex applies run under, so it can never return a value older than one the view already applied. It honors readiness (ErrNotReady before caught-up, wrapped ErrWatcherLost after a loss, ErrViewStopped after Stop) and poison (a poisoned key returns its *PoisonError, never a stale value).

func (*View[T]) List 

func (v *View[T]) List(prefix string, limit int) ([]KeyedEntry[T], error)

List returns projection entries whose key has the given prefix (empty prefix matches all), sorted by key, truncated to limit when limit > 0. The complete match set is sorted before the limit is applied (deterministic bounded reads). Poisoned keys are excluded — they have no servable value. It honors the same readiness gate as Get.

func (*View[T]) Poisoned 

func (v *View[T]) Poisoned() map[string]*PoisonError

Poisoned returns a copy of the current per-key poison records (G6). It is a diagnostic surface and is served in every state.

func (*View[T]) Restart 

func (v *View[T]) Restart() error

Restart re-bootstraps a failed view: it opens a fresh WatchAll, replays, reconciles ghost keys (keys absent from the fresh replay are removed), and only then reports caught-up again (G5). Subscribers terminated by the loss must re-attach for a coherent snapshot. The new watcher inherits the Start context — jetstream binds watcher lifetime to the WatchAll context, so a per-call context would kill the watcher when the call returned; if the Start context itself was cancelled, Restart fails.

func (*View[T]) SnapshotAndSubscribe 

func (v *View[T]) SnapshotAndSubscribe(ctx context.Context) (Snapshot[T], *Subscription[T], error)

SnapshotAndSubscribe atomically captures a snapshot and registers the subscription at one view sequence S under the projection mutex (G1): the snapshot holds every change applied at <= S, the subscription delivers exactly the changes > S. Capture holds the lock (bounded copy); delivery to the caller happens after unlock. ctx bounds the subscription: its cancellation detaches like Unsubscribe. Fails with ErrNotReady until the view is caught up.

func (*View[T]) Start 

func (v *View[T]) Start(ctx context.Context) error

Start opens the single WatchAll and begins bootstrap. ctx bounds the life of the watcher and ticker; its cancellation is a watcher-loss event (fail closed), not a clean shutdown — use Stop for that.

func (*View[T]) Stop 

func (v *View[T]) Stop()

Stop shuts the view down: every subscription receives an explicit terminal close (Err reports ErrViewStopped — never a silent hang), the watcher and ticker exit, and all further operations return ErrViewStopped.

func (*View[T]) Subscribe 

func (v *View[T]) Subscribe(ctx context.Context) (*Subscription[T], error)

Subscribe is the delta-only attach for trigger-shaped consumers: no snapshot, no second projection copy — the subscription delivers the changes after the attach sequence. Same readiness gate as SnapshotAndSubscribe.

func (*View[T]) WaitCaughtUp 

func (v *View[T]) WaitCaughtUp(ctx context.Context) error

WaitCaughtUp blocks until the view is caught up (nil), the view fails (wrapped ErrWatcherLost), the view stops (ErrViewStopped), or ctx is done.

type WatcherSource 

type WatcherSource interface {
	WatchAll(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error)
}

WatcherSource is the narrow method set the view needs from a KV bucket. jetstream.KeyValue satisfies it; unit tests inject a deterministic fake.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.