# Chapter 3 — Distributed Systems & CAP Theorem By the end of this chapter you will: - State the CAP theorem correctly (most people misquote it) - Distinguish CP from AP systems with examples - Choose between strong, eventual, and causal consistency - Reason about partitions without panicking --- ## The why > **Most engineers misquote CAP as "pick 2 of 3." That's wrong. CAP is about what you sacrifice during a network partition — and partitions are not optional.** Imagine three libraries in a city, each with their own copy of the book catalog. They share updates by sending runners between branches. One day, the bridge connecting Library A to Libraries B and C collapses — the runner can't cross. A customer walks into Library A and asks: "Do you have 'Designing Data-Intensive Applications'?" Library A has two choices: **Choice 1 — Consistency:** Refuse to answer. "I can't verify with the other branches right now. Please come back later." The customer is annoyed, but they'll never get wrong information. **Choice 2 — Availability:** Answer based on the local catalog. "Yes, we have one copy." But what if Library B just checked out the last copy, and the update runner hadn't crossed the bridge before it collapsed? The customer travels to Library A, only to find the book isn't actually there. The system was available (it answered) but inconsistent. Library A cannot be both perfectly consistent AND available while the bridge is down. This is the CAP theorem — not "pick two of three forever," but "when the network partitions, choose between consistency and availability." Now here's the part nobody tells you: the bridge is always broken. Network partitions happen constantly — a switch reboots, a cable is unplugged, a GC pause makes a node unreachable for 10 seconds. Distributed systems live in a perpetual state of potential partition. CAP isn't theoretical — it's the fundamental tension you navigate every day. --- ## Core concepts ### CAP, stated correctly The CAP theorem (Brewer, 2000; formally proved by Gilbert & Lynch, 2002) states: > A distributed data store can provide **at most two** of the following three guarantees simultaneously: **Consistency** (every read receives the most recent write), **Availability** (every request receives a non-error response), and **Partition Tolerance** (the system continues to operate despite network partitions). The critical nuance: **partition tolerance is not optional.** Networks are unreliable. You cannot choose "CA" — a system that is consistent and available but not partition-tolerant is logically impossible. If a partition occurs, a CA system would need to be both consistent (refuse stale reads) and available (respond to all requests) — a contradiction. So CAP really means: **during a network partition, do you choose Consistency or Availability?** - **CP (Consistent + Partition-tolerant):** During a partition, the system sacrifices availability to maintain consistency. Writes may be rejected. Reads may return errors. Examples: etcd, Consul, ZooKeeper, HBase, MongoDB (with strong consistency configured). - **AP (Available + Partition-tolerant):** During a partition, the system sacrifices consistency to remain available. It returns potentially stale data. Examples: Cassandra, DynamoDB (eventual consistency mode), CouchDB, Amazon S3 (read-after-write for new objects, eventual for overwrites). Saying "CAP means pick 2 of 3" implies you can pick CA and ignore P. You can't. Networks fail. A system that claims to be "CA" is a system that will fail catastrophically when its first network partition hits. Every distributed system must be P. The only real choice is between C and A when things go wrong. ### PACELC: CAP's more useful cousin PACELC (Abadi, 2012) extends CAP to cover what happens when there ISN'T a partition — which is most of the time: > If there is a **P**artition, trade off **A**vailability and **C**onsistency. > **E**lse, when the system is running normally, trade off **L**atency and **C**onsistency. This is the framework that actually describes real databases: | System | During Partition (PA/PC) | Normally (EL/EC) | Notes | |--------|------------------------|-------------------|-------| | DynamoDB | PA (available) | EL (low latency) | Always available, eventually consistent | | BigTable/HBase | PC (consistent) | EC (consistent) | Always consistent, trade latency | | Cassandra | PA/EL (tunable) | PA/EL (tunable) | Per-query consistency level | | MongoDB | PC/EC (default) | PC/EC | Configurable read/write concerns | | PostgreSQL (single node) | N/A | EC | Not distributed — CAP doesn't apply | ### Consistency models: a spectrum, not a binary "Consistency" in distributed systems isn't on/off. It's a spectrum of guarantees: **Strong consistency (linearizability):** After a write completes, all subsequent reads see that value. The system behaves as if there's only one copy of the data. This is the gold standard — and the most expensive. Implemented via consensus algorithms (Raft, Paxos) or single-leader replication with synchronous replication. **Sequential consistency:** Operations appear to execute in some total order, and operations from the same client appear in the order they were issued. Weaker than linearizability because the total order doesn't need to respect real-time ordering across clients. **Causal consistency:** Causally related operations are seen in order. If operation A causes operation B (e.g., a comment causes a notification), everyone sees A before B. Unrelated operations can be seen in any order. **Read-your-writes consistency:** A client always sees its own writes. If you update your profile picture, you see the new picture — but other users might see the old one for a few seconds. This is the minimum acceptable consistency for user-facing applications. **Eventual consistency:** Given enough time with no new writes, all replicas converge to the same value. This is the weakest guarantee — but the most scalable. Amazon S3, DNS, and CDNs are eventually consistent. ```typescript // Consistency levels in practice (DynamoDB-style) interface ConsistencyConfig { readConsistency: 'strong' | 'eventual'; writeConsistency: 'all' | 'quorum' | 'one'; } // Strong consistency: leader handles all reads/writes // Cost: higher latency, lower availability during partition async function stronglyConsistentRead(key: string): Promise { return leader.execute(`SELECT value FROM store WHERE key = $1`, [key]); } // Eventual consistency: read from any replica // Cost: may return stale data async function eventuallyConsistentRead(key: string): Promise { const replica = pickRandomReplica(); return replica.execute(`SELECT value FROM store WHERE key = $1`, [key]); } // Quorum writes: write to W nodes, read from R nodes, where W + R > N // This guarantees at least one node has the latest data for every read async function quorumWrite(key: string, value: string): Promise { const nodes = getTopology(key); // N = 5 replicas const writes = await Promise.allSettled( nodes.slice(0, 3).map(n => n.write(key, value)) // W = 3 ); const succeeded = writes.filter(w => w.status === 'fulfilled').length; if (succeeded < 3) throw new QuorumWriteFailedError(succeeded, 3); } ``` ### Consensus: how distributed nodes agree Consensus is the problem of getting multiple nodes to agree on a single value. It's the foundation of leader election, atomic broadcast, and distributed locks. **Raft (the understandable consensus algorithm):** Raft was designed specifically to be more understandable than Paxos. It works by: 1. **Leader election:** Nodes are in one of three states — Leader, Follower, or Candidate. The cluster elects a leader via randomized election timeouts. 2. **Log replication:** The leader accepts writes, appends them to its log, and replicates to followers. A write is "committed" when a majority of nodes have stored it. 3. **Safety:** The leader never overwrites its own log entries. A new leader must contain all committed entries from previous terms. ``` Raft cluster state machine: [Follower] ──timeout, no heartbeat──► [Candidate] ──receives majority votes──► [Leader] ▲ │ │ │ │ │ └──── discovers higher term ───────────┘ │ │ ◄────────────────────────── discovers higher term ──────────────────────────────┘ ``` **Paxos:** The original consensus algorithm. Mathematically proven correct but notoriously difficult to implement correctly. Google's Chubby lock service used Paxos. Most modern systems use Raft instead because it's simpler to implement and reason about. ### Split-brain: the distributed system's nightmare Split-brain occurs when a cluster partitions into two (or more) groups, and each group elects its own leader. Both leaders accept writes independently, creating divergent data that can't be automatically merged. ``` Before partition: [Node A (Leader)] ─── [Node B] ─── [Node C] [Node D] ─── [Node E] After partition (network cable cut): Group 1: [Node A (Leader)] ─── [Node B] ─── [Node C] Group 2: [Node D (Leader)] ─── [Node E] Both groups accept writes. Data diverges. 💥 ``` **How consensus prevents split-brain:** Raft requires a majority (quorum) for leader election. With 5 nodes, a leader needs 3 votes. If the network splits into groups of 3 and 2, only the group with 3 can elect a leader. The group with 2 can't form a quorum and becomes unavailable. This is a CP choice — the smaller partition sacrifices availability for consistency. ### Quorum reads and writes Quorum-based systems use voting to balance consistency and availability: - **N** = total number of replicas - **W** = number of replicas that must acknowledge a write - **R** = number of replicas that must respond to a read **The golden rule:** If **W + R > N**, every read is guaranteed to see the latest write (because at least one node overlaps between the read and write quorums). ``` N=5, W=3, R=3 → 3+3 > 5 ✓ (every read sees latest write) N=5, W=2, R=2 → 2+2 < 5 ✗ (reads may miss recent writes) N=5, W=4, R=2 → 4+2 > 5 ✓ (fast reads, slow writes) N=5, W=2, R=4 → 2+4 > 5 ✓ (fast writes, slow reads — good for write-heavy workloads) ``` --- ## Real-world example: How etcd uses Raft for Kubernetes Every Kubernetes cluster depends on etcd — a distributed key-value store that holds the entire cluster state (pods, services, config maps, secrets, RBAC rules). etcd's consistency model is the reason `kubectl apply` works correctly across a multi-master control plane. **etcd's architecture:** - Uses the Raft consensus algorithm for leader election and log replication - Typically 3 or 5 nodes for high availability (odd number prevents tie votes) - All writes go through the leader. Reads can go to any node (with a linearizable read option that goes through Raft) - A write is committed only after a majority of nodes acknowledge it **Why etcd must be CP:** Consider what happens if etcd were AP during a partition. Two control plane nodes in different partitions might accept conflicting `kubectl apply` commands — one scaling a deployment to 3 replicas, the other scaling to 5. The cluster state diverges. Pods are created and destroyed in conflicting ways. The system becomes unrecoverable without manual intervention. etcd's choice: during a partition, the minority partition refuses writes. Kubernetes API servers connected to the minority partition return errors. This is better than silently accepting conflicting state. **Real production numbers from etcd:** - A 3-node etcd cluster handles ~10,000 writes/second - Leader election takes ~2-5 seconds - A network partition of ~1 second is absorbed by Raft's heartbeat timeout (default 100ms, typically tuned to 500ms in production) - Recommended max data size: 8 GB (performance degrades beyond this) --- ## Common mistakes | Mistake | Why it's wrong | What to do instead | |---------|---------------|-------------------| | Claiming your system is "CA" | CA systems can't exist — if a partition occurs, you can't be both consistent and available. Your system will break | Acknowledge that your system is CP or AP. If you're on a single database, CAP doesn't apply — you don't have a distributed system | | Using "eventually consistent" as an excuse | "Eventually" could mean 50ms or 5 minutes. Users reasonably expect sub-second convergence for most data | Specify the convergence time SLA: "99th percentile staleness under 100ms." Monitor it | | Implementing your own consensus algorithm | Paxos is famously hard to implement correctly. Raft is simpler but still has edge cases. Even Jepsen-tested databases have bugs | Use etcd, ZooKeeper, or Consul for leader election and distributed coordination. Don't build consensus from scratch | | Assuming a 3-node cluster prevents split-brain | A 3-node cluster with one node isolated becomes a 2-node cluster — it can still elect a leader. But if the isolated node was the previous leader, you get split-brain for the election timeout period | Use 5 nodes for production. Odd numbers prevent ties. Always configure `minimum_master_nodes` / quorum properly | | Not testing partition behavior | Most teams never simulate network partitions. When the first real partition hits production, behavior is undefined | Run Chaos Engineering experiments: partition the network, observe behavior, verify your CP/AP choice holds | | Over-using strong consistency when eventual would work | "Likes" on a social media post don't need linearizability. Strong consistency costs latency and availability for no user benefit | Use strong consistency for financial transactions, inventory decrements, and access control. Use eventual consistency for counts, feeds, and analytics | | Setting W + R = N for "extra safety" | W=N means any single node failure blocks all writes. You've traded availability for consistency on every write, even without a partition | Use W=quorum, R=quorum. This gives you W+R>N with tolerance for (N-quorum) node failures | --- ## Production notes - **Run an odd number of Raft nodes.** With 3 nodes, you tolerate 1 failure. With 5 nodes, you tolerate 2 failures. An even number (4) tolerates the same 1 failure as 3 but costs more — the extra node provides no reliability benefit. - **Co-locate Raft nodes across failure domains.** If all 3 etcd nodes are in the same AWS availability zone and that AZ goes down, you lose the cluster. Spread across 3 AZs or 3 racks. - **Monitor Raft metrics.** Track: `raft_term` (should change only during elections), `raft_committed_index` (should advance monotonically), `raft_leader_changes_total` (spikes indicate instability), and `raft_proposals_failed_total` (non-zero means writes are being rejected). - **Set appropriate heartbeat and election timeouts.** Default Raft heartbeat is 100ms. For cross-region deployments, increase to 500ms-1s. Election timeout should be 5-10× the heartbeat. Too short = unnecessary elections on transient latency spikes. Too long = slow failure detection. - **Back up etcd regularly.** A corrupted etcd cluster means a destroyed Kubernetes cluster. Snapshot etcd every hour, store snapshots off-cluster (S3), and practice restoration. --- ## Performance notes - **Strong consistency costs 2-10× latency vs eventual.** A strongly consistent read in a multi-region system might take 100-300ms vs 5-20ms for eventual. Batch reads and precompute where possible. - **Leader-based consensus (Raft) throughput is bounded by the leader's disk I/O.** Every write must be durably stored on a majority. NVMe SSDs dramatically outperform EBS gp2 volumes for etcd workloads. - **Quorum writes add latency proportional to the slowest node in the quorum.** If W=3 and the 3rd node is 50ms slower, every write waits 50ms longer. Use `wtimeout` to cap write latency — at the cost of rejecting writes when nodes are slow. - **Gossip protocols (Cassandra, Dynamo-style) scale writes linearly with nodes.** Adding a node doesn't increase per-node write load. Raft-based systems don't scale write throughput with additional nodes — the leader is still the bottleneck. - **Batch writes to reduce Raft round trips.** Each Raft round trip is ~2-10ms. Batching 100 writes into one proposal gives ~100× throughput improvement for throughput-bound workloads. --- ## Security notes - **etcd endpoints must never be publicly accessible.** etcd holds every Kubernetes secret in plaintext (by default). An exposed etcd endpoint means full cluster compromise. Use network policies, security groups, and firewall rules to restrict access. - **Enable etcd authentication and TLS.** By default, etcd has no authentication. Enable client certificate authentication and TLS between peers. Kubernetes kubeadm setups do this automatically — self-managed etcd often doesn't. - **Encrypt Kubernetes secrets at rest in etcd.** Kubernetes secrets are base64-encoded (not encrypted) by default. Enable etcd encryption at rest using `EncryptionConfiguration` with AES-CBC or KMS plugins. - **Rotate etcd peer and client certificates before they expire.** An expired certificate causes all etcd nodes to reject connections. The cluster becomes unavailable. Set up certificate monitoring with 30-day warning alerts. - **Audit consensus operations.** Log all etcd writes with timestamps and source IPs. In a security incident, the etcd audit log shows exactly what changed and when — critical for forensics. --- ## Exercises ### Beginner 1. **CAP classification exercise:** Classify the following systems as CP, AP, or neither (explain why): PostgreSQL single-node, MongoDB replica set with `writeConcern: "majority"` and `readConcern: "local"`, DynamoDB with eventual consistency, etcd, Redis Cluster. Write 1-2 sentences per system explaining your classification. 2. **Consistency model matching:** Match each user-facing scenario with the minimum consistency model needed: (a) Viewing your own profile after editing it, (b) Seeing a friend's new post in your feed, (c) Checking your bank balance after a transfer, (d) Seeing a YouTube like count. Options: strong consistency, read-your-writes, eventual, causal. ### Intermediate 1. **Implement a simplified Raft leader election:** Write a TypeScript simulation of Raft leader election with 5 nodes. Each node has a randomized election timeout (150-300ms). Nodes transition between Follower, Candidate, and Leader. Simulate a leader failure and verify a new leader is elected within 2 election timeouts. Log all state transitions. 2. **Simulate split-brain and quorum:** Write a simulation of a 5-node cluster using majority quorum. Partition the network into groups of 3 and 2. Verify that only the group with 3 can commit writes. Reconnect the partition and verify that the group of 2 catches up from the leader's log. Then try a 2-2-1 split — verify no group can commit writes. ### Advanced 1. **Design a multi-region key-value store with tunable consistency:** Design a system with 3 regions (US, EU, APAC). Writes go to the local region's leader and replicate asynchronously to other regions. Reads can be strongly consistent (go to local leader) or eventually consistent (go to local replica). Define the API, consistency levels, and a strategy for handling conflicting writes to the same key (last-write-wins vs CRDT vs application-level merge). Include failure modes for each consistency level. 2. **Compare Paxos and Raft in depth:** Write a 2-page analysis comparing Paxos and Raft. Cover: leader election mechanisms, log replication, membership changes, understandability, production track record, and failure modes. Include at least 3 production incidents where consensus bugs caused outages (research Jepsen analyses). --- ## Interview questions ### Beginner 1. **Q: What is the CAP theorem, and why is "pick 2 of 3" misleading?** **A:** CAP says a distributed system can provide at most two of Consistency (all reads see latest write), Availability (all requests get a response), and Partition Tolerance (system works during network partitions). "Pick 2 of 3" is misleading because partition tolerance isn't optional — networks fail. The real choice is: during a partition, do you choose consistency (CP — reject requests to avoid returning stale data) or availability (AP — return potentially stale data to keep the system responsive)? You can't build a "CA" distributed system because partitions will happen. 2. **Q: What is split-brain, and how does Raft prevent it?** **A:** Split-brain is when a network partition causes a distributed cluster to split into two groups, each electing its own leader. Both leaders accept writes, creating divergent data that can't be automatically merged. Raft prevents split-brain by requiring a majority vote for leader election. A candidate must receive votes from more than half the nodes. If the cluster splits into two groups, at most one group has a majority — only that group can elect a leader. The minority group cannot commit writes. 3. **Q: What's the difference between strong consistency and eventual consistency?** **A:** Strong consistency (linearizability) guarantees that after a write completes, all subsequent reads see that value — the system behaves as if there's only one copy. Eventual consistency guarantees that if no new writes occur, all replicas will eventually converge to the same value — but reads might return stale data in the meantime. Strong consistency is needed for financial transactions and inventory management. Eventual consistency is acceptable for social media feeds, analytics, and DNS. ### Senior 1. **Q: Explain how quorum reads and writes (W + R > N) guarantee consistency, and what tradeoffs you make when tuning W and R.** **A:** With N replicas, a write to W replicas and a read from R replicas guarantees the read sees the latest write if W + R > N — because at least one replica overlaps between the read and write sets. **Tradeoffs:** (1) Write-heavy workload: set W=2, R=N (fast writes, slow reads — good for logging/analytics). (2) Read-heavy workload: set W=N, R=2 (fast reads, slow writes — good for configuration data). (3) Balanced: W=quorum, R=quorum. With N=5, quorum=3, giving W+R=6>5. Tolerates 2 node failures for reads, 2 for writes. The cost of W=N is that any single node failure blocks all writes — you've eliminated write availability entirely. The elegance of quorum is that you can tune per-operation: use W=1 for a "like" (low stakes), W=quorum for a purchase (medium stakes), W=N for a bank transfer (high stakes). 2. **Q: Describe PACELC and how it's more useful than CAP for describing real databases.** **A:** PACELC (Abadi, 2012) extends CAP to cover the normal (non-partitioned) case: **P**artition → trade **A**vailability vs **C**onsistency. **E**lse → trade **L**atency vs **C**onsistency. This maps directly to real databases. DynamoDB: PA/EL — during partition, it's available; normally, it trades consistency for low latency. HBase: PC/EC — always consistent, trades latency. The key insight: even when there's no partition, you're still making a tradeoff. A strongly consistent system like HBase adds latency on every operation (waiting for replication confirmation) even in perfect network conditions. PACELC explains why DynamoDB is fast (EL choice) and why HBase is slow (EC choice) — not just their behavior during disasters. 3. **Q: Your team is building a distributed task scheduler. Tasks must execute exactly once across N worker nodes. Design the distributed coordination using etcd. How do you handle worker failure, leader election, and exactly-once semantics?** **A:** **Leader election:** Workers campaign for leadership using etcd's lease + transaction API. Each worker creates a lease (e.g., 30s TTL) and tries to write to a `/scheduler/leader` key with `prevExist=false`. Only one succeeds — the leader. The leader periodically refreshes its lease. If the leader crashes, the lease expires, and other workers campaign. **Task distribution:** The leader watches `/scheduler/tasks` for new tasks. On new task, it assigns it to a worker by writing to `/scheduler/assignments/{taskId}` with the worker's ID. **Exactly-once:** The worker, on receiving a task, writes to `/scheduler/completed/{taskId}` using a compare-and-swap: only write if the key doesn't exist. This idempotency check prevents double execution. **Worker failure:** The leader watches worker leases. If a worker's lease expires (it crashed), the leader reassigns its in-progress tasks to other workers. The CAS on completion ensures even duplicate assignments execute once. **Limitations:** This gives at-least-once with an idempotency filter (effectively exactly-once for most failure modes). True exactly-once requires a two-phase commit between the work execution and the completion marker, which is significantly more complex. --- ## Summary The CAP theorem isn't a menu — it's a framework for understanding the fundamental tension in distributed systems. During a network partition, you choose between consistency (reject requests) and availability (serve potentially stale data). Since partitions are inevitable, the choice shapes your system's behavior under failure. PACELC extends this insight to normal operations: even without partitions, you're trading latency for consistency. Consistency is a spectrum — strong, sequential, causal, read-your-writes, eventual — and different parts of your application need different levels. Consensus algorithms like Raft and Paxos solve the distributed agreement problem, powering etcd, Consul, and ZooKeeper. Split-brain is the distributed system's nightmare, prevented by majority quorum. Understanding quorum (W + R > N) gives you a dial to tune consistency vs availability per-operation. The engineers who build reliable distributed systems don't avoid partitions — they design for them, test them, and know exactly what their system does when the bridge collapses. --- ## Quick recall - CAP: during a partition, choose Consistency or Availability. Partition tolerance is mandatory. - PACELC: during normal operation, trade Latency for Consistency. More useful than CAP for real systems. - "Pick 2 of 3" is a misleading oversimplification. There is no CA distributed system. - CP systems: etcd, ZooKeeper, HBase. AP systems: Cassandra, DynamoDB (eventual mode), CouchDB. - Consistency spectrum: strong > sequential > causal > read-your-writes > eventual. - Raft: leader election + log replication + safety. Used by etcd, Consul, TiKV. - Split-brain prevented by majority quorum. With 5 nodes, a leader needs 3 votes. Group of 2 can't elect. - Quorum formula: W + R > N guarantees every read sees latest write. Tune W and R per workload. - Use existing consensus implementations (etcd, ZooKeeper). Never build your own. - Test partition behavior with chaos engineering. Your CP/AP assumptions are theories until tested. --- ## Quiz 1. **What does the CAP theorem actually say about distributed systems?** A) You must pick exactly two of Consistency, Availability, and Partition Tolerance **B) During a network partition, you must choose between Consistency and Availability — Partition Tolerance is mandatory** C) Distributed systems are impossible D) CAP only applies to NoSQL databases 2. **Which of these is a valid consistency model that guarantees a user always sees their own updates?** A) Eventual consistency **B) Read-your-writes consistency** C) MongoDB default read concern D) CAP consistency 3. **How many nodes can fail in a 5-node Raft cluster while still allowing writes?** A) 4 B) 3 **C) 2** D) 5 (it's fully available) 4. **What does W + R > N guarantee in a quorum-based system?** A) The system is partition-tolerant B) Writes are faster than reads **C) Every read will see the most recent write (at least one node overlaps between read and write quorum)** D) The system is AP 5. **Why does etcd use the Raft consensus algorithm?** A) It's the fastest possible algorithm **B) Raft is designed to be more understandable than Paxos while providing the same strong consistency guarantees** C) It was invented by Google D) Kubernetes requires Paxos specifically 6. **What is split-brain in a distributed system?** A) A node with two network interfaces **B) When a network partition causes a cluster to split into two groups, each electing its own leader and accepting conflicting writes** C) A race condition in the application code D) Running two load balancers simultaneously 7. **Your system uses W=3, R=2, N=5. What happens when 3 nodes fail?** A) The system continues normally B) Reads succeed but writes fail **C) Both reads and writes fail — writes need 3 nodes (only 2 available), reads need 2 nodes (only 2 available) but with 2 surviving nodes, writes can't reach quorum so reads may return stale data** D) The remaining 2 nodes rebuild the cluster automatically

Chapter 3 — Distributed Systems & CAP Theorem

Part of L5