study › DDIA › Chapter 13 · A Philosophy of Streaming Systems

A Multishard Transfer Stays Exactly-Once Through a Crash — Without 2PC

A payment moves 11 from account A to account B — two different shards, no distributed transaction. Crash the processor mid-transfer, after crediting B and before debiting A, forcing redelivery. With dedup by request id the invariant holds — A=89, B=111, sum stays 200, each account touched once. Drop the dedup and the same crash double-credits B to 122, sum jumps to 211, money is created.

Concept

A payment moves 11 from account A (payer shard) to account B (payee shard) — two different shards, no distributed transaction between them. The transfer is written to a request log as one atomic append; a deterministic stream processor consumes that request and emits two events, a credit to B and a debit to A, each carrying the original request id. Now inject a crash mid-transfer — after the payee is credited, before the payer is debited and before the request is marked done — forcing the request to be redelivered. A reader predicts: without an atomic commit across both shards, this crash must either lose money or double-credit someone. Run it and the invariant holds — A=89, B=111, sum stays 200, each account touched exactly once — because the replay re-emits identical events and the shards dedupe by request id. Drop the dedup and the same crash double-credits B to 122, sum jumps to 211, money is created. Atomicity came from atomically writing one request, not from 2PC.

Provenance

Designing Data-Intensive Applications, 2nd ed. (Kleppmann & Riccomini, 2025), Chapter 13 — The Future of Data Systems, §"Exactly-once execution of an operation" and §"The end-to-end argument in databases" (Figure 13-2):

The fact that the message is delivered atomically along with its side effects [...] is what makes exactly-once semantics possible.

The book's construction: you do not need a distributed transaction to make a cross-shard operation exactly-once. You make the initial request an atomic event in a single log, derive all downstream effects deterministically from it, and make the consumers idempotent (dedupe by an operation id). At-least-once delivery plus idempotent application equals effectively-once.

Prerequisites

The surprise is that no step spans both shards atomically, yet the pair stays consistent. Four ideas carry it; each line notes where it shows up.

The log append is the only atomic action — writing the request to the request log is a single-object append, trivially atomic. Everything after is derived. It shows up as the one request-log append line before any event is emitted.
Deterministic derivation — the processor is a pure function of the request: same request in, byte-identical req-1:credit / req-1:debit out, every replay. The event ids are derived from the request id, never freshly generated. It shows up as the recovery run re-emitting the exact same event ids.
Idempotent consumers (dedupe by id) — each shard remembers the event ids it has applied and skips a repeat, so applying an event twice equals applying it once. It shows up as B: event req-1:credit already applied -> SKIP.
Why 2PC is avoided — a distributed transaction would make the debit and credit commit atomically together, but it blocks on a coordinator and holds locks across shards; the log-derive-dedupe construction gets the same all-or-nothing effect without any cross-shard commit. It shows up as the whole thing running with no coordinator, prepare, or commit phase.

Where to learn the prerequisites Exactly-once via atomic message + idempotence (#1–#3): DDIA §"Exactly-once execution of an operation" and §"Idempotence"; Kleppmann, "Exactly-once Semantics is Possible" (Confluent blog, 2017). Why 2PC blocks (#4): DDIA Ch. 8 §"Distributed Transactions and Consensus" — two-phase commit and the coordinator-failure problem it inherits. Flag the joint claim: atomic request-write + deterministic derivation + dedup carry the result together. Remove any one and the crash breaks something — that's exactly what Scenario 3 shows by removing the dedup.

Environment these numbers came from Deterministic, so the balances reproduce exactly. Captured on macOS 26.5.2 (Darwin 25.5.0), arm64, Python 3.15.0a8, standard library only — no Docker, no deps. No real concurrency or real messaging: the crash and redelivery are modelled explicitly in one process, so the result is fixed every run. Fixed workload: transfer 11 from A to B, both starting at 100 → sum 200.

Mental model

Instead of atomically committing across the payer and payee shards (a distributed transaction, 2PC), you atomically commit one request event, then derive every shard effect deterministically from it.

The atomic step is a single append to the request log. That's the only place all-or-nothing is needed, and a single-object append gives it for free.
The derivation — request → (credit event, debit event) — is a pure, deterministic function. Replaying it produces identical events with identical ids.
The application — each shard applying its event — is idempotent: it dedupes by event id, so a redelivered event changes nothing the second time.

A crash after the first effect just replays the derivation. Because replay is deterministic and application is idempotent, at-least-once redelivery becomes effectively-once, and the two shards end up consistent with no blocking commit protocol. The whole thing is one script, code/multishard_exactly_once.py.

Setup

cd study/ddia/ch13/code
python3 multishard_exactly_once.py

Do not read the output yet — make each prediction first.

Step 1 · the happy path

Predict. Transfer 11 from A (starts 100) to B (starts 100), no crash. The processor credits B then debits A. What are the two balances, and what is their sum?

What are the balances and their sum?

==================================================================== SCENARIO 1 -- happy path (no crash, dedup on) ==================================================================== initial: A=100, B=100, total=200 request-log append (the ONE atomic action): req-1 = transfer 11 A->B processor: derive events from request req-1 (transfer 11 A->B) processor: emit credit req-1:credit to B B: apply req-1:credit (+11) -> balance 111 processor: emit debit req-1:debit to A A: apply req-1:debit (-11) -> balance 89 processor: mark request req-1 done invariant: A=89 + B=111 = 200 (started at 200) -> HELD touches: A touched 1x, B touched 1x -> each account touched exactly once

A=89, B=111, sum 200. No surprise yet — the request appended, the processor derived a credit and a debit, both applied, done. The question is what a crash does to it.

Step 2 · crash mid-transfer, dedup ON

Predict. Same transfer, but the processor crashes after crediting B, before debiting A and before marking the request done. On recovery the request (still un-done) is reprocessed from the log — so the credit to B is emitted a second time. With per-shard dedup by event id, what are the final balances? Is B credited once or twice?

Does the crash lose money, double-credit, or neither?

==================================================================== SCENARIO 2 -- crash mid-transfer, dedup ON (redelivery is safe) ==================================================================== initial: A=100, B=100, total=200 request-log append (the ONE atomic action): req-1 = transfer 11 A->B --- first delivery: crash after crediting payee --- processor: derive events from request req-1 (transfer 11 A->B) processor: emit credit req-1:credit to B B: apply req-1:credit (+11) -> balance 111 !! CRASH: crashed after crediting payee, before debiting payer or recording completion !! request req-1 status = NOT done -> must be redelivered --- recovery: request still un-done, reprocess from the log (at-least-once) --- processor: derive events from request req-1 (transfer 11 A->B) processor: emit credit req-1:credit to B B: event req-1:credit already applied -> SKIP (balance stays 111) processor: emit debit req-1:debit to A A: apply req-1:debit (-11) -> balance 89 processor: mark request req-1 done invariant: A=89 + B=111 = 200 (started at 200) -> HELD touches: A touched 1x, B touched 1x -> each account touched exactly once

A=89, B=111, sum 200 — the invariant held, and each account was touched exactly once. The redelivered credit hit B's dedup set (req-1:credit already applied -> SKIP) and changed nothing; the debit, which the crash had prevented, ran once. A crash mid-transfer across two shards, no distributed transaction, and the money is conserved.

Step 3 · the SAME crash, dedup OFF

Predict. Identical crash and redelivery, but now the shards keep no memory of applied event ids. The credit to B is delivered twice and there's nothing to dedupe it. What is B's balance now? What is the sum — still 200?

With dedup gone, what is B's balance and the sum?

==================================================================== SCENARIO 3 -- SAME crash, dedup OFF (redelivery double-credits) ==================================================================== initial: A=100, B=100, total=200 request-log append (the ONE atomic action): req-1 = transfer 11 A->B --- first delivery: crash after crediting payee --- processor: derive events from request req-1 (transfer 11 A->B) processor: emit credit req-1:credit to B B: apply req-1:credit (+11) -> balance 111 !! CRASH: crashed after crediting payee, before debiting payer or recording completion !! request req-1 status = NOT done -> must be redelivered --- recovery: request still un-done, reprocess from the log (at-least-once) --- processor: derive events from request req-1 (transfer 11 A->B) processor: emit credit req-1:credit to B B: apply req-1:credit (+11) -> balance 122 processor: emit debit req-1:debit to A A: apply req-1:debit (-11) -> balance 89 processor: mark request req-1 done invariant: A=89 + B=122 = 211 (started at 200) -> VIOLATED (money created) touches: A touched 1x, B touched 2x -> an account was touched more than once

B=122, sum 211 — money was created out of nothing. The redelivered credit applied a second time (balance 122), B was touched twice, and the conservation invariant broke by exactly the transfer amount. Same log, same deterministic processor, same crash — the only removed ingredient was the dedup, and that alone is the difference between exactly-once and a double-credit.

What you should see

Happy path: A=89, B=111, sum 200, invariant held, each account touched once.
Crash + dedup: A=89, B=111, sum 200 — held through the crash; the redelivered credit is skipped and each account is still touched exactly once.
Crash − dedup: A=89, B=122, sum 211 — invariant violated, B touched twice, money created by exactly the 11 that was double-credited.

Why

The reader's prediction — "without an atomic commit across both shards a crash must lose money or double-credit" — assumes atomicity has to span the two shards. It doesn't. Look at where all-or-nothing is actually needed: only the request must be recorded atomically, and that's a single append to one log — a single-object write, atomic for free. Everything downstream is a deterministic function of that one durable fact. Because it's a function, re-running it on recovery yields byte-identical events: the recovery run re-emits req-1:credit and req-1:debit, the exact ids from the first run, not fresh ones. That determinism is what makes redelivery safe to attempt at all.

The second half is idempotence. At-least-once delivery means an event can arrive more than once (the crash guarantees it here). A shard that records the event ids it has applied turns a duplicate into a no-op: max-like, apply(x) then apply(x) equals apply(x). Deterministic re-emission gives the same id to retry against; the dedup set recognizes it; the effect lands exactly once. Compose the two and you get the book's equation: at-least-once delivery + idempotent (deduped) application = effectively-once, and it spans both shards without either one ever participating in a cross-shard commit. Scenario 3 removes just the idempotence and the composition collapses — the duplicate credit lands for real, and 200 becomes 211.

The boundary — the derivation must be deterministic and the effects dedupable This whole construction rests on replay producing identical events. Put one non-deterministic step in the derivation — a fresh uuid4() per event, a now() timestamp, a random shard pick — and the retry emits a different id that the dedup set has never seen, so it applies as new and you double-credit even with dedup on. Same failure mode as statement-based replication forwarding NOW() or RAND() (the Ch. 6 exercise): a non-deterministic statement replays to a different result. Determinism is not a nice-to-have here; it is the thing that makes the dedup key stable enough to catch the duplicate. Effects must also be things you can dedupe (carry a stable operation id); a side effect with no id — sending a raw email, poking a non-idempotent external API — falls outside the guarantee and needs its own idempotency key.

Go deeper

Why this beats 2PC. Add a third shard (a fee account) and a second concurrent transfer, and note the log-derive-dedupe version needs no coordinator, no lock held across shards, and survives a processor crash by replay — where 2PC would block every participant until the coordinator recovered. Measure nothing new; reason about the failure modes each avoids.
Break determinism on purpose. Change the event id from f"{rid}:credit" to include uuid4() or time.time(), rerun Scenario 2, and watch dedup fail to catch the redelivered credit — the invariant breaks with dedup still on. Cross-reference the Ch. 6 statement-based-replication exercise: same root cause.
The single-shard version. Collapse to one account and a retried "charge the card" request; this is the retry-double-charge problem, where the same atomic-request + idempotency-key idea prevents charging a customer twice on a network retry — the everyday face of Figure 13-2.

Sources: DDIA 2e, Ch. 13, §"Exactly-once execution of an operation", §"The end-to-end argument in databases" (Figure 13-2), §"Idempotence" · Kleppmann, "Exactly-once Semantics is Possible" (Confluent, 2017).