Distributed Systems II — Consistency & Delivery

The core problem: agreement over a flaky network

In Chapter 10 we scaled out — splitting work across many machines and copying (replicating) the same data to several nodes so reads are fast and a dead machine doesn't lose your data. "Just copy the data everywhere" sounds simple, and on a good day it is. The trouble is the network in between.

A network is not a reliable wire. Messages get dropped, delayed, reordered, and sometimes a whole group of machines gets cut off from another group — a network partition. Crucially, when a node sends a message and hears nothing back, it cannot tell whether the other node crashed, is just slow, or the reply got lost. The silence looks identical in all three cases.

That single picture is the whole chapter. The moment a partition happens, the copies can disagree, and each node has to decide — alone — how to behave. Everything that follows (CAP, consistency models, consensus, delivery, idempotency) is a different angle on this one problem: how do independent machines stay in agreement when the network won't cooperate?

The CAP theorem, in plain language

The CAP theorem is the most famous (and most misquoted) result in distributed systems. The three letters stand for:

• C — Consistency: every read sees the most recent write. All nodes show the same value at the same time.
• A — Availability: every request gets an answer (not an error or a hang), even if some nodes are unreachable.
• P — Partition tolerance: the system keeps working even when the network splits into groups that can't talk.

The popular phrasing — "pick 2 of 3" — is misleading. Across real networks, partitions will happen; P is not optional. So the honest reading is narrower and far more useful:

There's no universally "right" answer — only the right answer for this data. Part of reading a system's docs is spotting which way it leans, because that decision shows up in your data on the worst day.

Strong vs eventual consistency

"Consistency" isn't all-or-nothing; it's a spectrum of guarantees a system makes about what a read can see. Two anchor points matter most to a data engineer:

The phrase that trips people up is "eventually". It does not mean "broken" — it means "correct, after a short catch-up window." A classic example: you write a row to your database's primary, then immediately read it back from a read replica. The replica hasn't received the change yet, so your read returns the old row. Nothing is malfunctioning; the replica is eventually consistent and you read it a half-second too soon. This is the mechanism behind "I just updated it but the dashboard still shows the old number."

Consensus: how a cluster agrees on one truth

If nodes can disagree, how does a cluster ever decide a single value — like "node 3 is now the leader," or "this is record #847 in the log"? That's the consensus problem: getting a group of machines to agree on one thing despite some of them being slow, crashed, or briefly unreachable. The famous algorithms are Paxos and Raft. We'll skip the math entirely and keep just the intuition, because that's all you need to operate these systems.

Two ideas carry the whole thing:

• A majority must agree (a quorum). A value is only "real" once more than half the nodes have accepted it. With 5 nodes you need 3; the cluster survives losing 2. The magic is that any two majorities of the same group always share at least one node — so they can't commit two conflicting values. That overlap is what prevents a "split brain" where each side thinks it won.
• A leader coordinates. Rather than everyone proposing at once (chaos), the cluster elects one leader to order writes. If the leader vanishes, the survivors hold an election and a majority picks a new one. Writes pause for that brief election, then resume — this is the system choosing consistency over momentary availability.

You rarely call a consensus algorithm directly. But "needs a majority to make progress" explains a lot of real behavior — like why a 3-node cluster goes read-only when 2 nodes die, or why people run clusters with an odd number of nodes (so a majority is always well-defined).

Delivery semantics — the practical heart

Now the part you'll touch every single week. When one system sends a message to another over a flaky network — a producer to a queue, a queue to your pipeline — what guarantee do you get that the message arrives exactly as intended? There are three possible promises, and the differences are the source of a huge share of real pipeline bugs.

Why does at-least-once duplicate? Because of that same unreliable network and the silence problem from the start of the chapter. The sender delivers a message and waits for an acknowledgement ("got it"). If the work succeeded but the ack got lost on the way back, the sender hears nothing, assumes failure, and retries — delivering the same message a second time.

Idempotency — the superpower that tames at-least-once

Here's the key insight that makes peace with a duplicating world: if running an operation twice has the same effect as running it once, then duplicates stop mattering. An operation with that property is called idempotent. You met this in Course 2 when you built a pipeline you could safely re-run after a crash without doubling your data — that re-runnability was idempotency. Now you know the name and the why.

Compare two ways to load a payment event:

The three workhorse techniques — you'll reach for these constantly:

• Idempotency key. Give every event a stable unique id (an event_id, an order number, a hash of the payload). Carry it through the pipeline so a retry is recognizable as the same event, not a new one.
• Upsert instead of insert. "Insert-or-update by key" (MERGE / ON CONFLICT) makes a write land in the same final state no matter how many times it runs.
• Dedup on the key. Keep a record of ids you've already processed (a table, a set, a unique constraint) and drop anything you've seen before.

Wrapping up Foundations

Take a breath — you just finished the longest, deepest course in the curriculum. Look at the distance you've covered. You can now narrate the entire journey of data, end to end:

When you write SELECT … JOIN … GROUP BY now, you don't see magic — you see bytes moving from disk into memory, hashed and sorted by real algorithms (Ch. 2–4), maybe spilling when memory runs out (Ch. 8), possibly fanned across a cluster that had to agree on the result (Ch. 10–11). That narration — knowing what physically happens — is exactly the senior engineer's superpower the first chapter promised you'd build. It's the thing that turns "the system is being weird" into a specific, answerable question.

If any chapter here felt shaky, that's expected — Foundations is a reference you return to, not a wall you clear once. Distributed systems especially will click harder after you've built a pipeline that hits a duplicate event for real. Coming back is part of the plan.

✓ Check yourself

Answer from memory before opening each one.

1. A design document says: "CAP lets us pick two of three; we choose Consistency and Availability and skip Partition tolerance." What is wrong with that sentence, and what is the choice CAP actually forces — and when?

   Show answer

   Partitions are not something you can opt out of: across a real network they will happen, so P is not optional. The real choice applies only while a partition is in progress: stay consistent by refusing some requests (CP), or stay available by answering with what you have, possibly stale or conflicting (AP). When the network is healthy you get both. Which way to lean depends on the data — a balance leans C, a shopping cart leans A. See The CAP theorem, in plain language.

2. An integration test writes a row, immediately reads it back and asserts on the value. It passes on a developer's machine and fails about one run in fifty in staging, where reads are routed to replicas. The code is identical. What is happening, is anything malfunctioning, and which workloads should read from those replicas?

   Show answer

   The write goes to the primary and the read sometimes reaches a replica that has not received the change yet, so it returns the old row. Nothing is broken — a replica is eventually consistent, meaning correct after a short catch-up window, and the test occasionally reads a moment too soon. Read-your-own-write checks ("did the order go through?") belong on the primary; analytics belongs on replicas, because it doesn't need this-millisecond freshness and it takes load off the primary. See Strong vs eventual consistency.

3. A network fault splits a five-node consensus cluster into a group of three and a group of two. Which side can still commit writes, what does the other side do, and why can the two sides never commit conflicting values?

   Show answer

   Only the group of three: a value is committed once a majority — a quorum, 3 of 5 — accepts it, and if the old leader is on the wrong side the three elect a new one after a brief pause. The two cannot form a majority, so they cannot commit anything; they are choosing consistency over availability. Any two majorities of the same group share at least one node, so two conflicting values can never both gather a quorum — that overlap is what prevents split brain, and it is also why a three-node cluster stops accepting writes when two nodes die. See Consensus.

4. Consumer logs show event E processed successfully twice, thirty seconds apart. Producer logs show a timeout in between. Nobody's code has a bug. Reconstruct the sequence — and explain why the producer couldn't simply work out whether the first attempt had succeeded.

   Show answer

   The producer sent E, the consumer processed it, and the acknowledgement was lost on the way back. Hearing nothing, the producer timed out, assumed failure and resent — at-least-once delivery doing exactly what it promises: never lose, sometimes duplicate. It could not work out the truth because silence looks identical whether the other side crashed, is slow, or replied and the reply was lost. See Delivery semantics and the core problem.

5. An at-least-once queue feeds three different load statements: (a) UPDATE accounts SET balance = balance + 100 WHERE id = 7; (b) INSERT … ON CONFLICT (event_id) DO NOTHING; (c) UPDATE accounts SET balance = 450 WHERE id = 7. Which are safe when an event is redelivered, and how would you make the unsafe one safe?

   Show answer

   (b) and (c) are idempotent: the second insert is a no-op, and setting a row to a value lands in the same final state however many times it runs. (a) is not — each redelivery adds another 100. Make it safe by giving every event a stable idempotency key and deduplicating on it: record processed event_ids (a table or a unique constraint) and skip any you have already seen, recording the id in the same transaction as the balance change. At-least-once delivery plus an idempotent load is "effectively exactly-once". See Idempotency.

6. You are choosing delivery semantics for two streams: host CPU metrics sampled every second for a dashboard, and payment events feeding a ledger. A vendor also offers an "exactly-once" queue at a premium. What do you choose for each stream, and how do you evaluate the vendor's claim?

   Show answer

   Metrics can run at-most-once: losing an occasional sample is acceptable, and it is the fastest and simplest option. Payments need at-least-once — nothing may be lost — paired with an idempotent load so duplicates are harmless. Treat "exactly-once" with suspicion: end to end across independent systems it is impossible in the general case, and what is sold under that name is usually at-least-once plus deduplication that holds only inside the product's own boundary. It can be worth paying for within that boundary, but the moment you write to your own table you still need idempotency, so build that first. See "Exactly-once" is mostly a lie.

-- WAY 1: Upsert keyed by the event's unique id.
-- A redelivery overwrites the same row instead of adding a new one,
-- so the final state is identical no matter how many times it runs.
INSERT INTO orders (order_id, customer, amount, ordered_at)
VALUES (:order_id, :customer, :amount, :ordered_at)
ON CONFLICT (order_id) DO UPDATE
  SET customer = EXCLUDED.customer,
      amount   = EXCLUDED.amount,
      ordered_at = EXCLUDED.ordered_at;
-- (Requires a UNIQUE / PRIMARY KEY on order_id.)

-- WAY 2: Dedup — record processed ids and skip ones you've seen.
-- The blind insert simply does nothing on a repeat.
INSERT INTO orders (order_id, customer, amount, ordered_at)
VALUES (:order_id, :customer, :amount, :ordered_at)
ON CONFLICT (order_id) DO NOTHING;
-- Equivalently: keep a `processed_ids` table (or set) and check membership
-- before doing the work, inserting the id in the same transaction.

Takeaway

• When a network partition splits the copies, a system must choose to refuse some requests and stay consistent or answer and risk stale data. Partition tolerance is not optional.
• Choose the lean by the data: money and balances favor consistency, carts and feeds favor availability.
• Strong consistency means a read returns the latest write, and eventual consistency means copies converge after a delay, so know whether you are reading a primary or a replica.
• Consensus commits a value only when a majority agrees, which is why a three-node cluster stops accepting writes when two nodes fail.
• At-least-once delivery is the practical default and it duplicates messages, because a lost acknowledgement makes the sender retry. Exactly-once across independent systems is at-least-once plus deduplication.
• Make duplicates harmless with a stable event id, upserts keyed on it and a unique constraint, so re-running a load changes nothing.

That ends Foundations. Course 4 uses these ideas to build real pipelines with ingestion, dbt, orchestration and tests. → The DE Craft

Further reading

• Brewer's Conjecture and the Feasibility of Consistent, Available, Partition-Tolerant Web Services — Gilbert and Lynch, ACM SIGACT News, 2002. The formal proof behind the CAP statement.
• CAP Twelve Years Later: How the "Rules" Have Changed — Eric Brewer, IEEE Computer, 2012. Brewer's own correction of the pick-two reading.
• In Search of an Understandable Consensus Algorithm — Ongaro and Ousterhout, USENIX ATC 2014. The Raft paper: leader election and majority commit, as described in the consensus section.
• Consistency Models — Jepsen. A map of consistency models from eventual to strict serializable.
• Dynamo: Amazon's Highly Available Key-value Store — DeCandia et al., SOSP 2007. An availability-first design that accepts temporary disagreement.
• Message Delivery Semantics — Apache Kafka documentation. At-most-once, at-least-once and exactly-once as a real system defines them.
• Idempotent requests — Stripe API documentation. An idempotency key in production, so a retry is safe.