journal

type ClaimedInstance ¶

type ClaimedInstance struct {
	Instance saga.Instance
	LeaseID  idutil.SafeID
}

ClaimedInstance is a saga.Instance projection plus the fencing token the caller MUST echo to every lease-fenced mutator (Append, Heartbeat, MarkTerminal) for that instance. The LeaseID is minted by the ClaimPending sweep that returned the instance; once the lease is lost — reclaimed after expiry or superseded by a newer claim — every fenced operation presenting this LeaseID becomes a no-op (Heartbeat/MarkTerminal return ok=false) or a conflict error (Append).

Tip: pass ci.Instance.ID as instanceID and ci.LeaseID as leaseID to the fenced mutators (both are idutil.SafeID; order matters).

type Enqueuer ¶

type Enqueuer interface {
	Enqueue(ctx context.Context, instance saga.Instance) error
}

Enqueuer is the producer-facing narrow journal interface: saga enrollment only. A business producer (e.g. an HTTP slice that places orders) needs exactly Enqueue and none of the coordinator-only claim/append/commit surface, so it holds Enqueuer rather than JournalCore.

MemJournal and any PostgreSQL implementation satisfy Enqueuer automatically because both satisfy the full Journal (which embeds JournalCore, which contains Enqueue). No implementation change is required.

type Event ¶

type Event struct {
	Version int64 // per-instance position, assigned by Append; zero on input
	// GlobalSeq is the cross-instance global position — monotonic over the whole
	// journal, NOT per-instance like Version — assigned by the Journal on append
	// (in-memory from a global counter; PG from a BIGINT IDENTITY column, PR-PG).
	// It is the stable total order [GlobalReader] scans by and the value a
	// projection cursor checkpoints on (see [GlobalEvent]). Additive (#1609
	// PR-02): callers leave it zero on input, and a backend that does not yet
	// maintain a global sequence (PG until PR-PG) leaves it zero on the way out.
	GlobalSeq int64
	Kind      EventKind     // required, must be Valid
	StepName  idutil.SafeID // required for step kinds; empty for CompensationStarted / terminal kinds
	Payload   []byte        // opaque JSON object-or-null; copied defensively by the Journal
	CreatedAt time.Time     // stamped by the Journal on Append
}

Event is one durable, append-only record in a saga instance's history. The owning instance is identified by the Append(instanceID, ...) argument, so it is deliberately not duplicated as a field here. Version is assigned by Journal.Append (monotonic per instance, starting at 1) and CreatedAt is stamped by the Journal from its injected clock; callers leave both zero on the way in.

func (e Event) ValidateForAppend() error

type EventKind ¶

type EventKind uint8

EventKind classifies a durable saga journal event. The vocabulary is fully state-transition-bearing: every lifecycle move a saga makes has a corresponding event kind, so the append-only log alone replays the complete history — including WHICH terminal state was reached and WHEN the rollback phase began. The instance Status projection is a deterministic fold of these kinds, never a separate source of truth.

IMPORTANT: iota+1 ensures the zero value (0) is NOT a valid kind, so a forgotten/uninitialised Event.Kind never silently appears valid — the same discipline as saga.Status.

const (
	// Step-level events — appended via Append under the holder's lease; each
	// carries a StepName.
	KindStepStarted     EventKind = iota + 1 // step_started
	KindStepCompleted                        // step_completed
	KindStepFailed                           // step_failed
	KindStepCompensated                      // step_compensated — a compensation step result; legal only while Compensating

	// KindCompensationStarted marks the saga entering the rollback phase
	// (Running → Compensating). Appended via Append; saga-scoped (no StepName).
	// The Coordinator appends it when it DECIDES to compensate, before any
	// compensation runs — so a leader handoff mid-rollback reads Compensating,
	// not Running.
	KindCompensationStarted // compensation_started

	// Terminal events — appended ONLY via MarkTerminal, atomically with the
	// projection's terminal status flip. The terminal status is encoded in the
	// kind itself, so the log alone replays the final state.
	KindSagaSucceeded   // saga_succeeded
	KindSagaFailed      // saga_failed
	KindSagaCompensated // saga_compensated
	KindSagaExpired     // saga_expired

	// KindStepCompensationFailed — a compensation step's CompensateFunc returned
	// a non-nil error. Legal only while Compensating; status unchanged. Issued
	// per-step (carries StepName) and distinct from KindStepFailed (which is a
	// forward Step.Run failure during Running phase). Appended at iota position
	// 10 (after the terminal kinds) so existing kind wire values 1..9 stay
	// stable — adding a new kind in the middle would re-number persisted rows.
	// Introduced by #1181 to give reverse-compensation failures a wire-distinct
	// vocabulary (previously runCompensation co-opted KindStepFailed, which
	// the Compensating-phase projection rejected).
	KindStepCompensationFailed // step_compensation_failed

	// KindSagaCompensationFailed is the terminal journal event for
	// saga.StatusCompensationFailed. Written exclusively via MarkTerminal when
	// the compensation phase completed but at least one CompensateFunc returned a
	// non-nil error. Wire value 11 (appended after KindStepCompensationFailed at
	// position 10) to keep existing kind values 1..10 stable.
	// Introduced by #1210 to give the "rollback failed" terminal state a
	// dedicated event kind distinct from KindSagaFailed ("forward failed").
	KindSagaCompensationFailed // saga_compensation_failed
)

func TerminalEventKind(s saga.Status) (EventKind, bool)

func (k EventKind) IsTerminal() bool

func (k EventKind) String() string

func (k EventKind) Valid() bool

type GlobalEvent ¶

type GlobalEvent struct {
	GlobalSeq  int64         // == Event.GlobalSeq; the stable total-order position (cursor reads this)
	InstanceID idutil.SafeID // the saga instance this event belongs to
	Event      Event         // the durable event (Event.GlobalSeq == GlobalSeq)
}

GlobalEvent is one event returned by GlobalReader, carrying the global position and the owning instance ID alongside the Event itself. InstanceID is surfaced here because Event deliberately omits its owning instance (it is addressed by the Append argument in the per-instance API); a cross-instance scan must report it.

GlobalSeq is intentionally exposed at the envelope level (and mirrors the inner Event.GlobalSeq — the two are always equal) so a tailing consumer can read its cursor position without descending into the embedded Event, the same convention as a Kafka ConsumerRecord.offset or an outbox seq on the record wrapper. The downstream PR-03 SagaJournalSource cursor MUST read GlobalEvent.GlobalSeq (the public envelope path), not GlobalEvent.Event.GlobalSeq.

type GlobalReader ¶

type GlobalReader interface {
	// LoadSince returns events whose GlobalSeq is strictly greater than
	// afterGlobalSeq (exclusive lower bound = projection-checkpoint cursor
	// semantics), in ascending GlobalSeq order, at most limit events. Passing
	// afterGlobalSeq=0 scans from the beginning.
	//
	// Caught-up signal: a cursor equal to or greater than HeadSeq returns an
	// empty slice and a NIL error (so does an empty store) — a tailing loop polls
	// this to detect "no new events" and back off, distinguishing it from the
	// error cases purely by the nil error. limit MUST be > 0 and afterGlobalSeq
	// MUST be >= 0; a non-positive limit or negative cursor returns a KindInvalid
	// error (fail-closed, like ClaimPending's batchSize).
	LoadSince(ctx context.Context, afterGlobalSeq int64, limit int) ([]GlobalEvent, error)

	// HeadSeq returns the highest GlobalSeq currently assigned across all
	// instances — the upper bound a projection has yet to consume. An empty
	// journal returns 0 (no event ever carries GlobalSeq 0; the counter starts
	// at 1), so a fresh projection cursor of 0 means "caught up" iff HeadSeq==0.
	HeadSeq(ctx context.Context) (int64, error)
}

GlobalReader is the projection-facing global scan interface (#1609 PR-02): a cross-instance, append-order read of the event log, distinct from Reader (per-instance Load). A model-A CQRS projection (Axon Tracking Event Processor style) replays and tails the whole journal through it — the only way to observe saga terminal states (succeeded / compensated / failed), which are written ONLY to saga_events and never emitted to the outbox.

It is deliberately NOT folded into JournalCore: the single-sanctioned-holder seal (SAGA-JOURNAL-HOLDER-SEAL-01) restricts JournalCore field holders to the Coordinator, whereas the PR-04 projection Tailer must hold a global reader without driving sagas. The narrow type is therefore a CONSUMER view (it lets the Tailer hold a 2-method reader, structurally unable to claim/append/commit) — NOT a contract for read-only backends. Every GlobalReader implementation is itself the append-only saga_events store and hence a full Journal (MemJournal today; PGJournal at PR-PG); there is no store-less, read-only-only implementation in this design.

Events are ordered by Event.GlobalSeq, a stable total order across all instances. MemJournal assigns it from an in-process counter; the PG backend assigns it from a BIGINT IDENTITY column (PR-PG, deferred — until then PG does not implement GlobalReader). Every GlobalReader implementation is enrolled in the sagajournaltest.RunGlobalReaderConformance suite, enforced by archtest SAGA-GLOBALREADER-CONFORMANCE-ENROLL-01.

Position 0 is reserved: a conforming implementation never returns an event with GlobalSeq==0 (the counter starts at 1). The zero value only ever appears on events written by a backend that does not yet maintain a global sequence (none today — PG, the only such backend, does not implement GlobalReader until PR-PG). Consumers therefore use 0 as the "from the beginning" cursor and HeadSeq==0 as the empty-journal sentinel; they must never checkpoint a real position of 0.

type Heartbeater ¶

type Heartbeater interface {
	// Heartbeat extends the lease on a single claimed instance to
	// now+leaseDuration. leaseDuration MUST be > 0; a non-positive value returns a
	// KindInvalid error. It is lease-fenced: ok is false (with a nil error) when
	// leaseID no longer owns the instance, signaling the holder to stop driving
	// it. A never-enqueued instance also returns ok=false (nil error); callers
	// cannot distinguish it from a stale lease.
	Heartbeat(ctx context.Context, instanceID, leaseID idutil.SafeID, leaseDuration time.Duration) (ok bool, err error)
}

Heartbeater is the single lease-renewal method split out of Journal (#1209).

runtime/saga/executor owns the only sanctioned Heartbeat caller (the per-step heartbeat goroutine) and declares its OWN structurally-identical Heartbeater interface rather than importing this one: kernel/ cannot depend on runtime/, and the executor must never import kernel/saga/journal (SAGA-JOURNAL-HOLDER-SEAL-01). This kernel-side Heartbeater exists solely to compose the full Journal so that the PG / in-memory implementations satisfy it via interface embedding.

type Journal ¶

type Journal interface {
	JournalCore
	Heartbeater
}

Journal is the append-only durable event log plus lease-coordinated instance projection that L3 saga orchestration is built on. Two truths live behind one interface: the append-only event log is the source of truth for history/replay, while the projection (status, current version, lease) is the source of truth for coordination. An implementation keeps the two consistent by mutating the projection in the same critical section / transaction as the event append.

The interface is storage-dialect-neutral: no *sql.Tx or driver type appears in any signature, so the in-memory and PostgreSQL implementations satisfy the exact same contract and the same conformance suite (see kernel/saga/sagajournaltest).

Interface split: JournalCore + Heartbeater (#1209) ¶

Journal is the union of two narrower interfaces:

• JournalCore — enrollment, the append-only log, claim/projection, and terminal commit: everything a coordinator needs EXCEPT lease renewal.
• Heartbeater — the single Heartbeat lease-renewal method.

runtime/saga.Coordinator holds only JournalCore, so a centralized heartbeat loop becomes a compile error — the persisted field cannot express Heartbeat. Per-step lease renewal is funneled through runtime/saga/executor, which receives the full Journal value transiently at construction (NewExecutor). See archtests SAGA-COORDINATOR-NO-HEARTBEAT-LOOP-01 and SAGA-JOURNAL-HOLDER-SEAL-01. PG and in-memory implementations satisfy the full Journal, hence both halves.

Time ¶

Unlike the pure saga state machine (saga.AdvanceSaga takes an explicit now), the Journal owns lease-expiry arithmetic, so it sources time from a clock injected at construction rather than from a per-call now argument.

Lease fencing ¶

Enqueue is lease-free open enrollment: producers create instances before any leader exists. Every mutator that runs under a leader (Append, Heartbeat, MarkTerminal) is CAS-fenced by leaseID. The fencing return convention is deliberately asymmetric:

• Append returns a conflict error on a stale lease, because silently dropping a step event would desynchronise the leader's view of the instance version. A zombie leader must fail loudly.
• Heartbeat and MarkTerminal return ok=false (no error) on a stale lease, because losing a lease is an expected race during leader handoff, not a fault — the caller simply stops driving that instance.

Single sanctioned holder ¶

Only runtime/saga.Coordinator is intended to hold a JournalCore field; no struct in runtime/saga may persist the Heartbeat-bearing full Journal (or a bare Heartbeater) as a field. Both rules are enforced by the SAGA-JOURNAL-HOLDER-SEAL-01 archtest. Enqueue is the producer-facing entry point; a narrower producer interface may be split out if a real producer ever needs less than JournalCore.

type JournalCore ¶

type JournalCore interface {
	// Enqueue enrolls a new saga instance for orchestration. The instance MUST
	// pass saga.Instance.ValidateNew (Pending, CurrentStep 0, no timestamps
	// beyond StartedAt). Enqueue is lease-free: it writes the projection with
	// current version 0 and no lease, and writes no event (events describe step
	// execution, which has not begun). Re-enqueuing an existing ID returns a
	// KindConflict error with code errcode.ErrSagaDuplicateInstance.
	//
	// Load after Enqueue returns an empty, non-nil slice (see Load); a consumer
	// detects a not-yet-started instance by the empty event log.
	Enqueue(ctx context.Context, instance saga.Instance) error

	// Append durably records one Event for the instance under the holder's lease
	// and returns the per-instance version assigned to it (monotonic, starting
	// at 1). It is lease-fenced: leaseID MUST match the instance's current lease,
	// otherwise nothing is recorded and a KindConflict error is returned. The
	// event is validated via Event.ValidateForAppend (terminal event kinds are
	// rejected here — terminal status is committed via MarkTerminal). The Event's
	// Version and CreatedAt are assigned by the Journal; any caller-supplied
	// values are ignored.
	//
	// Append advances the projection's non-terminal Status as a deterministic fold
	// of the event kind, reusing the saga state machine (saga.AdvanceSaga) so every
	// move is transition-validated and an out-of-phase event is rejected without
	// mutation:
	//   - a forward step event (StepStarted/StepCompleted/StepFailed) moves a
	//     Pending instance to Running; on a Running instance it is a no-op; while
	//     Compensating it is rejected.
	//   - KindCompensationStarted moves Running → Compensating (the Coordinator
	//     appends it when it DECIDES to compensate, BEFORE any compensation runs,
	//     so a handoff mid-rollback reads Compensating, not Running); from Pending
	//     it is rejected.
	//   - KindStepCompensated is legal only while Compensating (status unchanged);
	//     elsewhere it is rejected.
	//   - KindStepCompensationFailed is legal only while Compensating (status
	//     unchanged, like KindStepCompensated); elsewhere it is rejected. It records
	//     that a per-step CompensateFunc returned a non-nil error — distinct from
	//     KindStepFailed (a forward-phase failure during Running).
	// Terminal event kinds are rejected by ValidateForAppend — terminal status is
	// committed only via MarkTerminal, which encodes the terminal state in the
	// event kind. Append also does NOT maintain the step cursor
	// (Instance.CurrentStep); see ClaimPending.
	//
	// An unknown instance returns KindNotFound with code errcode.ErrSagaNotFound;
	// a stale lease returns KindConflict with code errcode.ErrSagaStaleLease.
	// A terminal instance (whose lease was released by MarkTerminal) also
	// returns KindConflict with ErrSagaStaleLease — the projection holds no
	// lease, so the caller's leaseID cannot match. Callers needing to
	// distinguish "expired lease" from "already terminal" must consult Load.
	Append(ctx context.Context, instanceID, leaseID idutil.SafeID, event Event) (version int64, err error)

	// Load returns the full ordered event history for an instance (version
	// ascending), for replay / state reconstruction. A never-enqueued instance
	// returns a KindNotFound error with code errcode.ErrSagaNotFound; an
	// enqueued instance with no events yet returns an empty, non-nil slice.
	Load(ctx context.Context, instanceID idutil.SafeID) ([]Event, error)

	// ClaimPending atomically leases up to batchSize non-terminal instances that
	// are currently unleased or whose lease has expired, sets each one's lease to
	// expire at now+leaseDuration, and returns them stamped with a single batch
	// LeaseID that the caller MUST echo on every subsequent fenced call for each
	// claimed instance. When nothing is claimable it returns an empty slice, the
	// zero LeaseID, and a nil error.
	//
	// batchSize and leaseDuration MUST both be positive; a non-positive value
	// returns a KindInvalid error.
	//
	// The returned Instance carries the coordination projection: Status (the
	// coarse-grained Pending/Running/Compensating lifecycle the Journal maintains)
	// and lease, but NOT the fine-grained step cursor — Instance.CurrentStep is
	// not advanced by the Journal (it would require the definition's step count,
	// which the Journal does not hold) and remains 0. A caller resuming an
	// instance reconstructs the exact cursor by folding the event log from Load.
	//
	// Instance.Status is the current projection value (maintained by
	// Append/MarkTerminal) and is guaranteed up-to-date; callers MAY trust it
	// directly while still folding Load for the exact step cursor.
	ClaimPending(ctx context.Context, batchSize int, leaseDuration time.Duration) (claimed []ClaimedInstance, leaseID idutil.SafeID, err error)

	// MarkTerminal transitions the instance projection to finalStatus and appends
	// the matching terminal event atomically, so the log alone replays which
	// terminal state was reached. Terminal event kinds per finalStatus:
	//   - StatusSucceeded     → KindSagaSucceeded     (all steps committed)
	//   - StatusFailed        → KindSagaFailed        (forward failure, no rollback)
	//   - StatusCompensated   → KindSagaCompensated   (rollback completed cleanly)
	//   - StatusExpired       → KindSagaExpired       (overall timeout elapsed)
	//   - StatusCompensationFailed → KindSagaCompensationFailed (rollback itself
	//     failed; reachable only from StatusCompensating)
	//
	// finalStatus MUST be a terminal saga.Status reachable from the current
	// status (validated via saga.AdvanceSaga). It is lease-fenced: ok is false
	// (with a nil error) when leaseID no longer owns the instance. On success
	// the lease is released. A never-enqueued instance also returns ok=false
	// (nil error); callers cannot distinguish it from a stale lease.
	MarkTerminal(ctx context.Context, instanceID, leaseID idutil.SafeID, finalStatus saga.Status) (ok bool, err error)

	// RepoReady is a differentiated readiness check (kernel/healthz.RepoProber):
	// SQL-backed implementations exercise the saga_instances relation directly so
	// schema/migration drift surfaces independently of a pool-level ping;
	// in-memory implementations return nil.
	//
	// Probe registration (name + cellgen path) is the responsibility of the
	// Coordinator cell that holds the Journal (PR-03); a Journal implementation
	// only provides the RepoProber method and does not self-register.
	RepoReady(ctx context.Context) error
}

JournalCore is the Heartbeat-free core of Journal: enrollment, the append-only event log, claim/projection, and terminal commit. It is the interface runtime/saga.Coordinator persists — deliberately WITHOUT Heartbeat, so a centralized heartbeat loop cannot be built on a stored JournalCore field (the call would not type-check). See the "Interface split" section on Journal and archtest SAGA-COORDINATOR-NO-HEARTBEAT-LOOP-01.

Only runtime/saga.Coordinator may hold a JournalCore field, enforced by the SAGA-JOURNAL-HOLDER-SEAL-01 archtest. Business producers and query-side slices should hold narrower interfaces instead:

• Enqueuer — saga enrollment only (e.g. an HTTP slice that places orders).
• Reader — event-log read only (e.g. a status-query slice).
• ProducerReader — enrollment + read, for cells that need both without the coordinator-only claim/append/commit surface.

Note: SAGA-JOURNAL-HOLDER-SEAL-01 currently scans only runtime/saga for JournalCore field holders. Business cells (examples/, cells/) holding JournalCore are not yet machine-rejected (tracked in gh #1415).

type MemJournal ¶

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

MemJournal is an in-memory Journal implementation for tests and development. It is not a production substrate — the PostgreSQL-backed implementation (PR-04) owns the production path. Single sync.Mutex serializes all state mutations; this is acceptable at dev/test scale.

Two truths are kept consistent under the lock:

• The event log (events map) is append-only source of truth for history.
• The projection (instanceRow) is the authoritative coordination view: current status, event version counter, and lease fencing token.

func NewMemJournal(clk clock.Clock) (*MemJournal, error)

func (m *MemJournal) Append(_ context.Context, instanceID, leaseID idutil.SafeID, event Event) (int64, error)

func (m *MemJournal) ClaimPending(_ context.Context, batchSize int, leaseDuration time.Duration) ([]ClaimedInstance, idutil.SafeID, error)

func (m *MemJournal) Enqueue(_ context.Context, instance saga.Instance) error

func (m *MemJournal) HeadSeq(_ context.Context) (int64, error)

func (m *MemJournal) Heartbeat(_ context.Context, instanceID, leaseID idutil.SafeID, leaseDuration time.Duration) (bool, error)

func (m *MemJournal) Load(_ context.Context, instanceID idutil.SafeID) ([]Event, error)

func (m *MemJournal) LoadSince(_ context.Context, afterGlobalSeq int64, limit int) ([]GlobalEvent, error)

func (m *MemJournal) MarkTerminal(_ context.Context, instanceID, leaseID idutil.SafeID, finalStatus saga.Status) (bool, error)

func (m *MemJournal) RepoReady(_ context.Context) error

type ProducerReader ¶

type ProducerReader interface {
	Enqueuer
	Reader
}

ProducerReader composes the two narrow capabilities a business cell legitimately needs — enrollment (Enqueue) and event-log read (Load) — WITHOUT the coordinator-only lease/claim/commit methods (ClaimPending/Append/MarkTerminal/Heartbeat). The single-sanctioned-holder rule (only runtime/saga.Coordinator holds JournalCore) is unaffected: business cells now hold ProducerReader, structurally unable to drive a saga.

MemJournal and any PostgreSQL implementation satisfy ProducerReader automatically (they satisfy the full Journal which is a superset).

type Reader ¶

type Reader interface {
	Load(ctx context.Context, instanceID idutil.SafeID) ([]Event, error)
}

Reader is the query-facing narrow journal interface: event-log read only. A status-query slice folds the event log via Load and needs nothing else.

MemJournal and any PostgreSQL implementation satisfy Reader automatically because both satisfy the full Journal (which embeds JournalCore, which contains Load). No implementation change is required.