Skip to content
Menu

Documentation

Add a saga

Goal: a linear process that spans modules, such as "charge, then grant access, then notify", where a failure before the go/no-go point undoes the earlier steps and a failure after it is retried until it succeeds.

Use it when a later failure must undo an earlier module's effect, one module owns the process, and an operator must be able to see and repair a stuck instance.

Do not use it when the change touches one module's tables (one unit of work is enough), or when nothing has to be undone (publish an event and let other modules react). Read Sagas first: a saga has no isolation, and you must design for the intermediate states other requests see.

Prerequisites

  • The owning module exists (Add a module).
  • The command contracts the steps call exist, with idempotent compensation commands that accept "nothing to undo" (Add a contract). The generator gives you a stand-in so the saga runs before they exist.

Steps

1. Run the generator

sandloom add saga notes onboarding

It writes:

  • <module>/sagas.py with the definition (written as sagas.py.new if the file exists);
  • a stand-in command contract notes.onboarding_ledger.v1, appended to <module>/contracts.py with its DTO in <module>/schemas.py, and its stub and DTO copy in the shared DTO package, exactly as sandloom add contract does. The saga's command step calls it, so the generated code boots and runs.

It writes no tests and no route: you add both below.

2. Read what it wrote

"""The onboarding saga (ADR-0008): ``reserve``, then ``confirm``.

Replace the stand-in ledger contract (contracts.py) with the real provider's
command contract; the typed step names it through its generated reference.
"""

from datetime import timedelta
from typing import Any
from uuid import UUID

from pydantic import BaseModel

from demo.shared.contracts.notes_onboarding_ledger_v1 import NOTES_ONBOARDING_LEDGER_V1
from sandloom.sdk import Contract
from sandloom.sdk.sagas import Saga, local_step, step  # sandloom: advanced sagas

ledger = Contract(NOTES_ONBOARDING_LEDGER_V1)


class OnboardingInput(BaseModel):
    owner_id: UUID
    amount: int


def _amount(ctx: Any) -> dict[str, Any]:
    return {"amount": ctx.input.amount}


async def _confirm(session: Any, ctx: Any) -> dict[str, str]:
    return {"owner_id": str(ctx.input.owner_id)}


ONBOARDING = Saga(
    "onboarding",
    version=1,
    input=OnboardingInput,
    deadline=timedelta(minutes=10),
    steps=[
        step("reserve", call=ledger.record, args=_amount, compensate=ledger.record),
        local_step("confirm", apply=_confirm, kind="pivot"),
    ],
)
  • ledger = Contract(NOTES_ONBOARDING_LEDGER_V1) declares the command contract the steps call. call=ledger.record names the command through the typed reference, so a renamed method is a type error.
  • reserve is a command step: it calls the command exactly once at the provider (through its durable command log, keyed by the runtime) and names its compensation.
  • confirm is a local step marked kind="pivot": it runs in the transition's own transaction, and it is the go/no-go point. Before it, a refusal compensates; after it, steps are retried until they succeed.
  • deadline=timedelta(minutes=10) bounds the forward run.

The stand-in provider (contracts.py of a module that had none):

"""Contracts the notes module provides."""

from sandloom.sdk import command, contract

from .schemas import OnboardingLedgerStatus


@contract("notes.onboarding_ledger.v1")
class OnboardingLedger:
    """Contract ``notes.onboarding_ledger.v1``: public ``async def`` methods are reads,
    ``@command`` marks a command (deduplicated by its key, a saga step)."""

    async def status(self) -> OnboardingLedgerStatus:
        return OnboardingLedgerStatus(ready=True)

    @command(idempotency_key="operation_key")
    async def record(self, amount: int, operation_key: str) -> int:
        return amount

3. Replace the stand-in with your process

  1. Point each command step at the real provider: import its generated reference, declare x = Contract(X_Y_V1) at module level, and use call=x.<command> and compensate=x.<undo command>.
  2. Delete the stand-in class from contracts.py, its DTO (OnboardingLedgerStatus) from schemas.py and its stub from the shared DTO package (notes_onboarding_ledger_v1.py), then run sandloom generate (it rewrites the DTO copies), before anything calls them.
  3. Rewrite the input model. It holds ids and amounts, never personal data: it is stored on the instance row.
  4. Keep every args / compensate_args function pure, named and at module level: it reads only ctx.input and ctx.result(...). Arguments are recomputed on every attempt; a drift parks the instance.
  5. Choose the kinds: compensatable steps first, at most one pivot, then retriable steps. Use ctx.key("<step>") to pass a step's operation key to its compensation.
  6. Guard user-visible intermediate states with a status column that your routes respect until the saga decides (a semantic lock).

Changing the steps of a definition that already has running instances is a new version=, registered alongside the old one.

4. Start it from a route

start_saga writes the instance and its first nudge in the caller's unit of work, so the saga starts only if the business change commits. Answer 202 Accepted with a URL to poll:

from sandloom.sdk.sagas import start_saga, saga_status

handle = await start_saga(ONBOARDING, saga_id=row.id, input=OnboardingInput(...))

Starting the same saga_id again with equal input returns the existing instance; other input raises SagaInputConflict (409). saga_status(ONBOARDING, saga_id) is the read side. See Sagas for a full route.

Verify

The saga_harness fixture of the testing plugin drives the real saga protocol in a test, synchronously, and injects faults:

async def test_refund_when_confirm_refuses(saga_harness) -> None:
    saga_harness.fail_step("confirm", error="refused")
    run = await saga_harness.run(ONBOARDING, input=OnboardingInput(owner_id=uuid4(), amount=5))
    assert run.status == "compensated"

Every saga's test module should also contain the two one-liners await saga_harness.assert_compensations_idempotent(ONBOARDING, input=...) and await saga_harness.assert_resolves_before_compensating(ONBOARDING, input=...). Then:

sandloom test notes

In a running deployment, sandloom sagas list shows the instances that need attention and sandloom sagas show <id> the timeline and the remediation of a parked one.

Errors you may hit

  • SANDLOOM-B005: a step's contract is neither provided in the process nor bound in SERVICE_BINDINGS.
  • SANDLOOM-B006: the saga is defined outside the module's sagas submodule.
  • A startup refusal naming the saga, the step and a fix: a step's call or args does not match the provider's describe document.
  • SANDLOOM-B018: a command the saga calls has no idempotency_key.
  • SANDLOOM-R002: start_saga ran outside a unit of work.