Transactions, 2PC, sagas and idempotency¶
TL;DR¶
- A transaction makes writes atomic, isolated and durable on one node; across services you pick an atomic commit (2PC) or a saga of local transactions with compensations.
- Isolation is a dial: read committed allows lost updates, snapshot isolation allows write skew, serializable costs aborts or locks.
- Exactly-once processing is at-least-once delivery plus idempotency.
- Interviewers probe your isolation anomalies, a coordinator dying mid-commit, and a retry that must not charge twice.
Core concepts¶
ACID: what each letter buys you¶
Atomicity: all of a transaction's writes become visible or none do; recovery uses the write-ahead log (WAL) to roll an unfinished one back. Consistency: the database's invariants hold at commit; the application's stay the application's job. Isolation: concurrent transactions do not see each other's intermediate state. Durability: the commit record is forced to the WAL, or to a quorum of replicas, before the client hears "ok"; that forced write bounds a primary at ~5k-20k writes/s, so engines batch commits per fsync (group commit).
Isolation levels and the anomalies they allow¶
A dirty read sees an uncommitted write. A non-repeatable read reads a row twice and gets two values. A phantom re-runs a predicate query and finds rows appeared or vanished. A lost update is two read-modify-write cycles on one row: both read qty = 1, both write qty = 0, two customers own the last item. Write skew: two transactions read an overlapping set, each writes a different row, both commit, and a cross-row invariant breaks; two on-call doctors each see two on call and both leave.
| Anomaly | Read uncommitted | Read committed | Snapshot isolation (repeatable read) | Serializable |
|---|---|---|---|---|
| Dirty read | possible | prevented | prevented | prevented |
| Non-repeatable read | possible | possible | prevented | prevented |
| Phantom | possible | possible | prevented in practice | prevented |
| Lost update | possible | possible | aborted in PostgreSQL, possible in MySQL | prevented |
| Write skew | possible | possible | possible | prevented |
Read committed (the PostgreSQL, Oracle and SQL Server default) lets each statement see data committed before it started. PostgreSQL's repeatable read is snapshot isolation: one snapshot for the whole transaction, and the first committer wins, so the second writer of a row gets a serialization failure and retries. MySQL InnoDB's repeatable read does not abort the second writer, so read-then-update still loses updates unless you lock. Serializable equals some serial order, by locks (MySQL) or by aborting dangerous patterns (PostgreSQL); the price is retries and a lower commit rate.
How engines isolate: 2PL, MVCC and optimistic versions¶
Two-phase locking (2PL) holds shared read locks and exclusive write locks until commit, aborting a victim on deadlock: serializable, but readers and writers block each other. MVCC gives each write a new version tagged with its transaction id; readers pick the versions visible at their snapshot and never block writers. Snapshot isolation is MVCC reads plus write-conflict detection. Where an invariant matters, steer the engine: SELECT ... FOR UPDATE locks the rows you will write, and UPDATE stock SET qty = qty - 1 WHERE sku = ? AND qty > 0 moves the read-modify-write inside the engine. Optimistic concurrency is the lock-free form: read version = 7, write SET version = 8 WHERE id = ? AND version = 7, retry when zero rows change. It wins at low contention and crosses services (DynamoDB conditions, HTTP If-Match); on a hot row, retries burn throughput.
Two-phase commit and why it blocks¶
Both participants vote YES, then the coordinator's log decides, or fails to.
sequenceDiagram
autonumber
participant C as Coordinator
participant O as Orders DB
participant P as Payments DB
C->>O: PREPARE tx-4, alice -10
O->>O: force-write prepare record, keep the row lock
O-->>C: YES
C->>P: PREPARE tx-4, bob +10
P->>P: force-write prepare record, keep the row lock
P-->>C: YES
Note over C: commit point is the decision record in the coordinator log
alt coordinator logs COMMIT and survives
C->>O: COMMIT tx-4
O-->>C: ack, lock released
C->>P: COMMIT tx-4
P-->>C: ack, lock released
else coordinator crashes before logging a decision
Note over O,P: in doubt, voted YES, may neither commit nor abort alone, locks held
O->>O: tx-5 on alice is refused or waits
C->>O: after recovery, no decision logged, ABORT (presumed abort)
C->>P: ABORT tx-4
endPhase 1: PREPARE; each participant force-writes a prepare record, keeps its locks and votes, and a YES is a promise to no longer abort alone. Phase 2: the coordinator force-writes the decision (the commit point) and broadcasts it; participants apply, release locks and ack.
A participant that dies before voting is a timeout, counted as NO. One that dies after voting YES asks the coordinator on restart. A coordinator that dies before logging its decision leaves every YES voter in doubt, holding locks until it returns, because neither outcome is safe without the decision: 2PC is blocking. On restart, no decision logged means abort (presumed abort). Three-phase commit adds a pre-commit round so a replacement coordinator can finish, but only with bounded delays and no partitions; the practical fix is a replicated coordinator (Spanner runs 2PC over Paxos groups) or no 2PC at all.
The cost rules 2PC out between services: two round trips plus forced log writes, rows locked throughout. In a datacenter that is 2 x 500 µs = 1 ms plus fsyncs; across US regions 2 x 70 ms = 140 ms, so a contended row commits at most ~7 transactions/s.
Sagas: local transactions plus compensations¶
A saga is a sequence of local transactions T1..Tn, each committed in its own service, with compensations that semantically undo them (cancel, release, refund). If Ti fails, the compensations of T1..Ti-1 run in reverse. You lose atomicity and isolation: another request can read a reserved-but-unpaid order, so you add a semantic lock (a PENDING status readers respect), keep updates commutative and run the step most likely to fail first.
Compensatable steps can be undone. The pivot is the go/no-go step, the last one allowed to fail for a business reason (charge the card). Retriable steps after it must eventually succeed (ship, confirm), so they are retried and must be idempotent; one that never succeeds leaves the saga stuck and pages a human.
Orchestrated order saga: the pivot fails, compensations run in reverse.
sequenceDiagram
autonumber
participant S as Saga orchestrator
participant L as Saga log
participant O as Order service
participant I as Inventory service
participant P as Payment service
S->>L: saga o-2 started
S->>O: create order o-2 as PENDING
O-->>S: ok
S->>L: create_order done
S->>I: reserve 1 x sku-1 for o-2
I-->>S: ok
S->>L: reserve_inventory done
S->>P: charge 900 for o-2 (pivot)
P-->>S: declined
S->>L: charge_card failed, state COMPENSATING
S->>I: release reservation o-2 (compensation)
I-->>S: ok
S->>O: cancel order o-2 (compensation)
O-->>S: ok
S->>L: state COMPENSATED
Note over S,P: every call is keyed by o-2, so a restarted orchestrator can replay the log and repeat calls safelyChoreography has each service publish events and react to the others': no central component, but the flow is implicit and compensation logic scatters, so it suits two or three steps. Orchestration gives the saga a coordinator that calls each step and records progress in a saga log, so it can time out, resume after a crash and show an operator where it stopped; choose it from four steps up or when compensations must be audited.
Transactional outbox and CDC¶
Committing a row and then publishing an event is a dual write: crash between the two and the order exists without its event. The outbox pattern writes the business row and an outbox row in one local transaction, so the event exists exactly when the order does. A relay publishes outbox rows: a poller that marks rows sent, or change data capture (CDC), where a connector such as Debezium tails the replication log and publishes in commit order. The relay is at-least-once, a crash after publishing and before recording its position sends an event twice, so consumers keep an inbox of processed event ids.
Outbox row and business row in one commit; CDC publishes at least once; the consumer dedupes.
sequenceDiagram
autonumber
participant A as Order service
participant D as Orders DB with outbox table
participant R as CDC relay
participant K as Kafka topic orders
participant W as Email worker
participant X as Inbox table
A->>D: one local transaction, insert order and outbox row e-17
D-->>A: committed
D-)R: WAL change, outbox row e-17
R->>K: publish e-17
K-->>R: ack
Note over R: crash before the relay records its log position
R->>K: publish e-17 again (at-least-once)
K-)W: e-17
W->>X: seen e-17?
X-->>W: no
W->>W: send the email and insert e-17 into the inbox in one transaction
K-)W: e-17 duplicate
W->>X: seen e-17?
X-->>W: yes, skipIdempotency keys¶
The client generates a key per logical operation (a UUID kept across retries). The server keeps one record per scope:key, scoped to the tenant so clients cannot collide, holding a fingerprint of the request, a state and the response. begin has four outcomes: new (run the handler), completed (replay the stored response), in progress (a twin is running: 409) and mismatch (same key, different payload: 422). The in-progress claim stops two simultaneous retries from both charging; it expires after ~30 s so a dead worker does not lock the key forever, and a claim token rejects that worker's late completion. Keep records for the client's retry horizon, commonly 24 h: 10M writes/day x ~1 KB = 10 GB/day, fine in Redis with a TTL. The strongest placement is the business row's own transaction, so "effect without record" cannot happen; with a separate store, make the handler idempotent too, for example by forwarding your key to the payment provider.
Exactly-once is at-least-once plus idempotency¶
Exactly-once delivery is impossible: when the ack is lost, the sender cannot tell "never received" from "received, ack lost"; resend and you may duplicate, stop and you may lose. So choose at-least-once and make the effect idempotent: dedupe by id, use naturally idempotent writes (set, upsert) or compare-and-set on a version. Kafka's exactly-once semantics are this inside Kafka: an idempotent producer with per-partition sequence numbers plus transactions that commit output records and consumer offsets atomically; an email or HTTP call made by a consumer still needs its own idempotency.
Trade-offs¶
| Approach | Atomic across services | Lock or pending time | When the coordinator dies | Use when |
|---|---|---|---|---|
| Local transaction (one database) | n/a, one database | One fsync, ~1 ms | No coordinator | Rows that must agree live on one shard |
| 2PC (XA or in-database) | Yes | ~1 ms in-DC, ~140 ms cross-region, locks held | YES voters block until recovery | Shards of one database, DB plus queue in one DC |
| Orchestrated saga | No, compensations | Each step commits locally | Resumes from its log | 4+ steps, audited compensations |
| Choreographed saga | No, compensations | Each step commits locally | No coordinator; stuck flows hard to see | 2-3 steps |
| Outbox + idempotent consumers | Row and event together | Relay lag | Relay resends | Every event-emitting service |
First ask whether the rows that must agree can share a shard: an order and its lines under order_id, a wallet and its ledger under account_id, and a local transaction gives you every guarantee for free. When writes cross services, default to an orchestrated saga with an outbox: each service keeps its own transactions, the orchestrator keeps the flow visible, the outbox makes events reliable. Use 2PC only where a product runs it for you (a database spanning shards, XA between one database and one queue in one datacenter), and say that its availability is the product of the participants': two 99.9% participants give 99.8%, and a coordinator outage freezes locked rows. Choreography suits a short chain owned by one team; orchestration anything with compensations, timeouts or an operator view. In every option each step and consumer is idempotent, because the retry is not optional.
Python implementation¶
The protocol vocabulary:
class Vote(StrEnum):
YES = "yes"
NO = "no"
class Decision(StrEnum):
COMMIT = "commit"
ABORT = "abort"
class CrashPoint(StrEnum):
"""Where the coordinator dies during ``run``, for failure injection."""
BEFORE_DECISION_LOGGED = "before logging the decision"
AFTER_DECISION_LOGGED = "after logging the decision"
class NodeDownError(HandbookError):
"""The remote side is crashed: the caller sees a timeout instead of an answer."""
@dataclass(slots=True)
class TxRecord:
"""One entry of the coordinator's durable log."""
txid: str
participants: tuple[str, ...]
decision: Decision | None = None # None: prepares were sent, no decision logged yet
acked: set[str] = field(default_factory=set)
@property
def delivered(self) -> bool:
return self.decision is not None and set(self.participants) <= self.acked
Participant.prepare locks, validates, writes the prepare record and votes; recover asks the coordinator about in-doubt transactions:
class Participant:
"""A resource manager: a store of integer balances with per-key locks.
``_lock`` guards ``_data``, ``_locks``, ``_prepared`` and ``_decided``. ``_prepared`` plays
the role of the prepare records in the participant's WAL: it survives ``crash()``, which
is exactly why a recovered participant can still be in doubt.
"""
def __init__(self, name: str, data: Mapping[str, int] | None = None, floor: int = 0) -> None:
if not name:
raise ValidationError("participant name must be non-empty")
self.name = name
self._floor = floor # no value may drop below this (stock >= 0, balance >= 0)
self._data: dict[str, int] = dict(data or {})
self._locks: dict[str, str] = {} # key -> txid holding the lock
self._prepared: dict[str, dict[str, int]] = {} # txid -> values to apply on commit
self._decided: dict[str, Decision] = {}
self._down = False
self._crash_after_vote = False
self._lock = threading.Lock()
def get(self, key: str) -> int:
with self._lock:
return self._data.get(key, 0)
def in_doubt(self) -> list[str]:
"""Transactions this participant voted YES on and has not heard a decision for."""
with self._lock:
return sorted(self._prepared)
def locked_keys(self) -> dict[str, str]:
with self._lock:
return dict(self._locks)
def crash(self) -> None:
self._down = True
def crash_after_vote(self) -> None:
"""Arm a crash for the next ``prepare``: the vote gets out, then the node dies."""
self._crash_after_vote = True
def recover(self, coordinator: Coordinator) -> None:
"""Come back and resolve every in-doubt transaction by asking the coordinator."""
self._down = False
for txid in self.in_doubt():
try:
decision = coordinator.decision_for(txid)
except NodeDownError:
return # coordinator still down: stay in doubt, keep the locks
with self._lock:
self._apply_locked(txid, decision)
def prepare(self, txid: str, deltas: Mapping[str, int]) -> Vote:
"""Phase 1: lock the keys, validate, write the prepare record, vote.
A YES vote is a promise: the participant can no longer abort on its own, so it keeps
the locks until ``commit`` or ``abort`` arrives. A NO vote is a unilateral abort.
"""
with self._lock:
self._check_up()
if txid in self._prepared:
return Vote.YES # the coordinator retried; the first answer stands
if any(self._locks.get(key, txid) != txid for key in deltas):
return Vote.NO # another transaction holds a key: vote NO instead of waiting
new_values = {key: self._data.get(key, 0) + delta for key, delta in deltas.items()}
if any(value < self._floor for value in new_values.values()):
return Vote.NO # the write would break the local invariant
for key in deltas:
self._locks[key] = txid
self._prepared[txid] = new_values
if self._crash_after_vote:
self._crash_after_vote = False
self._down = True
return Vote.YES
def commit(self, txid: str) -> None:
"""Phase 2: apply the prepared values and release the locks (idempotent)."""
with self._lock:
self._check_up()
if txid not in self._prepared and self._decided.get(txid) is not Decision.COMMIT:
raise InvalidStateError(f"{self.name}: {txid} is not prepared")
self._apply_locked(txid, Decision.COMMIT)
def abort(self, txid: str) -> None:
"""Phase 2: drop the prepared values and release the locks (idempotent)."""
with self._lock:
self._check_up()
self._apply_locked(txid, Decision.ABORT)
def _apply_locked(self, txid: str, decision: Decision) -> None:
pending = self._prepared.pop(txid, None)
if pending is not None and decision is Decision.COMMIT:
self._data.update(pending)
for key in [key for key, holder in self._locks.items() if holder == txid]:
del self._locks[key]
self._decided[txid] = decision
def _check_up(self) -> None:
if self._down:
raise NodeDownError(f"{self.name} is down")
Coordinator.run collects votes, logs the decision and broadcasts it; crash_at injects a crash on either side of the commit point, recover replays the log:
class Coordinator:
"""The transaction manager. ``_log`` is its durable log and survives ``crash()``.
``_lock`` guards ``_log``, ``_participants`` and the crash flags, so transactions on
disjoint keys can run concurrently from several threads.
"""
def __init__(self) -> None:
self._participants: dict[str, Participant] = {}
self._log: dict[str, TxRecord] = {}
self._down = False
self._crash_at: CrashPoint | None = None
self._lock = threading.Lock()
def register(self, participant: Participant) -> None:
with self._lock:
self._participants[participant.name] = participant
def crash_at(self, point: CrashPoint | None) -> None:
"""Arm a crash for the next ``run`` (``None`` disarms)."""
with self._lock:
self._crash_at = point
def crash(self) -> None:
with self._lock:
self._down = True
def decision_of(self, txid: str) -> Decision | None:
"""Peek at the log (for demos and tests): ``None`` while undecided."""
with self._lock:
record = self._log.get(txid)
return record.decision if record else None
def pending(self) -> list[str]:
"""Decided transactions whose outcome has not reached every participant yet."""
with self._lock:
return sorted(txid for txid, record in self._log.items() if not record.delivered)
def decision_for(self, txid: str) -> Decision:
"""Answer a recovering participant.
Presumed abort: a transaction the log knows nothing about was never committed. A
transaction still in its voting phase may be aborted unilaterally, so that is what an
inquiry does to it.
"""
with self._lock:
self._check_up()
record = self._log.get(txid)
if record is None:
return Decision.ABORT
if record.decision is None:
record.decision = Decision.ABORT
return record.decision
def run(self, txid: str, plan: Mapping[str, Mapping[str, int]]) -> Decision:
"""Run 2PC for ``plan`` (participant name -> key deltas) and return the decision."""
if not plan:
raise ValidationError("a transaction needs at least one participant")
with self._lock:
self._check_up()
if txid in self._log:
raise ConflictError(f"{txid} already exists")
unknown = sorted(set(plan) - set(self._participants))
if unknown:
raise NotFoundError(f"unknown participants: {unknown}")
record = TxRecord(txid, tuple(plan))
self._log[txid] = record
votes = self._collect_votes(txid, plan)
decision = Decision.COMMIT if all(v is Vote.YES for v in votes.values()) else Decision.ABORT
self._maybe_crash(CrashPoint.BEFORE_DECISION_LOGGED, txid)
with self._lock:
if record.decision is None:
record.decision = decision # the commit point: durable before anyone is told
decision = record.decision
self._maybe_crash(CrashPoint.AFTER_DECISION_LOGGED, txid)
self._broadcast(record)
return decision
def recover(self) -> list[str]:
"""Replay the log: undecided transactions abort (presumed abort), decided ones are resent."""
with self._lock:
self._down = False
records = list(self._log.values())
resolved: list[str] = []
for record in records:
with self._lock:
if record.decision is None:
record.decision = Decision.ABORT
delivered = record.delivered
if not delivered:
self._broadcast(record)
resolved.append(record.txid)
return resolved
def _collect_votes(self, txid: str, plan: Mapping[str, Mapping[str, int]]) -> dict[str, Vote]:
votes: dict[str, Vote] = {}
for name, deltas in plan.items():
try:
votes[name] = self._participants[name].prepare(txid, deltas)
except NodeDownError:
votes[name] = Vote.NO # no vote within the timeout counts as NO
return votes
def _maybe_crash(self, point: CrashPoint, txid: str) -> None:
with self._lock:
if self._crash_at is not point:
return
self._crash_at = None
self._down = True
raise NodeDownError(f"coordinator crashed {point.value} of {txid}")
def _broadcast(self, record: TxRecord) -> None:
"""Phase 2: tell every participant; the ones that are down are retried on recovery."""
if record.decision is None:
raise InvalidStateError(f"{record.txid} has no decision to broadcast")
for name in record.participants:
if name in record.acked:
continue
participant = self._participants[name]
try:
if record.decision is Decision.COMMIT:
participant.commit(record.txid)
else:
participant.abort(record.txid)
except NodeDownError:
continue
with self._lock:
record.acked.add(name)
def _check_up(self) -> None:
if self._down:
raise NodeDownError("coordinator is down")
uv run python -m hld.two_phase_commit prints:
tx-1 every participant votes YES -> commit; alice=70 bob=130
tx-2 inventory votes NO (1 - 2 < 0) -> abort; orders rolled back, locks={}
tx-3 payments down during phase 1 -> abort; timeout counts as a NO vote
tx-4 coordinator crashed before logging the decision of tx-4
orders in doubt=['tx-4'] holding {'alice': 'tx-4'}; payments in doubt=['tx-4']
tx-5 touching alice meanwhile -> votes no: blocked by tx-4's lock
coordinator recovers, presumed abort for ['tx-3', 'tx-4']; alice=70 bob=130
tx-6 coordinator crashed after logging the decision of tx-6
log says commit, nobody was told; pending=['tx-6']
coordinator recovers, replays commit -> alice=50 bob=150
tx-7 payments dies right after YES -> commit; payments in doubt=['tx-7'], bob=150
payments recovers and asks the coordinator -> alice=45 bob=155
Step kinds, states and log entries:
class StepKind(StrEnum):
COMPENSATABLE = "compensatable" # may fail; undone by its compensation
PIVOT = "pivot" # the go/no-go step: the last one that may fail
RETRIABLE = "retriable" # after the pivot: retried until it succeeds
class SagaState(StrEnum):
RUNNING = "running"
COMPENSATING = "compensating"
COMPLETED = "completed"
COMPENSATED = "compensated"
STUCK = "stuck" # a retriable step or a compensation exhausted its attempts
class StepFailed(HandbookError):
"""A business failure inside a step (card declined, out of stock): do not retry."""
class TransientError(HandbookError):
"""A timeout or a 5xx from a downstream service: retry the same attempt."""
class OrchestratorCrash(HandbookError):
"""Failure injection: the orchestrator process dies between two log writes."""
@dataclass(frozen=True, slots=True)
class Step:
name: str
action: StepAction
compensation: StepAction | None = None
kind: StepKind = StepKind.COMPENSATABLE
@dataclass(frozen=True, slots=True)
class LogEntry:
step: str
event: str # started, done, failed, compensated
attempt: int = 1
detail: str = ""
@dataclass(slots=True)
class SagaRecord:
saga_id: str
context: SagaContext
state: SagaState = SagaState.RUNNING
entries: list[LogEntry] = field(default_factory=list)
def steps_with(self, event: str) -> list[str]:
return [entry.step for entry in self.entries if entry.event == event]
SagaOrchestrator runs forward, compensates in reverse, and resumes from the log by re-running the step in flight:
class SagaOrchestrator:
"""Drives one saga definition. ``_lock`` guards the crash-injection flag only."""
def __init__(self, steps: Sequence[Step], log: SagaLog, max_attempts: int = 3) -> None:
if not steps:
raise ValidationError("a saga needs at least one step")
if max_attempts < 1:
raise ValidationError("max_attempts must be at least 1")
self._steps = tuple(steps)
self._log = log
self._max_attempts = max_attempts
self._crash_after: str | None = None
self._lock = threading.Lock()
self._validate()
def _validate(self) -> None:
names = [step.name for step in self._steps]
if len(set(names)) != len(names):
raise ValidationError("step names must be unique")
has_compensatable = any(s.kind is StepKind.COMPENSATABLE for s in self._steps)
seen_pivot = False
for step in self._steps:
if step.kind is StepKind.COMPENSATABLE:
if seen_pivot:
raise ValidationError(f"{step.name}: no compensatable step after the pivot")
if step.compensation is None:
raise ValidationError(f"{step.name}: a compensatable step needs a compensation")
elif step.kind is StepKind.PIVOT:
if seen_pivot:
raise ValidationError("a saga has at most one pivot")
seen_pivot = True
elif has_compensatable and not seen_pivot:
raise ValidationError(f"{step.name}: retriable steps must follow the pivot")
def crash_after(self, step_name: str | None) -> None:
"""Arm a crash right after ``step_name`` runs, before its 'done' record is written."""
with self._lock:
self._crash_after = step_name
def start(self, saga_id: str, context: SagaContext) -> SagaState:
return self._drive(self._log.create(saga_id, context))
def resume(self, saga_id: str) -> SagaState:
"""Continue after a crash: a step that was in flight is run again (it must be idempotent)."""
return self._drive(self._log.get(saga_id))
def _drive(self, record: SagaRecord) -> SagaState:
if record.state is SagaState.RUNNING:
done = set(record.steps_with("done"))
for step in self._steps:
if step.name in done:
continue
if self._run_forward(record, step):
continue
stuck = step.kind is StepKind.RETRIABLE # past the pivot: nothing to undo
self._log.set_state(record, SagaState.STUCK if stuck else SagaState.COMPENSATING)
break
else:
self._log.set_state(record, SagaState.COMPLETED)
if record.state is SagaState.COMPENSATING:
self._compensate(record)
return record.state
def _run_forward(self, record: SagaRecord, step: Step) -> bool:
self._log.append(record, LogEntry(step.name, "started"))
if not self._attempt(record, step.name, step.action):
return False
self._maybe_crash(step.name)
self._log.append(record, LogEntry(step.name, "done"))
return True
def _compensate(self, record: SagaRecord) -> None:
done = set(record.steps_with("done"))
undone = set(record.steps_with("compensated"))
for step in reversed(self._steps):
if step.name not in done or step.name in undone or step.compensation is None:
continue
if not self._attempt(record, step.name, step.compensation):
self._log.set_state(record, SagaState.STUCK)
return
self._log.append(record, LogEntry(step.name, "compensated"))
self._log.set_state(record, SagaState.COMPENSATED)
def _attempt(self, record: SagaRecord, name: str, action: StepAction) -> bool:
for attempt in range(1, self._max_attempts + 1):
try:
action(record.context)
return True
except StepFailed as exc:
self._log.append(record, LogEntry(name, "failed", attempt, str(exc)))
return False
except TransientError as exc:
self._log.append(record, LogEntry(name, "failed", attempt, str(exc)))
return False
def _maybe_crash(self, step_name: str) -> None:
with self._lock:
if self._crash_after != step_name:
return
self._crash_after = None
raise OrchestratorCrash(f"orchestrator crashed after {step_name} ran, before its done record")
The demo services are keyed by order id, so a repeated call is a no-op:
class OrderService:
def __init__(self) -> None:
self._status: dict[str, str] = {}
self._lock = threading.Lock()
def set_status(self, order_id: str, status: str) -> None:
with self._lock:
self._status[order_id] = status
def status(self, order_id: str) -> str:
with self._lock:
return self._status.get(order_id, "NONE")
class InventoryService:
"""Reservations keyed by order id make reserve and release idempotent: a repeat is a no-op."""
def __init__(self, stock: Mapping[str, int]) -> None:
self._stock = dict(stock)
self._reservations: dict[str, tuple[str, int]] = {}
self._lock = threading.Lock()
def reserve(self, order_id: str, sku: str, qty: int) -> None:
with self._lock:
if order_id in self._reservations:
return
if self._stock.get(sku, 0) < qty:
raise StepFailed(f"{sku}: {self._stock.get(sku, 0)} left, {qty} wanted")
self._stock[sku] -= qty
self._reservations[order_id] = (sku, qty)
def release(self, order_id: str) -> None:
with self._lock:
reservation = self._reservations.pop(order_id, None)
if reservation is not None:
self._stock[reservation[0]] += reservation[1]
def stock(self, sku: str) -> int:
with self._lock:
return self._stock.get(sku, 0)
def reservations(self) -> int:
with self._lock:
return len(self._reservations)
class PaymentService:
def __init__(self, credit_limit: int) -> None:
self._limit = credit_limit
self._charged: dict[str, int] = {}
self._lock = threading.Lock()
def charge(self, order_id: str, amount: int) -> None:
with self._lock:
if order_id in self._charged:
return
if amount > self._limit:
raise StepFailed(f"card declined: {amount} over the {self._limit} limit")
self._charged[order_id] = amount
def charged(self, order_id: str) -> int:
with self._lock:
return self._charged.get(order_id, 0)
class ShippingService:
"""Times out ``fail_next`` times before it answers: the test bench for retriable steps."""
def __init__(self) -> None:
self._failures_left = 0
self._scheduled: set[str] = set()
self._lock = threading.Lock()
def fail_next(self, count: int) -> None:
with self._lock:
self._failures_left = count
def schedule(self, order_id: str) -> None:
with self._lock:
if order_id in self._scheduled:
return
if self._failures_left > 0:
self._failures_left -= 1
raise TransientError("shipping service timed out")
self._scheduled.add(order_id)
def order_saga(
orders: OrderService,
inventory: InventoryService,
payments: PaymentService,
shipping: ShippingService,
log: SagaLog,
max_attempts: int = 3,
) -> SagaOrchestrator:
"""Create order, reserve stock, charge the card (pivot), then ship and confirm (retriable)."""
def set_status(value: str) -> StepAction:
return lambda ctx: orders.set_status(ctx["order_id"], value)
steps = [
Step("create_order", set_status("PENDING"), set_status("CANCELLED")),
Step(
"reserve_inventory",
lambda ctx: inventory.reserve(ctx["order_id"], ctx["sku"], ctx["qty"]),
lambda ctx: inventory.release(ctx["order_id"]),
),
Step("charge_card", lambda ctx: payments.charge(ctx["order_id"], ctx["amount"]), kind=StepKind.PIVOT),
Step("schedule_shipment", lambda ctx: shipping.schedule(ctx["order_id"]), kind=StepKind.RETRIABLE),
Step("confirm_order", set_status("CONFIRMED"), kind=StepKind.RETRIABLE),
]
return SagaOrchestrator(steps, log, max_attempts=max_attempts)
uv run python -m hld.saga prints:
o-1 happy path -> completed; order=CONFIRMED stock=9; create_order=done reserve_inventory=done charge_card=done schedule_shipment=done confirm_order=done
o-2 card declined at pivot -> compensated; order=CANCELLED stock=9; create_order=done reserve_inventory=done charge_card=failed reserve_inventory=compensated create_order=compensated
o-3 orchestrator crashed after reserve_inventory ran, before its done record
log before resume: ['create_order:started', 'create_order:done', 'reserve_inventory:started']; reservations=2
resume re-runs reserve_inventory, still 2 reservations (idempotent) -> completed; order=CONFIRMED stock=8
o-4 shipping times out twice -> completed; order=CONFIRMED stock=7; create_order=done reserve_inventory=done charge_card=done schedule_shipment=failed schedule_shipment=failed schedule_shipment=done confirm_order=done
o-5 shipping down for good -> stuck; order=PENDING stock=6; create_order=done reserve_inventory=done charge_card=done schedule_shipment=failed schedule_shipment=failed schedule_shipment=failed
card charged 120 and kept: past the pivot nothing is undone; page a human
Each record holds a fingerprint, a state, a claim token and an expiry:
class Outcome(StrEnum):
NEW = "new" # the caller owns the key: run the handler, then complete()
REPLAY = "replay" # completed earlier: return the stored response, run nothing
IN_PROGRESS = "in_progress" # a twin request is running: 409, retry later
MISMATCH = "mismatch" # same key, different payload: 422, a client bug
class RecordState(StrEnum):
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
@dataclass(frozen=True, slots=True)
class Response:
status: int
body: Mapping[str, Any]
@dataclass(slots=True)
class Record:
key: str
fingerprint: str
state: RecordState
token: int # claim token: a new one per takeover, so a stale worker cannot complete
expires_at: float
response: Response | None = None
@dataclass(frozen=True, slots=True)
class Claim:
outcome: Outcome
token: int = 0
response: Response | None = None
def fingerprint(payload: Mapping[str, Any]) -> str:
"""Hash of the canonical JSON form, so a key cannot be silently reused for another request."""
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
return hashlib.sha256(canonical.encode()).hexdigest()[:16]
begin is one critical section with four outcomes; complete refuses a stale claim token:
class IdempotencyStore:
"""Records keyed by ``scope:key`` (scope = tenant or user), so two clients cannot collide.
``_lock`` guards ``_records`` and ``_next_token``. ``begin`` is one critical section, which
is what makes two simultaneous twins resolve to exactly one owner and one 409.
"""
def __init__(self, clock: Clock, ttl: float = 24 * 3600, in_progress_ttl: float = 30.0) -> None:
if ttl <= 0 or in_progress_ttl <= 0:
raise ValidationError("ttl and in_progress_ttl must be positive")
self._clock = clock
self._ttl = ttl
self._in_progress_ttl = in_progress_ttl
self._records: dict[str, Record] = {}
self._next_token = 1
self._lock = threading.Lock()
def __len__(self) -> int:
with self._lock:
return len(self._records)
def begin(self, scope: str, key: str, request_fingerprint: str) -> Claim:
"""Claim the key, or say why the caller must not run the handler."""
if not key:
raise ValidationError("idempotency key must be non-empty")
full_key = f"{scope}:{key}"
now = self._clock.now()
with self._lock:
record = self._records.get(full_key)
if record is not None and record.expires_at <= now:
del self._records[full_key] # completed and past its TTL, or an abandoned claim
record = None
if record is None:
token = self._next_token
self._next_token += 1
expires_at = now + self._in_progress_ttl
self._records[full_key] = Record(
full_key, request_fingerprint, RecordState.IN_PROGRESS, token, expires_at
)
return Claim(Outcome.NEW, token)
if record.fingerprint != request_fingerprint:
return Claim(Outcome.MISMATCH)
if record.state is RecordState.COMPLETED:
return Claim(Outcome.REPLAY, record.token, record.response)
return Claim(Outcome.IN_PROGRESS)
def complete(self, scope: str, key: str, token: int, response: Response) -> None:
"""Store the response; rejected when the claim was taken over by another worker."""
full_key = f"{scope}:{key}"
with self._lock:
record = self._records.get(full_key)
if record is None or record.token != token:
raise ConflictError(f"{full_key}: claim {token} is no longer current")
record.state = RecordState.COMPLETED
record.response = response
record.expires_at = self._clock.now() + self._ttl
def release(self, scope: str, key: str, token: int) -> None:
"""Drop a claim whose handler did nothing, so the client's retry is NEW again."""
full_key = f"{scope}:{key}"
with self._lock:
record = self._records.get(full_key)
if record is not None and record.token == token and record.state is RecordState.IN_PROGRESS:
del self._records[full_key]
def purge_expired(self) -> int:
"""What a background sweeper (or Redis TTL) does: forget expired records, return the count."""
now = self._clock.now()
with self._lock:
expired = [key for key, record in self._records.items() if record.expires_at <= now]
for key in expired:
del self._records[key]
return len(expired)
IdempotentHandler.execute wraps a handler and stores business failures so they replay too:
Handler = Callable[[Mapping[str, Any]], Response]
class IdempotentHandler:
"""Wraps a handler with side effects (charge a card, create an order) in the key protocol.
A ``HandbookError`` from the handler is a deterministic business failure: it is stored and
replayed like any other response, because a retry would fail the same way. Any other
exception means "unknown whether the side effect happened": the claim is kept, twins get
409 until ``in_progress_ttl`` passes, then a retry may take the claim over and run the
handler again, which is why the handler itself must tolerate a second run.
"""
def __init__(self, store: IdempotencyStore, handler: Handler) -> None:
self._store = store
self._handler = handler
def execute(self, scope: str, key: str, payload: Mapping[str, Any]) -> tuple[Outcome, Response]:
claim = self._store.begin(scope, key, fingerprint(payload))
if claim.outcome is Outcome.REPLAY and claim.response is not None:
return claim.outcome, claim.response
if claim.outcome is Outcome.IN_PROGRESS:
return claim.outcome, Response(409, {"error": "a request with this key is in progress"})
if claim.outcome is Outcome.MISMATCH:
return claim.outcome, Response(422, {"error": "key reused with a different payload"})
try:
response = self._handler(payload)
except HandbookError as exc:
response = Response(422, {"error": str(exc)})
self._store.complete(scope, key, claim.token, response)
return Outcome.NEW, response
uv run python -m hld.idempotency_store prints:
POST charge key=k1 new 201 {'charge_id': 'ch_1', 'amount': 120} charges=[120]
retry key=k1, same payload replay 201 {'charge_id': 'ch_1', 'amount': 120} charges=[120]
retry key=k1, amount changed mismatch 422 {'error': 'key reused with a different payload'} charges=[120]
POST key=k1 from another account new 201 {'charge_id': 'ch_2', 'amount': 120} charges=[120, 120]
POST key=k2, card declined new 422 {'error': 'card declined'} charges=[120, 120]
retry key=k2 replay 422 {'error': 'card declined'} charges=[120, 120]
POST key=k3 while a twin is in flight in_progress 409 {'error': 'a request with this key is in progress'} charges=[120, 120]
retry key=k3 after 31 s, claim expired new 201 {'charge_id': 'ch_3', 'amount': 120} charges=[120, 120, 120]
stalled worker completes late rejected: acct-7:k3: claim 4 is no longer current
24 h later: purge_expired removed 4 records, 0 left
retry key=k1 after the TTL new 201 {'charge_id': 'ch_4', 'amount': 120} charges=[120, 120, 120, 120]
In the interview¶
Name the kind of transaction the moment a write crosses a box: "Checkout spans order, inventory and payment, so it is an orchestrated saga: create, reserve, charge as the pivot, then ship and confirm as retriable steps, all keyed by order id, events through an outbox."
Phrases that signal depth: "snapshot isolation stops lost updates but not write skew"; "the coordinator's decision record is the commit point"; "effectively-once is at-least-once plus a scoped idempotency key with a TTL".
Two customers buy the last unit at the same instant under read committed. What happens?
A lost update: both read qty = 1, both write qty = 0. Fix it in the engine: UPDATE stock SET qty = qty - 1 WHERE sku = ? AND qty > 0, or FOR UPDATE, or a version column.
The coordinator crashes after every participant voted YES. What do the participants do?
Wait, in doubt: a YES vote gave up the right to abort and nobody knows whether commit was logged, so each holds its locks until the coordinator restarts and resends the decision or presumes abort.
Why not wrap the three services in one distributed transaction?
Locks held for two network round trips (1 ms in a datacenter, ~140 ms across regions), availability coupled to every participant, and XA support most services lack. A saga keeps transactions local.
Where do idempotency keys live and for how long?
In the business row's transaction when possible, which closes the crash window between effect and record; otherwise Redis with a TTL. Scope by client, store fingerprint and response, expire after the retry horizon, typically 24 h.
How do you publish the order-placed event exactly once?
You do not. Write it into an outbox table in the order's transaction, let a relay or CDC publish it at least once, and have consumers dedupe by event id.
Interview tip
Say which kind of transaction in the same breath: local, 2PC or saga. For a saga, name the pivot and the compensation of every step before it; pointing at the step after which "we refund instead of rolling back" sounds like you have shipped one.
Common mistakes¶
- Trusting repeatable read to stop lost updates everywhere: PostgreSQL aborts the second writer, MySQL does not. Fix: an atomic conditional
UPDATE,FOR UPDATEor a version column. - Compensating by deleting: removing the charge row leaves the ledger lying. Fix: a compensation is a new forward action (a refund entry, a cancelled status).
- An idempotency key with no scope or fingerprint: tenants collide, or a reused key replays the wrong response. Fix: key by
tenant:key, store the request hash, answer 422 on mismatch. - Retrying a POST without a key: a timeout plus a retry is two orders. Fix: the client mints the key before the first attempt and reuses it.
Common mistake
Believing a broker's "exactly-once" setting removes the need for idempotent consumers. Kafka's guarantee covers records and offsets inside Kafka; the email, the HTTP call and the database row your consumer produces happen twice on redelivery unless the consumer dedupes.
Self-check¶
Which anomaly does snapshot isolation allow?
Write skew: two transactions read an overlapping set, write disjoint rows, both commit, and a cross-row invariant breaks. Prevent it with serializable isolation or FOR UPDATE on the rows read.
What is the commit point of 2PC?
The moment the coordinator's decision record is durable: before it the transaction can still abort, after it commit is inevitable, and YES voters block until they learn which side the crash was on.
What is the pivot of a saga?
The last step allowed to fail for a business reason: steps before it have compensations, steps after it are retried until they succeed and must be idempotent.
Why must outbox consumers be idempotent?
The relay publishes at least once: a crash between publishing and recording its position republishes the event; consumers dedupe by event id.
Why does an idempotency record need an in-progress state?
Two retries can arrive within milliseconds; without a claim both run the handler, with it the second gets 409 and the next retry a replay.
Related¶
- CAP, PACELC and consistency models
- Design a payment system and digital wallet
- Messaging, queues and Kafka internals
- Design Amazon (e-commerce with inventory and flash sales)
- Consistency, replication and isolation tables
- Berenson et al., "A Critique of ANSI SQL Isolation Levels" (SIGMOD 1995)
- Garcia-Molina and Salem, "Sagas" (SIGMOD 1987)
- Gray and Lamport, "Consensus on Transaction Commit" (ACM TODS 2006)