Abstractions That Hide the Network

It is the last over of a cricket final, and the crowd at Ravi's tea stall outside Pune station has doubled. Asha scans his QR code to pay ₹120 for six cups of chai. The spinner turns for ten seconds, then her wallet app says "Payment failed. Try again." She taps again, and this time it works. That night she sees two debits of ₹120. The first payment had gone through; only its reply was lost in the overloaded mobile network. Nothing crashed, and no line of code had a typo. The app treated a call across the network like a call inside one program, and Asha paid for the difference, twice.

Asha's wallet app is this chapter's running example. Its bug came from an abstraction: a simple interface that hides a messy mechanism. A file hides disk sectors, TCP hides lost and reordered packets, and a remote call hides the network behind a function call. You build on the simple picture until the mess leaks through.

Every abstraction has places where it leaks, and good designs plan for exactly those places:

Distributed systems lean on three abstractions that leak like this: remote calls, consistency models (what a read may return when data has several copies) and failure models (what may break). The chapter follows Asha's payment through all three.

Start at the moment Asha's app gave up: its request left the phone, and no reply came back. On one machine, failure is mostly all or nothing: the program runs, or it crashes and you restart it. Across a network, one part can fail while the rest keeps running, and the parts still running often cannot tell what happened. This is partial failure, and it is the defining trait of a distributed system.

When a reply does not come, any of these may be true, and from the phone they all look the same:

In three of the five rows the money moves, yet the app sees the same silence in all five.

The caller's only tool is a timeout: wait a bounded time, then assume something went wrong. But a timeout cannot tell slow from dead, or "never ran" from "ran, reply lost". Clocks add one more doubt. Each machine has its own clock, and two clocks drift apart by milliseconds or more, so timestamps from different machines cannot be compared exactly.

The rest of the chapter is about living with this uncertainty. It starts with the call itself: what does a remote call hide, and what does it fail to hide?

Asha's app reached the payment service through a remote call. A remote procedure call (RPC) lets code call a function on another machine as if it were local: wallet.pay(asha, ravi, 120). A generated client stub encodes the arguments into bytes (JSON, or a compact binary format such as Protocol Buffers), sends them and waits. A server stub on the other machine decodes the request, calls the real function and sends the result back.

The syntax matches a local call. The behaviour does not, and each row below is a place where the network leaks through:

The Outcomes and Runs rows caused Asha's double charge, and the Versions row bites on every deploy. The next pages deal with them in that order, starting with the first decision every caller makes: how long to wait.

Asha's app waited ten seconds before giving up. Every remote call needs a timeout; without one, a stuck server holds your threads forever, and one slow dependency becomes your outage. Set it from the callee's measured latency. If 99 in 100 of its replies arrive within 100 ms (its p99, or 99th percentile), a timeout of 200 to 300 ms, two to three times that, catches real trouble without cutting off normal slow replies. Too short, and healthy calls fail and retry, adding load; too long, and users stare at a spinner. The timeout must also fit inside the caller's own deadline: a screen that must answer in 1 second cannot give one dependency 2 seconds.

After a timeout, the natural move is to retry. Retries hide brief glitches, such as a dropped packet or a restarting server, but they carry two dangers: a struggling server gets hit harder exactly when it can least cope, and the first attempt may have worked.

import random, time

def call_with_retries(send, attempts=4, base=0.2):
    for i in range(attempts):
        try:
            return send(timeout=0.3)            # never wait forever
        except (TimeoutError, ConnectionError):
            if i == attempts - 1:
                raise                           # give up cleanly
            cap = base * 2 ** i                 # 0.2 s, 0.4 s, 0.8 s
            time.sleep(random.uniform(0, cap))  # jitter

Exponential backoff with jitter: the wait limit doubles each time, and the random part stops thousands of phones retrying in lockstep.

Four rules keep retries from doing harm:

• Retry only errors that may be temporary: timeouts, dropped connections, 503 Service Unavailable. A 400 Bad Request will fail the same way every time.
• Cap both the number of attempts (3 or 4) and the total time spent trying.
• Retry at one layer only. If the app, its client library and the load balancer each try 3 times, one tap can become 27 requests. Availability, Reliability, Scalability and Latency adds retry budgets and circuit breakers.
• Retry only operations that are safe to repeat. A payment is not, as the next page shows.

Retrying is safe only for some operations. An operation is idempotent if doing it twice has the same effect as doing it once. "Set my display name to Asha" is idempotent. "Pay Ravi ₹120" is not: run it twice and ₹240 leaves the wallet. Reads are naturally idempotent; most writes that add, subtract or create something are not.

The fix is an idempotency key: a random id, such as a UUID, that the app creates once per payment, not once per attempt, and sends with every retry. The server stores each key with its result, so a repeat gets the first answer back instead of a second debit. A key made per attempt would change on every retry and catch nothing.

The scene hid two hard cases: the server may crash halfway, and two copies of a request may arrive at the same moment. Both are handled by committing the key and the debit in one database transaction, so neither can exist without the other.

def pay(req):
    key = req.headers["Idempotency-Key"]  # made by the app, once per payment
    old = keys.get(key)
    if old:
        return old.response                # a retry: replay the first answer
    with db.transaction():                 # key and debit commit together
        keys.insert(key)                   # unique index: a parallel copy fails here
        receipt = ledger.debit(req.user, req.amount)
        keys.save_response(key, receipt)
    return receipt

If two copies race, the unique index lets one insert succeed; the other waits, fails, then reads the saved answer. Keep keys longer than any retry window (say 24 hours), and reject a reused key that arrives with a different amount or payee.

Messaging systems name the same idea. Delivery guarantees come in three strengths:

Effectively once is the strongest guarantee you can build, and you build it from the two rows above it. No network can deliver a message exactly once, so systems that promise "exactly once" deliver at least once and make the repeats harmless, usually with an id that works like the idempotency key.

The same keys protect a consumer that receives a message twice, as Message Queues shows. All of this works because the key and the debit live in one database. The next page asks what happens when they cannot.

The transaction on the last page works because Asha's wallet balance lives in the app's own database. Topping up the wallet is different. When she adds ₹500 from her bank, the money moves inside a provider, a payment gateway or bank whose work cannot join your transaction. If your server crashes after calling the provider but before saving its answer, nobody knows whether she paid.

So record the attempt before calling out, and treat "unknown" as a state of its own:

Step 5 is the one people forget. After a timeout, the top-up may or may not have gone through: marking it failed invites Asha to pay again, and marking it paid credits money that never arrived. Asking the provider turns "unknown" into a fact. Design a Payment System builds a daily reconciliation, a comparison of your records against the provider's, on the same idea.

One row of the remote-call table is still open: the two sides of a call are deployed separately. During a rolling deploy, servers are replaced a few at a time, so old and new versions of the payment service run side by side for minutes or hours. Phones change even more slowly: some users run last year's wallet app, and nobody can force them to update. So every message format must work in both directions. New code must read old messages (backward compatibility), and old code must read new ones (forward compatibility).

Three habits keep both directions working:

• Add fields as optional: new servers treat a missing tip as zero, and old readers skip unknown fields (Protocol Buffers does; a strict JSON validator may reject them). But an old server that skips tip charges the wrong amount, so the app sends it only once every server understands it.
• Never reuse a field's name or number for something new, and never change its type in place. An amount that moves from rupees to paise needs a new field, such as amount_paise.
• Split risky changes across releases: ship code that reads both shapes, then start writing the new shape, then stop writing the old one. Database columns follow the same order: add the column, write both, backfill old rows, switch reads, and drop the old column later.

The slow order exists because any release may be rolled back. If version 2 writes a shape that only version 2 understands, rolling back to version 1 breaks on every record version 2 already wrote. Each step must be safe to undo, which is why one risky change becomes two or three dull releases.

So far every call had a single server answering it. Real data lives in several copies, which raises the next question: which copy answers, and what is it allowed to say?

As soon as data has more than one copy, a read can land on a copy that has not yet received the latest write. Each copy is a replica. A consistency model is the contract that says which values a read may return while replicas disagree. Stronger contracts are easier to program against and cost more to provide.

The wallet app keeps Asha's account on three replicas. Her balance starts at ₹500. She pays Ravi ₹120 (balance ₹380), then recharges her phone for ₹99 (balance ₹281). Each time her app or the support team's screen reads the balance, the read may land on a different replica. A moment after the recharge, the replicas might look like this:

Every replica catches up within moments. The question is what a reader may see in the meantime, and whether two readers may be told different stories. The next page lines up the answers.

Here are the four main models, strongest first, each applied to Asha's balance:

Read the table from top to bottom as a trade: each step down gives up a guarantee and gains speed and availability. "Strong consistency" usually means the top row, and eventual consistency is the weakest model in common use.

The cost comes from coordination. To be linearizable, one leader or a majority of replicas must order each write before it is confirmed, and a read must come from a leader that is sure it still leads, or must check with a majority. An eventually consistent write can be accepted by any nearby replica and copied to the others later, so it stays fast and keeps working when replicas cannot reach each other.

Causal is a useful middle: the refund and the recharge are unrelated, so readers may see them in either order, but nobody sees a refund before its payment.

Most reads in an app can use the weaker models, which leaves a practical question: which oddities will users actually notice? Two session guarantees, promises about one user's own sequence of reads and writes, fix the ones that hurt most.

• Read-your-writes: after Asha pays Ravi, her own history shows the payment. Without it, a lagging replica shows no payment, and she may pay again: the opening story by another route.
• Monotonic reads: once she has seen ₹281, she never sees ₹380 again.

Both are cheap. Monotonic reads come from sending each user's reads to the same replica. Read-your-writes comes from reading a user's own data from the leader for a short while after they write (say, a minute), or from a replica that has caught up with their last write. Causal consistency gives both.

Strong consistency has a price that no engineering removes. When a network fault splits the replicas into two groups, at most one side holds a majority. A linearizable store must refuse requests on the other side; an always-available store keeps answering there, possibly with stale data. This forced choice during a network partition is what the CAP theorem describes. For the wallet, the choice is easy: refuse the debit, because a refused payment can be retried, while money spent twice is hard to claw back.

PACELC extends CAP to normal days: if there is a Partition, trade Availability against Consistency; Else, trade Latency against Consistency. A write confirmed in both Mumbai and Delhi waits about 30 ms for the round trip, even when nothing is wrong.

With the models and their prices in view, the real question is which operation needs which. Choose per operation, not per company or per database: pay for strong guarantees where a wrong answer costs money or trust, and accept weaker ones everywhere else.

Here is how the wallet app's features sort out:

Notice that the first two rows ask for a kind of write, not a fresh read, and the home screen can show a slightly stale balance precisely because the debit does its own check.

Weaker models push work onto the application: it must tolerate stale reads and reordering, and in some stores it must merge two conflicting writes that different replicas accepted at the same time. The next page shows why the money rows need a write.

A tempting design reads Asha's balance from the primary, checks it in code, then writes the debit. Even a perfectly fresh read is out of date the moment it returns. Say she has ₹150. Her ₹120 tea payment and an automatic ₹99 phone recharge reach two servers at the same moment. Both read ₹150, both pass the check, and both debit. The wallet ends ₹69 below zero, however fresh the reads were.

The check and the change must be one atomic step that the store enforces:

UPDATE wallets
SET balance = balance - 120
WHERE user_id = 42 AND balance >= 120;
-- 1 row updated: paid. 0 rows: not enough money, nothing changed.

The database applies the two debits one at a time. Whichever goes second finds ₹30 or ₹51, matches 0 rows and fails cleanly.

Two other tools do the same job: a unique constraint, which makes a second claim on asha@wallet fail, and a transaction that locks the row before reading it (SELECT ... FOR UPDATE). A stricter isolation level such as serializable also catches the race, by aborting one transaction that the caller must then retry.

This completes the double-charge story. The unique index on the idempotency key stops two copies of the same payment; the conditional debit stops two different payments from overspending. Each tool guards against one race, and a payment service needs both.

All of them need one place that orders the writes to a row: a relational primary, or a store that sends each key's writes through one leader. In a leaderless store, two replicas can each accept a conflicting write, unless the store offers a compare-and-set (write only if the value is still what you read) built on consensus, which Consensus, Leader Election and Coordination explains.

So far the failures were lost replies and racing writes. A failure model is the list of faults a design promises to survive, and an algorithm is only correct under its model: a protocol built to handle crashes can be fooled by a node that lies. From the simplest to the hardest:

Each row down is harder to detect and costs more to survive.

The payment service is a crash-recovery case: after a restart, its disk survives and its memory does not. That is why the "in progress" record from the provider page is committed to disk before the call goes out. A record kept only in memory would vanish in exactly the crash it exists for.

Harder models also cost more machines. To survive f crashed nodes, a majority-based protocol needs 2f + 1 nodes: 3 survive 1 crash, 5 survive 2. Any two majorities share a node, so a decision one majority made cannot be lost by the next. To survive f Byzantine nodes takes 3f + 1: 4 nodes to survive one liar, plus far more messages.

Which rows does a real design plan for? Most systems inside one company's data centres design for crash-recovery, omission and timing faults, and detect them with timeouts and heartbeats (small "I am alive" messages sent every second or so). They skip full Byzantine protocols, because their machines trust each other and the cost is high, but they add cheap checks against corruption: checksums on stored data and messages, input validation, and authentication between services. Byzantine tolerance appears where parties do not trust each other, such as public blockchains, or where a wrong answer is dangerous, such as aircraft flight controls, where several computers vote on every output.

Because a timeout cannot tell slow from dead, every failure detector is a guess. Set it too short and healthy nodes are declared dead, causing needless failovers and, at worst, two nodes that both believe they lead (split brain). Set it too long and real failures take ages to handle. The hardest case is a node that is up but ten times slower: it passes basic health checks while dragging down every caller.

The usual defences are a majority vote to choose one leader, leases (leadership that expires unless renewed in time) and fencing (each new leader gets a higher number, and storage rejects writes carrying an older one). Leases also speed up reads: a leader may answer a linearizable read alone while its lease is valid, because a new leader must wait for the old lease to run out. That holds only if clocks tick at nearly the same rate and the leader is not paused mid-request, so leases carry a safety margin and writes still carry fencing numbers. Consensus, Leader Election and Coordination works through all three step by step.

Whenever a design gets an arrow between two services or a second copy of data, these lines show you know where the network leaks.