Saga

A saga runs a process that spans several aggregates. An aggregate changes one instance at a time, and no transaction spans two, so a process that touches several runs as a series of steps, each committed on its own. When a step fails after earlier ones succeeded, the saga undoes them with compensations.

A saga is an Effect workflow. Each step’s result is stored, so a run that’s interrupted resumes after its last finished step, and a run can wait hours or days for a signal or a timer. A policy usually decides when to start one.

The bank’s transfer is a saga: it debits one account, credits another, and refunds the debit if the credit fails.

CommandAggregatePolicySaga

A policy decides when the transfer runs; the saga carries it out, one step at a time.

Declare a saga

A saga has a key, the input it starts with, and every way it can be refused. The key identifies one run: starting the saga again with the same transfer id returns the same run.

export class ExecuteTransfer extends Domain.Saga("ExecuteTransfer", {
  key: TransferId,
  input: {
    from: AccountId,
    to: AccountId,
    amount: Cents,
    requestedBy: Actor,
    needsApproval: Schema.Boolean,
  },
  rejects: [NotOpen, NotOwner, Frozen, InsufficientFunds, DestinationClosed],
}) {}

Run its steps

Its Layer resolves the clients its steps use, then returns what each run does. Each step is an Activity: a named Effect whose result is stored. When the run replays, a finished step returns its stored result instead of running again.

yield* Activity.make({
  name: "credit",
  error: DestinationClosed,
  execute: accounts
    .send(to, new CreditTransfer({ transferId: key, amount }))
    .pipe(Effect.catchTag("NotOpen", () => Effect.fail(new DestinationClosed()))),
});

A step declares the errors it can fail with, so a refusal survives being stored. Here the destination account’s NotOpen becomes the saga’s own DestinationClosed.

A step that runs again is safe. A step can run twice: when it’s retried, or when its run is cut short after the step’s command committed but before its result was stored. The commands a step sends get ids derived from the run and the step, so the second run sends the same command ids, and the aggregate applies each command once. Keep each step to one command, so each stored result matches one fact.

Undo a step when a later one fails

A step can carry a compensation: what to do if the run fails after the step succeeded. The debit’s compensation refunds it.

yield* Activity.make({
  name: "debit",
  error: Schema.Union([NotOpen, NotOwner, Frozen, InsufficientFunds]),
  execute: accounts.send(from, new DebitTransfer({ transferId: key, amount, requestedBy })),
}).pipe(
  ExecuteTransfer.withCompensation(() =>
    Activity.make({
      name: "refund",
      execute: accounts.send(from, new RefundTransfer({ transferId: key, amount })),
    }),
  ),
);

When the credit is refused, the run fails, and every compensation registered so far runs, last step first.

CommandRejection

Take the step you can’t undo first

Debit before you credit. If the credit then fails, the money is refunded, so it can be in flight but never created out of nothing. In the same way, registration claims the username before the customer counts as registered: see Unique values.

Record the outcome

The Transfer aggregate shows where every transfer stands, so the saga reports back to it. The last step completes the transfer:

yield* Activity.make({
  name: "complete",
  execute: transfers.send(key, new CompleteTransfer()),
});

A failure is recorded by a finalizer. It’s registered first, so it runs after every compensation:

yield* Workflow.addFinalizer((exit) =>
  Exit.isSuccess(exit)
    ? Effect.void
    : Activity.make({
        name: "record-failure",
        execute: transfers.send(key, new FailTransfer({ reason: reasonOf(exit.cause) })),
      }).pipe(Effect.orDie),
);

Start it from a policy

Start a saga from a policy, on the event that records the request. The command that asked for the transfer commits TransferRequested and replies straight away; the policy starts the saga after that, and the saga runs on its own. Nobody waits for it.

CommandAggregateEventPolicySaga

The request is a fact before the process starts, so it's never lost, and the caller never waits for the process.

yield* ExecuteTransfer gives the saga’s client. start begins a run, or returns the one already started with that key, as a handle to the run. Policies run at least once, so key the saga by the aggregate it’s about: a policy that runs twice still starts one transfer.

case "TransferRequested":
  yield* executions.start(source, {
    from: event.from,
    to: event.to,
    amount: event.amount,
    requestedBy: event.requestedBy,
    needsApproval: event.needsApproval,
  });
  return;

List the saga on the model, next to its aggregates and policies, and its Layer with the other handlers.

sagas: [ExecuteTransfer],

Don’t wait for it in a policy

start returns as soon as the run is recorded. Its run handle also has result, which waits for the run to finish, but a run can take days, and a policy that waits holds up every later event on its key. Use result in tests and scripts. Anything that needs to react to how a saga ends should react to the events it reports instead, like TransferCompleted or TransferFailed.

Wait for a decision

A run can wait as long as it needs to. A large transfer waits until the Transfer decides it: an agent approves it, or its approval window closes first. The saga waits on a DurableDeferred, and suspends until someone completes it.

if (needsApproval) {
  // The Transfer decides between approval and expiry; nothing to do if it expired.
  if ((yield* DurableDeferred.await(Decision)) === "expired") return;
}

The policy passes on the decision with the run’s signal. get returns the run with a key:

case "TransferApproved":
  return yield* executions.get(source).signal(Decision, "approved");

The Transfer decides the race between approval and expiry, so there’s one source of truth for it. See Deadlines.

Follow it

TransferStatus is a view of the transfer and both accounts. A client watches it to see the saga finish: the saga reports every step to the Transfer, so the view has everything a client should see.

import * as Effect from "effect/Effect";
import * as Stream from "effect/Stream";
import { TransferId } from "@bank/domain/Transfer/Transfer.id";
import { TransferStatus } from "@bank/domain/Transfer/TransferStatus.view";
import type { CustomerClient } from "./flows.ts";

export const followTransfer = Effect.fn(function* (bank: CustomerClient, transferId: string) {
  yield* bank.TransferStatus.watch(TransferId.make(transferId)).pipe(
    Stream.takeUntil((status) => status?.completedAt !== null || status?.failedReason !== null),
    Stream.runForEach((status) =>
      Effect.log(status === null ? "pending" : TransferStatus.computed.status(status)),
    ),
  );
});

The saga’s own client reads a run too, which suits operators and tests more than apps. get returns the run with a key:

const run = executions.get(transferId);
const status = yield* run.status;

status is NotStarted, Running, Completed with its result, Rejected with its rejection, Interrupted, or Died with a defect. Runs read the same as Effect workflow runs, on every engine.

Test it

A story runs the whole saga in memory: the request, every step, the compensations, and its status at the end.

Test what a saga does, not how it’s built: the commands its steps send show up as events on the aggregates, and Story.saga checks how the run ended, with the same status its run handle reads.

Story.saga(ExecuteTransfer, transfer, { _tag: "Rejected", rejection: new DestinationClosed() }),

To test a saga on its own, complete what it waits for with Story.signal, and move its durable clocks with Story.clock:

Story.saga(ExecuteTransfer, transfer, { _tag: "Running" }),
Story.signal(ExecuteTransfer, transfer, Decision, "approved"),

A step that calls an external system waits until the story resolves the call, as a policy does. A run that dies fails the story, unless a step expects it with Story.saga(ExecuteTransfer, transfer, { _tag: "Died" }).

On Cloudflare, each run is a Durable Object named by its key, with its steps stored in its SQLite. DomainCloudflare.DurableObjects hosts sagas with the rest of the model, and Domain.InMemory runs them in memory.