saga

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 (d *Definition) Len() int

func (d *Definition) Validate() error

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(defs ...*Definition) (*InMemoryRegistry, error)

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

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(id, definitionID idutil.SafeID, now time.Time) Instance

func (i *Instance) ValidateNew() error

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 (p RetryPolicy) Validate() error

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 (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 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.