saga

Documentation

Overview

Package saga (see doc.go for the package narrative) is the runtime engine that drives saga instances forward. This file declares the post-commit signal sink the Coordinator calls from an AfterCommit hook.

Package saga is the GoCell runtime engine for L3 WorkflowEventual sagas (capability cap-15). It drives saga instances forward by reading the kernel/saga/journal.Journal, executing user-supplied Step bodies, then writing the resulting journal event(s) and outbox commands inside one short database transaction.

Leader election (single-process unsafe by default)

Without WithLeaderElect the Coordinator runs single-process with NO distributed leader election; it is NOT safe for multi-process deployment and emits slog.Warn(mode="unsafe_no_leader") at Start. Pass WithLeaderElect(distlock.Locker) (PR-05) to gate every drive behind a per-instance distributed lock keyed "saga:{definitionID}:{instanceID}": contending coordinators skip instances they cannot lock, a crashed leader's lock and journal lease both expire via TTL, and Start emits slog.Info(mode="leader_elect"). See leader_elect.go.

Layering

The Definition / Step / Resolver data primitives live in kernel/saga (a pure data package). The Coordinator + Dispatcher engine lives here. No cell wraps runtime/saga in PR-03 — the cell-side wiring (including RegisterReadiness) lands in PR-09.

Carry-over from PR-02 (#952)

• RepoReady probe registration: Coordinator satisfies healthz.RepoProber; cell-side registration deferred to PR-09 (tracked in #978).
• Enqueuer interface split: deferred until a real producer ships.

Note: The `saga_coordinator_ready` probe is NOT registered with any cell in PR-03; it becomes reachable via `/readyz` only after PR-09 wires it through cellgen's `RegisterReadiness`.

Observer (PR-#1181 / #1109)

The Coordinator and its internal Executor share a single executor.Observer (set via WithObserver; coordinator.go options.go). The interface carries six callbacks, split into two groups:

Executor-emitted (step-level):

• ObserveOutcome: called once per Execute/Compensate call at the terminal Result.
• ObserveRetry: called between step attempts (attempt N > 1).
• ObserveHeartbeatFailure: called on infra-error or stale-lease ticks.

Coordinator-emitted (tick/drive/leader-skip level):

• ObserveTick: called once per ClaimPending poll (empty / claimed / error).
• ObserveDrive: called once per instance driven (ok / error).
• ObserveLeaderSkip: called when acquireLead fails (contended / ctx_canceled / backend_error).

executor.NopObserver is the zero-cost default. executor.WithObserver(nil) is silently ignored (builder-noop option); the Executor keeps NopObserver. The Coordinator passes the WithObserver option through to the internal Executor via NewCoordinator's options — see options.go for WithObserver.

Both the Executor and the Coordinator guard observer calls with the same two layers of fail-closed protection:

1. Panic recovery: a panicking observer logs Warn with a redacted payload and execution continues (executor.recoverObserverPanic / Coordinator.recoverObserverPanic).
2. Bounded goroutine wait: the observer call runs on a fresh goroutine; the caller waits at most observerCallDeadline (default executor.DefaultObserverCallDeadline = 5s) before logging Warn and returning. For the Coordinator this prevents a hung observer from leaking the per-instance distlock (release() and inflightLocks.Delete run after ObserveDrive in tickOnce) and from blocking the shutdown drain (executor.callObserverBounded / Coordinator.safeObserve).

executor.HeartbeatFailureReason is the typed enum for ObserveHeartbeatFailure reason values: HeartbeatFailureInfraError (transient backend error) and HeartbeatFailureStaleLease (another coordinator owns the lease).

executor.IsLeaseLost is the public predicate compensation walks use to detect that RunWithHeartbeat returned because the lease was lost.

During runCompensation, the heartbeat goroutine is maintained via executor.RunWithHeartbeat. If the heartbeat reports a stale lease, the compensation context is canceled (errLeaseLost) and the walk stops; the instance will be re-claimed by another coordinator on its next tick.

Saga terminal states (PR-#1210)

Five terminal states encode why a saga finished:

• StatusSucceeded — all forward steps committed
• StatusFailed — forward-phase failure with no rollback (never entered Compensating)
• StatusCompensated — forward failure followed by clean rollback
• StatusCompensationFailed — rollback itself failed; ops intervention required
• StatusExpired — overall saga timeout elapsed at any non-terminal stage

StatusFailed and StatusCompensationFailed are distinct on purpose: the former says "we never tried to undo", the latter says "we tried and could not". Dashboards and ops runbooks should branch on this distinction. See kernel/saga.Status.String() for the wire labels (snake_case).

PR-03 deferred scope

• Coordinator-level Start API for producers (typed producer facade): deferred to PR-07/PR-09.

(Per-tick concurrent driving of claimed instances — formerly tracked here as #983 — is now delivered; see the tickLoop section below.)

Metrics emission is complete (#1109): the Coordinator and Executor fan all observability out through executor.Observer, which runtime/observability/ metrics.SagaCollector turns into six counters — saga_step_outcome_total, saga_step_retry_total, saga_heartbeat_failed_total (step-level) plus saga_tick_total, saga_drive_total, and saga_leader_elect_skip_total (coordinator-level; leader-elect skip carries reason=contended/ctx_canceled/ backend_error, where backend_error is the lock-acquire failure rate). slog remains the per-instance correlation channel (leader-elect skips and ctx-cancel are Debug, backend I/O errors Warn). The counters are only emitted when a real metrics Provider is wired via WithObserver; saga is not yet in a production cell (examples/orderfulfillment uses NopProvider), so production Provider wiring lands with the saga-as-cell migration (PR-09, #978).

Coordinator lifecycle

NewCoordinator validates required deps (journal/txRunner/outboxEmit/registry non-nil; clock panics if nil via clock.MustHaveClock). Start launches a single tickLoop goroutine:

• tickLoop: calls ClaimPending each PollInterval, drives each claimed instance through one step (Run outside tx, Append + Emit + AfterCommit Kick inside tx). Per-step lease maintenance (heartbeat) is owned by the Executor's per-step heartbeat goroutine (see runtime/saga/executor).

Stop is idempotent. Ready() returns a channel closed once Start transitions to running. RepoReady delegates to journal.RepoReady so the wrapping cell (PR-09) can register it via cellgen-emitted RegisterReadiness.

tickLoop drives the instances claimed in one tick concurrently — each led instance runs driveOne in its own goroutine and the tick wg.Waits for the whole batch before claiming the next (#983). Peak concurrency = ClaimBatchSize, which is therefore also the drive-concurrency / external-step-IO bound: lower ClaimBatchSize when steps open many external connections. Claim count and drive fan-out are deliberately one knob because a claimed instance holds a journal lease kept alive only by the Executor heartbeat that starts inside driveOne; decoupling them (claim a large batch, drive with lower concurrency) needs a resident worker pool and is deferred (#978). See Amendment 2026-06-07 of the saga ADR (docs/architecture/202606021000-adr-saga-l3-orchestration-engine.md).

Step.Run is the only StepFunc callsite

Inside this package, saga.StepFunc is invoked exclusively from safeRun (a recover-guarded helper). safeRun is called BEFORE txRunner.RunInTx, so user code never executes with a database transaction held open. This invariant is locked by the SAGA-STEP-RUN-OUTSIDE-TX-01 archtest.

Coordinator is the single sanctioned JournalCore holder

Coordinator is the only struct in this package that holds a journal.JournalCore field — the Heartbeat-free core of journal.Journal, so a centralized heartbeat loop is a compile error (#1209). No struct in this package may persist the Heartbeat-bearing full journal.Journal (or a bare journal.Heartbeater) as a field. Locked by SAGA-JOURNAL-HOLDER-SEAL-01 and SAGA-COORDINATOR-NO-HEARTBEAT-LOOP-01 archtests.

ref: dtm dtmsvr/cron.go (CronTransOnce structure) ref: temporalio sdk-go internal_task_pollers.go ref: ThreeDotsLabs/watermill message/router.go

Index

• Constants
• type Config
  • func DefaultConfig() Config
  • func (c Config) Validate() error
• type Coordinator
  • func NewCoordinator(j journal.Journal, tx persistence.TxRunner, em koutbox.Emitter, ...) (*Coordinator, error)
  • func (c *Coordinator) Ready() <-chan struct{}
  • func (c *Coordinator) RepoReady(ctx context.Context) error
  • func (c *Coordinator) Start(ctx context.Context) error
  • func (c *Coordinator) Stop(ctx context.Context) error
• type Dispatcher
• type NoopDispatcher
  • func (NoopDispatcher) Kick(context.Context)
• type Option
  • func WithConfig(cfg Config) Option
  • func WithDispatcher(d Dispatcher) Option
  • func WithLeaderElect(locker distlock.Locker) Option
  • func WithLogger(l *slog.Logger) Option
  • func WithObserver(o executor.Observer) Option
  • func WithTracer(t wrapper.Tracer) Option
• type StepCompletedEvent

Constants

const LeaderElectModeLabel = "leader_elect"

LeaderElectModeLabel is the slog "mode" field value emitted at Start() when the Coordinator runs with distlock leader election — the multi-process-safe counterpart to UnsafeModeLabel.

const ProbeCoordinatorReady healthz.ProbeName = "saga_coordinator_ready"

ProbeCoordinatorReady is the healthz ProbeName for the saga coordinator journal readiness probe. Cell-side registration (RegisterReadiness) is tracked in #978 (saga-as-cell migration follow-up); this PR lands the const declaration so the PROBENAME-SEALED-FUNNEL-01 archtest golden inventory can lock the declared form. Wiring is not performed here to avoid scope creep — the coordinator is not yet a first-class Cell.

const UnsafeModeLabel = "unsafe_no_leader"

UnsafeModeLabel is the slog field value emitted at Start() to alert operators that this Coordinator runs without distributed leader election.

Types

type Config

type Config struct {
	// PollInterval is how often tickLoop calls ClaimPending. Default 200ms.
	PollInterval time.Duration
	// ClaimBatchSize is the maximum number of instances claimed per tick.
	// Default 16. It is also the per-tick drive-concurrency bound: tickOnce
	// drives every claimed instance that passes the leader gate in its own
	// goroutine, so peak concurrent driveOne (and concurrent external step IO)
	// per tick ≈ ClaimBatchSize. Lower it when steps open many external
	// connections — claim count and drive fan-out are intentionally the same
	// knob, because a claimed instance holds a journal lease that is only kept
	// alive by the Executor heartbeat that starts inside driveOne; claiming more
	// than are driven concurrently would let the excess leases go stale while
	// parked. Decoupling claim batch from drive concurrency requires a resident
	// worker pool (claim-on-free-slot) and is deferred (ADR §8 / #978).
	ClaimBatchSize int
	// LeaseDuration is how long a claimed lease is held. Default 30s.
	// Also used as the per-instance distlock TTL in leader-elect mode AND
	// forwarded to the internal Executor for heartbeat lease renewal.
	LeaseDuration time.Duration
	// HeartbeatInterval is how often the internal Executor's per-step
	// heartbeat goroutine renews the lease. Default executor.DefaultHeartbeatInterval
	// (= 10s = LeaseDuration/3). Must satisfy
	// HeartbeatInterval * executor.HeartbeatLeaseSafetyFactor < LeaseDuration
	// so at least one heartbeat lands before expiry.
	HeartbeatInterval time.Duration
}

Config holds tunable parameters for the Coordinator engine.

NewCoordinator initializes c.cfg = DefaultConfig() BEFORE running options, so callers that never invoke WithConfig get the documented defaults. If WithConfig is supplied, it replaces the whole struct (assigns c.cfg = cfg); there is no partial-merge and no per-field substitution. Validate() then runs once and rejects zero values for PollInterval / ClaimBatchSize / LeaseDuration / HeartbeatInterval — those four fields MUST be positive. To override only some fields, start from the defaults:

cfg := saga.DefaultConfig()
cfg.PollInterval = 500 * time.Millisecond
c, err := saga.NewCoordinator(..., saga.WithConfig(cfg))

HeartbeatInterval + LeaseDuration flow into the internally-constructed Executor as the single source of truth — #1181 F5 deleted the WithExecutor option that previously allowed callers to inject an Executor with a different journal / lease; claim and heartbeat are now guaranteed same-source by construction.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with documented defaults.

func (Config) Validate

func (c Config) Validate() error

Validate returns nil iff all fields are positive and the heartbeat / lease ratio is safe.

type Coordinator

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

Coordinator is the in-process saga engine. It drives pending saga instances forward: claiming a batch, replaying event history, executing one Step.Run body outside any database transaction, then atomically recording the outcome (Append + outbox Emit) inside a short RunInTx call.

By default it runs in a single process with NO distributed leader election (unsafe mode); Start() emits slog.Warn(mode="unsafe_no_leader"). Pass WithLeaderElect(distlock.Locker) to make it multi-process safe: each claimed instance is driven only after winning a per-instance distributed lock, and Start() emits slog.Info(mode="leader_elect") instead. See leader_elect.go.

Single sanctioned journal holder

Coordinator is the only struct in this package that holds a journal.JournalCore field — the Heartbeat-free core of journal.Journal. The narrow type makes a centralized heartbeat loop a compile error (c.journal.Heartbeat is undefined); per-step lease renewal is funneled through the internal executor. Locked by SAGA-JOURNAL-HOLDER-SEAL-01 (only Coordinator may hold JournalCore; no struct may persist the Heartbeat-bearing full Journal) and SAGA-COORDINATOR-NO-HEARTBEAT-LOOP-01 (#1209).

func NewCoordinator

func NewCoordinator(
	j journal.Journal,
	tx persistence.TxRunner,
	em koutbox.Emitter,
	reg ksaga.Resolver,
	clk clock.Clock,
	opts ...Option,
) (*Coordinator, error)

NewCoordinator validates required deps and applies opts. Nil required deps return errcode.KindInvalid + ErrValidationFailed. A nil or typed-nil clock panics via clock.MustHaveClock (programmer error).

func (*Coordinator) Ready

func (c *Coordinator) Ready() <-chan struct{}

Ready returns a channel that is closed once Start transitions to running.

func (*Coordinator) RepoReady

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

RepoReady implements healthz.RepoProber. The probe is binary (ready / not ready), driven by two layers in order:

1. Coordinator lifecycle — the probe MUST report not-ready while the instance is stopped / starting / stopping. A coordinator that has not reached coordRunning cannot drive new claims even if the journal is fine; conversely a stopping coordinator is draining inflight work and should be drained out of load balancers.
2. Journal storage — when running, delegate to journal.RepoReady so an underlying PG/mem outage flips readiness off without needing a separate probe name (failure domains differ from the pool-level postgres_ready; see .claude/rules/gocell/observability.md §"Cell 级别 Repo Readiness Probe").

The unsafe single-process mode (UnsafeModeLabel) is signaled at Start() via slog.Warn — it does NOT toggle readiness, because PR-03's contract is "unsafe but ready to drive a saga in a single process". PR-05 leader-elect is the path for multi-process safety; this probe stays binary.

Cell-side registration (RegisterReadiness funnel) is the Coordinator cell holder's responsibility (PR-09).

func (*Coordinator) Start

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

Start launches tickLoop and blocks until ctx is canceled or Stop is called.

func (*Coordinator) Stop

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

Stop signals shutdown, drains inflight driveOne goroutines, then cancels the loops and waits for them to exit. Idempotent.

Drain budget = the passed-in ctx (same budget as the rest of Stop). While draining, tickOnce short-circuits to stop accepting new claims and the per-step Executor heartbeats keep leases valid so inflight steps don't lose their lease mid-flight. If a non-cooperative Step.Run ignores ctx.Done() and exceeds the drain budget:

• cancel() fires anyway and Stop returns (best-effort).
• The orphaned step goroutine continues until it returns naturally; the Executor heartbeat goroutine has by then exited, so the journal lease expires.
• In leader-elect mode, Stop orphans every in-flight distlock (orphanInflightLocks): renewal is stopped without a shutdown-time release round-trip, so the distlock key expires on its lease TTL (~1×TTL from the last successful renewal, best-effort) and a competitor coordinator can take over — bounded-TTL handoff, I/O-free so it cannot hang on an unreachable backend during shutdown. After Stop, per-instance distlock keys linger in the backend for up to Config.LeaseDuration before expiring (vs the previous immediate-release behavior); a coordinator restarting within that window will skip those instances until the lease lapses — a liveness cost, not a safety degradation (the journal lease_id CAS continues to fence late commits). A still-running wedged step keeps its lock until TTL while the journal lease_id CAS continues to fence its late commits (the PR-05 efficiency-lock model, leader_elect.go). Cooperative steps already released via tickOnce; orphanInflightLocks idempotently covers the drain-budget-exhausted case.
• This is the inherent limit of cooperative cancellation in Go: the step goroutine itself cannot be killed. Step authors are responsible for selecting on ctx.Done() inside blocking primitives — see ksaga.StepFunc godoc.

type Dispatcher

type Dispatcher interface {
	Kick(ctx context.Context)
}

Dispatcher is the post-commit signal sink. After a successful saga step commit, the Coordinator calls Kick once from a kernel/persistence.RegisterAfterCommit hook so a downstream component (typically the outbox Relay) can immediately poll instead of waiting for its next periodic tick.

PR-03 ships only NoopDispatcher. A real wiring (e.g., poke outbox.Relay) will land in a later PR when Relay grows a Notify API.

Lifecycle: Kick is called from a hookCtx that has the database transaction stripped (see kernel/persistence.RunAfterCommitHooks). Implementations MUST NOT call back into the database — Kick is signal-only.

type NoopDispatcher

type NoopDispatcher struct{}

NoopDispatcher is a default Dispatcher that does nothing. It is safe to use in tests and as the default Coordinator dispatcher when no downstream is wired.

func (NoopDispatcher) Kick

func (NoopDispatcher) Kick(context.Context)

Kick implements Dispatcher with intentional no-op behavior — NoopDispatcher is the default sink when no downstream component is wired (PR-03 ships this only; outbox Relay integration lands in a later PR). The Coordinator still calls Kick once per successful commit; dropping it costs nothing because the downstream poller will catch up via its own periodic tick.

type Option

type Option func(*Coordinator)

Option configures a Coordinator.

func WithConfig

func WithConfig(cfg Config) Option

WithConfig replaces the entire Config struct (c.cfg = cfg). No partial merge, no per-field default substitution: NewCoordinator runs cfg.Validate() after the options loop and Validate rejects zero values for PollInterval / ClaimBatchSize / LeaseDuration / HeartbeatInterval. Callers that want defaults for unset fields must start from DefaultConfig():

cfg := saga.DefaultConfig()
cfg.PollInterval = 500 * time.Millisecond
saga.WithConfig(cfg)

Must be called before Start.

func WithDispatcher

func WithDispatcher(d Dispatcher) Option

WithDispatcher sets the post-commit kick target. Both bare-nil and typed-nil are silently ignored; the Coordinator falls back to NoopDispatcher (set in NewCoordinator). Must be called before Start.

func WithLeaderElect

func WithLeaderElect(locker distlock.Locker) Option

WithLeaderElect enables distlock-based leader election. Before driving each ClaimPending-d instance, the Coordinator acquires a per-instance distributed lock keyed "saga:{definitionID}:{instanceID}"; instances whose lock is held by another process are skipped this tick (no-lock → skip), and the lock is released after the drive. The lock TTL is Config.LeaseDuration; distlock is auto-renewed by distlock's shared manager goroutine and the journal lease is renewed by the Executor's per-step heartbeat goroutine, both during the hold.

distlock here is an efficiency lock, NOT the correctness boundary: it shrinks the window in which two coordinators concurrently run a Step.Run body, but if distlock renewal fails mid-drive the drive continues — correctness is guaranteed by the journal lease_id CAS fencing (a stale leader's Append / MarkTerminal is rejected at commit). The distlock TTL and the journal lease share the same LeaseDuration but start at different instants (acquireLead vs ClaimPending), so on crash the distlock key may expire slightly before the journal lease; this is safe — the journal CAS still fences the old leader.

This is a strong-dependency wiring option (see runtime-api.md Option 范式分层): a nil locker — both bare-nil and typed-nil — sets the leaderElectNil sentinel and is rejected at NewCoordinator with KindInvalid/ErrValidationFailed. Without this option the Coordinator runs in single-process unsafe mode and Start() logs mode=unsafe_no_leader.

ref: runtime/distlock/locker.go (Lock-as-Resource; Acquire is non-blocking, returns ErrLockTimeout on contention) + temporal service/history shard ownership.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger replaces the structured logger. A nil l is silently ignored; the Coordinator falls back to slog.Default() (set in NewCoordinator). Must be called before Start.

func WithObserver

func WithObserver(o executor.Observer) Option

WithObserver attaches an executor.Observer — the saga-wide observability sink. It receives both step-level events (outcome / retry / heartbeat-failure, from the internally-constructed Executor) and coordinator-level events (tick / drive / leader-skip, from the Coordinator itself). A typed-nil observer is silently ignored; both the internal Executor and the Coordinator's own observe path fall back to executor.NopObserver{}. Must be called before Start.

Category: builder-noop (cf. runtime-api.md). The Observer is stored on the Coordinator (used by safeObserve) and wired into the Executor NewCoordinator constructs internally — a single sink satisfies both producers.

#1181 F8 typed-nil guard: WithObserver uses validation.IsNilInterface so a nil *SagaCollector is rejected and NopObserver is kept.

The supplied Observer MUST be safe for concurrent use: a tick drives its claimed instances in parallel, so its methods are invoked concurrently from multiple goroutines (see executor.Observer godoc).

For why the Coordinator owns Executor construction (rather than accepting an injected Executor), see runtime/saga/coordinator.go Coordinator struct comments (#1181 F5 design): the Executor must use the Coordinator's journal so claim and heartbeat are guaranteed same-source by construction, not by convention.

func WithTracer

func WithTracer(t wrapper.Tracer) Option

WithTracer sets the Tracer used for both the per-instance `saga.coordinator.driveOne` span owned by Coordinator AND the per-step `saga.executor.step.run` / `saga.executor.step.compensate` spans owned by the internally-constructed Executor. A typed-nil tracer is silently ignored; the Coordinator falls back to wrapper.NoopTracer{} (set in NewCoordinator). Must be called before Start.

Category: builder-noop (cf. runtime-api.md) — tracing is an optional adapter wiring. Coordinator does NOT accept a separately-injected Executor; the same tracer reaches both layers via NewCoordinator's internal Executor construction, so the trace parent/child relationship is guaranteed by type-system construction (single source of tracing config).

type StepCompletedEvent

type StepCompletedEvent struct {
	InstanceID idutil.SafeID `json:"instanceId"`
	Step       idutil.SafeID `json:"step"`
}

StepCompletedEvent is the outbox payload published after a successful step commit. It is emitted to the topic returned by stepCompletedTopic.