How to design a digital wallet

A digital wallet holds user balances and moves money between users and merchants entirely inside a system you own: no card network, no outside company in the middle. Owning both sides sounds like it should make the problem easier. It makes it harder, because a wrong number is now yours alone to explain.

In this lesson, we’ll design that wallet from the balance outward. By the end you’ll be able to explain why a balance is computed and not stored, size the write ceiling of a single hot account row from first principles, and move money across two machines without a distributed transaction. Three ideas carry the rest:

• why a balance is computed from a log instead of stored, and what that computation costs;
• why a single popular account row is a hard throughput ceiling, and where that ceiling comes from;
• how to move money between two machines without a distributed transaction, and why two-phase commit is the wrong tool even when you own both databases.

What goes in and what comes out

The input is a transfer request with five fields: a sender account, a receiver account, an amount in whole pennies, a currency, and an idempotency key (a caller-supplied string that marks a retry as the same request, not a new one).

There are three outputs:

1. a transfer record whose state is one of SENDER_DEBITED, COMPLETED, or COMPENSATED;
2. four immutable bookkeeping rows called ledger entries: two on the sender’s machine, two on the receiver’s;
3. an updated balance that both parties can read.

A balance read is separate: it takes an account id and returns a number plus the position in the log that number was computed at.

Vocabulary, before anything else

Five terms recur throughout.

• Ledger: an append-only list of rows. Rows are added, never edited and never deleted. It is the system’s only record of where money is.
• Minor units: whole pennies. $5.00 is written 500, never 5.00. Binary floating-point cannot represent 0.1 exactly, so money arithmetic on floats silently loses fractions (money is never a float shows the loss).
• Double-entry: every movement of money is recorded as two or more signed lines that add to zero. Money leaving one account is money arriving in another, written once in two halves. So SUM(amount_minor) = 0 holds over the whole ledger at every instant, and any nonzero result names a broken transaction (double-entry derived).
• Idempotency: doing something twice leaves the world as it would be after doing it once. The four rules for building it are in the idempotency rules in one place, extended across an external boundary in idempotency across a boundary you cannot roll back, and restated for this chapter in Saga and why compensate is not rollback.
• PSP, or payment service provider: the outside company that talks to card networks on your behalf. It is central to the payment system chapter; this design deliberately does not have one.

Why this is different from the payment system

The payment system chapter put the ledger at the edge of the company, where a card network holds half the state and you cannot see it. This chapter puts the ledger inside, where you own every byte. That changes the problem: 10,000 writes per second land on a single merchant’s account row, and a row lock serializes all of them.

A row lock is the database’s way of stopping two transactions from changing the same row at once: the first transaction to touch the row holds it until it commits, and everyone else waits. Serialize means exactly that: the writes happen one after another, never at the same time.

Three numbers are new here, and the table says where each comes from:

Framing: what decision, and what breaks

One question has to be settled before anything else: how is a balance represented? The traffic numbers, the interface, and the architecture all either feed that decision or follow from it. It is harder here than in most systems, because a wallet that owns both sides of every transfer has nobody else to blame for a wrong number.

No third party, and why that hurts

There is no third party in this design. The wallet is the bank, so a transfer is two writes you control end to end.

That sounds easier, but it removes any excuse for being eventually consistent: the arrangement where different copies of the data disagree for a while and are guaranteed only to agree in the end. A user who sees a balance and cannot spend it has been shown a wrong number, and there is no payment provider to blame.

The decision: how a balance is represented

There are exactly three candidates.

1. A number in a column, updated in place. Fast, and it cannot be the truth: a money movement has two halves, and if the process dies between them, both rows are individually plausible and no constraint fires: the corruption lives in the relationship between two rows, which the schema does not record (what a single balance column loses).
2. A fold over an append-only log of entries. To fold a list is to walk it start to finish carrying a running result (here, adding up every signed entry) so a balance becomes a computation, not a stored value. Correct, auditable, able to answer questions about any past instant, and unusably slow to read at scale.
3. Both. The log is the truth, and the column holds a memoized copy of the fold (computed once and saved, so it need not be recomputed) maintained in the same transaction as the entries it summarizes, with a background job that re-folds the log and compares.

Option 3 is the answer. Everything hard in the rest of the chapter follows from it: how often you save a running total (snapshots); what happens when one account’s fold is extended at 10,000/s against a row that sustains 1,333/s (the hot account); and what happens when the two accounts in a transfer live on different machines (transfers across shards).

What breaks: the hot account

A hot account takes vastly more traffic than the rest. Not the average account: the average account gets a handful of entries a day and every design here would work for it. One merchant during a flash sale takes a third of the system’s peak write traffic onto a single primary key, and every mechanism that was comfortable at the average becomes a queue at that peak.

Requirements

Three words recur below. A credit is money arriving in an account and a debit is money leaving it. A hold is money set aside (reserved so it cannot be spent twice) while some pending thing resolves.

Functional

• Hold a per-user, per-currency balance. Credit it, debit it, place and release holds.
• Transfer between two internal accounts atomically from the user’s point of view: the user never observes a state where the money is half-moved.
• Return a balance and a paginated statement.
• Reproduce the balance as of any past instant, from the log alone.

Non-functional. Two shorthands appear in the table. p99 is the 99th percentile: the latency that 99 requests in 100 come in under, the slow tail, not the typical case. RPO is recovery point objective: how much recently accepted work you are willing to lose when a machine dies. RPO 0 means none, which forces a second copy of every write before it is acknowledged; its cost sets the hardest limit in the chapter (what one row can do).

Back-of-envelope

Four numbers carry the architecture: transfers per second, entries per second, petabytes of storage, and (the one that decides transfers across shards) the fraction of transfers that touch two machines.

Traffic. Assume 1 B transfers/day, $5.00 average: a consumer wallet, many small payments. Dividing by ~100,000 s/day (the round stand-in for 86,400) gives 10,000 transfers/s average and 30,000/s at a 3x peak. At 500 minor units each, that is $5 B/day moving through the system.

Entries per transfer. Each split of the data across machines is a shard; a transfer whose two accounts live on different shards is cross-shard. A same-shard transfer writes two entries (debit sender, credit receiver). A cross-shard transfer writes four, because the money makes a stop in a holding account on the way:

1. debit the sender (sender’s shard);
2. credit an in-transit account (sender’s shard);
3. debit the in-transit account (receiver’s shard);
4. credit the receiver (receiver’s shard).

The saga section explains why the stop exists, and the choice shows almost every transfer is cross-shard, so size on four: 4 B entries/day, 120,000 entries/s at peak.

Storage. Each entry row is 240 B on disk (derived field by field in the payment system). At 4 B entries/day x 240 B that is ~960 GB/day; over 7 years, 2.45 PB of ledger, and 7.4 PB across 3 replicas. Unlike the payment system, this is a storage problem.

Tier split. That much data forces a tier split: recent data on fast, expensive storage, older data on slow, cheap storage. Statements older than 90 days are something a user deliberately asks for, not something a page load needs, so they live in columnar object storage: bulk cloud storage holding files column by column instead of row by row, which compresses far better. That leaves ~86 TB hot (90 days) and ~2.37 PB cold.

Shard count. At 4 TB usable NVMe per node (the fast solid-state disk a database primary sits on), 86 TB hot needs ~22 nodes. Round up to a power of two (32 shards) because doubling the count is the only resharding step (redistributing data across a new number of shards) that moves half the keys instead of nearly all of them (consistent hashing, sharding).

Is throughput the binding constraint instead? The binding constraint is the resource that runs out first and therefore sets the size. A cross-shard transfer is two local writes, so peak is 60,000 shard-txns/s, ~1,875 per shard. A tuned relational primary on 32 cores sustains roughly 10,000 small writes/s, so each shard runs at ~19% utilization and throughput alone would need only ~6 shards. Storage binds at 22, throughput at 6: storage wins, and 32 shards leave the write path ~81% idle, exactly the headroom the hot account will consume.

The number that decides section 4. With 32 shards and random account pairs, a transfer stays on one shard only if the receiver lands on the sender’s shard, 1 chance in 32. So 96.9% of transfers are cross-shard (1 − 1/32). Whatever cross-shard coordination costs, you pay it on the common case, not on an edge case.

API sketch

The interface makes three commitments visible: money is an integer count of pennies, held money is a separate number from available money, and every call that changes something takes an idempotency key.

POST /v1/transfers
Idempotency-Key: 3ab1...             <- canonical hash of the body (idempotency rule R1)
{ "from": "acc_1", "to": "acc_2", "amount_minor": 500, "currency": "USD" }
201 { "transfer_id": "tr_9", "state": "SENDER_DEBITED" }
409 { "error": "insufficient_funds", "available_minor": 320 }

GET  /v1/accounts/{id}/balance
200 { "available_minor": 12500, "held_minor": 500, "as_of_seq": 88213 }

GET  /v1/accounts/{id}/entries?from=2026-07-01&to=2026-07-31&cursor=...
POST /v1/holds        Idempotency-Key: ...   { "account": "acc_1",
                                               "amount_minor": 2000, "ttl_s": 900 }
DELETE /v1/holds/{id} Idempotency-Key: ...

Three details are doing real work.

The idempotency key is on the hold endpoints too, not only on /transfers: that is rule R4, “put the key on every call that changes something” (the idempotency rules). A retried POST /v1/holds that places the hold twice takes twice the money out of available.

The available_minor and held_minor fields are separate because they are separate accounts. A hold is a zero-sum transfer between a user’s available and held sub-accounts, not a boolean flag on a row. That keeps the zero-sum invariant intact (the invariant) and makes “why is my money gone” answerable from the statement, because the hold is a visible line on it.

The as_of_seq value is the sequence number the balance was computed at. Each account’s entries are numbered 1, 2, 3, … in commit order. A client that reads a balance, then reads again and sees a lower as_of_seq, has been served by a replica (a copy of the database kept for reads) that is behind the primary. Returning the fold position costs nearly nothing and turns an invisible anomaly into a detectable one.

Data model

Four tables: ledger_entries is the log that is the truth, account_balances is the memoized fold, balance_snapshots holds periodic running totals used only for rebuilds, and transfer_outbox makes cross-shard delivery reliable.

CREATE TABLE ledger_entries (        -- append only, sharded by account_id
  account_id    BIGINT NOT NULL,
  seq           BIGINT NOT NULL,     -- per-account, gapless, assigned on commit
  transfer_id   UUID   NOT NULL,
  amount_minor  BIGINT NOT NULL CHECK (amount_minor <> 0),
  currency      CHAR(3) NOT NULL,
  created_at    TIMESTAMPTZ NOT NULL,
  prev_hash     BYTEA NOT NULL,      -- hash chain, see below
  PRIMARY KEY (account_id, seq)
);

CREATE TABLE account_balances (      -- the memoized fold, one row PER BUCKET
  account_id    BIGINT NOT NULL,
  bucket        SMALLINT NOT NULL,   -- 0 for cold accounts, 0..15 when hot
  amount_minor  BIGINT NOT NULL,
  last_seq      BIGINT NOT NULL,
  PRIMARY KEY (account_id, bucket)
);

CREATE TABLE balance_snapshots (     -- rebuild and audit only, not the read path
  account_id    BIGINT NOT NULL,
  seq           BIGINT NOT NULL,
  amount_minor  BIGINT NOT NULL,
  PRIMARY KEY (account_id, seq)
);

CREATE TABLE transfer_outbox (       -- written in the SAME txn as step 1
  transfer_id   UUID PRIMARY KEY,
  to_shard      SMALLINT NOT NULL,
  payload       JSONB NOT NULL,
  published_at  TIMESTAMPTZ
);

One choice does most of the work: putting account_id first in both primary keys. PRIMARY KEY (account_id, seq) clusters an account’s entire history contiguously: the rows are physically stored in key order, so all of one account’s entries sit next to each other on disk. That is what makes both the statement scan and the fold sequential instead of random, and sequential reads are roughly two orders of magnitude cheaper. PRIMARY KEY (account_id, bucket) does the same for the buckets of the hot account, so reading a split balance is as cheap as reading a single one.

One more term: an outbox is a table you write a to-be-sent message into in the same transaction as the data change that prompted it. Because the message row and the debit commit together, a crash can never leave the debit committed and the message lost. A separate relay process reads the table and publishes what it finds.

High-level architecture

The whole system is one write path fanning into 32 shards, plus background jobs hanging off each shard.

flowchart TD
    C["Wallet client"] --> API["Transfer API<br/>Idempotency-Key required"]
    API --> RT["Router<br/>shard = hash of account_id<br/>32 shards"]
    RT --> S1[("Shard 1<br/>ledger_entries append-only<br/>account_balances by bucket")]
    RT --> SN[("Shard 32")]
    S1 --> OB["Outbox relay<br/>same txn as the debit"]
    OB --> Q["Queue<br/>at-least-once delivery"]
    Q --> W["Saga step 2 worker<br/>idempotent on transfer_id"]
    W --> SN
    W --> DLQ["Dead-letter to human review"]
    S1 --> SNAP["Snapshotter<br/>every 20 M entries"]
    S1 --> ARCH["Archiver<br/>older than 90 days to Parquet"]
    ARCH --> OBJ[("Object storage<br/>2.37 PB cold")]
    S1 --> AUD["Invariant checker<br/>per-shard sum is zero<br/>in-transit residue equals money in flight"]

The wallet client (the phone app) calls the Transfer API, which rejects any mutating request without an idempotency key. The router picks a shard by hashing account_id, spreading accounts evenly. Each shard stores ledger_entries and account_balances in one database on purpose: that is the only way the log and its memoized fold can be written in a single transaction.

Four background jobs hang off every shard:

• Outbox relay: publishes the step-2 messages written in the same transaction as the debit, onto a queue that promises at-least-once delivery (message queue).
• Snapshotter: every 20 M entries, writes a running total so a rebuild never starts from the beginning of time.
• Archiver: moves entries older than 90 days to Parquet files in object storage. Parquet is a columnar file format, values grouped by column, which compresses far better.
• Invariant checker: verifies per shard that entries sum to zero and that the in-transit residue equals the money genuinely in flight.

Downstream of the queue is the saga step 2 worker: it consumes relayed messages and is idempotent on transfer_id, so a repeat delivery changes nothing. A saga is a chain of small local transactions standing in for one big distributed one (built below). Anything permanently unprocessable goes to a dead-letter store and on to human review.

Deep dive 1 — balance as a fold

The framing chose option 3: the log as truth, with a memoized fold beside it. Here we justify it: why the log alone is too slow to read, how often we write down a running total, and how we catch the saved number when it goes wrong.

The fold, and why you cannot do it on read

A balance is not a stored number. It is fold(+, 0, entries): start from 0 and add every signed entry for that account, in order. That definition is what makes the ledger reproducible: balance(a, T) is the same fold restricted to entries with seq <= T, so any past balance is recoverable exactly, and an auditor’s question about last March is a query, not an archaeology project.

def fold(entries, snapshot=None):
    """A balance is a fold over the log: start from 0 and add every signed
    entry in order. A snapshot is (seq, amount) and lets the fold start
    partway down the log instead of at the beginning of time."""
    start_seq, running = (0, 0) if snapshot is None else snapshot
    for seq, amount in entries:
        if seq > start_seq:
            running += amount
    return running

log = [(1, 500), (2, -200), (3, 750)]
assert fold(log) == 1050
assert fold(log, (2, 300)) == 1050     # 300 + 750: only the tail is replayed

Pricing the fold. Two costs per entry: one main-memory reference (~100 ns, latency numbers) and ~400 ns to decode the row and add it, so ~500 ns/entry, i.e. 2 M entries/s per core. Disk is not the limit: the (account_id, seq) clustering makes the read sequential, delivering 240 B rows at ~4 M/s, faster than the CPU consumes them. So CPU binds at 2 M entries/s: the number to carry for the rest of the chapter.

Apply it to the two extremes. A cold user at 10 entries/day over 7 years is ~25,550 entries, folding in 12.8 ms, near the 10 ms budget, so for most accounts you would not need a stored balance at all. A large merchant at 100 M entries/year over 7 years is 700 M entries, folding in 350 seconds, nearly six minutes to answer “what is my balance”. That is why account_balances exists: the fold’s cost grows with lifetime activity, and lifetime activity is unbounded.

Snapshots, derived from a recovery target

A snapshot is one saved pair of (sequence number, balance at that point), letting a rebuild start partway down the log. If account_balances is lost or wrong, you rebuild from the log, and rebuild time is what the snapshot interval controls.

State the target (a balance must be recoverable in 10 s) and multiply by the 2 M entries/s fold rate: snapshot every 20 M entries, per account. It is a count, not a clock. In wall-clock terms the hot merchant (10,000 entries/s) is snapshotted every ~33 minutes; the cold user (10/day) effectively never, and does not need to be. A nightly schedule would get both ends wrong: it snapshots 500 M accounts that will never be read, and still leaves the merchant 8 hours of log to replay.

The snapshot table is tiny: 4 B entries/day ÷ 20 M = 200 snapshots/day system-wide, ~13 KB at 64 B each.

The materialized balance is a cache of a fold

One distinction carries the rest of the design: a saved balance cannot be stale, but it can be wrong, and those need different answers. Materialized means precomputed and stored, not derived on demand.

account_balances is updated in the same transaction as the entries that change it. That is the whole cache-invalidation story, and it is why this cache needs no invalidation logic. The balance can never be stale, because there is no window in which the entries exist and the balance does not: they commit together or neither does.

It can still be wrong: a bad migration, a bug in the update code, a corrupted page. Wrong is what the audit job catches. The audit re-folds the log from scratch and compares to the stored balance: 4 B entries/day at 2 M/s is 2,000 core-seconds, ~62.5 core-seconds per shard per day. Run it nightly on a replica and no one notices.

Pair it with the zero-sum check, which catches a movement recorded on one side only (what checking the invariant costs). Per shard, scanning the day’s ~125 M entries (~30 GB) at 1 GB/s is 30 s, all 32 shards in parallel, so the check stays a 30-second job no matter how many shards you add. Sharding made the correctness check cheaper, not harder: the one place in this design where distribution helps.

Deep dive 2 — the hot account

A single database row imposes a hard write ceiling, and the flash-sale merchant sails straight through it. The ceiling follows from first principles; splitting a balance across several rows gets past it; and the split introduces two concurrency bugs, both invisible unless you test with threads.

What one row can actually do

The 1,333 writes/s figure is not a hardware fact: it is set by a durability promise. A transfer crediting a merchant takes a row lock on that merchant’s balance row and holds it until commit. The hold time is not the query time: it is everything between acquiring the lock and releasing it at commit. Four things happen in that window:

• Index descent (~10 us): walking the index from root to the row. On a hot row the pages are in memory, so no disk read.
• Row update plus write-ahead log (WAL) record (~40 us): the WAL is a sequential log the database appends every change to before touching the data pages, so a crash can be replayed forward.
• Group commit fsync (~200 us): fsync forces buffered writes onto physical disk; group commit means several transactions share one call, so this is amortized.
• Synchronous replica acknowledgement (~500 us): one round trip to a second machine and back. This is RPO 0 being paid for.

Sum: 750 us lock hold, two-thirds of it the replica ack. That is not tunable: it is the price of RPO 0, and it sits inside the lock because the lock cannot release before commit.

Locked updates to one row are strictly serial, so one second divides into 750 us slots: 1,000,000 / 750 = 1,333 writes/s. The flash-sale merchant needs 10,000, 7.5x short, and it does not fail gracefully.

“7.5x short” understates it, because lock waiting is a queue: once demand approaches capacity, waiting time grows without bound. The ten-thousandth transfer of that second is queued behind ~7,500 others with no latency ceiling. Worse, every other transfer on that shard slows too, because each waiter holds a database connection from a fixed-size pool while it waits: once the waiters have taken every connection, unrelated work cannot get one either (pooling).

Sharded sub-balances

Split the hot account’s balance into K rows, called buckets, whose sum is the true balance. Credits pick a bucket by hashing the transfer id, so K rows absorb the writes and each bucket’s lock is independent.

flowchart TD
    CR["Credit<br/>hash(transfer_id) picks one bucket<br/>never fails"] --> B0
    CR --> B1
    CR --> B15
    subgraph HOT["Hot account balance = sum of all buckets"]
      B0["bucket 0"]
      B1["bucket 1"]
      B15["bucket 15"]
    end
    DB["Debit<br/>locks all buckets in ascending id,<br/>checks the TOTAL, rebalances"] --> HOT

Sizing K: 10,000 / 1,333 = 7.5, so K ≥ 8. Take K = 16 for headroom → 625/s per bucket, 47% utilization: below roughly half capacity, waiting time grows in proportion to load instead of running away. K is a power of two so the bucket is hash(transfer_id) & 15 (one bitwise op) and K can be doubled later while every existing entry keeps a valid bucket.

Two properties must hold, and neither is optional:

1. The check and the write happen under one lock set (all the locks a single operation holds at once). Otherwise the total is true when read and false when the subtraction lands, and the account goes negative.
2. The locks are taken in ascending bucket id. Otherwise two debits acquire them in opposite orders and deadlock: thread one holds bucket 0 waiting for 15 while thread two holds 15 waiting for 0. Neither can proceed, so neither will ever release. The cure is lock ordering: pick one global order in advance and require every path to take locks in that order. A deadlock needs a cycle, and a cycle requires someone to go backwards through the order, which the rule forbids.

from threading import Lock

BUCKETS = 16

def bucket_for(transfer_id):
    return hash(transfer_id) % BUCKETS

class ShardedBalance:
    """Credits fan out across buckets and never fail. A debit must check the
    TRUE total (the sum of all buckets), and the check and the write have to
    happen under one lock set, or the check is only a guess."""

    def __init__(self, n=BUCKETS):
        self.buckets = [0] * n
        self.locks = [Lock() for _ in range(n)]

    def credit(self, transfer_id, amount):
        i = bucket_for(transfer_id)
        with self.locks[i]:                       # one row, no order to get wrong
            self.buckets[i] += amount

    def debit(self, transfer_id, amount):
        i = bucket_for(transfer_id)
        for j in range(len(self.buckets)):        # ASCENDING id, always
            self.locks[j].acquire()
        try:
            if sum(self.buckets) < amount:        # total is true under the locks
                raise ValueError("insufficient funds")
            need = amount - self.buckets[i]
            for j, v in enumerate(self.buckets):  # rebalance, itself zero-sum
                if j == i or need <= 0:
                    continue
                move = min(v, need)
                self.buckets[j] -= move
                self.buckets[i] += move
                need -= move
            self.buckets[i] -= amount
        finally:
            for j in reversed(range(len(self.buckets))):
                self.locks[j].release()

Both bugs need real threads to appear; single-threaded, the buggy version passes. Drop the locks and keep if sum(...) < amount as a bare precondition, and 32 concurrent debits against a balance of 1,600 (covering 16 of them) end at −1,600: every thread read a total that covered its 100, and all 32 then subtracted. Keep the locks but grab the chosen bucket first and scan the others afterwards (the shape the rebalance loop naturally suggests) and two debits whose buckets sit at opposite ends deadlock immediately.

The rebalance loop is not decoration: it is the only reason a debit against a bucketed account behaves like a debit against a single balance, instead of failing whenever the money happens to sit in a bucket other than the chosen one. And locking the whole range on a debit is not expensive, for the reason debits are the hard direction gives: on the account this design exists for, credits are hot and debits are a nightly batch.

What sharding costs the read path

The obvious objection: a balance read now touches 16 rows instead of one. Databases read from disk in fixed-size pages (typically 8 KB), and the page is the unit of I/O, so the real question is how many pages those rows span. A bucket row is 64 B, so 16 buckets are ~1 KB, an eighth of one page. Clustered under PRIMARY KEY (account_id, bucket), they sit on one leaf.

A B-tree is the sorted, tree-shaped index a relational database walks from root to a leaf (the bottom page holding the data), typically four page reads at realistic sizes (B-trees). One descent lands on the leaf holding all 16 buckets, so the bucketed read costs the same four page reads as a single-row read, but only because of the clustering. Key it (bucket, account_id) instead and one account’s buckets scatter across 16 places, cost 16 descents, and the objection would be right.

Rows examined do grow: at peak, 30,000 x 16 = 480,000 rows/s at ~200 ns each is ~96 M ns/s, under 10% of one core across the whole fleet. Negligible, but worth checking, not assuming.

The cost that hardware cannot pay is consistency: the 16 rows must all be read as of one instant, or their sum is a number that never existed. An isolation level decides which concurrency surprises a transaction may observe. The weakest in common use, READ COMMITTED, never shows uncommitted data but lets two reads inside one transaction differ. A single SELECT SUM(amount_minor) ... WHERE account_id = ? is one statement and sees one consistent view, so the plain balance read is safe there. A read-modify-write across buckets is not: between two statements a rebalance can move money out of a bucket you read into one you have not, so you count it twice, or the other way, and never. Use REPEATABLE READ (re-reading gives the same answer) or an explicit SELECT ... FOR UPDATE locking the whole range (the anomaly matrix).

Debits are the hard direction

Money in and money out are not symmetric. Credits split across buckets trivially; debits do not. Addition commutes, so a credit can go into any bucket and never fails: there is no precondition. A debit must answer “are there sufficient funds”, and the funds are spread across 16 rows. Five problems follow.

The fourth row is the one the ordering rule usually misses. A same-shard transfer is a single local transaction touching two account rows. The natural way to write it (debit the sender, then credit the receiver) orders the locks by the direction of the transfer, which is exactly the order guaranteed to collide with a transfer going the other way.

from threading import Lock

BALANCES = {"acc_1": 10_000, "acc_2": 10_000}
ROW_LOCKS = {a: Lock() for a in BALANCES}

def transfer_same_shard(src, dst, amount):
    # A single local transaction still deadlocks when two of them lock the same
    # two rows in opposite orders. Lock by ascending account_id, never in
    # transfer order, so A->B and B->A take the two locks in the SAME order.
    first, second = sorted((src, dst))          # the guard
    with ROW_LOCKS[first], ROW_LOCKS[second]:
        BALANCES[src] -= amount
        BALANCES[dst] += amount

Replace sorted((src, dst)) with src, dst and two threads hammering acc_1 -> acc_2 and acc_2 -> acc_1 deadlock on the first iteration.

The rule is not “order your buckets”. It is “there is exactly one global order in which locks may be taken, and every path takes it”: buckets by id, account rows by account_id, and a shard’s rows before another shard’s if any path ever touches both.

The fix works because the workload is one-directional. A hot merchant is hot in one direction only: it receives 10,000 payments/s and pays out in nightly batches. So credits fan out across 16 buckets, and debits all take bucket 0, which sees a handful of writes a day, so locking the full range on a debit costs nothing. The design works because the workload is asymmetric. For an account genuinely hot in both directions (an exchange’s settlement account) the answer is to stop pretending it is one account: give it explicit per-region or per-desk accounts and reconcile between them with real ledger transfers, so the concurrency shows up in the books where you can query it, instead of hidden inside a lock.

Promotion and demotion

Bucketing every account would multiply the row count by 16 for the 99.99% of accounts that see under one write per day and buy them nothing. So split adaptively, at a threshold derived from the ceiling: promote an account to K = 16 when its one-minute write rate exceeds 133/s (10% of the 1,333 serial ceiling), early enough that the queue is still well-behaved. Demote after an hour below 13/s. The gap between promote and demote is deliberate and has a name, hysteresis: making the way out different from the way in stops an account hovering near one threshold from flapping every minute.

Both transitions are ordinary zero-sum transactions (a split moves the balance out of bucket 0 into 16 buckets; a merge moves it back), so a bug in the splitter shows up in the nightly audit instead of in a customer complaint. Memory for ~1,000 hot accounts x 16 rows x 64 B is ~1 MB. The adaptive path is free; the universal path would not have been.

Deep dive 3 — transfers across shards

Distributed money movement offers one central trade: one atomic transaction spanning two machines, or availability, and here you cannot have both. Since 96.9% of transfers touch two shards, whatever this costs, it is the common case.

2PC, and the failure that matters

Two-phase commit (2PC) is the classic protocol for making two databases commit or abort together. A coordinator (one process running the protocol) asks both shards to PREPARE: “get ready and promise you can commit”. If both vote yes, the coordinator records the decision durably and tells them both to commit. It gives real atomicity: the transfer happens on both shards or neither, with no observable in-between. This is not a straw man.

The throughput cost. 2PC adds the coordinator’s own log fsync (~200 us) and the commit round trip (~500 us) to the 750 us local hold → 1,450 us → 690 writes/s per row, 48% of the single-row ceiling. Even with 16 buckets that is 11,040/s against a 10,000/s peak, 10% headroom against a target that already assumed a 3x peak. That is not a margin; it is a coincidence.

The failure mode is worse than the throughput cost. A shard that has voted yes is PREPARED: it has promised it can commit, so it may not abort on its own and must keep its locks until told the outcome. That is what makes 2PC a blocking protocol, a participant can be left with no legal move: it cannot commit (not told to), cannot abort (promised not to), and cannot release the locks. If the coordinator dies after collecting votes and before the decision is durable, those locks stay held for as long as recovery takes, indefinitely if its log is lost. A 30-second coordinator outage during the flash sale blocks 10,000 x 30 = 300,000 balance rows: 300,000 people who cannot spend their own money because of a machine they have never heard of. 2PC does not fail by being slow; it converts one crash into a system-wide freeze on exactly the accounts that were most active.

Saga, and why compensate is not rollback

A saga is a sequence of local transactions, each committing on its own, with a stored record of intent linking them. Nothing spans two machines; nothing is held open waiting. If a later step proves impossible, earlier steps are undone by running new transactions that reverse their effect: this is compensation, and the difference from rollback is the end of this section.

A saga replaces one distributed transaction with two local ones and a durable intent between them:

1. Shard A: debit the sender, credit in_transit@A. Commit, and write the outbox row in the same transaction.
2. Shard B: debit in_transit@B, credit the receiver. Commit, keyed on (transfer_id, step) so a repeated delivery does nothing.

flowchart LR
    S["Sender<br/>shard A"] -->|"step 1: debit sender,<br/>credit in-transit"| IT["In-transit account"]
    IT -->|"step 2: debit in-transit,<br/>credit receiver"| R["Receiver<br/>shard B"]
    OB["Outbox on shard A<br/>written in step 1's txn"] -->|"at-least-once"| Q["Queue"]
    Q --> W["Step 2 worker<br/>idempotent on transfer_id, step"]
    W -.->|"runs step 2"| IT

Keyed means an actual uniqueness constraint written in the same transaction as the entries (rule R3), not a sentence in a design doc and not an if in application code that races with itself.

The in-transit account makes this respectable, not a hack: it is an ordinary account that holds money while it is between owners, so a stalled transfer has its money sitting somewhere with a name and a queryable balance. Each step is independently zero-sum, so the invariant holds on every shard at every instant, including mid-saga. The money is never nowhere.

The blind spot is that zero-sum cannot see a duplicate. A posting is one balanced group of entries written together. A replayed step 2 pays Bob twice with two perfectly balanced postings, so total() is still 0 and shard_total("B") is still 0: every zero-sum check passes while a stranger’s money is gone. Double-entry catches asymmetry (a movement recorded on one side only); a duplicate is symmetric, twice.

LEDGER = []       # append-only: (shard, transfer_id, step, posting_id, account, amount)
POSTINGS = {}     # (transfer_id, step) -> posting_id  -- PK, written in the same txn
STATE = {}        # transfer_id -> saga state
TERMINAL = {"COMPLETED", "COMPENSATED"}

class LateStep(Exception):
    """A step arriving after the saga reached a terminal state."""

def post(shard, transfer_id, step, entries):
    """Three guards:
    1. zero-sum, so a half-written movement has no representation;
    2. one posting per (transfer_id, step): delivery is at-least-once, and a
       replayed step pays the receiver twice while every zero-sum check passes;
    3. nothing lands on a transfer already terminal, or a compensation and a
       straggling step 2 both apply and money nobody sent is created."""
    if sum(a for _, a in entries) != 0:
        raise ValueError("step does not balance")
    key = (transfer_id, step)
    if key in POSTINGS:                          # redelivery: benign no-op
        return "duplicate"
    if STATE.get(transfer_id) in TERMINAL:
        raise LateStep(f"{transfer_id} is {STATE[transfer_id]}")
    posting_id = len(POSTINGS) + 1
    POSTINGS[key] = posting_id                   # uniqueness constraint, same txn
    for account, amount in entries:
        LEDGER.append((shard, transfer_id, step, posting_id, account, amount))
    return "posted"

def compensate(shard, transfer_id, entries):
    # The reversing entries and the terminal transition are ONE transaction.
    # Once it commits, a straggling step 2 has nowhere to land.
    if post(shard, transfer_id, "compensate", entries) == "posted":
        STATE[transfer_id] = "COMPENSATED"

The post() guards ordered so that a repeated delivery of a step already seen is a no-op, while a never-seen step arriving after the transfer finished is an incident (LateStep). Those are different events and must produce different outcomes, which is why the dedupe check comes before the terminal-state check. Now watch the blind spot directly, appending a duplicate by hand to bypass the guard:

post("A", "tr1", "debit_sender",    [("alice", -500), ("in_transit@A", 500)])
post("B", "tr1", "credit_receiver", [("in_transit@B", -500), ("bob", 500)])

# A duplicate step 2, bypassing post()'s uniqueness guard entirely:
for account, amount in [("in_transit@B", -500), ("bob", 500)]:
    LEDGER.append(("B", "tr1", "credit_receiver", 999, account, amount))

assert sum(a for *_, a in LEDGER) == 0        # zero-sum: still perfect
# ...but Bob now holds 1,000 for a 500 transfer, in_transit@B is debited twice
# for one credit, and only a COUNT of distinct posting_ids per (transfer_id,
# step) -- or a negative in-transit residue -- reveals it.

Four things follow.

1. The zero-sum invariant is structurally blind to duplication, for the same reason it works. The invariant holds because the ledger is append-only and every posting balances, so a second copy of a balanced posting preserves it exactly. This is not a gap to patch; it is a different failure class needing a different check.

2. So monitor what the invariant cannot see. Two checks, neither a sum over amounts: SELECT transfer_id, step, COUNT(DISTINCT posting_id) ... GROUP BY 1,2 HAVING COUNT(DISTINCT posting_id) > 1 (one row is a paid-twice incident, named), and the in-transit residue must be non-negative. The residue is the sum of the in-transit accounts across both shards; normally it is the money currently between steps. A duplicated step 2 debits in_transit@B twice for one credit, so the residue goes negative, a cheaper alert than any sum.

3. A terminal state is a guard, not a label. Compensation and step 2 are two writers racing over one transfer, so compensation must claim the transfer in the same transaction that posts the reversal. Without that claim, a straggler lands after the refund: Dave gets 300 back, Eve is credited 300 anyway, every stated check passes, and 300 units nobody sent now exist. The guard is bidirectional: a step arriving after COMPLETED is refused by the same line.

4. Redelivery is routine; a late never-seen step is an incident. At-least-once delivery makes redelivery constant. Turning routine redeliveries into exceptions is how a team learns to ignore the exception that mattered.

Timing. The trigger is the transactional outbox: the message row committed with the debit in step 1, and the relay publishes what it finds. Delivery is at-least-once (message queue), which is why step 2 must be idempotent. The delay before the receiver sees money is relay poll (~100 ms) + queue hop (~50 ms) + step-2 commit (~0.75 ms) ≈ 150 ms, almost all of it the poll interval. Not instant, and the product needs to know that.

Compensating is not rolling back. The same event, a transfer that does not complete, plays out differently down the two paths:

A rollback erases a state that never existed for anyone; a compensation is a new, visible, auditable event undoing a real old one. If Alice’s balance dropped by 500 for four seconds, a payment she attempted then may have been declined, and no compensation un-declines it. That is the honest cost of the saga. Two rules keep it rare:

• Order the steps so the step that can fail comes first. The debit can fail on insufficient funds; the credit cannot fail for any business reason. Debiting first means business failures happen before anything is visible, so compensation is only ever needed for infrastructure failures.
• Infrastructure failures are retried, not compensated. Step 2 retries against a live shard until it succeeds. If shard B is down for an hour, the money sits visibly in in_transit for an hour and the transfer completes. Compensation is terminal only when step 2 is permanently impossible (a closed receiving account), and that ends in human review.

The choice

The answer is the saga, for four reasons, strongest first.

1. The blocking window is unacceptable for this workload. 300,000 locked balances from one coordinator crash is worse than any consistency anomaly here. Availability of people’s own money is the product.
2. 2PC halves the hot-row ceiling (1,333 → 690/s) and leaves only 1.1x headroom after bucketing. The saga keeps the full 1,333 per bucket (16 x 1,333 = 21,328/s, 2.13x headroom) because every step is a plain local transaction with no extra phases inside the lock.
3. At 96.9% cross-shard, 2PC’s cost is not amortized over anything. It is the price of essentially every transfer.
4. Double-entry removes saga’s usual objection. The standard complaint is that the intermediate state is anomalous; here it is a named account with a queryable balance and an alertable residue. The invariant never breaks; it just has a leg in in_transit.

You accept two product facts, not bugs: reversals are visible on statements, and the receiver’s credit lands ~150 ms after the sender’s debit. The one case where 2PC would win is a same-shard transfer (1/32 of traffic), and there it is not 2PC at all, just a single local transaction. So route both accounts onto the same shard whenever you can (a user and their own sub-accounts), and that fraction becomes free.

Append-only, and why corrections are new entries

Every mechanism so far has appended to the ledger; none has edited it. The ledger takes INSERT and nothing else. UPDATE and DELETE are revoked at the grant level (the database’s own permission system) so nobody holds the privilege, including the application’s own database account. This is enforcement, not a code-review convention, and the difference matters at 3 a.m.

Three reasons, in descending order of how often each is the one that matters:

1. Time travel is the product. balance(a, T) = SUM(amount_minor) WHERE account_id = a AND seq <= T is exact and cheap only because rows never change. One UPDATE and every historical balance the system ever reported becomes unreproducible, including the ones printed on statements customers already have.
2. A correction carries information. “This entry was wrong, here is the reversal, at this time, by this operator, for this reason” is several facts. Editing the row keeps none of them.
3. The invariant is only checkable over an immutable log. Yesterday’s checkpointed sum is a constant because the past cannot change. Allow updates and every check becomes a full scan of all history.

So a correction is a new zero-sum transaction: it reverses the original and, if needed, posts the right one, carrying corrects: <transfer_id> so the pair is joinable later. The account’s history grows; it never rewrites.

The hash chain

Beyond permissions, chain each entry to the one before it: prev_hash = SHA256(prev_hash || row) per shard, each row carrying a cryptographic fingerprint computed from the previous row’s fingerprint concatenated with its own contents. Because every fingerprint depends on the entire history behind it, any silent edit changes every fingerprint after the edited row, and the chain stops verifying from that point.

That is what tamper-evident means, and note what it does not mean: you cannot prevent a determined operator with disk access from editing a row, but you can guarantee the edit is detectable. Publishing the day’s final fingerprint (the head hash) somewhere outside the system makes the break provable to someone who has no reason to trust you.

The cost is one hash over every entry byte. SHA-256 runs at ~1 GB/s per core, and at peak 120,000 x 240 = 28.8 MB/s: 2.9% of one core to make the ledger tamper-evident. There is no argument against paying that.

Read paths: balance versus statement

Reading money back splits into two questions (“what is my balance?” and “show me my statement”) that look similar and have nothing in common mechanically. One term: cursor-paginated means results come back a page at a time with an opaque marker meaning “continue from here”, never by page number: the only correct way to page through a growing log, because with page numbers a new entry shifts every later row and the reader sees a duplicate or a gap.

A month of a normal user’s statement (10 entries/day) is ~300 entries, ~72 KB, contiguous on disk thanks to clustering, read in under a millisecond. Older months come from Parquet in object storage, partitioned by (month, account_bucket); that read takes seconds, acceptable for something a user explicitly asked for. The column-store layout also makes the archive far smaller than 240 B/row, because currency and account_id compress to nearly nothing once the rows are sorted (storage).

The one caching decision

Do not cache the balance in Redis (the usual in-memory key-value store). The reason is not performance: the source is already a four-page read against pages almost certainly in the buffer pool (the database’s own in-memory cache of recently used pages), so an external cache saves almost nothing. The reason is that a stale balance is not a slow answer, it is a wrong one. Every cache introduces a window in which the ledger and the cache disagree; in most domains that is a tolerable trade, but here it is the bug: the “sees a balance and cannot spend it” failure, reintroduced deliberately. Cache the statement page if you like. Never cache the number the user is about to spend against.

Bottlenecks and scaling

Every ceiling above in one place. Only the first row is a limit the design has to work around; the rest have comfortable margins.

Resharding

Resharding is the one genuinely painful operation. How much data moves depends on the placement scheme. With plain hash(account_id) mod N, going from 32 to 64 shards changes the modulus, so about half of all keys land somewhere new. With a hash ring (shards and keys both placed on a notional circle, each key owned by the next shard clockwise) adding one shard steals keys from its neighbour alone, about 1/(N+1) of the total (consistent hashing).

Either way, the hard constraint is that an account must not move part-way through a transfer. So the procedure is four steps:

1. Freeze new sagas for that account.
2. Drain the in-flight ones, measurable, not guessed: the in_transit residue filtered to that account, going to zero.
3. Copy the data and flip the routing entry.
4. Unfreeze.

The drain is bounded by the step-2 retry deadline, not by how long the copy takes, so you can predict it before you start.

Failure modes

Every way the design breaks, the signal that reveals it, and the response. Three of these are invisible to the check most people would reach for.

Alternatives rejected

Each row is a design commonly proposed for this system, with the one-line reason it does not survive the numbers above.

Assumption ledger

Every number above rests on something assumed. This table separates assumptions safe to assume from ones worth verifying with the business, and marks which ones actually change the design. Load-bearing means the architecture changes if the assumption is wrong; otherwise being wrong only moves a number.

Conclusion

A wallet you own end to end trades the payment system’s external uncertainty for internal contention. The load-bearing decisions:

• The ledger is an append-only log; a balance is a fold over it, memoized in the same transaction so it can never be stale. Snapshot every 20 M entries so a rebuild fits a 10-second budget.
• A single balance row sustains ~1,333 writes/s, mostly the RPO-0 replica ack. A hot account splits its balance across 16 buckets: credits fan out, debits check the total under one lock set taken in ascending bucket id.
• 96.9% of transfers are cross-shard. Move money with a saga (debit, in-transit, credit), not 2PC: 2PC halves the hot-row ceiling and one coordinator crash freezes hundreds of thousands of balances. Every saga step is zero-sum, but zero-sum is blind to a duplicated step, so dedupe on (transfer_id, step).
• Corrections are new reversing entries, never UPDATE, which keeps every past balance reproducible and the invariant checkable.

One line to remember: the log is the truth and the balance is a fold you memoize; every hard problem here is that one choice meeting a hot row or a cross-shard hop.

Cheat sheet

Further reading

• Pat Helland, Life beyond Distributed Transactions: an Apostate’s Opinion: the case for local transactions plus reconciliation over distributed commit.
• Pat Helland, Immutability Changes Everything (ACM Queue): why append-only logs make systems auditable and reproducible.
• Martin Fowler, Event Sourcing: the pattern behind balance-as-a-fold.
• Chris Richardson, microservices.io: the Saga and Transactional Outbox patterns.
• Jim Gray and Leslie Lamport, Consensus on Transaction Commit: two-phase commit and why it blocks.

Related: Payment System owns double-entry, money types, and the outside world; Hotel Reservation owns idempotency keys and the double-booking race; Distributed Message Queue owns the outbox’s delivery semantics; database internals owns isolation levels and the anomaly matrix.