projection

type Apply ¶

type Apply = cellvocab.ProjectionApply

Apply is the business event→state projection hook: given a consumed ProjectionEvent, mutate the read-model. The transaction is ambient — Apply obtains it via persistence.TxFromContext(ctx) exactly like outbox.Writer.Write, because the Coordinator invokes Apply inside persistence.TxRunner.RunInTx so the read-model mutation and the checkpoint advance commit atomically (exactly-once delivery; the harness never calls Apply twice for the same offset).

Apply MUST NOT open its own transaction or connection. A transient failure returns a plain error (the Coordinator requeues). A permanent failure returns an error wrapping outbox.NewPermanentError(err); the Coordinator classifies it as DispositionReject and routes to the DLX — the same vocabulary as the ConsumerBase handler convention (see .claude/rules/gocell/eventbus.md). Decided in ADR §3 Q2 against the eventhorizon read-modify-write entity shape and the explicit tx-handle parameter.

It is an alias of cellvocab.ProjectionApply (also aliased by kernel/cell.ProjectionApply), the single underlying type that lets the bootstrap drain pass a cell-recorded hook to Coordinator.Subscribe with no named-type conversion.

ref: JasperFx/marten async-daemon IDocumentOperations apply shape.

type CheckpointStore ¶

type CheckpointStore interface {
	// LoadOffset returns the last committed offset for the projection, or 0 if
	// none has been recorded yet (cold start).
	LoadOffset(ctx context.Context, cellID, projectionID string) (int64, error)
	// SaveOffset advances the projection's committed offset within the ambient
	// transaction carried by ctx.
	SaveOffset(ctx context.Context, cellID, projectionID string, offset int64) error
}

CheckpointStore persists a projection's consumed offset. It is the framework's own offset table — it does NOT touch any business read-model schema (the CellTx-offset design keeps the harness clear of the GAP-8 seal; ADR §4).

The offset is an opaque, monotonically increasing cursor over the projection's input stream. Its concrete mapping to a stream position is owned by the replay source defined in PR-01 (the harness compares a replayed event's position against the stored checkpoint to skip already-applied events) — it is NOT an outbox.Entry field (Entry carries no sequence number today). LoadOffset returns 0 for an unknown (cellID, projectionID) pair (cold start = offset 0).

Both methods are ambient-tx: SaveOffset participates in the caller's transaction via persistence.TxFromContext(ctx) (no raw db handle — enforced by PROJECTION-CHECKPOINT-TX-BOUND-01), so the offset advance commits together with the Apply mutation.

Implementations: mem (PR-01) + postgres (PR-02); both verified by the shared projectiontest.RunCheckpointConformance template (PR-01). Decided in ADR §3 Q1 (caller-provided tx + harness-internal SaveOffset, mirroring outbox.Writer and Axon's JdbcTokenStore same-tx commit).

type Coordinator ¶

type Coordinator struct {
	// contains filtered or unexported fields
}

Coordinator wires an event subscription to an Apply function, managing exactly-once delivery via a CheckpointStore. Each consumed event is processed inside a TxRunner.RunInTx so that the Apply mutation and the checkpoint advance commit atomically.

Per-projection model (PR-03) ¶

Each Coordinator is per-projection (projectionID moves from Subscribe to NewCoordinator). Subscribe may only be called once; a second call returns an error. This is a Hard constraint enforced by a subscribed atomic.Bool CAS. A cell hosting N projections creates N Coordinators.

Gate mechanism (PR-03) ¶

During a rebuild, the live event handler parks on an unbuffered channel gate (liveGate). An OPEN gate is a CLOSED channel (reads pass immediately); a SHUT gate is a fresh open (non-closed) channel (reads block). Only the single rebuild goroutine calls shutGate/openGate.

ref: JasperFx/marten async-daemon ProjectionDaemon. ref: AxonFramework TrackingEventProcessor.

func NewCoordinator(clk clock.Clock, cfg CoordinatorConfig) (*Coordinator, error)

func (c *Coordinator) Close(ctx context.Context) error

func (c *Coordinator) Phase() Phase

func (c *Coordinator) Probes() ([]healthz.Probe, error)

func (c *Coordinator) Rebuild(ctx context.Context) error

Stop → Reset → Replay → Catchup → Live

func (c *Coordinator) Snapshot(ctx context.Context) (Snapshot, error)

func (c *Coordinator) Subscribe(
	ctx context.Context,
	spec contractspec.ContractSpec,
	apply Apply,
	opts ...Option,
) error

type CoordinatorConfig ¶

type CoordinatorConfig struct {
	Registrar    SubscribeRegistrar
	CellID       string
	ProjectionID string
	TxRunner     persistence.TxRunner
	Store        CheckpointStore
	Cursor       Cursor
	Replay       ReplaySource
	Tracer       wrapper.Tracer
	Metrics      *Metrics // optional; nil = instruments disabled
}

CoordinatorConfig bundles the Coordinator's dependencies. It mirrors the kernel many-dependency convention NewConsumerBase(claimer, ConsumerBaseConfig, clk): required deps live in a config struct validated with hand-written nil guards, while clk is a separate positional parameter to NewCoordinator.

clk is deliberately NOT a field here: CLOCK-POSITIONAL-INJECTION-01 mandates a positional clock parameter and forbids a Clock field on input config structs.

Metrics is optional (nil disables all instruments). Every other field is required; a nil/empty value is rejected by NewCoordinator. Grouping the deps in a struct (rather than 9 positional params) removes the silent-swap hazard of the two adjacent string fields CellID/ProjectionID.

ref: open-source consensus — Watermill cqrs.EventProcessorConfig, Axon TrackingEventProcessor.Builder; kernel kernel/outbox.ConsumerBaseConfig.

type Cursor ¶

type Cursor interface {
	Position(entry ProjectionEvent) (int64, error)
}

Cursor maps a consumed event to its monotonic stream position. The ProjectionEvent carrier exposes no sequence field — the position is supplied by the replay source.

Position invariants (required of every implementation) ¶

1. Monotonic: within the same (cellID, projectionID) stream, Position is non-decreasing across events delivered in order. DISTINCT events MUST get STRICTLY INCREASING positions — if two distinct events shared a position, the Coordinator's pos <= checkpoint guard would silently skip the second after the first commits the checkpoint (a projection gap), so a coarser cursor is unsound. The ONLY valid equality is RE-DELIVERY of an already-applied event, which returns its same previously-assigned position; that guard handles idempotent re-delivery. (RunCursorConformance asserts strict increase across distinct seeded entries accordingly.)

2. 1-based: every valid event position is ≥ 1. The value 0 is reserved to mean "no checkpoint / cold start" (the default returned by CheckpointStore on first read). A Cursor must never return 0 for a real event.

3. Gap-allowed: the position sequence may have gaps (e.g. jumping from 3 to 7). The Coordinator applies both ends of the gap as they arrive; events for the missing positions 4–6 will be skipped if they arrive later because their position is ≤ the checkpoint. This is expected and correct — gaps arise when the event source does not emit every sequence number.

4. Resolution error semantics: an error returned by Position is treated as transient by default (Coordinator requeues the event for retry). Wrap the error in outbox.NewPermanentError to signal that the event is unrecoverable and should be routed to the dead-letter exchange.

PR-01 shipped this interface and the MemCursor test fake; the production outbox-journal-backed cursor (adapters/postgres.PGProjectionCursor, reading outbox_entries.seq) landed in PR-04c (#1368). Note the v1 limitation: positions come from the transient outbox relay, so the cursor/replay are gated to dev/preview until a durable projection journal lands (#1504).

ref: Axon TrackingToken (position is a property of the token store / stream).

type MemCheckpointStore ¶

type MemCheckpointStore struct {
	// contains filtered or unexported fields
}

MemCheckpointStore is an in-process CheckpointStore backed by a plain map. It is suitable for tests and demos; it does NOT participate in any database transaction (ctx is accepted but unused — ambient-tx binding is the caller's concern). Durability is process-lifetime only.

Thread-safe: concurrent SaveOffset / LoadOffset calls are protected by a RWMutex.

func NewMemCheckpointStore() *MemCheckpointStore

func (m *MemCheckpointStore) LoadOffset(_ context.Context, cellID, projectionID string) (int64, error)

func (m *MemCheckpointStore) SaveOffset(_ context.Context, cellID, projectionID string, offset int64) error

type MemCursor ¶

type MemCursor struct {
	// contains filtered or unexported fields
}

MemCursor is an in-process Cursor paired with a MemReplaySource: it resolves an entry's monotonic position by looking it up in the source. Suitable for tests and demos only.

Position returns the 1-based insertion index of the entry in the source (same value as MemReplaySource.Position). If the entry is not present in the paired source, Position returns 0 and a permanent error — retry cannot fix a missing source entry.

ref: Axon TrackingToken (position is a property of the token store / stream).

func NewMemCursor(src *MemReplaySource) (*MemCursor, error)

func (c *MemCursor) Position(entry ProjectionEvent) (int64, error)

type MemReplaySource ¶

type MemReplaySource struct {
	// contains filtered or unexported fields
}

MemReplaySource is an in-process append-only ReplaySource backed by a plain slice. It is suitable for tests and demos only; it does NOT participate in any database transaction and provides no durability.

Position scheme ¶

Position is 1-based insertion index: the first Append call assigns position 1, the second assigns position 2, etc. This scheme is deterministic, gap-free, and agrees with the paired test memCursor.Position(e) implementation that uses positionOf(e) to look up the 1-based index.

Event identity ¶

Position resolution matches events by their EventID (the ProjectionEvent carrier's identity accessor), which the carrier contract requires to be unique across the whole replay source — so the lookup is collision-free even when test events are created in the same nanosecond and does not rely on any pointer or timestamp identity.

Thread-safe: all methods are protected by a RWMutex.

func NewMemReplaySource() *MemReplaySource

func (m *MemReplaySource) Append(entry ProjectionEvent)

func (m *MemReplaySource) Head(_ context.Context) (int64, error)

func (m *MemReplaySource) Position(e ProjectionEvent) int64

func (m *MemReplaySource) Replay(ctx context.Context, fromOffset int64, fn func(ProjectionEvent) error) error

type Metrics ¶

type Metrics struct {
	// ReplayLag is a gauge tracking the event lag in seconds:
	// lag = now − lastApplied.OccurredAt(). Labels: {cell, projection}.
	ReplayLag kernelmetrics.GaugeVec
	// RebuildDuration is a histogram observing rebuild wall-clock duration in
	// seconds. Labels: {cell, projection}.
	RebuildDuration kernelmetrics.HistogramVec
	// PendingEvents is a gauge tracking replay.Head() − checkpoint: the number
	// of events not yet reflected in the projection. Labels: {cell, projection}.
	PendingEvents kernelmetrics.GaugeVec
}

Metrics holds the optional pre-bound instruments the Coordinator records to. A nil *Metrics (or nil individual fields) disables that instrument. Build via RegisterMetrics and inject into NewCoordinator as an optional parameter (nil = instruments disabled, cloned from kernel/reconcile/metrics.go pattern).

func RegisterMetrics(p kernelmetrics.Provider) (*Metrics, error)

type OnReset ¶

type OnReset = cellvocab.ProjectionResetHook

OnReset is the business hook invoked during a projection rebuild Reset phase. It allows the projection owner to clear its read-model state (e.g. TRUNCATE a view table) before the harness resets the checkpoint offset to 0 and begins replay. The transaction is ambient — OnReset obtains it via persistence.TxFromContext(ctx), exactly like Apply. Both OnReset and the SaveOffset(0) call share the same transaction; if OnReset returns an error, the whole transaction rolls back and the read-model is left untouched.

Passing nil is valid (projection has no read-model table to clear — offset reset alone suffices). OnReset is set via WithOnReset option on Subscribe.

It is an alias of cellvocab.ProjectionResetHook (also aliased by kernel/cell.ProjectionResetHook).

ref: AxonFramework @ResetHandler — called during TrackingEventProcessor reset to let the projection clear application state before replay.

type Option ¶

type Option func(*subscribeOptions)

Option configures a projection subscription. It is the frozen functional-option seam consumed by Coordinator.Subscribe (PR-01); concrete option constructors (e.g. starting offset, fail-open policy) are added alongside the Coordinator. Declared here so the Subscribe API surface is fixed by the ADR rather than drifting when the implementation lands.

Available option constructors:

• WithOnReset — registers an OnReset hook invoked during the rebuild Reset phase.

Passing no opts (an empty or nil slice) is valid.

func WithOnReset(fn OnReset) Option

type Phase ¶

type Phase uint8

Phase is the lifecycle state of a projection, exposed by the Coordinator (PR-03) so operators (readyz / metrics) and business read paths can observe where a projection is. The four rebuild phases mirror Axon's processor lifecycle (Stop → Reset → Replay → Catch-up); PhaseLive is the steady-state value outside a rebuild.

Business read-503 opt-in ¶

rebuild does NOT block business reads by default — stale reads during rebuild are the industry consensus (Axon / Marten / Commanded). A business read path MAY consult Phase() and return 503 of its own accord; the harness never forces it. PhaseLive means the read-model is fresh and serving; any other phase means a rebuild is in progress. See ADR §5.

Phase() is a best-effort point-in-time snapshot, not a freshness lock: a read path that observes PhaseLive and then queries the read-model has no atomic ordering against a rebuild starting between the two calls. Business code requiring strict freshness must maintain its own read quiescence.

Frozen membership ¶

The const set is frozen by archtest PROJECTION-STATE-PHASE-FROZEN-01 (the operational contract for readyz/metrics/503 must not drift silently). The rebuild state machine + transition table that drive these phases land in PR-03.

ref: kernel/outbox/state.go (enum + String pattern). ref: AxonFramework EventProcessor lifecycle (Stop/Reset/Replay/Catch-up).

const (
	// PhaseLive: steady-state consuming, read-model fresh and serving.
	//
	// IMPORTANT: iota+1 makes the zero value (0) an invalid Phase, so a
	// forgotten/uninitialised Phase cannot silently appear as PhaseLive.
	PhaseLive Phase = iota + 1 // = 1
	// PhaseStopped: rebuild step 1 — consume loop halted, checkpoint claim
	// released so the reset/replay can rewrite the offset row safely.
	PhaseStopped
	// PhaseReset: rebuild step 2 — business OnReset hook (TRUNCATE/DROP its
	// read-model) + harness resets the checkpoint offset to 0, one CellTx.
	PhaseReset
	// PhaseReplay: rebuild step 3 — replaying the event stream from offset 0,
	// each event in its own CellTx (Apply + checkpoint).
	PhaseReplay
	// PhaseCatchup: rebuild step 4 — replay reached the head offset captured at
	// rebuild start; now consuming newly arriving events until caught up, then
	// transition back to PhaseLive.
	PhaseCatchup
)

func (p Phase) String() string

func (p Phase) Valid() bool

type ProjectionEvent ¶

type ProjectionEvent = cellvocab.ProjectionEvent

ProjectionEvent is the minimal typed read-only carrier the harness applies: ReplaySource.Replay's callback, Cursor.Position, and Apply all carry a ProjectionEvent rather than the concrete outbox.Entry, so the outbox event source and a saga-journal event source (EPIC #1609 PR-03) share one typed funnel. It is an alias of cellvocab.ProjectionEvent — the carrier type lives in the cellvocab leaf because kernel/projection imports kernel/cell (whose ProjectionApply mirror references the same carrier), so the type must live in a package both import without a cycle. See cellvocab.ProjectionEvent for the method contract and the not-sealed rationale; carrier-shape enforcement is archtest PROJECTION-EVENT-CARRIER-TYPED-01.

type ReplaySource ¶

type ReplaySource interface {
	// Replay iterates events with position strictly greater than fromOffset in
	// ascending position order, calling fn for each. fromOffset==0 replays all
	// events. If fn returns an error, Replay stops immediately and returns that
	// error; events already passed to fn are NOT retried.
	Replay(ctx context.Context, fromOffset int64, fn func(ProjectionEvent) error) error

	// Head returns the highest available position in the source, or 0 if the
	// source is empty. Head is used to determine the rebuild cutoff and to
	// compute the pending_events metric (Head − checkpoint).
	Head(ctx context.Context) (int64, error)
}

ReplaySource is the read-model event store interface consumed by a Coordinator rebuild. It delivers past events in deterministic ascending position order so the rebuild can apply them exactly once from offset 0 (or any resume point).

Position scheme ¶

Positions are 1-based monotonically increasing integers. Position 0 means "no events / cold start" (same sentinel as CheckpointStore.LoadOffset cold start). Each event returned by Replay has a position > fromOffset, and positions across multiple Replay calls on the same source are stable (deterministic, not ephemeral).

Transaction semantics ¶

ReplaySource.Replay is NOT transactional: it iterates and calls fn for each qualifying event. The Coordinator wraps each fn call in RunInTx so that the Apply and SaveOffset commit atomically. Replay does not open or join any ambient transaction — it is the caller's responsibility to manage tx boundaries around fn.

Replay duration bound (v1) ¶

In v1, Replay is bounded only by ctx cancellation: there is no max-entries or max-duration parameter. Operators bound replay duration by setting a timeout on the rebuild context (the ctx passed to Rebuild, which is forwarded as the runRebuild ctx). Per-batch limiting is deferred to a future version when large-history projections require it.

The carrier is the typed ProjectionEvent interface (EPIC #1609 PR-01): the outbox-backed source (adapters/postgres.PGProjectionReplaySource) and a future saga-journal source both satisfy it; neither leaks the concrete outbox.Entry.

ref: AxonFramework EventStore — ordered event stream replay by position/token. ref: JasperFx/marten IDocumentSession.Events.QueryAllRawEvents — append-only event store replay.

type Snapshot ¶

type Snapshot struct {
	Phase            Phase
	PendingEvents    int64
	ReplayLagSeconds float64
}

Snapshot is a point-in-time view of a projection's lifecycle and replay lag, returned by the rebuild control-plane endpoint as the 202 response body ({phase, pendingEvents, replayLagSeconds}).

Phase is always populated (read from the in-memory atomic — no I/O, never fails). PendingEvents and ReplayLagSeconds are best-effort: on a checkpoint-store / replay read error they are zero and Snapshot returns a non-nil error. Callers (the rebuild handler) log the error but MUST NOT fail an already-admitted rebuild over a degraded snapshot read — the rebuild has been accepted regardless of whether its current lag could be read.

ReplayLagSeconds == 0 is ambiguous by design and must be read together with PendingEvents: it means either "caught up" (PendingEvents == 0) OR "lag unknown" (PendingEvents > 0 but nothing applied yet — cold start / fresh rebuild, so there is no last-applied time to measure from). The lag readyz probe distinguishes these (it leaves the lag gauge unwritten during the startup-grace case rather than reporting a false-healthy zero).

type SubscribeRegistrar ¶

type SubscribeRegistrar interface {
	Subscribe(
		spec contractspec.ContractSpec,
		handler outbox.EntryHandler,
		consumerGroup string,
		cellID string,
		opts ...cell.SubscriptionOption,
	) error
}

SubscribeRegistrar is the minimal cell.Registrar surface the Coordinator needs: it registers the projection's wrapped event subscription. The Coordinator holds this narrow interface rather than the full cell.Registrar so that (a) the dependency is honest (the Coordinator only ever calls Subscribe) and (b) a holder injected at wiring time — e.g. the bootstrap projection drain's capture adapter — need not stub the other Registrar methods (no nil-embed foot-gun). cell.Registrar satisfies this interface structurally.

The signature mirrors cell.Registrar.Subscribe exactly so a *cell.RegistryRecorder (or the drain's capture adapter) is assignable without conversion.