saga

Defines typed saga workflows: steps, their undo and compensation, the typed ports that connect them, and validation of the whole graph.

Use this module to describe a workflow once, then run it with saga/execution (in memory) or saga/durable (with saved checkpoints). A Workflow(input, output, error, undo_error) is a pure description: define evaluates its builder exactly once, validates step names, attempt budgets, timeouts and port ownership, and records the static step graph. No run evaluates the builder again, so one Workflow value serves every run, each with its own input.

Dependencies are typed Port values, not names: perform schedules a step on a port and returns a port for its output, and map, both, all and choose combine ports. The compiler checks the wiring.

A step returns Ok(output) or a typed Error(error). effect gives a step a stable EffectKey to send downstream as an idempotency key. undo registers the action that reverses a completed step when a later step fails; compensate decides, after a failed attempt, whether to retry, continue with a substitute output, abort, or hold; unknown_when names the returned errors after which the step’s effect is unknown, and such an error holds the run for reconciliation unless on_unknown opts into rollback. embed and map_errors compose one workflow into another. saga/durable adds what a step needs to recover after a restart.

import saga
import saga/execution

pub fn checkout() {
  saga.define("checkout", fn(order) {
    order
    |> saga.perform(
      saga.step("reserve_inventory", reserve)
      |> saga.undo(fn(undo) { release(undo.output) }),
    )
    |> saga.perform(
      saga.effect("charge_payment", fn(reservation, key) {
        charge(reservation, idempotency_key: saga.idempotency_key(key))
      })
      |> saga.unknown_when(is_maybe_sent),
    )
  })
}

pub fn run_checkout(workflow, order: Order) {
  execution.run(workflow, order, execution.config())
}

Types

AttemptFailure

</>

Why one attempt did not succeed: the step’s own run returned an application error, the attempt task crashed, or it timed out. The union is closed.

pub type AttemptFailure(e) {
  Returned(error: e)
  Crashed(crash: Crash)
  TimedOut
}

Constructors

```
Returned(error: e)
```
```
Crashed(crash: Crash)
```
```
TimedOut
```

Crash

</>

A reified native exception: which class was raised, and a formatted reason for logs. Produced only when a task crashes outside its own Result.

pub type Crash {
  Crash(class: CrashClass, reason: String)
}

Constructors

Crash(class: CrashClass, reason: String)

CrashClass

</>

The three native exception classes a crashed task can raise.

pub type CrashClass {
  ErrorClass
  ExitClass
  ThrowClass
}

Constructors

```
ErrorClass
```
```
ExitClass
```
```
ThrowClass
```

DefinitionError

</>

A defect in a workflow’s authored shape, found by define before any runtime resource exists. All errors are collected, not just the first. Later releases may add variants, for example for further graph checks: describe them with describe_definition_error.

pub type DefinitionError {
  EmptyWorkflowName
  EmptyStepName(scope: List(String))
  InvalidMaxAttempts(step: StepAddress, value: Int)
  InvalidTimeout(step: StepAddress, value: duration.Duration)
  ForeignPort(step: StepAddress)
  OrphanStep(step: StepAddress)
}

Constructors

```
EmptyWorkflowName
```
```
EmptyStepName(scope: List(String))
```

InvalidMaxAttempts(step: StepAddress, value: Int)

InvalidTimeout(step: StepAddress, value: duration.Duration)

```
ForeignPort(step: StepAddress)
```
```
OrphanStep(step: StepAddress)
```
A step was created during the builder (via perform/embed) but its output port never reaches the workflow’s final output, so it would never run.

EffectKey

opaque </>

The context of one action of a step, for external systems: its keys and the correlation of its run. Read it with the accessors below; the type is opaque, so a later release can add to it without breaking callers.

idempotency_key is the same for every attempt of the step within one run of a local workflow, or one durable execution across restarts. Send it as the downstream idempotency key, so a retry after an unknown outcome cannot repeat the effect.
attempt_number is the 1-based attempt number.
attempt_key is unique to this attempt; use it to record or look up one attempt.
correlation_of is the correlation of the run that performs the action: the one set with execution.with_correlation or durable.with_correlation, which is also the correlation of the run’s saga/telemetry events. A durable execution without one carries correlation.from_key(id) of its execution id; a local run without one gets a fresh correlation.unique() at start. Every run has one, so a step passes it on to the clients it calls, for example http_gun.with_correlation, so their events join the run’s.

A local run derives the idempotency key from its run id, which is new for every execution.run; a durable execution derives it from the id given to durable.start_or_reconnect, so it survives restarts. The text of a key is opaque: compare and store it, never parse it.

saga.step receives only the step’s input: use saga.effect for a step that needs the context. undo and compensate receive it as key.

pub opaque type EffectKey

FailedAttempt

</>

What a compensate decider receives about a failed attempt: the step’s input, why the attempt failed, its 1-based number, how many further attempts the budget allows, and the failed attempt’s EffectKey.

Saga builds this record. Read its fields by label (failed.failure, or FailedAttempt(failure:, ..) in a pattern); do not build or match it positionally. It stays a public record so that a decider reads failed.failure directly.

pub type FailedAttempt(i, e) {
  FailedAttempt(
    input: i,
    failure: AttemptFailure(e),
    attempt: Int,
    attempts_left: Int,
    key: EffectKey,
  )
}

Constructors

FailedAttempt(
  input: i,
  failure: AttemptFailure(e),
  attempt: Int,
  attempts_left: Int,
  key: EffectKey,
)

OnUnknown

</>

What a run does when a step ends with an error that unknown_when classified and no compensate decision settles it. The union is closed.

pub type OnUnknown {
  Reconcile
  RollBack
}

Constructors

```
Reconcile
```
End the run execution.Unresolved with the error as evidence and undo nothing, so the effect can be reconciled first. The default.
```
RollBack
```
Fail the run with execution.StepFailed (or RetryLimitReached) and undo the completed steps, as for a known error.

Port

opaque </>

A typed reference to a value produced in the workflow graph being built: the workflow’s own input, or the output of a perform, map, both, all, choose or embed. Two consumers of the same Port value depend on the same node, so that node executes once per run.

A port reads its value in three stages: the graph is composed once, at define; the coordinator reads the dependency values from the run’s store for each attempt; and the task that runs the attempt computes any map transformation, under rescue, so a panicking or slow map becomes an attempt crash instead of reaching the coordinator.

pub opaque type Port(a, e, u)

Recovery

</>

An explicit decision about a failing attempt, returned from a compensate decider. Retry/RetryAfter request another attempt (if the budget allows; the delay is capped by execution.with_max_retry_delay); Continue accepts a replacement output with its own undo; Abort fails the run and permits rollback of completed steps; AbortAfterCleanupFailure additionally records a cleanup error that happened while deciding; Hold leaves prior effects unresolved with no rollback authority. The union is closed.

pub type Recovery(o, e, u) {
  Retry
  RetryAfter(delay: duration.Duration)
  Continue(output: o, undo: Undo(u))
  Abort(error: e)
  AbortAfterCleanupFailure(error: e, cleanup_error: u)
  Hold(evidence: e)
}

Constructors

```
Retry
```
```
RetryAfter(delay: duration.Duration)
```
```
Continue(output: o, undo: Undo(u))
```
```
Abort(error: e)
```

AbortAfterCleanupFailure(error: e, cleanup_error: u)

```
Hold(evidence: e)
```

Step

opaque </>

A pure description of one unit of work: how to run it, and optionally how to undo a completed run or decide on a failed one. Step values compose with perform inside a define builder; they never run until a Workflow starts.

A failed attempt (Failed below) carries its recovery decision already bound to the concrete error it produced and to the decider in effect. This keeps map_step_errors sound: mapping a step only translates values the step itself produced, so map_error: e1 -> e2 is only ever called forward, never inverted. decide_crash covers Crashed and TimedOut, which the coordinator detects itself and which carry no error value. resolve does the same for a durable restart: it is the recovery resolver with the decider bound in.

pub opaque type Step(i, o, e, u)

StepAddress

</>

A step’s recorded location: nested scope (from embed and choose), a name, and the 1-based occurrence rank among steps sharing the same scope and name.

pub type StepAddress {
  StepAddress(scope: List(String), name: String, occurrence: Int)
}

Constructors

StepAddress(scope: List(String), name: String, occurrence: Int)

StepDescriptor

</>

A static, read-only description of one step: its address, its dependencies, and which capabilities it was authored with. Never exposes executable closures or values.

pub type StepDescriptor {
  StepDescriptor(
    address: StepAddress,
    depends_on: List(StepAddress),
    undoable: Bool,
    compensates: Bool,
    max_attempts: Int,
    timeout: option.Option(duration.Duration),
  )
}

Constructors

StepDescriptor(
  address: StepAddress,
  depends_on: List(StepAddress),
  undoable: Bool,
  compensates: Bool,
  max_attempts: Int,
  timeout: option.Option(duration.Duration),
)

Undo

</>

Whether a successful (or Continued) output has an undo action.

pub type Undo(u) {
  NoUndo
  UndoWith(run: fn() -> Result(Nil, u))
}

Constructors

```
NoUndo
```
```
UndoWith(run: fn() -> Result(Nil, u))
```

UndoRequest

</>

What an undo action receives: the step’s input, the output that succeeded, and the undo action’s own EffectKey.

Saga builds this record. Read its fields by label (undo.output, or UndoRequest(output:, ..) in a pattern), which keeps compiling when a later release adds a field; do not build or match it positionally. It stays a public record so that the common callback stays fn(undo) { release(undo.output) }.

pub type UndoRequest(i, o) {
  UndoRequest(input: i, output: o, key: EffectKey)
}

Constructors

UndoRequest(input: i, output: o, key: EffectKey)

Workflow

opaque </>

A pure, named workflow description with input, output, business error, and undo-error types.

A builder function runs exactly once per define, in the calling process, to validate the workflow and build its graph; embed runs a workflow’s builder once more as part of another workflow’s define. No run evaluates a builder: every run replays the built graph against its own value store, so concurrent runs of one Workflow never share values.

pub opaque type Workflow(i, o, e, u)

Values

address_to_string

</>

pub fn address_to_string(address: StepAddress) -> String

Renders checkout/charge_payment#2 (the occurrence suffix is only shown when there is more than one occurrence at that scope + name).

all

</>

pub fn all(
  first: Port(a, e, u),
  rest: List(Port(a, e, u)),
) -> Port(List(a), e, u)

Combines port values in list order. The required first Port excludes an empty input; callers with a list of Ports must handle the empty case.

attempt_key

</>

pub fn attempt_key(key: EffectKey) -> String

A key unique to this attempt, to record or look up one attempt.

attempt_number

</>

pub fn attempt_number(key: EffectKey) -> Int

The 1-based number of the attempt.

both

</>

pub fn both(
  first: Port(a, e, u),
  second: Port(b, e, u),
) -> Port(#(a, b), e, u)

Combines two ports into one pair, without adding a scheduled node. Both original ports still execute according to their own dependencies.

choose

</>

pub fn choose(
  input: Port(i, e, u),
  name: String,
  decision: Port(Bool, e, u),
  when_true: fn(Port(i, e, u)) -> Port(o, e, u),
  when_false: fn(Port(i, e, u)) -> Port(o, e, u),
) -> Port(o, e, u)

Constructs both branches once, and executes only the selected branch. The decision is a scheduled value, shared by all nodes in that branch.

compensate

</>

pub fn compensate(
  step: Step(i, o, e, u),
  max_attempts max_attempts: Int,
  with decide: fn(FailedAttempt(i, e)) -> Recovery(o, e, u),
) -> Step(i, o, e, u)

Attaches an explicit recovery decision for a failing attempt (an application error, a crash, or a timeout), with a total attempt budget. With no compensate, a failure aborts after exactly one attempt.

A crash stays visible whatever decide returns. decide receives Crashed/TimedOut for an attempt that never returned, whose effect may or may not have happened. Whatever it returns, that attempt is named in the outcome’s execution.unknown_effects, as is a returned error that unknown_when classified. In particular, Abort(error) after a crash ends the run with execution.StepFailed(step, error), the same cause as an aborted typed error: choose an error that says the attempt crashed if the caller must tell the two apart from the cause.

correlation_of

</>

pub fn correlation_of(key: EffectKey) -> correlation.Correlation

The correlation of the run that performs the action; see EffectKey.

define

</>

pub fn define(
  name: String,
  build: fn(Port(i, e, u)) -> Port(o, e, u),
) -> Workflow(i, o, e, u)

Builds and validates a named workflow written in source code. The builder runs once, in the calling process, before any runtime resource exists: define checks step names, attempt budgets, timeouts, that every port used belongs to this evaluation, and that every step reaches the output.

A defect is a bug in the source, so define panics with a message that names the workflow and every offending step; any test that builds the workflow catches it. Use try_define when names, budgets or timeouts come from runtime data.

describe

</>

pub fn describe(
  workflow: Workflow(i, o, e, u),
) -> List(StepDescriptor)

Static, read-only descriptors for every step, in builder call order.

describe_definition_error

</>

pub fn describe_definition_error(
  error: DefinitionError,
) -> String

Describes a definition error for logs.

effect

</>

pub fn effect(
  name: String,
  run: fn(i, EffectKey) -> Result(o, e),
) -> Step(i, o, e, u)

Creates a step whose run function also receives the attempt’s EffectKey. Send saga.idempotency_key(key) to the external system as its idempotency key: it stays the same across retries, so a retry after an unknown outcome cannot repeat the effect. saga.correlation_of(key) is the run’s correlation, for the step’s own clients.

embed

</>

pub fn embed(
  input: Port(i, e, u),
  workflow: Workflow(i, o, e, u),
) -> Port(o, e, u)

Composes workflow into the graph being built for another workflow, sharing the same run and journal (not an independent child). The embedded steps are addressed under a nested scope named after workflow, so repeated embeds and name collisions stay distinct in describe, address_to_string and telemetry. An inner builder may use ports of the outer workflow; a port from an unrelated definition is reported as ForeignPort.

idempotency_key

</>

pub fn idempotency_key(key: EffectKey) -> String

The key to send downstream as an idempotency key. It is the same for every attempt of the step in one run of a local workflow, or one durable execution across restarts.

map

</>

pub fn map(
  port: Port(a, e, u),
  with: fn(a) -> b,
) -> Port(b, e, u)

Applies a pure transformation to a port’s value. map is not memoized: it re-runs in every task that consumes the resulting port. with is only ever invoked from the innermost (task-level) call of fetch — see Port’s doc comment — so a panic or a slow computation inside with never reaches the coordinator.

map_errors

</>

pub fn map_errors(
  workflow: Workflow(i, o, e1, u1),
  error map_error: fn(e1) -> e2,
  undo_error map_undo_error: fn(u1) -> u2,
) -> Workflow(i, o, e2, u2)

Adapts a whole workflow’s error and undo-error types. The mapped workflow reuses the original’s built graph and wraps each step’s results; its builder never runs again for a standalone run, and runs once inside another workflow’s define when the mapped workflow is embedded.

map_step_errors

</>

pub fn map_step_errors(
  step: Step(i, o, e1, u1),
  error map_error: fn(e1) -> e2,
  undo_error map_undo_error: fn(u1) -> u2,
) -> Step(i, o, e2, u2)

Adapts a step’s error and undo-error types into a unified workflow vocabulary. Persistence added by saga/durable before the mapping is kept: its resolver’s answers are mapped forward like the step’s own results. A compensate decider written in the old vocabulary keeps deciding the step’s failures, including a failure that a resolver added before the mapping establishes after a restart. A resolver added after the mapping answers in the new vocabulary, so a Failed answer from it needs a decider in the new vocabulary too: without one the execution suspends with durable.InvalidCheckpoint.

name

</>

pub fn name(workflow: Workflow(i, o, e, u)) -> String

The workflow’s declared name.

on_unknown

</>

pub fn on_unknown(
  step: Step(i, o, e, u),
  policy: OnUnknown,
) -> Step(i, o, e, u)

Sets what the run does when this step ends with an error that unknown_when classified (default Reconcile); see unknown_when. Crashes and timeouts are not affected.

perform

</>

pub fn perform(
  input: Port(i, e, u),
  step: Step(i, o, e, u),
) -> Port(o, e, u)

Schedules step to run once its input port is ready, producing a new port for its output. Reusing the same input Port value from two perform calls creates two independent nodes (and therefore two occurrences); reusing the resulting Port value across consumers shares the one node.

step

</>

pub fn step(
  name: String,
  run: fn(i) -> Result(o, e),
) -> Step(i, o, e, u)

Creates a step from its name and its run function. With no further modifiers, a failure is terminal after one attempt and nothing is undone on rollback. The run function receives only the step’s input; use effect to read the attempt’s keys and the run’s correlation.

timeout

</>

pub fn timeout(
  step: Step(i, o, e, u),
  limit: duration.Duration,
) -> Step(i, o, e, u)

Bounds one attempt of this step to limit, overriding the run’s execution.with_step_timeout default in either direction. define rejects a limit below 1 millisecond.

try_define

</>

pub fn try_define(
  name: String,
  build: fn(Port(i, e, u)) -> Port(o, e, u),
) -> Result(Workflow(i, o, e, u), List(DefinitionError))

Builds and validates a named workflow like define, but returns every defect as a DefinitionError instead of panicking, for a workflow whose step names, attempt budgets or timeouts come from runtime data. describe_definition_error renders one.

undo

</>

pub fn undo(
  step: Step(i, o, e, u),
  run: fn(UndoRequest(i, o)) -> Result(Nil, u),
) -> Step(i, o, e, u)

Attaches an undo action, run only if this step’s attempt already succeeded and the run later rolls back. It receives the step’s input, the output that succeeded and the undo’s own EffectKey, and runs at most once: a failed, crashed or timed-out undo is reported in the settlement and never retried.

unknown_when

</>

pub fn unknown_when(
  step: Step(i, o, e, u),
  classify: fn(e) -> Bool,
) -> Step(i, o, e, u)

Marks the returned errors after which the step’s effect is unknown: a timeout or connection loss after the request may have been sent, for example. The attempt is named in the outcome’s execution.unknown_effects with execution.ActionReturnedUnknown, so a retried success becomes CompletedWithUnknownEffects. Calling it twice marks the errors either classifier matches.

A matching error still reaches the compensate decider as Returned(error), and the decider’s decision applies: Abort rolls the run back, Hold ends it Unresolved. When the step has no decider, or its decider asked for a retry the attempt budget no longer allows, the step ends as on_unknown says: by default (Reconcile) the run ends execution.Unresolved with the error as evidence and undoes nothing, so an uncertain payment never releases the stock reserved before it. on_unknown(RollBack) fails the run and undoes the completed steps instead. Durable runs apply the same rule to an error that a recovery resolver reports after a restart.