saga

Documentation

Overview

Package saga provides the L3 (WorkflowEventual) saga orchestration state machine: the Status lifecycle, the Instance record, and the pure transition functions (AdvanceSaga, AdvanceStep, Transition) that advance an instance. It is the kernel vocabulary that the runtime Coordinator and the durable Journal build on.

Lifecycle

A saga instance moves through a fail-closed state machine. All transitions are validated; an illegal move returns an error without mutating the instance.

Pending ──► Running ──► Succeeded                    (happy path)
   │           │
   │           ├──► Compensating ──► Compensated      (rollback succeeded)
   │           │           └──────► Failed            (compensation itself failed)
   │           └──► Failed                            (failed before any step committed)
   │
   └──► Failed | Expired                              (rejected / expired while queued)

Expired is reachable from every non-terminal state (Pending, Running,
Compensating): the overall timeout (owned by the Coordinator) is orthogonal
to step progress.

Pending: created, no step has run. Running: executing steps forward; CurrentStep advances via AdvanceStep. Compensating: a step failed with prior committed work; undoing in reverse. Succeeded: all steps committed (terminal). Failed: terminal; reachable from Running (nothing to undo) or Compensating (second-order failure). Compensated: forward failed, rollback completed cleanly (terminal). Expired: overall timeout elapsed (terminal).

Time injection

Every mutator takes an explicit now time.Time; the state machine never reads the wall clock. This keeps transitions pure and deterministically testable, and lets the Coordinator source time from a single clock. A monotonicity guard rejects a now that precedes StartedAt or the previous UpdatedAt.

Not in this package (deferred, with their consumers)

• Definition / Step / StepFunc / CompensateFunc / RetryPolicy / Resolver / InMemoryRegistry live in this same package (kernel/saga). The Coordinator engine and the per-step Executor that execute them live in runtime/saga and runtime/saga/executor.
• The backoff / jitter algorithm that consumes RetryPolicy, the per-step context.WithTimeout enforcement, and the heartbeat goroutine live in runtime/saga/executor; this package carries only the static config (RetryPolicy value + Step.Timeout / Definition.Timeout).
• The durable append-only Journal lands in kernel/saga/journal.
• Instance.LogValue (slog.LogValuer for redacted logging) lands in runtime/saga together with the State payload it needs to redact; Instance carries only IDs / status / timestamps (no sensitive fields), so the default slog reflection is safe in the interim.

References

ref: dtm-labs/dtm dtmsvr/storage/sql/saga_branch.go — saga branch state + compensation ordering. ref: temporalio/temporal service/history/workflow/state_rebuilder.go — append-only workflow state machine. ref: kernel/command (in-repo) — L4 state-machine template: iota+1 zero-invalid status, map-based transition table, fail-closed Transition, now-injected AdvanceCommand, NewEntry + ValidateNew construction.

Index

• func AdvanceSaga(inst *Instance, to Status, now time.Time) error
• func AdvanceStep(inst *Instance, stepCount int, now time.Time) error
• func Transition(from, to Status) error
• type CompensateFunc
• type Definition
  • func (d *Definition) Len() int
  • func (d *Definition) Validate() error
• type InMemoryRegistry
  • func NewInMemoryRegistry(defs ...*Definition) (*InMemoryRegistry, error)
  • func (r *InMemoryRegistry) Lookup(id idutil.SafeID) (*Definition, bool)
• type Instance
  • func NewInstance(id, definitionID idutil.SafeID, now time.Time) Instance
  • func (i *Instance) ValidateNew() error
• type Resolver
• type RetryPolicy
  • func (p RetryPolicy) Validate() error
• type Status
  • func (s Status) CanTransitionTo(target Status) bool
  • func (s Status) IsTerminal() bool
  • func (s Status) String() string
  • func (s Status) Valid() bool
  • func (s Status) ValidTransitions() []Status
• type Step
• type StepFunc

Functions

func AdvanceSaga

func AdvanceSaga(inst *Instance, to Status, now time.Time) error

AdvanceSaga validates a status transition and applies the timestamp side effects that the kernel owns. It is fail-closed: an illegal transition or a clock regression returns an error WITHOUT mutating the instance.

Side effects on success:

• every transition sets UpdatedAt = now
• a terminal target additionally sets CompletedAt = now

Time is injected via now; the state machine never reads the wall clock.

func AdvanceStep

func AdvanceStep(inst *Instance, stepCount int, now time.Time) error

AdvanceStep advances the forward step cursor by one within a Running saga. stepCount is the total number of steps in the definition (the Coordinator passes def.Len()); this keeps AdvanceStep self-contained and free of any dependency on the SagaDefinition type. Fail-closed: rejects when the saga is not Running, when stepCount is not positive, when CurrentStep is out of the valid cursor range [0, stepCount-1), or on clock regression. The range check is the guardian of CurrentStep for replayed/loaded instances (PR-02/PR-04), not just freshly constructed ones.

AdvanceStep is NOT used for the final step: once CurrentStep reaches stepCount-1, the Coordinator instead transitions Running → Succeeded via AdvanceSaga.

func Transition

func Transition(from, to Status) error

Transition validates a state transition from → to and returns an error if the transition is not allowed. This is a pure validation function; it does NOT mutate any state.

Types

type CompensateFunc

type CompensateFunc func(ctx context.Context, inst *Instance, committedState []byte) error

CompensateFunc undoes one previously-committed step during the Compensating phase. It MUST be pure-reverse: it receives the step's committed state and reverses that step's effect through the same domain ports the forward Run used, never touching the transaction/outbox layer (the Coordinator owns the transaction; Compensate runs in the application domain only). This purity is statically enforced by archtest SAGA-STEP-COMPENSATE-PURE-01 — a Compensate body that calls *sql.Tx / pgx.Tx / outbox.Writer / outbox.Emitter / persistence.TxRunner fails the build.

type Definition

type Definition struct {
	ID          idutil.SafeID
	Steps       []Step
	Timeout     time.Duration
	RetryPolicy RetryPolicy
}

Definition is the static recipe for a saga. ID is the DefinitionID stored on every Instance enrolled under this definition. Steps execute in slice order. Timeout is the overall ceiling (Expired); the Coordinator enforces total-saga timeout (see Coordinator.driveOne).

RetryPolicy is the saga-wide default backoff policy; each Step inherits it unless the Step sets its own (zero value => inherit). Both additions are forward-compatible new optional fields.

func (*Definition) Len

func (d *Definition) Len() int

Len returns the number of steps.

func (*Definition) Validate

func (d *Definition) Validate() error

Validate returns nil iff:

• ID is non-empty SafeID (use idutil.SafeID.Validate)
• len(Steps) >= 1
• each Step.Name is non-empty AND passes idutil.SafeID.Validate
• Step.Name values are distinct within the Definition
• each Step.Run is non-nil (StepFunc); Step.Compensate may be nil
• Step.Timeout >= 0
• each Step.RetryPolicy and Definition.RetryPolicy pass RetryPolicy.Validate
• Definition.Timeout >= 0

Errors: errcode.KindInvalid + ErrValidationFailed, or errcode.KindConflict + ErrConflict for duplicate step names. Runtime data (IDs, step index) is carried in WithDetails or WithInternal; Message is a const literal.

type InMemoryRegistry

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

InMemoryRegistry is a trivial map-backed Resolver for tests and the PR-09 example. Construction validates each Definition via Definition.Validate; the constructor returns an error on the first invalid definition. Lookup is read-only and race-safe (no mutation after construction).

func NewInMemoryRegistry

func NewInMemoryRegistry(defs ...*Definition) (*InMemoryRegistry, error)

NewInMemoryRegistry validates each definition then registers it by ID. Nil definitions return errcode.KindInvalid + ErrValidationFailed. Duplicate IDs return errcode.KindConflict + ErrConflict.

func (*InMemoryRegistry) Lookup

func (r *InMemoryRegistry) Lookup(id idutil.SafeID) (*Definition, bool)

Lookup implements Resolver.

type Instance

type Instance struct {
	// ID uniquely identifies this execution (framework-generated).
	ID idutil.SafeID
	// DefinitionID names the saga definition being executed. It is a plain
	// identifier; the SagaDefinition type itself lives with the Coordinator.
	DefinitionID idutil.SafeID

	Status Status
	// CurrentStep is the 0-based forward cursor. It is meaningful only once
	// Status >= Running; AdvanceStep advances it.
	CurrentStep int

	StartedAt   time.Time  // creation instant (caller-provided, not wall-clock)
	UpdatedAt   *time.Time // last transition; nil until the first advance
	CompletedAt *time.Time // set when a terminal status is reached
}

Instance represents a single execution of a saga definition.

The lifecycle: Pending → Running → Succeeded, with Compensating/Compensated and Failed/Expired terminals (see Status). Status and timestamps are owned by the state machine (NewInstance, AdvanceSaga, AdvanceStep); callers MUST NOT mutate them directly.

func NewInstance

func NewInstance(id, definitionID idutil.SafeID, now time.Time) Instance

NewInstance creates an Instance in Pending status. This is the only sanctioned constructor; it enforces the creation-time invariants that Instance.ValidateNew re-asserts:

• Status = StatusPending (callers cannot create a non-Pending instance)
• CurrentStep = 0
• StartedAt = now (caller-provided, making the constructor deterministic)
• UpdatedAt / CompletedAt = nil

func (*Instance) ValidateNew

func (i *Instance) ValidateNew() error

ValidateNew checks that an instance is in its initial (just-created) state. It is a creation-time validator, NOT suitable for an instance already advanced through the lifecycle. It ensures callers cannot bypass the state machine by constructing an Instance with an arbitrary status, cursor, or timestamps.

type Resolver

type Resolver interface {
	Lookup(definitionID idutil.SafeID) (*Definition, bool)
}

Resolver resolves a Definition for a claimed Instance. Coordinator looks up by Instance.DefinitionID. Returns ok=false when the DefinitionID isn't registered (Coordinator treats this as terminal-Failed).

The single-method interface is named Resolver (not "Registry") to follow Go's "-er for single-method interfaces" convention (io.Reader, fmt.Stringer, etc.); the implementing struct is still InMemoryRegistry because it owns registration state, but the abstract role consumers depend on is "resolve a definition by ID".

Interface (not concrete map) so PR-07's contractgen can emit typed SagaResolver_<Name> values, and PR-08 archtest can check holders use the interface, not a closure.

type RetryPolicy

type RetryPolicy struct {
	// MaxAttempts is the total number of Run invocations (including the first).
	// NOTE: 0 means "inherit", not unlimited — saga has no unlimited-retry option.
	MaxAttempts  int
	BaseInterval time.Duration
	MaxInterval  time.Duration
}

RetryPolicy configures exponential-backoff retry for a step. It is a value type (not a pointer) so the zero value means "inherit": a Step's zero RetryPolicy inherits the Definition's, and a Definition's zero RetryPolicy falls back to the executor's package defaults. The runtime executor (runtime/saga/executor) resolves the effective policy and computes the backoff; this type only carries the configuration.

Field semantics (note the deliberate deviation from Temporal's RetryPolicy, where MaxAttempts==0 means unlimited):

• MaxAttempts: total Run invocations including the first. 0 => inherit; a resolved 1 means a single attempt with no retry. Saga has no "unlimited" option — retries are bounded, then compensation runs.
• BaseInterval: first backoff base (the delay before the first retry). 0 => inherit / executor default.
• MaxInterval: backoff cap. 0 => inherit / executor default.

func (RetryPolicy) Validate

func (p RetryPolicy) Validate() error

Validate returns nil iff MaxAttempts >= 0, both intervals are >= 0, and (when both are non-zero) MaxInterval >= BaseInterval. Errors are errcode.KindInvalid + ErrValidationFailed with a const-literal message.

type Status

type Status uint8

Status represents the lifecycle state of an L3 (WorkflowEventual) saga instance. A saga orchestrates a sequence of steps with compensation: on forward failure it rolls back committed steps in reverse.

ref: dtm-labs/dtm saga branch state (CompensateFailed maps to StatusCompensationFailed); temporalio/temporal workflow state (no direct equivalent — Temporal compensates via activities/saga pattern, not built-in terminal status).

const (
	// StatusPending: created, no step has run yet.
	//
	// IMPORTANT: iota+1 ensures the zero value (0) is NOT a valid status.
	// A forgotten/uninitialised Instance.Status will not silently appear valid.
	StatusPending      Status = iota + 1 // = 1
	StatusRunning                        // executing steps forward
	StatusCompensating                   // a step failed; running Compensate in reverse
	StatusSucceeded                      // all steps committed (terminal)
	StatusFailed                         // forward-phase failure with no rollback (never entered Compensating; terminal)
	StatusCompensated                    // forward failed, rollback completed cleanly (terminal)
	StatusExpired                        // overall timeout elapsed (terminal)
	// StatusCompensationFailed is reached when the compensation phase itself
	// encounters a step failure: at least one CompensateFunc returned a non-nil
	// error. It is distinct from StatusFailed (which signals a forward-phase
	// failure with no compensation, or an explicit no-compensate decision) so
	// operators can distinguish "rollback failed" from "forward failed" by
	// inspecting the terminal status alone, without reading the event log.
	// Reachable only from StatusCompensating (terminal).
	StatusCompensationFailed // = 8
)

func (Status) CanTransitionTo

func (s Status) CanTransitionTo(target Status) bool

CanTransitionTo reports whether s can transition to target.

func (Status) IsTerminal

func (s Status) IsTerminal() bool

IsTerminal reports whether s is a terminal (final) state. Terminal states: Succeeded, Failed, Compensated, Expired, CompensationFailed.

func (Status) String

func (s Status) String() string

String returns a human-readable label for the Status.

func (Status) Valid

func (s Status) Valid() bool

Valid reports whether s is a recognized Status value.

func (Status) ValidTransitions

func (s Status) ValidTransitions() []Status

ValidTransitions returns the set of states reachable from s, as a defensive copy. Returns nil for terminal states.

type Step

type Step struct {
	Name       idutil.SafeID
	Run        StepFunc
	Compensate CompensateFunc // optional; nil => no rollback for this step
	Timeout    time.Duration  // per-step; 0 => inherit Definition.Timeout
	// RetryPolicy overrides Definition.RetryPolicy for this step.
	// NOTE: 0 means "inherit", not unlimited — saga has no unlimited-retry option.
	RetryPolicy RetryPolicy
}

Step is a single named unit inside a Definition. Name MUST be a valid SafeID and flows into journal Event.StepName.

Compensate is the optional pure-reverse rollback for this step; nil means the step has nothing to undo (forward-failure is terminal, and the reverse compensation walk skips it). RetryPolicy overrides Definition.RetryPolicy for this step; its zero value inherits (mirroring Timeout's "0 => inherit Definition.Timeout").

type StepFunc

type StepFunc func(ctx context.Context, inst *Instance, prevState []byte) (newState []byte, err error)

StepFunc executes one saga step body. ctx carries the instance's deadline; inst is read-only metadata (mutating it is a programmer error — the Coordinator owns transitions). prevState is the opaque JSON payload of the previous step's success event (nil for the first step). Returns new opaque JSON state, or a non-nil error to signal step failure.

Cooperative cancellation contract

Step.Timeout / Definition.Timeout are enforced via context deadline only — Go has no preemptive goroutine cancellation, so a Step.Run that ignores ctx.Done() blocks its driver's goroutine until it returns naturally. Step authors MUST select on ctx.Done() inside any IO / blocking primitive.

Best-effort recovery when a step ignores cancellation: the driver (the Coordinator's drain on Stop, or the per-step Executor once Execute returns) stops extending the lease; once the lease expires another coordinator (or a restart) re-claims the instance. There is no preemptive kill.