import logging
from typing import Awaitable, Callable, List, Tuple
log = logging.getLogger("saga")
Compensation = Callable[[], Awaitable[None]]
class CompensationError(Exception):
def __init__(self, failures):
self.failures = failures
joined = ", ".join(f"{n}: {e!r}" for n, e in failures)
super().__init__(f"compensation failed for [{joined}]")
class Saga:
def __init__(self, name: str):
self.name = name
self._stack: List[Tuple[str, Compensation]] = []
self._committed = False
def add_compensation(self, name: str, undo: Compensation) -> None:
self._stack.append((name, undo))
async def run_step(self, name: str, do: Awaitable, undo: Compensation):
result = await do
# only register the undo once the forward step has actually succeeded
self.add_compensation(name, undo)
return result
def commit(self) -> None:
self._committed = True
async def _compensate(self) -> None:
failures = []
while self._stack:
name, undo = self._stack.pop()
try:
log.info("compensating step %s", name)
await undo()
except Exception as exc: # keep unwinding; collect for later
log.exception("compensation for %s failed", name)
failures.append((name, exc))
if failures:
raise CompensationError(failures)
async def __aenter__(self) -> "Saga":
return self
async def __aexit__(self, exc_type, exc, tb) -> bool:
if exc_type is None:
self.commit()
return False
log.warning("saga %s failed with %r; rolling back", self.name, exc)
await self._compensate()
return False # re-raise the original exception
import logging
from saga import Saga
log = logging.getLogger("provisioning")
async def provision_tenant(tenant_id: str, plan: str, compute, dns, billing):
async with Saga(f"provision:{tenant_id}") as saga:
vm = await saga.run_step(
"create_vm",
compute.create_vm(tenant_id, plan),
lambda: compute.delete_vm_if_exists(tenant_id),
)
record = await saga.run_step(
"create_dns",
dns.upsert_record(f"{tenant_id}.app.example.com", vm.ip),
lambda: dns.delete_record_if_exists(f"{tenant_id}.app.example.com"),
)
# this step raises -> DNS record then VM are torn down in reverse
hold = await saga.run_step(
"billing_hold",
billing.place_hold(tenant_id, plan),
lambda: billing.release_hold(tenant_id),
)
return {"vm": vm.id, "dns": record.name, "hold": hold.id}
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from provisioning import provision_tenant
@pytest.mark.asyncio
async def test_billing_failure_rolls_back_vm_and_dns():
compute = SimpleNamespace(
create_vm=AsyncMock(return_value=SimpleNamespace(id="vm-1", ip="10.0.0.5")),
delete_vm_if_exists=AsyncMock(),
)
dns = SimpleNamespace(
upsert_record=AsyncMock(return_value=SimpleNamespace(name="t1.app.example.com")),
delete_record_if_exists=AsyncMock(),
)
billing = SimpleNamespace(
place_hold=AsyncMock(side_effect=RuntimeError("card declined")),
release_hold=AsyncMock(),
)
with pytest.raises(RuntimeError, match="card declined"):
await provision_tenant("t1", "pro", compute, dns, billing)
# forward billing never succeeded, so its compensation must NOT run
billing.release_hold.assert_not_awaited()
# completed steps are undone in reverse order
dns.delete_record_if_exists.assert_awaited_once_with("t1.app.example.com")
compute.delete_vm_if_exists.assert_awaited_once_with("t1")
This snippet implements the saga pattern for a multi-step operation that touches several external systems which cannot participate in a single database transaction. When step three fails, there is no ROLLBACK to lean on, so each already-completed step must be compensated — the created VM deleted, the DNS record removed, the billing hold released. The core idea is to record a compensating action immediately after each forward step succeeds, then, on failure, replay those compensations in reverse (LIFO) order.
In saga.py, the Saga class is an async context manager built around a stack of (name, coro_factory) pairs. add_compensation pushes a callable that undoes the step just performed; run_step is a convenience that executes a forward action and registers its compensation atomically so there is never a window where a step succeeded but its undo is unrecorded. The magic lives in __aexit__: if the block exits with an exception, _compensate is invoked, popping the stack and awaiting each undo in reverse. Compensations are best-effort — failures are collected into CompensationError rather than raised immediately, because aborting rollback halfway would leave even more orphaned resources. On a clean exit the stack is simply commit()-ed and discarded.
Compensating actions must be idempotent and tolerant of partial state: a delete may run against a resource that was never fully created, so each undo swallows not-found conditions. This is the central trade-off of sagas versus real transactions — there is no isolation and no atomicity, only eventual cleanup, so intermediate states are briefly visible to other observers.
provisioning.py shows the pattern in use. provision_tenant opens a Saga and calls run_step for the VM, DNS, and billing operations. Because the injected billing.place_hold raises, control unwinds through __aexit__, and the DNS record and VM are torn down in reverse order automatically — the caller never writes explicit cleanup. Each forward call is paired with a lambda capturing the identifiers it needs to compensate.
The pattern is worth reaching for whenever a workflow spans multiple non-transactional resources and a half-finished workflow is worse than none. The main pitfalls are non-idempotent compensations, forgetting to register an undo before the next step runs, and compensations that themselves depend on state a later failing step never produced.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
module EmailNormalization
extend ActiveSupport::Concern
included do
attr_accessor :soft_warningsSoft Validation: Normalize + Validate Email
class CreateDeadJobs < ActiveRecord::Migration[7.1]
def change
create_table :dead_jobs do |t|
t.string :jid, null: false
t.string :queue, null: false
t.string :klass, null: falseBackground Job Dead Letter Queue (DLQ) Table
import axios from 'axios';
export type NormalizedErrors = {
fields: Record<string, string>;
formLevel: string | null;
};Frontend: normalize and display server validation errors
Share this code
Here's the card — post it anywhere.
