Overview

It is 10:00:00 and the stadium show goes on sale. Two million fans press "Hold seat" in the BoxOffice app at roughly the same second, and 60,000 seats have to be shared among them — each seat sold exactly once. One fan, on a train, presses the button, the spinner turns for five seconds, and the app shows "Something went wrong, try again". She presses again. Did the first request hold a seat? Is she about to hold two? Will she be charged twice?

On a single machine that question would not exist: the function either returned or threw. BoxOffice, however, runs on dozens of machines — gateways, several replicas of the inventory service, a replicated PostgreSQL, Redis, a Kafka cluster — and the only thing connecting them is a network that can lose, delay and duplicate messages. That makes BoxOffice a distributed system, and this class is about what that changes.

A distributed system is a set of independent computers (nodes) that cooperate only by exchanging messages over a network, with no shared memory and no shared clock, and that appear to their users as a single coherent system¹. Leslie Lamport's half-joking definition is the most useful one for an engineer: "a distributed system is one in which the failure of a computer you didn't even know existed can render your own computer unusable"².

Roadmap

1. This overview — why distribute, what it costs, failure models, the unreliable network (8 fallacies), timeouts, delivery semantics and idempotency.
2. Replication & Consistency — single-leader, multi-leader and leaderless replication; the consistency spectrum from linearizable to eventual; session guarantees such as read-your-writes.
3. CAP & PACELC — the consistency-versus-availability choice during a partition, the latency-versus-consistency choice the rest of the time, and multi-region designs.
4. Consensus & Quorums — majorities, Raft (election and log replication), R + W > N, leases and fencing tokens, and the tools that do it for you.
5. Distributed Transactions — atomicity across services: two-phase commit and why it blocks, sagas with compensations, try–confirm–cancel reservations, and the transactional outbox.
6. Time & Clocks — NTP and clock drift, Lamport and vector clocks, hybrid logical clocks, TrueTime, and how to order events in practice.
7. Quiz — check your understanding.

Slides

Click the slides and use ← → to move through them; F goes full screen and P opens the presenter view, with the notes. Open the slides in a new tab

Why distribute?

Nobody distributes a system for fun: a single process with a single database is easier to write, test, debug and reason about. We pay the price of distribution only when one machine can no longer deliver a non-functional requirement:

Vertical scaling — a bigger machine — is the first answer and often the right one for a while. It stops working for three reasons: there is a largest machine you can buy; the price grows faster than the capacity at the top end; and, however big it is, it is still one machine — a single point of failure (SPOF).

At scale, failure is the norm

The second argument hides a paradox: distributing for fault tolerance means using more machines, and more machines means more failures. If each machine fails on a given day with a small probability \(p\), independently, the probability that at least one of \(N\) machines fails that day is \(1 - (1-p)^N\):

With an (illustrative) 0.1% chance of failure per machine per day, three machines see a failure on about 0.3% of days, but a fleet of 3,000 machines sees at least one failure on about 95% of days. Google's early reports describe exactly this experience: at fleet scale, hardware failures are a daily routine, not an exception³. The conclusion shapes the rest of the course: a distributed system must treat failure as a normal event and keep working through it — it cannot hope to avoid it.

What distribution costs

BoxOffice is a distributed system

Follow one click on "Hold seat A-12" through BoxOffice. Every arrow is a message over the network; every box is a process that can fail, pause or slow down on its own:

flowchart LR
    fan([Fan's phone]) -->|"POST /holds"| gw[gateway]
    gw -->|"validate token"| auth[auth]
    gw -->|"forward"| inv[inventory]
    inv -->|"UPDATE seat ... (transaction)"| pg[("PostgreSQL primary")]
    pg -->|"replicate WAL"| rep[("PostgreSQL replica")]
    inv -->|"cache seat map"| redis[("Redis")]
    inv -->|"SeatHeld event"| kafka[["Kafka (3 brokers)"]]
    kafka -->|"consume"| notif[notification]

Nothing in that picture shares memory or a clock. If Redis is slow, inventory waits; if the replica lags, a read from it is stale; if Kafka is unreachable, the hold may be committed in PostgreSQL but the SeatHeld event never published. Each of these is a partial failure, and the rest of this page is about naming them precisely.

Failure models

A failure model states what a faulty component may do. It matters because every algorithm is only correct under some model: a protocol that tolerates crashes can be fooled by a node that lies.

Two more ideas complete the vocabulary:

• Partial failure: on one machine, a program usually works or crashes as a whole. In a distributed system, part of it fails while the rest keeps running, and nobody has a complete view of which part. Gray failures, studied by Microsoft in Azure, are the worst case — the system's own monitoring disagrees with its users⁴.
• Timing assumptions: in a synchronous model, messages and processing have known upper bounds; in an asynchronous model, there are no bounds at all. Real networks are partially synchronous: they behave well most of the time, with occasional unbounded delays (a congested switch, a GC pause, a VM migration). Practical algorithms (Raft, for example) are designed to be safe under asynchrony and to make progress only when the network behaves.

The network is unreliable

The 8 fallacies of distributed computing

In the 1990s, L. Peter Deutsch and colleagues at Sun Microsystems listed the assumptions that newcomers to distributed systems make — all of them false⁵:

Every one of them shows up in later classes: fallacy 1 in Reliability, 2 in Estimation, 3 in Traffic & Edge, 4 in Network & Security, 5 in Cloud-native.

Slow or dead?

When inventory sends a request to payment and no reply arrives, it cannot know which of these happened:

sequenceDiagram
    participant I as inventory
    participant P as payment
    I->>P: authorise R$ 450 (request 1)
    Note over I,P: (a) request lost  (b) payment crashed<br/>(c) reply lost  (d) payment just slow
    I-->>I: no reply after 2 s: timeout
    I->>P: authorise R$ 450 (retry)
    P-->>I: OK
    Note over I,P: in cases (c) and (d) the fan was charged twice

From the caller's side, a lost request, a crashed server, a lost reply and a very slow server all look the same: silence. The only practical tool is a timeout — after a deadline, assume failure. But a timeout does not tell you whether the operation ran, so two things follow:

1. Choosing the timeout is a trade-off. Too long and the caller's threads pile up waiting (the cascading failure of Reliability); too short and you declare healthy-but-slow nodes dead, retry, and add load exactly when the system is struggling. A common starting point is slightly above the downstream's p99.9 latency, within the caller's own latency budget. Systems such as Cassandra and Akka go further and use adaptive failure detectors (the φ accrual detector) that learn the normal distribution of heartbeat delays⁷.
2. Retries must be safe. If the retry can apply the operation a second time, it must be harmless to do so — the operation must be idempotent.

Delivery semantics and idempotency

At most once, at least once, exactly once

Whenever a message crosses an unreliable network, the sender chooses a delivery semantics:

"Exactly-once delivery" over an unreliable network is impossible in general — the sender can never be sure the receiver did not process a message whose acknowledgement was lost. What systems such as Kafka call exactly-once is exactly-once processing: duplicates may arrive, but their effect is applied once, through deduplication or transactions⁸. The Messaging class covers this in depth.

Idempotent operations

An operation is idempotent if applying it several times has the same effect as applying it once: \(f(f(x)) = f(x)\).

Two techniques make non-idempotent operations safe to retry:

1. Rewrite it as a state transition with a condition (compare-and-set): "set the seat to HELD if it is FREE", "set version 8 if the current version is 7". A duplicate fails the condition and changes nothing.
2. Idempotency keys: the client generates a unique key per intended operation and sends it with every attempt; the server stores the key with the result and, on a duplicate, returns the stored result instead of executing again. Stripe's public API popularised the Idempotency-Key header⁹, and there is an IETF draft standardising it.

sequenceDiagram
    participant F as Fan's app
    participant O as order
    participant DB as PostgreSQL
    F->>O: POST /orders, Idempotency-Key: 7f3a…
    O->>DB: INSERT key 7f3a… (unique) + order 981
    DB-->>O: committed
    O--xF: 201 Created (reply lost on the train)
    F->>O: retry: POST /orders, Idempotency-Key: 7f3a…
    O->>DB: INSERT key 7f3a…
    DB-->>O: unique violation: key exists
    O->>DB: SELECT stored response for 7f3a…
    O-->>F: 201 Created, order 981 (same order, no duplicate)

The key and the business change must be stored in the same transaction; otherwise a crash between the two writes reopens the window for duplicates:

CREATE TABLE idempotency_keys (
    key           UUID PRIMARY KEY,
    request_hash  TEXT        NOT NULL,   -- reject the same key with a different body
    response      JSONB,                  -- what we answered the first time
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

BEGIN;
INSERT INTO idempotency_keys (key, request_hash)
VALUES ('7f3a…', 'sha256:…')
ON CONFLICT (key) DO NOTHING;            -- 0 rows inserted → duplicate: return stored response
INSERT INTO orders (id, fan_id, total) VALUES (981, 42, 450.00);
UPDATE idempotency_keys SET response = '{"orderId": 981}' WHERE key = '7f3a…';
COMMIT;

// ---- strategy 3: optimistic, compare-and-set on a version column ---------------
private void holdOptimistic(long eventId, List<Long> seatIds, UUID holdId) {
    var seats = read(eventId, seatIds, "");                      // no lock
    requireAvailable(seats, seatIds);
    for (var seat : seats) {
        int changed = jdbc.sql("""
                UPDATE inventory.seats
                SET status = 'HELD', hold_id = :holdId, version = version + 1
                WHERE id = :id AND version = :version AND status = 'AVAILABLE'""")
                .param("holdId", holdId).param("id", seat.id()).param("version", seat.version())
                .update();
        if (changed == 0) {
            // someone changed this seat after we read it: give up (the exception rolls back)
            throw new ResponseStatusException(HttpStatus.CONFLICT, "seat " + seat.id() + " was just taken");
        }
    }
}

public Result charge(String idempotencyKey, ChargeRequest req) {
    Optional<PaymentOut> done = find(idempotencyKey);
    if (done.isPresent()) {
        return new Result(done.get(), false); // a retry: answer, do not charge again
    }
    // remote call OUTSIDE any database transaction: never hold a connection while waiting
    PspClient.Charge charge = psp.charge(idempotencyKey, req.amount(), "order-" + req.orderId());
    String status = "approved".equals(charge.status()) ? "AUTHORIZED" : "DECLINED";
    jdbc.sql("""
            INSERT INTO payments (id, idempotency_key, order_id, amount, status, provider_ref)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT (idempotency_key) DO NOTHING
            """)
            .params(UUID.randomUUID(), idempotencyKey, req.orderId(), req.amount(), status, charge.id())
            .update();
    return new Result(find(idempotencyKey).orElseThrow(), true);
}

Putting it together: the concept map

The three root problems — partial failure, an unreliable network, and no global clock — explain almost every design decision in the rest of the course. The map below connects each problem to the techniques of this class (blue), to the subpage that teaches it, and to the palette components where you will meet it in the playground (green):

flowchart LR
    PF["Partial failure"]:::problem --> REP["Replication<br/>(keep copies)"]
    NET["Unreliable network"]:::problem --> IDEM["Timeouts, retries,<br/>idempotency"]
    CLK["No global clock"]:::problem --> ORD["Ordering:<br/>logical / hybrid clocks"]
    REP --> CM["Consistency models<br/>linearizable … eventual"]
    CM --> CAP["CAP / PACELC<br/>C vs A, L vs C"]
    CAP --> CONS["Consensus, quorums,<br/>fencing"]
    IDEM --> TX["Sagas, outbox,<br/>2PC"]
    CONS --> TX
    ORD --> CONS
    REP -.-> sql["sql · nosql"]:::comp
    CM -.-> cache["cache · cdn"]:::comp
    CONS -.-> kafka["kafka · scheduler"]:::comp
    IDEM -.-> pay["payment · queue"]:::comp
    TX -.-> stream["stream · kafka"]:::comp
    CAP -.-> dns["dns (multi-region)"]:::comp
    classDef problem fill:#e5484d22,stroke:#e5484d
    classDef comp fill:#1e9e5a22,stroke:#1e9e5a
    classDef tech fill:#2f6fed22,stroke:#2f6fed
    class REP,IDEM,ORD,CM,CAP,CONS,TX tech

Key takeaways

What breaks next

BoxOffice now has the vocabulary to survive the network between its machines — timeouts, idempotent retries, one consistency model per data domain, consensus to choose the primary. It has not yet decided who receives the traffic. At 10:00:00 two million phones send a request in the same second: about 100,000 requests/s at the edge (the estimate), while one catalog instance measured in lab 2 keeps its p99 target only up to a few hundred requests per second. Most of those requests must end before any service, the rest must reach a healthy replica, and bots must not get there first: class 4 — Traffic & Edge.

Related

• Playground: System Design Playground — every component in this course can fail; turn on chaos (data-chaos) in any scenario.
• Hands-on: Hands-on 7 — Async (idempotent consumer, outbox), Hands-on 9 — Resilience & chaos (timeouts, retries, fault injection with Toxiproxy).
• Exercises: Event-driven (idempotent consumer), Design doc.
• Case studies: Ticketing (BoxOffice flash sale), Payment system, Distributed cache.
• Classes: Estimation (latency numbers), Databases — replication, Messaging — semantics, Reliability.
• Interview prep: Interview.