saga/durable

Runs a saga.Workflow with saved checkpoints, so that a run survives the loss of its runner, its caller or its VM, and can be recovered, read or cancelled later.

Use this module when a run must outlive the process or VM that started it. It runs the same workflow and the same runner as saga/execution and returns the same execution.Outcome; local runs need none of it.

Make every step recoverable: recoverable gives a step its codecs and the resolver that establishes an interrupted attempt’s effect; restore_undo, resolve_undo and resolve_compensation cover undo and compensation.
new checks the workflow once and attaches root codecs and a compatibility stamp; with_version changes the workflow version (default "1") and with_config sets the run configuration.
start_or_reconnect saves an execution under a caller-chosen id, or reconnects to the one saved there, and returns its Run handle.
drive runs it until it finishes or suspends, within a timeout; read and cancel act on it without a runner.

One saga/storage.Storage serves every execution of a store: saga/storage/memory, saga/storage/file and the saga_postgres package are adapters, and saga/storage/conformance checks another.

import gleam/dynamic/decode
import gleam/json
import saga
import saga/codec
import saga/durable
import saga/storage/memory

let order = codec.json("order-1", fn(id) { Ok(json.string(id)) }, decode.string)
let text = codec.text()
let charge =
  saga.effect("charge", fn(order, key) { charge(order, saga.idempotency_key(key)) })
  |> saga.undo(fn(undo) { refund(undo.output, saga.idempotency_key(undo.key)) })
  |> durable.recoverable(version: "1", input: order, output: text, resolve: lookup_charge)
  |> durable.resolve_undo(lookup_refund)
let workflow = saga.define("checkout", saga.perform(_, charge))
let persistence =
  durable.new(workflow, input: order, output: text, error: text, undo_error: text)
let assert Ok(store) = memory.start()
let assert Ok(run) =
  durable.start_or_reconnect(persistence, memory.storage(store), id: "checkout-123", input: "order-123")
let outcome = durable.drive(run, timeout: duration.seconds(30))

Who stops what. drive runs the execution in a runner process and waits at most timeout. On timeout, when the process that called drive exits, and when the runner is killed or crashes, the run stops: in-flight attempts are killed, the claim is released, and the last checkpoint stays the recovery authority, so the next drive resumes. Only the loss of the runner’s VM leaves the claim to the storage’s own expiry, such as a lease. Only cancel cancels. Waking a runner after a restart is the caller’s job: unfinished lists the executions that wait for one.

See the durability design (docs/design/durability/design.typ) for the storage contract and recovery rules.

Types

Boundary

</>

Which saved value a codec failed on. The union is closed.

pub type Boundary {
  RunInput
  RunOutput
  RunError
  RunUndoError
  StepInput(step: saga.StepAddress)
  StepOutput(step: saga.StepAddress)
}

Constructors

```
RunInput
```
```
RunOutput
```
```
RunError
```
```
RunUndoError
```
```
StepInput(step: saga.StepAddress)
```
```
StepOutput(step: saga.StepAddress)
```

CheckpointProblem

</>

Why a saved checkpoint cannot be restored. Later releases may add variants.

pub type CheckpointProblem {
  Malformed
  ForeignExecution(saved: String)
  GraphMismatch
  ConcurrencyBelowInFlight(in_flight: Int, max_concurrency: Int)
  UndoNotRestorable(step: saga.StepAddress)
  CompensationInputMissing(step: saga.StepAddress)
  DeciderMissingAfterMapping(step: saga.StepAddress)
}

Constructors

```
Malformed
```
The bytes are not a checkpoint this version of saga wrote.
```
ForeignExecution(saved: String)
```
The storage returned the checkpoint of execution saved.
```
GraphMismatch
```
The saved graph has a different number of steps.
```
ConcurrencyBelowInFlight(in_flight: Int, max_concurrency: Int)
```
More attempts were in flight when it was saved than the configured max_concurrency allows now.
```
UndoNotRestorable(step: saga.StepAddress)
```
A saved undo cannot be rebuilt: restore_undo returned NoUndo.
```
CompensationInputMissing(step: saga.StepAddress)
```
A compensation decision was saved before its input was.
```
DeciderMissingAfterMapping(step: saga.StepAddress)
```
A resolver added after saga.map_step_errors answered Failed, and the step has no compensate decider in the mapped vocabulary.

Error

</>

Why a durable operation failed or an execution suspended. Later releases may add variants: branch on error_kind and log describe_error.

pub type Error {
  StorageFailure(error: storage.Error)
  IncompatibleDefinition
  InputMismatch
  CodecFailure(boundary: Boundary, error: codec.CodecError)
  RecoveryRequired(required: Required)
  InvalidCheckpoint(problem: CheckpointProblem)
  CheckpointTooLarge(bytes: Int, limit: Int)
  SuspensionNotSaved(cause: Error, recording: Error)
  InvalidConfig(errors: List(execution.ConfigError))
  InvalidTimeout(timeout: duration.Duration)
  RunnerLost
  DriveTimedOut
}

Constructors

```
StorageFailure(error: storage.Error)
```
The storage refused or failed an operation.
```
IncompatibleDefinition
```
The saved execution belongs to another workflow, version or codec set.
```
InputMismatch
```
start_or_reconnect found the id saved with a different input.

CodecFailure(boundary: Boundary, error: codec.CodecError)

A codec could not save or restore a value at boundary.

```
RecoveryRequired(required: Required)
```
A resolver could not establish an interrupted action’s effect: find out what happened to required.key, then drive again.
```
InvalidCheckpoint(problem: CheckpointProblem)
```
The saved checkpoint cannot be restored by this definition.
```
CheckpointTooLarge(bytes: Int, limit: Int)
```
A checkpoint grew past the limit (with_max_checkpoint_bytes).
```
SuspensionNotSaved(cause: Error, recording: Error)
```
The execution suspended with cause, and saving that suspension failed with recording; the previous checkpoint stays.

InvalidConfig(errors: List(execution.ConfigError))

The run configuration is invalid.

```
InvalidTimeout(timeout: duration.Duration)
```
drive was given a timeout below 1 millisecond.
```
RunnerLost
```
The runner process was killed or crashed before reporting, or the caller’s exit stopped it. Its claim was released; drive again.
```
DriveTimedOut
```
drive reached its timeout; the runner stopped and the execution resumes from its last checkpoint at the next drive.

ErrorKind

</>

The closed classification of an Error, for deciding what to do:

Busy: another runner owns the execution; try again later.
Transient: the runner or the store failed; drive again.
NeedsReconciliation: an effect is unknown; resolve it first.
Incompatible: the saved execution belongs to another definition or input.
Defect: a bug in the definition, configuration, codecs or adapter.

pub type ErrorKind {
  Busy
  Transient
  NeedsReconciliation
  Incompatible
  Defect
}

Constructors

```
Busy
```
```
Transient
```
```
NeedsReconciliation
```
```
Incompatible
```
```
Defect
```

Evidence

</>

What a recovery resolver established about an action that was interrupted before saga saved its result:

Completed(value): it happened, with this result;
Failed(error): it happened and failed with this error;
NotSent: it provably never happened, which authorizes saga to run it again under the same key;
MaybeSent: its effect is unknown, so the execution suspends with RecoveryRequired until a later drive can establish it.

The union is closed.

pub type Evidence(o, e) {
  Completed(o)
  Failed(e)
  NotSent
  MaybeSent
}

Constructors

```
Completed(o)
```
```
Failed(e)
```
```
NotSent
```
```
MaybeSent
```

Persistence

opaque </>

A checked persistence capability for an existing workflow: root codecs, a compatibility stamp and the run configuration. It does not construct another graph.

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

Required

</>

The action whose effect must be established before the execution can continue: its step, which action (StepAttempt(n), StepCompensation(n) or StepUndo), and its EffectKey. Read it by label (Required(key:, ..)); a later release may add fields.

pub type Required {
  Required(
    step: saga.StepAddress,
    action: execution.Action,
    key: saga.EffectKey,
  )
}

Constructors

Required(
  step: saga.StepAddress,
  action: execution.Action,
  key: saga.EffectKey,
)

Run

opaque </>

One saved execution: its persistence, its storage and its id.

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

Status

</>

A saved execution’s state: not finished, suspended with a saved reason, or finished with its outcome.

pub type Status(o, e, u) {
  Pending
  Suspended(reason: Error)
  Finished(outcome: execution.Outcome(o, e, u))
}

Constructors

```
Pending
```
```
Suspended(reason: Error)
```

Finished(outcome: execution.Outcome(o, e, u))

Values

cancel

</>

pub fn cancel(run: Run(i, o, e, u)) -> Result(Nil, Error)

Records cancellation for an unfinished execution. A running drive observes it at its next checkpoint and rolls back. Does nothing for a finished execution.

describe_error

</>

pub fn describe_error(error: Error) -> String

Describes an error for logs.

drive

</>

pub fn drive(
  run: Run(i, o, e, u),
  timeout timeout: duration.Duration,
) -> Result(execution.Outcome(o, e, u), Error)

Runs the execution until it finishes, suspends or timeout passes, and returns its outcome once it is saved. A finished execution returns its saved outcome at once.

The run happens in a runner process. When timeout passes, drive stops the runner and returns DriveTimedOut; when the calling process exits, the runner stops by itself; when the runner itself is killed or crashes while the caller lives, drive returns RunnerLost. In every case the runner’s in-flight attempts are killed, its claim is released at once, and the last checkpoint stays: this is not cancellation, and the next drive resumes, asking each interrupted attempt’s resolver what happened. Only when the runner’s whole VM is lost does the claim stay until the storage notices, such as a lease that expires. Concurrent drives of one execution contend through the storage: all but one return StorageFailure(Busy), which callers retry after at least the storage’s owner-loss window (a lease-based adapter’s lease).

error_kind

</>

pub fn error_kind(error: Error) -> ErrorKind

Classifies an error; see ErrorKind.

id

</>

pub fn id(run: Run(i, o, e, u)) -> String

The execution id given to start_or_reconnect.

new

</>

pub fn new(
  workflow: saga.Workflow(i, o, e, u),
  input input: codec.Codec(i),
  output output: codec.Codec(o),
  error error: codec.Codec(e),
  undo_error undo_error: codec.Codec(u),
) -> Persistence(i, o, e, u)

Checks that every step of workflow can be restored and attaches the root codecs and a compatibility stamp built from the workflow version ("1" until with_version changes it), the graph and every codec version.

A workflow that cannot be persisted is a bug in the source: a step without recoverable, a compensating step without restore_undo, or an empty step or codec version. new then panics with a message naming every offending step and codec.

read

</>

pub fn read(
  run: Run(i, o, e, u),
) -> Result(Status(o, e, u), Error)

Reads the saved state of an execution without a live runner.

reconnect

</>

pub fn reconnect(
  persistence: Persistence(i, o, e, u),
  storage: storage.Storage,
  id id: String,
) -> Result(Run(i, o, e, u), Error)

Reconnects to the execution saved under id, for example one that unfinished listed, without knowing its input. Fails with StorageFailure(NotFound) when there is none.

recoverable

</>

pub fn recoverable(
  step: saga.Step(i, o, e, u),
  version version: String,
  input input: codec.Codec(i),
  output output: codec.Codec(o),
  resolve resolve: fn(i, saga.EffectKey) -> Evidence(o, e),
) -> saga.Step(i, o, e, u)

Makes a step recoverable after a restart: version names the step’s behavior (change it when the step’s callbacks change meaning), input and output save its values, and resolve establishes the effect of an attempt that was admitted but whose result was not saved. resolve receives the attempt’s input and EffectKey, must be safe to call repeatedly, and is called instead of repeating the attempt. Every step of a durable workflow needs this capability; new panics naming each step that lacks it. The order relative to saga.compensate, saga.unknown_when and saga.map_step_errors does not matter, except that a Failed answer from a recoverable added after map_step_errors needs a compensate decider written in the mapped vocabulary (see saga.map_step_errors).

resolve_compensation

</>

pub fn resolve_compensation(
  step: saga.Step(i, o, e, u),
  resolve: fn(i, saga.EffectKey) -> option.Option(
    saga.Recovery(o, e, u),
  ),
) -> saga.Step(i, o, e, u)

Establishes the decision of a saga.compensate decider that was interrupted, from the failed attempt’s input and EffectKey: Some(decision) applies that decision under the original attempt budget, and None keeps the execution suspended. Saga never repeats the decider itself after a restart, so a decider with an external effect records its decision under saga.attempt_key(key).

resolve_undo

</>

pub fn resolve_undo(
  step: saga.Step(i, o, e, u),
  resolve: fn(saga.UndoRequest(i, o)) -> Evidence(Nil, u),
) -> saga.Step(i, o, e, u)

Establishes the effect of an undo that was interrupted: Completed(Nil) when the undo happened, Failed(error) when it failed, NotSent to run the saved undo again, and MaybeSent to suspend. Without a resolver an interrupted undo suspends the execution.

restore_undo

</>

pub fn restore_undo(
  step: saga.Step(i, o, e, u),
  restore: fn(saga.UndoRequest(i, o)) -> saga.Undo(u),
) -> saga.Step(i, o, e, u)

Declares how to rebuild a compensating step’s undo from its saved input and output, after a restart or after a Continue decision. Every persistent step with saga.compensate needs it, even when the answer is saga.NoUndo. The factory must be pure: saga may call it while saving a checkpoint, to check that the undo can be rebuilt.

start_or_reconnect

</>

pub fn start_or_reconnect(
  persistence: Persistence(i, o, e, u),
  storage: storage.Storage,
  id id: String,
  input input: i,
) -> Result(Run(i, o, e, u), Error)

Saves a new execution with input under id, or reconnects to the one already saved there, and returns its handle. Reconnecting succeeds only for a compatible definition and the same encoded input; a different input returns InputMismatch. Does not run anything; call drive. Choose ids that name the workflow ("checkout:" <> order_id) when one store serves several workflows.

unfinished

</>

pub fn unfinished(
  storage: storage.Storage,
  limit limit: Int,
) -> Result(List(String), Error)

The ids of up to limit executions that are pending or suspended and that no live runner owns: the executions that wait for a drive. Reconnect to each with reconnect. Saga itself never drives them.

with_config

</>

pub fn with_config(
  persistence: Persistence(i, o, e, u),
  config: execution.Config,
) -> Persistence(i, o, e, u)

Sets the configuration every drive of this persistence runs with (default execution.config()).

with_correlation

</>

pub fn with_correlation(
  run: Run(i, o, e, u),
  correlation: correlation.Correlation,
) -> Run(i, o, e, u)

Carries correlation in every saga/telemetry event of this handle’s drives, and in the saga.EffectKey of every step callback and resolver that runs under it.

The first drive decides, and the execution keeps it. The first drive of an execution saves the correlation it uses with the checkpoint: this handle’s, or, without one, correlation.from_key of the execution id. Every later drive, on any handle, reads the saved value back, so the events and steps of one execution report one correlation. A value set after the first drive is ignored. An execution saved before the correlation was recorded reads as correlation.from_key(id).

with_max_checkpoint_bytes

</>

pub fn with_max_checkpoint_bytes(
  persistence: Persistence(i, o, e, u),
  bytes: Int,
) -> Persistence(i, o, e, u)

Bounds the size of one saved checkpoint (default 16 MiB). Every commit rewrites the whole checkpoint; a larger one suspends the execution with CheckpointTooLarge.

with_version

</>

pub fn with_version(
  persistence: Persistence(i, o, e, u),
  version: String,
) -> Persistence(i, o, e, u)

Sets the workflow version (default "1"). Change it whenever the workflow’s behavior changes: a saved execution with another version is refused with IncompatibleDefinition instead of being misread. Panics on an empty version.