The Engineer's Fleet

Three good reasons

• Scale. More requests, or more data, than one machine can handle.
• Availability. Any single machine will eventually fail. Keeping copies elsewhere lets the service continue.
• Latency. Light in optical fibre covers about 200 kilometres per millisecond. London to Sydney is about 17,000 kilometres, so no request can make that round trip in under about 170 milliseconds, and real routes take longer. If users are on the other side of the world, some of your system has to be near them.

There are organisational reasons too, such as letting separate teams own and deploy separate parts, and legal ones, such as keeping European customers' data in Europe. Those are real, but notice that none of the reasons is "it's more interesting" or "we might be big one day".

Scale up before you scale out

A single modern server can have hundreds of cores, terabytes of memory and storage that serves millions of operations a second. A well-tuned relational database on one large machine, with a couple of read replicas and a cache in front, carries far more traffic than most products will ever see. Scaling up is boring and expensive per machine. Scaling out is exciting and expensive in engineering time, forever.

A useful map of the options is the scale cube from Abbott and Fisher's The Art of Scalability. Scale along the X axis by running identical copies behind a load balancer, which is easy for stateless services. Scale along the Y axis by splitting the system by function into services (chapter 17). Scale along the Z axis by splitting the data itself, so each machine holds part of it (chapter 5). The X axis is cheap. The other two are where the difficulty in this volume lives, because they involve state.

Orders of magnitude

Distributed design is mostly arithmetic about delays. Memorise these rough figures for current hardware, and then measure your own, because they vary.

The lesson is in the right-hand column. A function call that becomes a network call inside a data centre gets thousands of times slower; one that crosses an ocean, millions. That's why "just split it into services" so often makes things slower, and why chatty conversations between regions are the first thing to design out.

Availability is arithmetic too

Availability is usually quoted in "nines". Each extra nine cuts the downtime you're allowed by a factor of ten.

Dependencies on the critical path multiply: if a request needs ten services that are each up 99.9% of the time, the request succeeds only about 99% of the time. Redundancy works the other way: two independent copies that are each up 99% of the time give 99.99% together. But only if they fail independently, and they rarely do. Copies share a deploy pipeline, a configuration system, a region, a bug. Correlated failure is why "we have three replicas" so often turns out to mean "we had one failure, three times".

def series(*availabilities: float) -> float:
    """Every component is needed: availabilities multiply."""
    total = 1.0
    for a in availabilities:
        total *= a
    return total


def parallel(availability: float, copies: int) -> float:
    """Any one copy is enough, assuming failures are independent."""
    return 1 - (1 - availability) ** copies


series(*[0.999] * 10)    # 0.9900: ten "three nines" dependencies make two nines
parallel(0.99, 2)        # 0.9999: two "two nines" copies make four, if independent

What distribution costs

The rest of this volume is the invoice. Some parts will fail while others carry on (chapter 2). There's no shared clock (chapter 3). Copies of data disagree (chapters 4 to 7). Agreeing on anything needs careful protocols (chapters 8 to 10). And the operational burden grows with every moving part: deploying, observing, debugging and paying for many machines instead of one. Distribute when the alternative is worse. It often isn't, yet.

Partial failure

On one machine, failure is mostly all-or-nothing: the process crashes and everything stops. Across machines, some parts fail and others don't, and no participant has a complete picture. Send a request to another service and wait. Any of these may have happened:

• The request was lost before it arrived.
• It arrived, and the server crashed before doing anything.
• It arrived, was fully processed, and the response was lost.
• It's still being processed, slowly, perhaps behind a garbage-collection pause.

From the sender's side, all four look identical: silence. That single fact is the root of most of this volume. It's why operations need idempotency keys, why "exactly once" is so hard, and why you can't simply ask whether a machine is dead.

The eight fallacies

In the 1990s, Peter Deutsch and colleagues at Sun Microsystems listed the false assumptions that programmers new to distributed systems invariably make; James Gosling later added the eighth. Every one of them is still being made today.

Failure models

To reason about a protocol, you have to say what kinds of failure it tolerates. The common models, from mildest to worst:

• Crash-stop. A node halts and never returns.
• Crash-recovery. A node halts and may come back later, with whatever it wrote to disk but nothing that was in memory. This is what real servers do, and most protocols in this volume assume it.
• Omission. Messages are lost.
• Byzantine. A node behaves arbitrarily: it lies, sends conflicting messages, or is controlled by an attacker. Systems inside one organisation usually assume this doesn't happen; blockchains and some aerospace systems must not.

Timing needs a model too. In a synchronous system, messages and pauses have known upper bounds. In an asynchronous one, there are no bounds at all. Real systems are partially synchronous: usually well-behaved, with bounded delays, but occasionally not, for unbounded stretches. Good protocols stay correct even when timing goes wrong, and only need the good periods to make progress.

Timeouts are guesses

The only practical way to decide that a node has failed is to stop waiting for it. But a timeout can't distinguish "dead" from "slow". Make it too short and you declare healthy-but-busy nodes dead, trigger failovers and retries that add load exactly when the system is struggling, and cause the outage you feared. Make it too long and users wait, and recovery is slow.

Derive timeouts from measured latency, such as a little above the 99.9th percentile, not from round numbers someone typed years ago. Adaptive failure detectors, such as the phi accrual detector used by Cassandra and Akka, go further: they learn the normal distribution of heartbeat intervals and report a suspicion level rather than a yes or no.

The underlying impossibility has a famous form, the two generals problem. Two armies must attack together, and can only communicate by messengers who may be captured. No protocol can ever make both generals certain they've agreed, because the last messenger might always be lost. The practical response isn't to find certainty. It's to design for its absence: idempotent operations, retries with the same key, and reconciliation afterwards.

Make "unknown" a real outcome

The most useful habit this chapter can give you is to stop treating a timeout as a failure. It's a third outcome, and code should say so.

from enum import Enum


class Outcome(Enum):
    SUCCEEDED = "succeeded"
    FAILED = "failed"        # the server definitely refused: report it, or fix and retry
    UNKNOWN = "unknown"      # it may or may not have happened: find out, don't guess


def charge(client, request, timeout_s: float):
    try:
        response = client.post("/charges", json=request, timeout=timeout_s,
                               headers={"Idempotency-Key": request["idempotency_key"]})
    except (Timeout, ConnectionLost):
        return Outcome.UNKNOWN, None
    if response.status_code >= 500:
        return Outcome.UNKNOWN, None     # the server may have failed after committing
    if response.status_code >= 400:
        return Outcome.FAILED, response.json()
    return Outcome.SUCCEEDED, response.json()

An UNKNOWN result must be resolved, never assumed: retry with the same idempotency key, or ask the server about the operation by its id. The user sees "processing", not "failed", because telling someone their payment failed when it went through is how double payments happen.

Partitions happen, and short ones cause long outages

A network partition splits machines into groups that can't talk to each other, while each group is healthy on its own. Peter Bailis and Kyle Kingsbury's 2014 survey, The Network Is Reliable, collected many of them from real companies: failed switches, misconfigured firewalls, saturated links, cloud network faults. They're rare per link and routine across a large fleet.

GitHub's incident of 21 October 2018 shows how a short partition becomes a long outage. Maintenance cut connectivity between its East Coast data centre and the rest of its network for 43 seconds. In that window, its database orchestration tool, which used Raft to decide leadership among its own nodes, promoted replicas on the West Coast to primaries. When the network returned, writes had landed in both places, and the primaries were now across the continent from the applications using them. Untangling it took 24 hours and 11 minutes of degraded service. The lesson isn't that automatic failover is bad. It's that failover is itself a risky operation, and deciding to fail over because of a brief silence can cost far more than waiting would have.

Gray failure

Not every failure is a clean crash. A disk becomes slow. One network card drops one packet in a hundred. A node answers health checks perfectly while failing half of real requests. Microsoft researchers named this gray failure: the system's own failure detectors think everything is fine while users suffer. The defence is to measure what users experience, compare nodes with their peers, and treat an outlier as a suspect even when it passes its checks.

Physical clocks lie

Every server keeps time with a quartz oscillator that runs slightly fast or slow. A drift of 50 parts per million, which is unremarkable, adds up to more than four seconds a day. The Network Time Protocol corrects this by periodically adjusting the clock towards a reference, typically to within a few milliseconds inside a data centre, and much worse across the internet or on an overloaded host. Corrections can make a clock jump, including backwards. Virtual machines can be paused and resumed with a stale clock. Leap seconds have caused real outages.

Every platform offers two kinds of clock, and you must use the right one.

• Wall-clock time, such as Python's time.time(), is the time of day. It's right for showing people when something happened, and it can jump.
• Monotonic time, such as time.monotonic(), only ever moves forward, from an arbitrary starting point. It's right for measuring durations and enforcing timeouts, and meaningless across machines.

The most expensive misuse of wall clocks is last write wins: two replicas each accept a write, and the one with the later timestamp is kept. If one machine's clock runs ahead, its writes keep winning even when they happened first, and the other writes vanish silently. Many databases do exactly this by default.

Happened-before

In 1978, Leslie Lamport showed how to reason about order without any clock at all. Event a happened before event b if they're on the same process and a came first; or if a is the sending of a message and b is its receipt; or if there's a chain of such steps from a to b. If neither happened before the other, the events are concurrent: neither could have influenced the other, whatever the wall clocks say.

Lamport clocks

A Lamport clock is a counter on each node. Increment it for every event, attach it to every message, and on receipt set it to one more than the larger of your value and the message's. If a happened before b, then a's timestamp is smaller. Breaking ties by node id gives a total order that every node agrees on and that respects causality.

class LamportClock:
    def __init__(self):
        self.time = 0

    def tick(self) -> int:
        """A local event, including sending a message."""
        self.time += 1
        return self.time

    def receive(self, sent_time: int) -> int:
        """Receiving a message stamped with the sender's clock."""
        self.time = max(self.time, sent_time) + 1
        return self.time

The limitation runs one way only. A smaller timestamp doesn't mean the event happened first; the two events might be concurrent. Lamport clocks can order things consistently, but they can't tell you whether two events conflict.

Vector clocks

A vector clock keeps one counter per node. Each node increments its own entry for every event and, on receiving a message, takes the element-wise maximum of the two vectors before incrementing. Comparing two vectors then tells you everything: if every entry of one is less than or equal to the other's, it happened before; if each has an entry larger than the other's, they're concurrent. That's how systems in the tradition of Amazon's Dynamo detect conflicting writes instead of silently discarding one.

def increment(clock: dict[str, int], node: str) -> dict[str, int]:
    updated = dict(clock)
    updated[node] = updated.get(node, 0) + 1
    return updated


def merge(a: dict[str, int], b: dict[str, int]) -> dict[str, int]:
    return {n: max(a.get(n, 0), b.get(n, 0)) for n in a.keys() | b.keys()}


def compare(a: dict[str, int], b: dict[str, int]) -> str:
    nodes = a.keys() | b.keys()
    a_le_b = all(a.get(n, 0) <= b.get(n, 0) for n in nodes)
    b_le_a = all(b.get(n, 0) <= a.get(n, 0) for n in nodes)
    if a_le_b and b_le_a:
        return "equal"
    if a_le_b:
        return "before"
    if b_le_a:
        return "after"
    return "concurrent"          # a genuine conflict: keep both, and merge

The price is size. A vector grows with the number of participants, which is fine for a handful of replicas and a problem if every client is a participant. Refinements such as dotted version vectors keep it manageable.

Hybrid logical clocks and TrueTime

People want timestamps that look like real time, for reading a snapshot "as of 10:00" or for debugging. Hybrid logical clocks combine the physical clock with a logical counter: timestamps stay close to wall-clock time, but when a message arrives from a machine whose clock is ahead, the counter preserves causality anyway. CockroachDB, YugabyteDB and MongoDB all use them.

Google's Spanner takes the opposite approach: make the physical clock trustworthy. Its TrueTime service uses GPS receivers and atomic clocks to give every machine not a time but an interval, guaranteed to contain the true time, typically a few milliseconds wide. When a transaction commits, Spanner waits until the interval has passed before revealing the commit. That commit wait guarantees that timestamp order matches real-time order across the world. The lesson generalises: bounded clock error lets you use time for ordering, and you pay for the bound by waiting it out.

Practical rules

• Measure durations and timeouts with a monotonic clock.
• Don't decide which write wins by comparing wall-clock timestamps from different machines, unless you've accepted losing data.
• For order within a system, use versions and sequence numbers issued by one authority, such as a database, or logical clocks.
• Store timestamps in UTC, and remember which clock produced them.
• Time-ordered identifiers, such as UUIDv7, sort roughly by creation time. That's convenient for indexes, and not a causal order.
• To follow one request across services, use a trace id (chapter 20), not the timestamps in each service's logs.

# A: measuring how long a request took
started = time.time()
handle(request)
metrics.observe("request_seconds", time.time() - started)

# B: two servers each save an edit; keep the latest
def resolve(edit_a, edit_b):
    return edit_a if edit_a.updated_at > edit_b.updated_at else edit_b

Single-leader replication

The most common arrangement, used by PostgreSQL, MySQL and most relational databases. One node, the leader or primary, accepts all writes and records them in a log. Followers apply the leader's log in the same order, so they hold the same data, a little later. Reads can go to any node.

The crucial choice is when the leader tells the client "done".

• Synchronous: after at least one follower has the write. An acknowledged write survives the leader's death, but every write waits for a follower, and a slow or dead follower stalls writes.
• Asynchronous: as soon as the leader has it. Writes are fast, but if the leader dies, writes it acknowledged and hadn't yet shipped are lost.
• Semi-synchronous: one follower synchronous, the rest asynchronous. The usual compromise: no acknowledged write lives on only one machine.

Failover, promoting a follower when the leader dies, is where the trouble lives. You must decide the leader is really dead (chapter 2), choose the most up-to-date follower, redirect every client, and make sure the old leader can't come back and keep accepting writes. Get the last part wrong and you have split brain: two leaders, both taking writes, diverging. Chapters 8 and 9 cover how consensus and fencing prevent it.

Replication lag and what users see

With asynchronous followers, reads from a follower may return data that's seconds or, under load, minutes old. Usually harmless, but it produces three specific anomalies that users notice and report as bugs.

• Read-your-writes. A user saves a change, and the next page load doesn't show it, as above. Fix: read data the user has recently changed from the leader, or remember the log position of their last write and read only from a follower that has reached it.
• Monotonic reads. A user refreshes and sees a new comment, refreshes again and it's gone, because the second read hit a follower further behind. Fix: route each user's reads to the same replica, for example by hashing their id.
• Consistent prefix. An observer sees a reply before the question it answers, when the two were written to different partitions that replicate independently. Fix: write causally related data to the same partition, or track causal dependencies.

def save_invoice(session, leader, invoice):
    position = leader.execute_returning_log_position(
        "UPDATE invoices SET ... WHERE id = %s", (invoice.id,))
    session["min_position"] = max(session.get("min_position", 0), position)


def load_invoice(session, followers, leader, invoice_id):
    needed = session.get("min_position", 0)
    for follower in followers:
        if follower.replayed_position() >= needed:      # it has seen our last write
            return follower.get_invoice(invoice_id)
    return leader.get_invoice(invoice_id)               # nobody has caught up yet

In PostgreSQL, the leader's position comes from pg_current_wal_lsn() and a follower's from pg_last_wal_replay_lsn(). The same idea works with any database that exposes a log position.

Multi-leader replication

Sometimes more than one node must accept writes: a leader in each region so writes stay local, an app that works offline and syncs later, or a document several people edit at once. Every leader then accepts writes independently, and conflicts are inevitable: the same field changed in two places before either change reached the other.

Conflicts must be resolved somehow, and the options get better as they get more work. Last write wins by timestamp throws data away (chapter 3). Merging per field keeps both changes when they touch different fields. Application-specific rules can merge meaningfully, such as taking the union of line items added in two places. And conflict-free replicated data types, CRDTs, are data structures designed so that replicas can always merge, in any order, any number of times, and converge on the same value. They power collaborative editors and offline-first apps.

class GCounter:
    """A grow-only counter. Each replica increments only its own slot."""

    def __init__(self, replica_id: str):
        self.replica_id = replica_id
        self.counts: dict[str, int] = {}

    def increment(self, amount: int = 1) -> None:
        self.counts[self.replica_id] = self.counts.get(self.replica_id, 0) + amount

    def value(self) -> int:
        return sum(self.counts.values())

    def merge(self, other: "GCounter") -> None:
        for replica, count in other.counts.items():
            self.counts[replica] = max(self.counts.get(replica, 0), count)

Merging takes the maximum of each slot, an operation that gives the same answer whatever the order and however often it's repeated. So replicas that exchange state over an unreliable network, with duplicates and reordering, still converge on the true total. Richer CRDTs handle sets, maps and text in the same spirit.

Leaderless replication

Amazon's 2007 Dynamo paper popularised a design with no leader at all, used today by Cassandra and others. A write goes to all N replicas of a key and succeeds when W of them acknowledge it. A read asks R replicas and takes the newest version. If W + R > N, every read set overlaps every write set, so at least one replica consulted on a read has the latest acknowledged write. A common setting is N = 3, W = 2, R = 2, which tolerates one replica being down for both reads and writes.

Stale replicas are repaired in the background: on reads, when a coordinator notices an old version, and by anti-entropy processes that compare replicas using Merkle trees. When some replicas are unreachable, a sloppy quorum may store writes on substitute nodes and hand them back later, which is called hinted handoff; that keeps writes available but weakens the overlap guarantee. And even strict quorums don't make a system linearizable on their own: concurrent writes, partially failed writes and clock-based conflict resolution all leave gaps. Quorums give you a high probability of fresh reads, not a promise.

Partitioning and replication together

Partitioning, also called sharding, splits a dataset so that each piece lives on a different node, letting you store more data and handle more writes than one machine can. It's independent of replication: in practice each partition is also replicated, so every node holds several partitions, leading some and following others.

By key range

Assign contiguous ranges of keys to each partition: invoices A to F here, G to M there. Range scans stay efficient, because neighbouring keys live together. Bigtable, HBase, Spanner and CockroachDB work this way, splitting a range automatically when it grows too large.

The danger is hot spots from keys that arrive in order. If the key starts with a timestamp, every new write lands in the newest range, and one node takes all of today's traffic while the rest sit idle. Put something with spread, such as a tenant id, before the timestamp.

By hash

Hash each key and assign partitions by hash value. Load spreads evenly whatever the keys look like, at the cost of range scans, which now touch every partition. But never place keys with hash(key) % N, where N is the number of nodes. Change N from 4 to 5 and about 80% of all keys move to a different node, all at once.

Consistent hashing

Consistent hashing, introduced at MIT in 1997, places both nodes and keys on a circle by hashing them. Each key belongs to the first node clockwise from it. Adding a node moves only the keys it takes over from its neighbour; removing one hands its keys to the next.

With only a few points on the circle, the arcs between nodes come out uneven, so one node may own twice as much as another. The fix is virtual nodes: each physical node appears at many points. That evens out the load, and when a node leaves, its keys scatter across many survivors rather than landing on one neighbour.

import bisect
import hashlib


def _hash(value: str) -> int:
    return int.from_bytes(hashlib.sha256(value.encode()).digest()[:8], "big")


class HashRing:
    def __init__(self, nodes=(), vnodes: int = 100):
        self.vnodes = vnodes
        self._ring: list[tuple[int, str]] = []          # sorted (position, node) pairs
        for node in nodes:
            self.add(node)

    def add(self, node: str) -> None:
        for i in range(self.vnodes):
            bisect.insort(self._ring, (_hash(f"{node}#{i}"), node))

    def remove(self, node: str) -> None:
        self._ring = [(position, n) for position, n in self._ring if n != node]

    def node_for(self, key: str) -> str:
        index = bisect.bisect(self._ring, (_hash(key),))  # first point clockwise
        return self._ring[index % len(self._ring)][1]

Run it on 20,000 team ids with four nodes, then add a fifth. About 16% of keys move, all of them to the new node, against about 80% with the modulo approach. Alternatives with similar properties include rendezvous hashing, where each key goes to the node with the highest hash of the key and node combined, and Google's jump consistent hash, which needs no stored ring at all.

Many systems skip rings entirely and use a fixed number of partitions, far more than the number of nodes, say 1,024. Keys map to partitions by hash, and partitions map to nodes through a small table. Adding a node moves whole partitions, and the key-to-partition mapping never changes. Kafka topics, Elasticsearch indexes and Redis Cluster's 16,384 hash slots all work this way.

Hot keys

Even perfect hashing can't split a single key. One enormous tenant, one celebrity account, or one viral invoice sends a disproportionate share of traffic to one partition. Remedies, roughly in order of preference: cache the hot key's reads; give the whale its own partition or database, as The Engineer's Shipyard suggested for large tenants; and, for write-heavy counters, split the key into several sub-keys such as views:inv_1#0 to #9, writing to a random one and summing all of them on read.

Secondary indexes

Partitioning by team id makes "all invoices of team X" cheap. What about "all overdue invoices across every team", or "invoices for this client email"? There are two designs.

• Local indexes: each partition indexes its own data. Writes stay simple, but a query by the secondary attribute must ask every partition and combine the results, called scatter-gather, and it's only as fast as the slowest partition.
• Global indexes: the index is partitioned by the indexed value itself. Reads touch one index partition, but a single write may update index partitions on other nodes, usually asynchronously, so the index lags.

Often the better answer is to not ask the operational database at all: stream changes into a search engine or analytics warehouse built for cross-partition questions (chapter 13).

Rebalancing and routing

Moving partitions between nodes is expensive: it copies data over the network while the system carries live traffic. Do it gradually and with throttling, and be wary of fully automatic rebalancing triggered by failure detection. A node that's merely slow can be declared dead, trigger a flood of data movement, and make everything slower, a feedback loop chapter 14 returns to. Clients need to know where each partition lives. They can ask a routing tier, ask any node, which forwards the request, or cache the mapping themselves. The authoritative mapping is usually kept in a consensus-backed store (chapter 8).

Why we need contracts

With several clients and several replicas, a stream of reads and writes can interleave in many ways. A consistency model says which of those interleavings, or histories, are allowed. Engineers talk loosely about "strong" and "eventual" consistency, but the precise models differ in ways that matter, and databases' marketing often claims more than the database delivers. Jepsen, Kyle Kingsbury's testing project, has repeatedly found popular databases violating the models their documentation promised.

Linearizability

The strongest single-object model, and the one people usually mean by "strong consistency". Every operation appears to take effect instantaneously at some moment between its start and its end, and once any read has returned a new value, every later read returns it too. The system behaves as if there were a single copy of the data, even though there are several.

That's what makes certain things possible at all: a lock that only one client can hold, a leader that only one node can be, a username only one person can claim, a balance that can't go below zero. The cost is coordination on every operation, and, during a partition, the side that can't reach a majority must refuse requests (chapter 7).

client A:  write(x = 1)  |-------------|                          (0 to 10 ms)
client B:                     read(x) -> 1  |-------|             (5 to 15 ms)
client C:                                          read(x) -> 0   (20 to 25 ms)

C started after A's write had finished and after B had already seen 1,
yet C read the old value. A single copy could never behave this way.

Weaker, cheaper models

Causal consistency deserves particular attention. Operations related by cause and effect, in the sense of chapter 3, are seen in the same order everywhere; unrelated operations may be seen in different orders. That rules out the anomalies users find most confusing, and it's the strongest model that can keep working on both sides of a partition. Session guarantees, identified by the Bayou project in 1994, are smaller promises made to each client: read-your-writes, monotonic reads, monotonic writes, and writes-follow-reads. They fix most of the anomalies from chapter 4 cheaply.

Eventual consistency guarantees very little on its own, but CRDTs strengthen it to strong eventual consistency: replicas that have received the same updates hold the same state, immediately, whatever order the updates arrived in.

Consistency is not isolation

Two vocabularies grew up separately, and they're easy to confuse. Isolation levels, such as serializable, snapshot isolation and read committed, describe how transactions over several objects interleave. Consistency models such as linearizability describe single objects in real time. A database can be serializable without being linearizable: its transactions behave as if they ran one at a time, but not necessarily in real-time order, so a read-only transaction might see a slightly old snapshot. The combination of both is called strict serializability, which is roughly what Spanner calls external consistency.

You may already have what you need

A single-leader database gives you linearizable operations on the leader for free, including uniqueness. Claiming a team's public URL slug takes one statement, and exactly one caller can ever win.

INSERT INTO team_slugs (slug, team_id)
VALUES ('studio-north', 'team_42')
ON CONFLICT (slug) DO NOTHING
RETURNING team_id;       -- a row back means you have it; no row means someone else does

The same operation against an eventually consistent store is genuinely hard, because two replicas can each accept the slug before hearing about the other. That's often the deciding question when choosing a datastore: which of your operations need a single authoritative answer?

Choosing per operation

Most systems need several models at once. Money, uniqueness, and permissions changes need linearizable or serializable operations. A user's own profile and settings need read-your-writes. Comments and conversations need causal order. Like counts, view counts and activity feeds are fine with eventual consistency. Paying the cost of the strongest model everywhere is as much a design mistake as using the weakest one for payments.

History 1:  A writes x = 1 from 0 to 10 ms.
            B reads x from 5 to 15 ms and gets 1.
            C reads x from 20 to 25 ms and gets 0.

History 2:  A writes x = 1 from 0 to 10 ms.
            B reads x from 2 to 4 ms and gets 0.
            C reads x from 5 to 8 ms and gets 1.

History 3:  A writes x = 1 from 0 to 10 ms.
            B reads x from 2 to 6 ms and gets 1.
            B reads x again from 7 to 9 ms and gets 0.

def claim_slug(slug, team_id):
    if not replica_db.slug_exists(slug):          # read from a follower
        leader_db.insert_slug(slug, team_id)
        return True
    return False

What CAP actually says

Eric Brewer proposed the idea in 2000, and in 2002 Seth Gilbert and Nancy Lynch proved it with precise definitions. C means linearizability, chapter 6's strongest model. A means every request received by a node that hasn't failed gets a non-error response. P means the network may lose any number of messages between nodes. The theorem: during a partition, a system can't have both C and A. A node that can't reach the others must either answer, possibly with stale data, or refuse, and lose availability.

What it doesn't say

• "Pick two out of three" is misleading. You don't get to choose whether the network partitions. The only real choice is what to do when it does.
• "CA" isn't an option for a distributed system. A system can be consistent and available only while it runs on one node or ignores partitions, and then the first partition makes the choice for you, usually badly.
• CAP's definitions are narrow. Its availability means every non-failed node answers, which few real systems attempt. Its consistency is only linearizability. Many useful systems are neither CAP-consistent nor CAP-available: a single-leader database with asynchronous failover can lose acknowledged writes and is unavailable on the old leader's side. Martin Kleppmann's 2015 essay makes this case under the blunt title Please stop calling databases CP or AP.
• The choice isn't global. Brewer's own 2012 retrospective stressed that partitions are rare, that systems can choose differently for different operations and data, and that the interesting work is in how you detect a partition, operate during it, and recover afterwards.

PACELC: the trade-off you make every day

Daniel Abadi pointed out in 2010 that CAP describes a rare situation and ignores a constant one. His formulation: if there's a Partition, choose between Availability and Consistency; Else, even when everything is healthy, choose between Latency and Consistency. Waiting for replicas to agree takes round trips, and round trips take time, on every single request.

The classifications are about defaults. Many databases let you choose per request, which is the real lesson: consistency is a property of an operation, not of a product.

from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement

balance = SimpleStatement(
    "SELECT balance_minor FROM credits WHERE team_id = %s",
    consistency_level=ConsistencyLevel.QUORUM)    # a majority must agree: slower, fresher

activity = SimpleStatement(
    "SELECT * FROM activity WHERE team_id = %s LIMIT 20",
    consistency_level=ConsistencyLevel.ONE)       # any replica: fast, maybe stale


# DynamoDB: eventually consistent by default; opt in per read.
table.get_item(Key={"team_id": team_id}, ConsistentRead=True)

A partition, from the inside

Suppose the invoicing app runs a five-node consensus-backed database across three regions, and the link to one region fails. The two nodes in that region can't reach a majority. They refuse writes, and consistent reads, until the link returns: consistency is kept, and availability is lost for users routed there. The other three nodes form a majority and carry on normally. A leaderless store in the same situation would let both sides keep accepting writes, and reconcile the conflicting versions afterwards: availability is kept, and consistency must be repaired later.

Neither behaviour is right in general. The useful question is, for each operation, which would hurt the business more during a partition: an error, or a wrong answer?

The problem

Several nodes must agree on a value, or on an ordered log of values, despite crashes and lost messages. A solution needs two properties. Safety: nodes never decide differently. Liveness: they eventually decide. In 1985, Fischer, Lynch and Paterson proved that no deterministic algorithm can guarantee both in a fully asynchronous system if even one node may crash. This FLP result doesn't make consensus impossible in practice. It means real algorithms keep safety always, and use timeouts to make progress during the periods when the network behaves, which in the partially synchronous world of chapter 2 is nearly always.

You depend on consensus whenever you use etcd, which stores Kubernetes' state, ZooKeeper, Consul, the metadata layer of Kafka, or distributed databases such as CockroachDB and TiDB, which run a consensus group for every range of data. Since Kafka 4.0 in March 2025, Kafka no longer needs ZooKeeper at all; it runs its own Raft-based metadata quorum, called KRaft.

Majorities

Consensus algorithms decide by majority. With 2f + 1 nodes, any f can fail and the rest still form a majority. And any two majorities of the same group always share at least one node, so two conflicting decisions can never both be approved: some node would have had to vote for both. Three nodes tolerate one failure; five tolerate two. Adding a fourth node to three doesn't help, because a majority of four is three, so it still tolerates only one failure, with more nodes that can fail.

Raft

Leslie Lamport's Paxos, from 1989, was the first practical solution, and is famously hard to understand. Diego Ongaro and John Ousterhout designed Raft in 2014 with understandability as an explicit goal, and it has become the default in new systems. It has three parts.

Leader election. Time is divided into numbered terms, each with at most one leader. A leader sends regular heartbeats. A follower that hears nothing before its election timeout, randomised, for example between 150 and 300 milliseconds in the original paper, so that nodes rarely time out together, starts a new term, votes for itself, and asks the others for votes. Each node grants at most one vote per term, and only to a candidate whose log is at least as up to date as its own. A candidate with votes from a majority becomes leader. If the vote splits, the candidates time out and try again in a new term.

Log replication. Clients send commands to the leader, which appends them to its log and sends them to followers. Each message includes the index and term of the entry just before the new ones, and a follower rejects entries that don't follow on from what it has, so the leader backs up until the logs match and then overwrites any divergent tail. An entry is committed once the leader has stored it on a majority, and only then is it applied to the state machine and acknowledged to the client. Every node applies the same commands in the same order, so every node reaches the same state. That's a replicated state machine.

Safety. The voting rule is what ties it together. A committed entry is on a majority; a new leader needs votes from a majority; the two majorities overlap; and no node votes for a candidate whose log is behind its own. So every new leader already holds every committed entry.

def log_is_up_to_date(cand_last_term, cand_last_index, my_last_term, my_last_index):
    if cand_last_term != my_last_term:
        return cand_last_term > my_last_term      # a later term wins, whatever the length
    return cand_last_index >= my_last_index       # same term: the longer log wins


def handle_request_vote(state, req):
    if req.term < state.current_term:
        return VoteReply(term=state.current_term, granted=False)   # a stale candidate
    if req.term > state.current_term:
        state.current_term, state.voted_for = req.term, None       # a new term: step down
        state.role = "follower"
    already_voted = state.voted_for not in (None, req.candidate_id)
    if not already_voted and log_is_up_to_date(
            req.last_log_term, req.last_log_index,
            state.last_log_term(), state.last_log_index()):
        state.voted_for = req.candidate_id
        state.persist()                  # the vote must survive a crash before we reply
        state.reset_election_timer()
        return VoteReply(term=state.current_term, granted=True)
    return VoteReply(term=state.current_term, granted=False)

Note the persist() before replying. If a node could forget its vote in a crash, it might vote twice in the same term, and two leaders could be elected. Details like this are why the Raft paper's one-page summary of the rules, its Figure 2, is worth reading line by line.

What consensus is for, and what it costs

Use a consensus-backed store for the small amount of data that must be agreed exactly: who the leader is, which node owns which partition, configuration, locks and leases (chapter 9). Every committed write costs at least one round trip to a majority, plus a disk write on each, so consensus groups are small, usually three or five nodes, and are kept out of the high-volume data path unless, as in distributed databases, the design shards data across many independent groups.

Across regions, that round trip is tens or hundreds of milliseconds, which is why globally consistent databases feel slower than local ones: that's PACELC's latency cost, made concrete.

Why distributed locks exist

You often need exactly one of something: one worker processing a particular job, one instance running the nightly scheduler, one node acting as primary for a shard. On a single machine, a mutex does this. Across machines, you need a lock that survives the holder crashing, so it must expire. A lock with an expiry is a lease: granted for, say, ten seconds, kept alive by renewing it, and automatically released if the holder stops renewing, so a dead holder can't block everyone forever.

The problem with expiry

A lease protects you only if the holder notices when it has expired. It often can't. A process can be frozen by a long garbage-collection pause, by its virtual machine being suspended and migrated, by heavy swapping, or simply by being descheduled, and then resume with no idea that time has passed. Network packets can be delayed. Clocks can jump. From the frozen process's point of view, it checked the lease a moment ago and it was valid.

t = 0 s    Client 1 acquires the lease, valid until t = 10 s
t = 1 s    Client 1 is frozen by a long garbage-collection pause
t = 10 s   The lease expires
t = 11 s   Client 2 acquires the lease and writes to the file
t = 15 s   Client 1 resumes, still believes it holds the lease, and writes too
           Two writers, one file: corrupted

No amount of care by Client 1 fixes this, because the pause can happen between any check and the write that follows it. The protection has to live in the resource.

Fencing tokens

Have the lock service hand out a number that increases every time the lease is granted: 33 to Client 1, 34 to Client 2. Every write to the protected resource carries the holder's token, and the resource remembers the highest token it has seen and rejects anything lower. When Client 1 wakes and writes with token 33, the storage has already seen 34, and refuses.

class StaleLeaseholder(Exception):
    pass


class FencedStore:
    def __init__(self):
        self.highest_token = 0
        self.data = {}

    def write(self, key: str, value, token: int) -> None:
        if token < self.highest_token:
            raise StaleLeaseholder(f"token {token} is older than {self.highest_token}")
        self.highest_token = token
        self.data[key] = value

In a database, the same check is a condition on the write itself, such as UPDATE shard_owner SET … WHERE shard = %s AND epoch <= %s. The tokens must come from something linearizable: a consensus-backed store such as etcd, whose revision numbers only ever increase, ZooKeeper's versions, or a database sequence. Google's Chubby lock service, described in 2006, called them sequencers; primary-election schemes call them epochs or generation numbers. The idea is the same.

Efficiency or correctness?

Martin Kleppmann's 2016 essay How to do distributed locking made a distinction worth remembering. Ask what happens if the lock fails and two holders run at once.

• If it merely wastes effort, such as two workers generating the same report, a lock is an efficiency optimisation. A simple lease in Redis, with occasional duplicates, is fine.
• If it corrupts data or moves money twice, the lock is protecting correctness, and it needs a consensus-backed service plus fencing tokens checked by the resource.

The same essay criticised Redlock, an algorithm for locking across several Redis nodes, for depending on bounded pauses and clock drift and for producing no fencing token; Redis's creator published a response, and the debate is worth reading for how carefully each side states its assumptions. The practical conclusion hasn't changed: timing assumptions are exactly what real systems violate.

The same bug, at planet scale

On 19 and 20 October 2025, a race of exactly this shape took down a large part of Amazon Web Services' US-East-1 region for around fifteen hours. Two independent automation workers updated the DNS records for DynamoDB's regional endpoint. One checked that its plan was newer than the one already applied, then ran into unusual delays. By the time it applied the plan, its check was stale, and it overwrote a newer plan with an older one; a cleanup process then deleted that old plan, leaving the endpoint with no DNS records at all. Services that depended on DynamoDB, including parts of EC2, failed in cascade. The general lesson is the one in this chapter: a check made at one moment protects nothing at a later moment. The version comparison must happen atomically, at the point of the write.

Better still: don't need a lock

• Conditional writes. Compare-and-set with a version number (The Engineer's Shipyard, chapter 2) makes the database the arbiter, with no lock at all.
• Idempotent operations. If running twice is harmless, two runners are harmless.
• Single-writer ownership. Assign each partition to exactly one consumer, as Kafka's consumer groups do (chapter 11), so concurrent writers can't arise.
• A platform primitive. Kubernetes controllers elect leaders with Lease objects stored in etcd; use what your platform already trusts rather than inventing a scheme.

with distributed_lock(f"invoice:{invoice_id}", ttl=10):
    invoice = db.get_invoice(invoice_id)
    if invoice.status == "sent":
        db.execute("UPDATE invoices SET status = 'paid' WHERE id = %s", (invoice_id,))
        db.insert_payment(invoice_id, amount_minor)

with db.transaction():
    updated = db.execute(
        "UPDATE invoices SET status = 'paid', version = version + 1 "
        "WHERE id = %s AND status = 'sent'",
        (invoice_id,)).rowcount
    if updated:
        db.insert_payment(invoice_id, amount_minor, payment_id=payment_id)

The problem

Upgrading a team's subscription in the grown-up invoicing app touches several services, each with its own database: billing charges the card, entitlements unlocks the features, notifications sends the receipt, and the CRM records the upgrade. No single database transaction covers all four. If the charge succeeds and the entitlement update fails, the customer has paid for nothing. Something has to make the whole operation all-or-nothing, or close enough.

Two-phase commit

The classic protocol has a coordinator and several participants. In the first phase, the coordinator asks each participant to prepare: do all the work, write it durably, take the necessary locks, and vote. A yes vote is a promise: "I can commit, and I will if you tell me to, even if I crash and restart first." In the second phase, if every vote was yes, the coordinator records its decision and tells everyone to commit; if any vote was no, everyone aborts.

That window is the protocol's central weakness: two-phase commit is a blocking protocol. Rows stay locked while participants are in doubt, so other transactions queue behind them, and a coordinator that never recovers leaves them stuck until someone intervenes by hand. It also costs at least two round trips and several durable writes per transaction, and every participant must support it. Your payment provider won't join your transaction, and nor will an email service.

Two-phase commit is still used successfully where those problems are solved. Distributed databases such as Spanner run it between their own shards, with each participant and the coordinator replicated by consensus, so a crashed machine doesn't stop the protocol. Inside one such database, let it do this for you. What you shouldn't do is build it yourself across independent services.

Sagas

A saga, a term from a 1987 paper by Hector Garcia-Molina and Kenneth Salem, replaces one big transaction with a sequence of local transactions, each committed on its own. Every step has a compensating action that semantically undoes it: a refund for a charge, a release for a reservation, a cancellation for a booking. If a step fails permanently, the saga runs the compensations for the steps already completed, in reverse order.

from dataclasses import dataclass
from typing import Callable


@dataclass
class Step:
    name: str
    action: Callable[[dict], None]
    compensate: Callable[[dict], None]


def run_saga(steps: list[Step], context: dict) -> bool:
    completed: list[Step] = []
    for step in steps:
        try:
            step.action(context)
            completed.append(step)
        except PermanentFailure:
            for done in reversed(completed):
                done.compensate(context)     # must be idempotent, and retried until it works
            return False
    return True


upgrade_subscription = [
    Step("record pending upgrade", record_pending_upgrade, cancel_pending_upgrade),
    Step("charge prorated amount", charge_card, refund_charge),
    Step("unlock plan features", apply_entitlements, revert_entitlements),
]

A production orchestrator persists the saga's progress after every step, so a crash mid-saga resumes rather than starting over, and it retries transient failures before declaring a step failed. Workflow engines such as Temporal exist to provide exactly this durability.

Two design rules make sagas manageable.

• Put irreversible steps last. Some actions can't be compensated: an email can't be unsent, a parcel can't be unshipped. Order the saga so that everything before a certain step can be undone, and everything after it is guaranteed to succeed eventually with retries. That step is called the pivot. The confirmation email belongs after it.
• Design for visible intermediate states. A saga isn't isolated: other requests can see the half-finished state, a charge without features, for example. Use explicit pending states that the rest of the system understands, rather than pretending the operation is atomic.

Orchestration or choreography?

In an orchestrated saga, a central coordinator tells each service what to do, as in the code above. The flow is explicit and easy to follow, test and monitor. In a choreographed saga, each service reacts to events from the others: billing emits PaymentCaptured, entitlements reacts and emits PlanUpgraded, and so on. Coupling is lower, but the flow exists only as the sum of many subscriptions, and understanding it means reading all of them. Start with orchestration for anything with more than two or three steps.

The plumbing: outbox and idempotent consumers

Every step of a saga changes a local database and sends a message, which is the dual-write problem from The Engineer's Shipyard: use the transactional outbox. Messages are delivered at least once, so every consumer must be idempotent. The simplest way is a table of processed message ids, written in the same transaction as the effect.

def handle_payment_captured(db, message):
    with db.transaction():
        first_time = db.execute(
            "INSERT INTO processed_messages (message_id) VALUES (%s) ON CONFLICT DO NOTHING",
            (message.id,)).rowcount
        if not first_time:
            return                                # a redelivery: already handled
        db.execute("UPDATE upgrades SET status = 'paid' WHERE id = %s", (message.upgrade_id,))

Pat Helland's essay Life beyond Distributed Transactions argues this is how large systems end up working anyway: entities that are each consistent on their own, communicating by messages, with idempotence and reconciliation instead of global atomicity.

Queues

In a message queue such as RabbitMQ or Amazon SQS, producers add messages and competing consumers take them. Each message goes to one consumer, which acknowledges it when done, and then it's deleted. A consumer that dies without acknowledging has its message redelivered to someone else. Queues are ideal for spreading independent tasks across workers: resize this image, send this email. They make weak promises about order, because several consumers work in parallel and redeliveries jump the queue, though some offer ordered variants, such as SQS FIFO queues with message groups.

Logs

A log, such as Apache Kafka, Redpanda or Amazon Kinesis, is an append-only sequence of messages that isn't deleted when read. Each topic is split into partitions. Within a partition, messages keep their order and are numbered by offset. Consumers don't remove messages; each consumer group records how far it has read, and can rewind to reprocess history. Messages are kept for a configured period, or forever with log compaction, which keeps only the latest message for each key, turning the log into a durable, replayable table.

Within a consumer group, each partition is read by exactly one member, which gives two properties at once: order within a partition, and parallelism across partitions. The number of partitions therefore caps the useful number of consumers. A producer chooses the partition by hashing the message key, so everything about one invoice, keyed by its id, stays in order.

Jay Kreps' 2013 essay The Log argued that this one abstraction, an ordered, replayable record of what happened, unifies databases, replication and data integration. That's why logs sit at the centre of so many architectures, and why the line between queues and logs keeps blurring: Kafka 4.0 added early support for queue-style consumption.

Delivery guarantees

Where a consumer records its progress, relative to doing the work, decides what happens when it crashes.

• At most once: record progress, then process. A crash in between loses the message.
• At least once: process, then record progress. A crash in between processes the message again. This is the normal, sensible default.
• Exactly once: possible only inside a closed system. Kafka's transactions, introduced in 2017, let a consumer read from Kafka, write results to Kafka, and commit its offset, all atomically, so within Kafka nothing is lost or duplicated. The moment processing touches the outside world, such as sending an email or calling an API, you're back to at least once.

What you can always achieve is effectively once: at-least-once delivery plus idempotent processing (chapter 10), so a duplicate changes nothing.

from confluent_kafka import Consumer, Producer

producer = Producer({
    "bootstrap.servers": BROKERS,
    "enable.idempotence": True,     # the broker drops duplicates from producer retries
    "acks": "all",                  # wait for every in-sync replica
})
producer.produce("invoice-events", key=invoice_id, value=json.dumps(event))
producer.flush()

consumer = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "reminder-service",
    "enable.auto.commit": False,    # we decide when a message counts as done
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["invoice-events"])
while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue
    handle(json.loads(msg.value()))  # must be idempotent: a crash below means a replay
    consumer.commit(message=msg)

Operational realities

• Consumer lag, how far a group is behind the newest message, is the metric to alert on. Adding consumers beyond the partition count does nothing; add partitions early, because changing the count later changes which partition each key maps to.
• Poison messages that always fail will block their partition forever. After a few attempts, move them to a dead-letter topic with the error attached, and carry on.
• Hot partitions come from skewed keys, exactly as in chapter 5.
• Schemas are contracts. Many teams consume your events. Register schemas and enforce compatibility rules, as you would for a public API (The Engineer's Shipyard, chapter 3).

Commands and events

A command asks for something to happen: SendInvoice. It's addressed to one handler, named in the imperative, and may be refused. An event records that something happened: InvoiceSent. It's named in the past tense, broadcast to anyone interested, and can't be refused, because it's already true. Keeping the two apart in names and in design avoids a lot of confusion about who is responsible for what.

Four things people mean by "event-driven"

Martin Fowler's 2017 essay separates four patterns that share the name.

• Event notification. A small event says something changed; consumers call back for details. Loose coupling, but the call-back couples them again.
• Event-carried state transfer. The event carries the data, so consumers keep their own copies and never call back. More autonomy and resilience, at the cost of duplicated, eventually consistent data.
• Event sourcing. The events are the system of record.
• CQRS. Separate models for writing and for reading.

The benefits are real. Producers don't wait for consumers. New consumers can be added without touching the producer. A slow consumer doesn't slow the producer. The costs are also real: the flow of the system is no longer visible in any one place, data is eventually consistent everywhere, events become contracts that are hard to change, and debugging a problem means tracing it across services (chapter 20).

CQRS

Command Query Responsibility Segregation separates the model that handles commands, which validates them and enforces the business rules, from one or more models that answer queries, each shaped for the query it serves. The invoicing app's write model is normalised and strict. Its dashboard reads from a denormalised table of totals per team, its search reads from a search index, and finance reads from a warehouse, all kept up to date from the stream of changes. Each read model can use the storage that suits it and can be rebuilt from scratch.

You may already do a light version: a read replica, or a materialised view refreshed from the main tables. Full CQRS is worth it when read and write shapes differ sharply and read load dominates.

Event sourcing

Instead of storing each invoice's current state and overwriting it, store the sequence of events that happened to it, and compute the current state by replaying them. The event store is append-only; state is a fold over history.

from dataclasses import dataclass


@dataclass
class Invoice:
    id: str
    status: str = "draft"
    total_minor: int = 0
    paid_minor: int = 0
    version: int = 0


def apply(invoice: Invoice, event: dict) -> Invoice:
    kind = event["type"]
    if kind == "InvoiceCreated":
        invoice.total_minor = event["total_minor"]
    elif kind == "InvoiceSent":
        invoice.status = "sent"
    elif kind == "PaymentRecorded":
        invoice.paid_minor += event["amount_minor"]
        invoice.status = "paid" if invoice.paid_minor >= invoice.total_minor else "partially_paid"
    elif kind == "InvoiceVoided":
        invoice.status = "void"
    invoice.version += 1
    return invoice


def load(store, invoice_id: str) -> Invoice:
    invoice = Invoice(id=invoice_id)
    for event in store.events_for(invoice_id):
        invoice = apply(invoice, event)
    return invoice


def record_payment(store, invoice_id: str, amount_minor: int) -> None:
    invoice = load(store, invoice_id)
    if invoice.status in ("draft", "void"):
        raise InvalidCommand(f"can't record a payment on a {invoice.status} invoice")
    store.append(invoice_id, {"type": "PaymentRecorded", "amount_minor": amount_minor},
                 expected_version=invoice.version)   # fails if someone appended meanwhile

What you gain: a complete audit trail for free, the ability to answer "what did this invoice look like last Tuesday?", and read models that can be rebuilt, or new ones created, by replaying history, which turns a buggy projection into a fix-and-replay rather than a data repair. The expected_version on append gives optimistic concurrency, exactly as in The Engineer's Shipyard.

What you pay: events are permanent, so every schema mistake lives forever and old versions must be translated as they're read, a technique called upcasting. Aggregates with long histories need snapshots to load quickly. Deleting personal data for privacy laws needs special techniques, such as encrypting each person's data with their own key and destroying the key, called crypto-shredding. And most developers need time to think this way. Event sourcing suits domains where history is the domain: ledgers, bookings, anything audited. For ordinary records, it's usually too much.

Change data capture

A lighter way to get a stream of changes is to read the database's own replication log with a tool such as Debezium and publish every row change. It needs no changes to application code. But raw row changes expose your internal schema to every consumer, so publishing deliberate events through an outbox table, which Debezium can also relay, keeps the contract under your control.

def upcast(event: dict) -> dict:
    if event["type"] == "InvoiceCreated" and event.get("schema", 1) == 1:
        return {
            "type": "InvoiceCreated",
            "schema": 2,
            "total_minor": to_minor_units(str(event["amount"]), "USD"),
            "currency": "USD",              # every version 1 event was in dollars
        }
    return event

Batch processing

A batch job reads a fixed input, such as yesterday's events or a snapshot of a table, and produces an output: a report, a search index, training data. Google's 2004 MapReduce paper made the model famous: a map step transforms each record and emits key-value pairs, the framework groups everything by key across the cluster, called the shuffle, and a reduce step combines each group. Modern engines such as Apache Spark generalise it into arbitrary graphs of operations, and much batch work now happens as SQL inside a data warehouse.

Batch has a wonderful property: inputs are immutable and jobs are deterministic, so recovery is simply running the job again. A bug in yesterday's report is fixed by fixing the code and re-running it.

Event time and processing time

A stream processor works on data that never ends, and must produce results before it has seen everything. Two clocks matter. Event time is when something happened, stamped at the source. Processing time is when the system gets to it. They can differ by milliseconds or by days: a phone that records usage offline uploads it when it finds a signal, and a failed consumer catches up on an hour of backlog. Any result about when things happened, such as invoices sent per hour, must use event time, or it changes depending on how busy the system was.

Windows

To aggregate an unbounded stream, you group it into finite windows of event time.

• Tumbling windows are fixed and consecutive: invoices sent per hour.
• Sliding or hopping windows overlap: payments in the last five minutes, updated every minute.
• Session windows group bursts of activity separated by quiet gaps: one user's visit to the dashboard.

Watermarks and late data

When is the window from 10:00 to 11:00 complete? Never, strictly: an event from 10:59 might still be on a phone somewhere. A watermark is the system's running estimate that no more events older than some time T will arrive. When the watermark passes the end of a window, the window's result is emitted. An event that arrives after its window was emitted is late, and must be dropped, sent to a side output for correction, or used to update the published result. Google's 2015 Dataflow paper, the basis of Apache Beam, framed this as a three-way trade-off between correctness, latency and cost, which you choose per pipeline.

from collections import defaultdict


class TumblingCounter:
    """Counts events per window of event time, tolerating stragglers."""

    def __init__(self, window: int = 60, allowed_lateness: int = 30):
        self.window, self.lateness = window, allowed_lateness
        self.open = defaultdict(int)       # window start -> running count
        self.newest = 0                    # highest event time seen so far
        self.late = []                     # events that arrived after their window closed

    def watermark(self) -> int:
        return self.newest - self.lateness     # we expect nothing older than this

    def add(self, event_time: int) -> list[tuple[int, int]]:
        start = event_time - event_time % self.window
        if start + self.window <= self.watermark():
            self.late.append(event_time)       # its window has already been emitted
            return []
        self.open[start] += 1
        self.newest = max(self.newest, event_time)
        return self._emit_closed()

    def _emit_closed(self) -> list[tuple[int, int]]:
        closed = sorted(s for s in self.open if s + self.window <= self.watermark())
        return [(s, self.open.pop(s)) for s in closed]

This watermark is the simplest kind: "the newest event minus a fixed allowance". Real engines compute watermarks from their sources, and can emit early, speculative results and then refine them as late data arrives.

State and exactly-once in streams

Stream processors keep state: counts, windows, joins. To survive crashes, engines such as Apache Flink periodically snapshot every operator's state together with the input offsets it corresponds to, using a variant of the Chandy–Lamport distributed snapshot algorithm. After a crash, everything rolls back to the last snapshot and replays from those offsets. That gives exactly-once updates to the state. Results written to the outside world are exactly-once only if the sink is transactional or idempotent, which is the lesson of chapter 11 again.

Do you need a stream processor?

Probably not yet. Many "real-time" needs are satisfied by a scheduled query every five minutes over a warehouse fed by change data capture. Reach for stream processing when freshness is worth the operational cost: fraud detection that must act within seconds, alerting, live operational dashboards, or feeding derived data to user-facing features. Architectures once kept a batch layer and a streaming layer side by side, called the lambda architecture; many now use a single streaming path with replay from the log for recomputation, called the kappa architecture.

How failures cascade

A dependency gets slow. Every caller's requests to it now take longer, so each caller holds more threads and connections open at once. Callers run out of them and slow down in turn, and their callers pile up behind them. Within minutes, one slow database makes the whole product unresponsive, including features that never touch it. Slowness spreads further than failure does, because a fast error releases resources and a slow success holds them.

Timeouts and deadlines

Every call to another machine needs a timeout (chapter 2). Better still, pass a deadline down the chain of calls: not "each hop gets five seconds" but "the user's request must finish by 12:00:03.200". Each service subtracts its own work and passes the remainder on, and anything that would finish after the deadline is abandoned rather than wasted. gRPC propagates deadlines this way by default; with HTTP, pass the remaining budget in a header.

Retries, with a budget

The previous volumes covered retrying transient failures with exponential backoff and jitter. At the scale of many services, two further rules matter.

• Retry at one layer. If four layers each make up to four attempts, one user request can become 4 × 4 × 4 × 4 = 256 requests to the bottom service, arriving precisely when it's already failing.
• Use a retry budget. Allow retries only while they're a small fraction of traffic, for example no more than 10% of requests from a client. When the dependency is healthy, occasional failures get retried; when it's down, retries stop adding load. Google's SRE book describes this as a per-client retry budget.

Circuit breakers

Michael Nygard's Release It! popularised the circuit breaker. It watches calls to a dependency. While things are healthy, the circuit is closed and calls pass through. After enough consecutive failures it opens, and calls fail immediately without touching the dependency, which gets room to recover while callers stop wasting threads on it. After a cooldown it lets a probe call through, the half-open state: success closes the circuit, failure opens it again.

import time


class CircuitOpen(Exception):
    pass


class CircuitBreaker:
    def __init__(self, failure_threshold=5, cooldown_s=30.0, clock=time.monotonic):
        self.failure_threshold, self.cooldown_s, self.clock = failure_threshold, cooldown_s, clock
        self.failures = 0
        self.opened_at = None                       # None means the circuit is closed

    def call(self, fn, *args, **kwargs):
        if self.opened_at is not None and self.clock() - self.opened_at < self.cooldown_s:
            raise CircuitOpen()                     # fail fast; don't pile on
        # Closed, or cooling down is over (half-open): let this call through.
        # A production breaker admits one probe at a time while half-open.
        try:
            result = fn(*args, **kwargs)
        except Exception:
            self.failures += 1
            if self.opened_at is not None or self.failures >= self.failure_threshold:
                self.opened_at = self.clock()       # trip, or re-trip after a failed probe
            raise
        self.failures, self.opened_at = 0, None     # success closes the circuit
        return result

When the circuit is open, have a fallback ready: cached data, a default, a degraded feature, or a clear error. An open breaker with no fallback just moves the failure.

Bulkheads, load shedding and backpressure

• Bulkheads, named after a ship's watertight compartments, give each dependency its own pool of threads or connections. A slow PDF service can exhaust its own pool, and invoice pages keep working.
• Load shedding means refusing work you can't finish, early and cheaply, with 503 and Retry-After, rather than accepting it and timing out later. Shed by priority: keep payments and sign-ins working, drop analytics and exports first. Adaptive concurrency limits, which adjust the number of requests in flight based on observed latency, do this automatically.
• Backpressure means a slow consumer pushes back on its producer instead of silently buffering. Every queue in the system should be bounded. An unbounded queue turns overload into ever-growing latency and, eventually, running out of memory.

Hedged requests

Jeff Dean and Luiz André Barroso's 2013 article The Tail at Scale described a cheap cure for slow outliers. Send a request to one replica; if it hasn't answered within the 95th-percentile latency, send the same request to a second replica, and use whichever answers first. Only about 5% of requests are sent twice, and the slowest requests become dramatically faster; in their measurements on Bigtable, the 99.9th-percentile latency fell from over a second to well under a hundred milliseconds. Use it only for idempotent reads, and cancel the loser.

Metastable failures

Sometimes a system stays down after the cause has gone. A 2021 paper by Nathan Bronson and colleagues named these metastable failures. A brief trigger, such as a load spike or a cache node restarting, pushes the system into overload, and something then sustains the overload even after the trigger ends: retries from timed-out clients, a cache that can't refill because the database is too busy to answer, or work queued for requests nobody is waiting for any more. The system is stable in a bad state. Recovery needs a deliberate cut in load, by shedding traffic, disabling retries or draining queues, below the level that triggered it, then bringing traffic back gradually.

Where balancing happens

A layer 4 load balancer works on connections: it picks a backend for each TCP connection and forwards bytes without understanding them. It's fast and simple, and it balances connections, not requests, which matters when one long-lived connection carries thousands of requests. A layer 7 balancer understands HTTP: it can route by path or header, balance each request separately, retry, and terminate TLS. Most application traffic goes through layer 7 proxies, such as Envoy, NGINX or a cloud load balancer, or through a sidecar proxy in a service mesh.

Choosing a backend

• Round robin takes turns. Fine when every request costs about the same.
• Least outstanding requests sends each request to the backend with the fewest in flight, which adapts to slow requests and slow machines.
• Consistent hashing sends the same key to the same backend, for cache affinity, at the cost of hot spots (chapter 5).
• The power of two choices: pick two backends at random, and send to the less loaded of the two. Michael Mitzenmacher showed that this tiny change makes the busiest backend's load dramatically closer to the average than purely random choice, and it avoids the stampede that happens when every balancer independently sends everything to the one "least loaded" backend. Envoy's least-request balancing uses it.

import random


def max_load(servers: int, requests: int, choose, rng) -> int:
    load = [0] * servers
    for _ in range(requests):
        load[choose(load, rng)] += 1
    return max(load)


def pick_random(load, rng):
    return rng.randrange(len(load))


def pick_two(load, rng):
    a, b = rng.randrange(len(load)), rng.randrange(len(load))
    return a if load[a] <= load[b] else b


rng = random.Random(1)
max_load(100, 10_000, pick_random, rng)   # typically 120 to 135: some servers 30% over the mean
max_load(100, 10_000, pick_two, rng)      # typically 101 or 102: almost perfectly even

Health: active, passive and gray

Active health checks probe each backend on a schedule and remove those that fail. Passive checks, called outlier detection in Envoy, watch real traffic and eject a backend whose error rate or latency is far worse than its peers', which catches the gray failures of chapter 2 that pass every probe. Use both, and cap how many backends can be ejected at once, so a problem that affects every backend, such as a bad dependency, doesn't make the balancer eject all of them.

Service discovery

Instances come and go constantly, so clients must look them up. The options range from DNS, which is simple but cached, sometimes for longer than you'd like; to a registry such as Consul or etcd that clients query or subscribe to; to platform-managed discovery such as Kubernetes Services; to a service mesh, where sidecar proxies handle discovery, balancing, mutual TLS and retries. A mesh is convenient, but its retries are one more layer of retries, so configure them with chapter 14 in mind.

Graceful shutdown

Every deploy removes instances, and doing it carelessly drops requests. The order matters: stop advertising the instance as ready, wait long enough for balancers to notice and stop sending new work, finish the requests already in flight within a deadline, close connections to databases and queues, and only then exit.

Containing the blast radius

The most important architectural question about failure isn't whether something will break, but how much breaks with it. Four techniques limit the damage.

• Zones. Run in at least three availability zones. Amazon's Builders' Library describes static stability: provision enough capacity in the surviving zones in advance that losing a whole zone needs no new machines and no action by the control plane, which may itself be struggling during the incident.
• Cells. Split the whole stack, services and data alike, into independent copies called cells, each serving a subset of customers. A bad deploy, a poison request or a corrupted database affects one cell.
• Shuffle sharding. Give each customer a random combination of a few workers from a larger pool. Two customers rarely share their whole combination, so a customer whose requests crash workers takes down its own shard while almost everyone else keeps at least one healthy worker.
• Staged rollouts, one cell or zone at a time, as in The Engineer's Shipyard, chapter 17.

And when parts are down, degrade gracefully: serve cached or slightly stale data, hide recommendations, switch to read-only mode, queue writes for later. A product that's partly working is very different, for its users, from one that's down.

Where caches sit

Between a user and your database there may be five caches: the browser's, a CDN at the edge, an in-process cache inside each application server, a shared cache such as Redis or Memcached, and the database's own buffer pool. Each one closer to the user is faster and serves more traffic, and is harder to invalidate. This chapter is mostly about the shared cache, where the distributed problems live.

Reading and writing through a cache

The common pattern is cache-aside: on a read, try the cache; on a miss, read the database and store the result. The alternatives are read-through, where the cache itself loads from the database, write-through, where writes go to the cache and the database together, and write-behind, where the cache acknowledges writes and flushes them to the database later, which is fast and loses data when the cache fails.

With cache-aside, when data changes, delete the cache entry rather than setting it to the new value. Two writers that each set the cache can finish in the opposite order to their database writes and leave the older value cached. Deletes don't have that problem, and Facebook's Memcache paper chose them for the same reason: a delete can be repeated safely. Deleting isn't enough on its own, though, as the figure shows.

Leases

The Memcache paper, from Facebook in 2013, fixes this with leases. On a miss, the cache hands the reader a token for that key. A delete for the key cancels any outstanding token, and a set is accepted only with a token that's still valid. In the figure, the delete in step 4 cancels the reader's token, so step 5 is rejected and the next reader loads the new price.

The same tokens solve a second problem. The cache issues a token for a given key at most once every few seconds, ten by default in the paper, and tells other readers who miss in the meantime to wait briefly and try again. So when a popular key is deleted, one reader goes to the database while the rest wait a few milliseconds for it to refill the cache, instead of all of them querying the database at once.

Stampedes

That rush of simultaneous misses is a cache stampede, also called a thundering herd or dog-pile. It happens whenever a hot entry expires or is deleted, and it can overload the database at exactly the moment the cache has nothing to give it. There are four standard defences, and they combine well.

• Coalesce requests. Within one process, let the first miss for a key do the work and have concurrent misses wait for its result. Go's singleflight package does exactly this. Across processes, leases or a short lock in the cache do the same.
• Serve stale while refreshing. Keep the expired value, return it immediately and refresh in the background. HTTP has the same idea built in: RFC 5861's stale-while-revalidate, plus stale-if-error for serving the old copy when the origin is failing.
• Refresh early, at random. Let a request occasionally refresh a value shortly before it expires, with a probability that rises as expiry approaches, so one request refreshes it and the others never miss.
• Add jitter to TTLs, so that entries written together, for example after a deploy or a bulk import, don't all expire together.

import math
import random
import time


def xfetch(cache: dict, key, recompute, ttl_s: float, beta: float = 1.0,
           now=time.monotonic, rand=random.random):
    """Refresh a cached value early, with a probability that rises near its expiry."""
    entry = cache.get(key)
    if entry is not None:
        value, delta, expires_at = entry
        # log() of a number in (0, 1] is at most zero, so this acts as if it were a little
        # later than it is: a random head start, scaled by delta, the recompute time.
        if now() - delta * beta * math.log(1.0 - rand()) < expires_at:
            return value
    start = now()
    value = recompute()
    delta = now() - start
    cache[key] = (value, delta, now() + ttl_s)
    return value

This is XFetch, from a 2015 paper by Andrea Vattani, Flavio Chierichetti and Keegan Lowenstein. With a recomputation that takes 200 ms, a single request made one second before expiry refreshes the value with a probability under 1%; half a second before, about 8%; 200 ms before, 37%; 50 ms before, 78%. On a key read hundreds of times a second, some request almost certainly refreshes it just before it expires, while every other request keeps reading the cached value. Raising beta above 1 refreshes earlier.

When the cache becomes load-bearing

A cache with a 97% hit rate means the database sees only 3% of reads, and the database gets sized for that. Now the arithmetic runs the other way. If the hit rate falls to 90%, the database's load more than triples; if the cache is emptied by a restart, a flush or a change to the key format, the load rises more than thirtyfold. The database was never built for that, so it falls over, and the cache can't refill from a database that's down. This is the metastable failure of chapter 14 in its most common form.

• Protect the database with a limit on concurrent queries from cache misses, so a cold cache causes slow responses rather than a dead database.
• Warm up gradually. Facebook let a new, empty cache cluster read from a warm cluster's cache instead of from the database until its hit rate recovered. Restart cache nodes one at a time, and never flush everything at once.
• Replace failed nodes carefully. Rehashing a failed node's keys onto the survivors can overload them too. The Memcache paper sent the failed node's traffic to a small, separate pool called Gutter, whose entries expire quickly, instead.
• Change key formats with a version prefix and a gradual rollout, never all at once.

Hot keys and missing keys

A hot key, such as a celebrity's profile or the home page's configuration, can receive more traffic than one cache node can serve, and consistent hashing puts every request for it on that one node. Keep very hot keys in a small in-process cache with a short TTL, or store several copies under different suffixes, such as home:1 to home:8, and have readers pick one at random.

At the other extreme, requests for keys that don't exist miss the cache every time and go to the database, which a scraper or an attacker probing random IDs will do deliberately. Cache "not found" results too, with a short TTL, or keep a Bloom filter of the keys that do exist and turn away the rest before they reach the database.

Invalidating across machines and regions

An application that writes to the database and then deletes the cache entry can crash between the two steps, and a region that deletes its own cache entry can't easily reach the caches in other regions. Facebook's answer was to drive invalidations from the database's commit log: a daemon beside each database reads committed changes and sends the matching deletes to the caches in its region. That's change data capture, from chapter 12, applied to caches. Invalidations then happen only after the change is committed, and survive application crashes. In other regions the same daemons run beside the database replicas, so caches there are invalidated only once the change has replicated, and no reader can refill them with the old value from a replica that hasn't caught up. Whatever the mechanism, give every entry a TTL, as the safety net that eventually corrects whatever the invalidations missed.

def update_price(product_id, new_price):
    db.execute("UPDATE products SET price = %s WHERE id = %s", (new_price, product_id))
    cache.set(f"price:{product_id}", new_price, ttl=3600)

The main styles

What distribution costs

Everything in the first sixteen chapters is the price of putting a network between two pieces of code. A function call that always worked becomes a request that can fail, time out or succeed without telling you (chapter 2). Availabilities multiply along a chain of calls (chapter 1). A join becomes an API call, and a transaction becomes a saga (chapter 10). Data gets copied between services and is eventually consistent (chapter 12). Every interface between services needs versioning, as in The Engineer's Shipyard, chapter 3, and every service needs its own deploys, dashboards, alerts and on-call rotation.

What it buys

In return, each service can be deployed without coordinating with the others, scaled on its own, written in a different language, and isolated for security or compliance, so that only the payment service, for example, is in scope for a card-data audit. Nearly all of these benefits are really about people: microservices are mainly a way for many teams to work without waiting for each other. A team of eight rarely has that problem, and pays the full cost anyway.

Conway's law

In 1968 Melvin Conway observed that the systems an organisation designs end up mirroring its communication structure: four teams building a compiler produce a four-pass compiler. You can resist this or use it. The "inverse Conway manoeuvre" shapes the teams to match the architecture you want. Team Topologies, by Matthew Skelton and Manuel Pais, builds on the idea: most teams should be stream-aligned, owning one slice of the product end to end, supported by platform teams that make shared infrastructure self-service, with service boundaries drawn so that each team can hold its part in its head.

The modular monolith

Martin Fowler observed that almost every successful microservice system he'd heard of started as a monolith that grew too big, while systems built as microservices from scratch often got into serious trouble. The practical middle ground is a modular monolith: one deployable, divided into modules that each own their tables and expose a small public interface, with no reaching into another module's internals. Shopify took this route with its very large Rails application, and built a tool, Packwerk, to enforce the boundaries. If a module later needs to become a service, its boundary is already there.

The boundaries only stay real if a machine checks them. Tools exist for most languages, such as import-linter for Python, Packwerk for Ruby, ArchUnit for Java and dependency-cruiser for JavaScript, but the idea fits in a page.

import ast
import pathlib
import sys

# Each package under app/ is a module. This lists which others each may use.
ALLOWED = {
    "billing": {"customers"},
    "customers": set(),
    "notifications": {"billing", "customers"},
}


def imports(path: pathlib.Path):
    tree = ast.parse(path.read_text(), filename=str(path))
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                yield alias.name, node.lineno
        elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
            for alias in node.names:
                yield f"{node.module}.{alias.name}", node.lineno


def violations(root: pathlib.Path):
    for path in sorted(root.rglob("*.py")):
        rel = path.relative_to(root).parts
        if len(rel) < 2:
            continue
        owner = rel[0]
        for name, line in imports(path):
            parts = name.split(".")
            if parts[0] != root.name or len(parts) < 2 or parts[1] == owner:
                continue
            target = parts[1]
            if target not in ALLOWED.get(owner, set()):
                yield f"{path}:{line}: {owner} may not depend on {target}"
            elif len(parts) > 2 and parts[2] != "api":
                yield f"{path}:{line}: {owner} imports {name}; use {target}.api"


if __name__ == "__main__":
    found = list(violations(pathlib.Path("app")))
    print("\n".join(found) or "module boundaries respected")
    sys.exit(1 if found else 0)

# app/billing/models.py:1: billing imports app.customers.models.Customer; use customers.api
# app/customers/api.py:1: customers may not depend on billing

The distributed monolith

The worst outcome is services that can't change independently: they share a database and read each other's tables, they must be deployed together, or one request passes through a long chain of synchronous calls. That's all the cost of distribution with none of the independence. If two services always change together, they're one service.

How services talk

Synchronous calls, over HTTP or gRPC, are simple to write and reason about, but they couple the caller's availability and latency to the callee's. Asynchronous messages and events, through a queue or a log (chapter 11), decouple services in time, so the consumer can be down for an hour without the producer noticing, at the cost of eventual consistency and harder debugging. Use synchronous calls when the caller needs the answer to continue, and events for everything that can happen afterwards. When a service keeps needing another's data, it can keep its own read-only copy, updated from that service's events, instead of calling it on every request.

Serverless, and choosing the grain

Serverless functions remove servers from your concerns and cost nothing when idle, which suits spiky and event-driven work. They have cold starts, time limits of minutes rather than hours, and a per-call price that can exceed a server's at steady high load. In 2023, Prime Video's engineers described moving one monitoring service, which checks video streams for defects, from serverless functions orchestrated by AWS Step Functions, passing video frames between steps through S3, into a single process that passes them in memory. Its infrastructure cost fell by over 90%. The lesson isn't that one style is right, but that the split must match how data moves: steps that exchange large amounts of data many times a second belong in one process.

Migrating with a strangler fig

Rewriting a large system from scratch rarely works. Fowler's strangler fig, named after a vine that grows around a tree until it replaces it, migrates one piece at a time: put a routing layer in front of the old system, rebuild one capability in the new one, send that capability's traffic there, and delete the old code. Each step is small and reversible. Running the new code in shadow, on copies of real traffic with its results compared but not used, catches differences before any user sees them.

Start from the domain

Eric Evans's Domain-Driven Design, published in 2003, starts from an observation: in most business software, the hard part isn't the technology but the domain, meaning the rules of lending, logistics, insurance or whatever the software is for. The design has two halves. Strategic design finds the large boundaries in the business. Tactical design is a set of patterns for modelling inside one boundary. The strategic half matters most for distributed systems, because it tells you where the seams are.

A shared language

Developers and domain experts should use the same words for the same things, in conversation, in documents and in the code. Evans calls this the ubiquitous language. If the business says "the account is on hold" and the code says user.status_flag = 3, every conversation needs a translation, and translations are where misunderstandings get in. When the code uses the experts' words, an expert can read a test and say whether it's right.

Bounded contexts

A shared language breaks down at the scale of a whole company, because the same word means different things in different departments. To sales, a "customer" is a lead with a stage in the pipeline; to billing, a legal entity with a tax ID and a payment method; to support, a person with open tickets. One Customer class serving all three becomes a huge object that every team edits and no team owns.

A bounded context is a boundary within which one model applies and each word has one meaning. Sales, billing and support each have their own customer model, sharing only an identifier, and translate explicitly where they meet. Bounded contexts are the best candidates for the modules and services of chapter 17, because a boundary drawn where the language changes is one the business already respects.

How contexts relate

Evans's context map names the relationships between contexts, and naming them makes the politics visible.

• Customer and supplier. The upstream team plans its changes around the downstream team's needs.
• Conformist. The downstream team simply adopts the upstream model, typically because it has no influence over it.
• Anticorruption layer. The downstream team translates the upstream model into its own at the boundary, so a legacy system's or a vendor's quirks don't leak into its code. This is the right default for any third-party API.
• Open host service and published language. The upstream team offers one well-documented protocol for everyone, instead of a custom integration per consumer.
• Shared kernel. Two teams co-own a small piece of the model. Keep it very small, since every change needs both teams.

Where to invest

DDD sorts the parts of a business into three kinds of subdomain. The core is what makes this business different from its competitors, and deserves your best engineers and your most careful modelling. Supporting subdomains are specific to the business but don't set it apart, so build them simply. Generic subdomains, such as authentication, payments and email, are the same everywhere, so buy them or use open source. Much wasted effort goes into building a generic subdomain lovingly while the core is neglected.

Event storming

Alberto Brandolini's event storming workshop is a fast way to find the contexts. Domain experts and engineers stand at a long wall and write domain events, things that happened, in the past tense, such as "Order placed" or "Payment failed", on sticky notes, in time order. Then they add the commands that cause each event, the people who issue them, the policies that react ("whenever a payment fails, notify the customer"), and the open questions. Boundaries show up where the vocabulary changes and where work passes from one department to another.

Inside a context: the tactical patterns

• Entities have an identity that persists as their attributes change, such as a customer or an order.
• Value objects are defined entirely by their value and are immutable, such as an amount of money, an address or a date range. Two equal amounts are interchangeable.
• Aggregates are clusters of objects that are changed together, as one unit, through a single root object. The aggregate is the boundary of consistency: its rules hold at the end of every transaction.
• Domain events record that something the experts care about has happened.
• Repositories load and save whole aggregates, so the rest of the code never handles half of one.

from dataclasses import dataclass
from decimal import Decimal


@dataclass(frozen=True)
class Money:                              # value object: immutable, equal by value
    amount: Decimal
    currency: str

    def __add__(self, other: "Money") -> "Money":
        if other.currency != self.currency:
            raise ValueError("cannot add different currencies")
        return Money(self.amount + other.amount, self.currency)

    def times(self, n: int) -> "Money":
        return Money(self.amount * n, self.currency)


@dataclass(frozen=True)
class OrderPlaced:                        # domain event, named in the past tense
    order_id: str
    total: Money


class Order:                              # aggregate root: every change goes through it
    MAX_LINES = 50

    def __init__(self, order_id: str, currency: str):
        self.id, self.currency = order_id, currency
        self.lines: list[tuple[str, int, Money]] = []
        self.status = "draft"
        self.events: list[OrderPlaced] = []

    def add_line(self, sku: str, qty: int, unit_price: Money) -> None:
        if self.status != "draft":
            raise ValueError("order can no longer be changed")
        if qty <= 0 or len(self.lines) >= self.MAX_LINES:
            raise ValueError("invalid line")
        self.lines.append((sku, qty, unit_price))

    def total(self) -> Money:
        total = Money(Decimal("0"), self.currency)
        for _, qty, price in self.lines:
            total = total + price.times(qty)
        return total

    def place(self) -> None:
        if self.status != "draft" or not self.lines:
            raise ValueError("only a draft with lines can be placed")
        self.status = "placed"
        self.events.append(OrderPlaced(self.id, self.total()))

# The repository saves the order and its new events in one transaction (the outbox of
# chapter 10). Payment and stock are separate aggregates, which react to OrderPlaced.

Adding a line to a placed order, or adding euros to dollars, fails inside the model, wherever the call comes from. The rules live in one place instead of being re-checked, or forgotten, in every handler.

Aggregates in a distributed system

Aggregates matter far beyond object modelling, because an aggregate is the natural unit for almost everything in the earlier chapters. It's the unit of a transaction, and of optimistic locking with a version number. It's the natural partition key (chapter 5), since everything one transaction touches then lives on one shard. It's one stream in an event-sourced system (chapter 12). Rules that span aggregates, such as "reserve stock when an order is placed", are enforced eventually, through domain events, sagas and the outbox of chapter 10. Vaughn Vernon's advice for designing them is short: keep aggregates small, refer to other aggregates by ID, change one aggregate per transaction, and use eventual consistency between them.

When not to use it

An application that mostly stores and displays records, with few rules, doesn't need aggregates and repositories, and neither do generic subdomains. The tactical patterns are worth their weight only where the rules are complex. The strategic ideas, a shared language and clear contexts, are worth using almost everywhere.

The method

Whether it's a design review at work or a system design interview, the same steps work, in this order.

1. Clarify the requirements. The functional ones, what the system does, and the non-functional ones: scale, latency, availability, consistency, durability, cost and compliance. Ask which operations matter most, and which failures are unacceptable.
2. Estimate. Requests per second at average and peak, data size and growth, bandwidth. The numbers decide which problems you actually have.
3. Define the interface and the data. The main API calls and the main entities, with their access patterns: what's looked up by what.
4. Sketch the simplest design that meets the numbers. Often that's one application tier, one database and a cache.
5. Find what breaks, and deepen there. The bottleneck, the hot spot, the consistency requirement, the operation that can't be lost. This is where the earlier chapters come in.
6. Plan for failure and operations. For each component: what happens when it's slow, down or wrong? How will you know? How do you deploy it?
7. State the trade-offs. Every choice gives something up. Say what, and why that's acceptable here.

Estimating

Estimates only need to be right to within a factor of a few, which makes them quick. A day has 86,400 seconds, so a million events a day is about 12 a second. Peaks run at two to ten times the average. Here's an invoicing platform that sends 60 million invoices a month, three emails per invoice, and keeps a 150 KB PDF of each invoice for seven years, with a peak of five times the average.

The write rate is small, and one database handles it easily, so there's no case for sharding the invoice data. The email rate needs a queue and a provider, nothing heroic. The number that shapes the design is storage: three-quarters of a petabyte of PDFs, before replication. That belongs in object storage, with lifecycle rules that move old files to cheaper tiers, and it raises a question for the business: must every PDF be kept, or can it be regenerated from the invoice data when someone asks for it?

A worked design: a distributed rate limiter

Requirements. A public API runs on 40 servers behind a load balancer. Each API key may make 100 requests a second on average, with bursts of up to 200. The check must add at most a millisecond or two, should be accurate to within a few percent, and must not take the API down if it fails.

Estimate. At a peak of 50,000 requests a second, that's 50,000 checks a second, each a small read and write on one key: comfortably within a small Redis cluster, with keys spread across shards.

Where the state lives. Each server could enforce its own share, 100 ÷ 40 = 2.5 requests a second per key, with no network call at all. But that's wrong whenever a client's traffic is spread unevenly across servers, which is common with long-lived connections, and the share changes whenever the fleet scales. A central store is accurate, at the cost of one round trip, about half a millisecond within a data centre, and of a new dependency.

The algorithm. A fixed window counter, one counter per key per second, is simplest, but allows twice the limit across a window boundary. A sliding log of timestamps is exact but stores every request. A token bucket stores two numbers per key and expresses exactly this requirement: a steady rate plus a burst.

Atomicity. The check reads the bucket, computes and writes it back. If two servers do that concurrently, both can spend the last token, a lost update from chapter 6. Redis runs a Lua script atomically, so the whole check goes into one script, which also uses Redis's clock instead of each server's.

-- token_bucket.lua    KEYS[1]: the bucket    ARGV: rate per second, burst, cost
local rate  = tonumber(ARGV[1])
local burst = tonumber(ARGV[2])
local cost  = tonumber(ARGV[3])

-- Use the Redis server's clock, so API servers with skewed clocks agree (chapter 3).
local t   = redis.call("TIME")
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)

local state  = redis.call("HMGET", KEYS[1], "tokens", "ts")
local tokens = tonumber(state[1]) or burst
local ts     = tonumber(state[2]) or now

-- Refill for the time since the last call; never go backwards if the clock does.
tokens = math.min(burst, tokens + math.max(0, now - ts) / 1000 * rate)

local allowed, wait_ms = 0, 0
if tokens >= cost then
  tokens = tokens - cost
  allowed = 1
else
  wait_ms = math.ceil((cost - tokens) / rate * 1000)
end

redis.call("HSET", KEYS[1], "tokens", tokens, "ts", now)
-- A full bucket is the same as no bucket, so let idle keys expire.
redis.call("PEXPIRE", KEYS[1], math.ceil(burst / rate * 1000))
return {allowed, wait_ms}

import redis

r = redis.Redis(host="ratelimit.internal", socket_timeout=0.005)   # 5 ms
token_bucket = r.register_script(open("token_bucket.lua").read())


def check(api_key: str, local_limiter, rate=100, burst=200) -> tuple[bool, int]:
    """Returns (allowed, milliseconds until a token is available)."""
    try:
        allowed, wait_ms = token_bucket(keys=[f"rl:{api_key}"], args=[rate, burst, 1])
        return allowed == 1, wait_ms
    except redis.RedisError:            # slow or down: fail open, up to a local cap
        ok = local_limiter.allow(api_key)
        return ok, 0 if ok else 1000

Deepening. With the basic design in place, the questions that matter are about the edges.

• Abusive clients. A rejected request still costs a round trip to Redis, so a client hammering the API with an exhausted key loads the shard that owns its key. Each server remembers denials locally until wait_ms has passed, and turns those requests away without asking Redis.
• Redis failing. The short timeout bounds the added latency. On an error, the server falls back to its local share of the limit, fails open within that cap, and alerts. An outage of the limiter becomes a period of rough limits, not an outage of the API.
• Failover. Redis replicates asynchronously (chapter 4), so a failover can lose the last few updates and briefly give some clients extra requests. For a rate limiter that's acceptable. For an account balance it wouldn't be, which is why that would need a different design.
• More regions. A single global count would put a cross-region round trip, tens to hundreds of milliseconds (chapter 1), on every request. Give each region its own share of the limit instead, or synchronise regional counts in the background and accept brief overshoot.
• Telling clients. Reject with 429 Too Many Requests and a Retry-After taken from wait_ms, as in The Engineer's Shipyard, chapter 16.

Trade-offs. This design gives up a little accuracy, during failovers and fallbacks, for speed and availability, which is the right choice for a rate limiter and would be the wrong one for a ledger. Saying so explicitly is the last step of the method, and the one most often skipped.

Why ordinary tests miss them

A distributed system's behaviour depends on things a test doesn't control: thread scheduling, network timing, clock readings and random choices. A bug that needs one particular interleaving might show up once in a million runs, in production, and never again while you watch. The first step is making the code controllable. Every example in this book took its clock, and sometimes its random number generator, as a parameter, so that a test could supply fake ones. Do the same with the network: put calls to other machines behind an interface that a test can replace with one that delays, drops, duplicates or reorders.

Test the failure paths

Most tests check that things work. Distributed systems need tests that things fail well: that a timeout is handled, a duplicate is ignored, an out-of-order message is harmless, and a crash between two steps leaves nothing half done. Property-based testing, with a library such as Hypothesis for Python, generates hundreds of inputs and shrinks any failure to the smallest example, which suits properties such as "merging in any order gives the same result" from chapter 4. For integration tests, Testcontainers runs real databases and brokers, and Toxiproxy sits between your service and them to add latency, cut connections or limit bandwidth on demand.

Jepsen: checking what databases promise

Kyle Kingsbury's Jepsen tests a real cluster from the outside. Clients perform operations while a "nemesis" partitions the network, kills and pauses nodes, and skews clocks. Every operation's start, end and result goes into a history, and a checker then searches for an order of the operations that's consistent with the database's claimed consistency model: Knossos for linearizability, Elle for transaction isolation. If no such order exists, the database broke its promise, and the history shows how. Jepsen has found violations in many well-known databases, often where the documentation promised more than the system delivered. Before trusting a database with data that matters, read its Jepsen analysis if one exists.

Deterministic simulation testing

FoundationDB's engineers built their database to run inside a simulator. In simulation, the network, the disks and the clock are fakes, many nodes run in a single thread, and every choice, from message delays to which node crashes next, comes from one random seed. So a whole cluster's run is a function of its seed. The simulator injects faults far more often than reality does and runs time faster than a real clock, so it covers years of bad luck in hours. When an invariant fails, rerunning the same seed reproduces the failure exactly, as many times as debugging needs.

TigerBeetle's simulator, the VOPR, works the same way: a seed and a commit reproduce any failure it finds. Antithesis, a company founded by FoundationDB engineers, runs ordinary software inside a deterministic hypervisor, to get the same property without rewriting it. The core idea fits in a few lines.

import random


def simulate(seed: int, dedupe: bool) -> bool:
    """One run: deliver five payments over a network that duplicates and reorders."""
    rng = random.Random(seed)             # every random choice comes from the seed
    payments = [f"pay_{i}" for i in range(5)]
    in_flight = []
    for p in payments:
        in_flight.append(p)
        if rng.random() < 0.2:            # 20% of messages are delivered twice
            in_flight.append(p)
    rng.shuffle(in_flight)                # and in any order
    applied, seen = [], set()
    for message in in_flight:
        if dedupe and message in seen:
            continue
        seen.add(message)
        applied.append(message)
    return sorted(applied) == payments    # invariant: each payment applied exactly once


failing = [s for s in range(10_000) if not simulate(s, dedupe=False)]
len(failing)                              # 6761: about two runs in three break the invariant
simulate(failing[0], dedupe=False)        # False, and False on every rerun: it replays exactly
all(simulate(s, dedupe=True) for s in range(10_000))   # True

A real simulator runs the system's actual code against simulated networks and disks, not a toy model, but the properties are the same ones shown here: a search over thousands of seeds, a clear invariant, and exact replay of any failure found.

Formal methods: checking the design

Simulation tests the code you've written. TLA+, created by Leslie Lamport, checks a design before you write it. You describe the system's states and the steps it can take, and the TLC model checker explores every reachable state of a small instance, checking your invariants in each one. Engineers at AWS have used it since 2011. On DynamoDB's replication and group-membership system it found three bugs whose shortest demonstrations took up to 35 steps, orderings no test suite would stumble on.

Here is the lost update from chapter 6 as a model: two processes each read a shared counter and then write back one more than they read.

------------------------------ MODULE Counter ------------------------------
EXTENDS Naturals

Procs == {"p1", "p2"}
VARIABLES counter, copy, pc
vars == <<counter, copy, pc>>

Init == /\ counter = 0
        /\ copy = [p \in Procs |-> 0]
        /\ pc = [p \in Procs |-> "read"]

Read(p)  == /\ pc[p] = "read"
            /\ copy' = [copy EXCEPT ![p] = counter]
            /\ pc' = [pc EXCEPT ![p] = "write"]
            /\ UNCHANGED counter

Write(p) == /\ pc[p] = "write"
            /\ counter' = copy[p] + 1
            /\ pc' = [pc EXCEPT ![p] = "done"]
            /\ UNCHANGED copy

AllDone == \A p \in Procs : pc[p] = "done"
Done    == AllDone /\ UNCHANGED vars      \* finished: stutter rather than deadlock

Next == (\E p \in Procs : Read(p) \/ Write(p)) \/ Done
Spec == Init /\ [][Next]_vars

\* The invariant TLC checks in every reachable state.
NoLostUpdate == AllDone => counter = 2
=============================================================================

Checking NoLostUpdate, TLC reports a violation almost immediately and prints the trace: p1 reads 0, p2 reads 0, p1 writes 1, p2 writes 1. The model is small enough to check every ordering, and that's the point: a model of a real protocol, such as a replication scheme or a payment workflow, is usually a few hundred lines, far smaller than the code, and can be checked exhaustively before the code exists. AWS also uses lighter-weight methods on the code itself, such as property-based tests and model checking of real implementations, as it did for a storage node in S3.

Chaos engineering

Netflix's Chaos Monkey, from around 2011, randomly terminated production instances during working hours, so every team had to build services that survived losing a machine. The discipline that grew from it is careful rather than random. Define the system's steady state by a business metric, such as successful checkouts a minute. Form a hypothesis that it holds under some fault. Inject the fault in a controlled way, with a small blast radius and an abort switch, and look for a difference. Game days rehearse larger scenarios, such as losing a zone, with the people and runbooks involved. AWS Fault Injection Service, Gremlin and Chaos Mesh provide the tooling.

Which tool, when

import random
import time

import requests


def send_with_retry(url, payload):
    for attempt in range(5):
        try:
            return requests.post(url, json=payload, timeout=2)
        except requests.RequestException:
            time.sleep(min(30, 2 ** attempt) * random.random())
    raise RuntimeError("gave up")

def send_with_retry(url, payload, post=requests.post, sleep=time.sleep,
                    rand=random.random, attempts=5):
    for attempt in range(attempts):
        try:
            return post(url, json=payload, timeout=2)
        except OSError:                  # requests' exceptions are OSErrors too
            if attempt == attempts - 1:
                raise
            sleep(min(30, 2 ** attempt) * rand())


def test_gives_up_after_five_attempts():
    calls, sleeps = [], []

    def failing_post(url, **kwargs):
        calls.append(url)
        raise ConnectionError("down")

    with pytest.raises(ConnectionError):
        send_with_retry("https://example.test", {}, post=failing_post,
                        sleep=sleeps.append, rand=lambda: 1.0)
    assert len(calls) == 5
    assert sleeps == [1, 2, 4, 8]