# DDIA Ch. 13: a multishard transfer stays exactly-once through a crash — without 2PC

## 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.

1. **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.
2. **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.
3. **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`.
4. **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

Deterministic, so the balances reproduce exactly:

- 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.

## Steps

### 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?

**Observe.**

```
====================================================================
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?

**Observe.**

```
====================================================================
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?

**Observe.**

```
====================================================================
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

1. **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.
2. **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.
3. **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.