Distributed Systems Core — CAP/PACELC, Sharding, Replication, Multi-Region

Why this session matters

Part of a 24-topic learning series on engineering, ML, and LLM systems. Each session is a 90-minute deep-dive on one topic — designed so anyone can pick it up cold. Every two topics are followed by a revision session with recall prompts and hands-on drills.

Part 1: CAP, PACELC, Quorums — How Distributed Systems Actually Trade Off

Why this session matters

It builds on the rhythm of one focused topic, paced so you have time to actually absorb it rather than rush.

Agenda

• CAP — what it actually says (and what it doesn’t)
• PACELC — the 'else' Brewer left out and why it matters in practice
• Consensus 101 — Raft in one diagram + the role of quorums
• Real systems on the CAP map — Dynamo, Cassandra, Spanner, Postgres
• Practical guidance — pick consistency at the operation level, not the system level

Pre-read (skim before the session)

• Martin Kleppmann — Distributed Systems lectures
• Brewer 2012 — CAP twelve years later
• Abadi 2012 — Consistency Tradeoffs in Modern Distributed DB Design
• Raft paper — In Search of an Understandable Consensus Algorithm

Deep dive

1. CAP — what it actually says

Brewer's CAP theorem: in the presence of a network partition, you must choose between consistency (every read sees the latest write) and availability (every request gets a non-error response).

What people get wrong:

• CAP is not a system-wide property. It applies per operation, per partition event.
• "AP" doesn’t mean inconsistent forever; it means you'll eventually converge.
• "CP" doesn’t mean infinitely consistent; it means you reject writes when you can't guarantee it.

During normal operation (no partition), modern systems are both consistent AND available. CAP only fires during a partition.

2. PACELC — the 'else' clause

Abadi 2012 extended CAP:

If there is a Partition, trade Availability vs Consistency. Else (during normal operation), trade Latency vs Consistency.

This is closer to how engineers actually feel the trade-off. Linearisable systems pay latency for every write (cross-region round-trips, consensus rounds); eventually-consistent systems are fast all the time but converge later.

  Partition?       Choose
  ----------       ------
     yes    ———▶   A  or  C
      no    ———▶   L  or  C

3. Quorums

A quorum is the minimum number of replicas that must respond for a read or write. For N replicas, common settings:

• W = writes acked by W replicas
• R = reads gather R replicas
• Strong consistency if W + R > N (writes and reads always overlap).

Examples:

• N=3, W=3, R=1 — strong, slow writes, fast reads. Postgres replication w/ sync replica.
• N=3, W=2, R=2 — strong, balanced. Cassandra QUORUM.
• N=3, W=1, R=1 — fast, eventually consistent. Cassandra ONE.

4. Raft in one diagram

Raft is the most popular consensus algorithm in production (etcd, Consul, TiKV, CockroachDB, KRaft, MongoDB internals).

All nodes hold a replicated log.
One leader at a time.

  Client → Leader: "append entry X"
  Leader → Followers: AppendEntries(X)
  Followers respond OK
  Once majority OK, leader commits, applies to state machine,
  responds to client.

Elections:
  Followers heartbeat-timeout → candidate → request votes →
  if majority votes, becomes leader.

Properties:

• Safety — at most one leader per term; committed entries never lost.
• Liveness — makes progress as long as a majority is up and can talk.
• Linearisability — the log defines a single global order of operations.

Failure handling: leader dies → election (~150–300 ms typical) → new leader; clients retry.

5. Real systems on the CAP/PACELC map

The one big shift in the last decade: Spanner / CockroachDB make CP across regions practical thanks to synchronised clocks and Raft. You no longer have to give up consistency to scale geographically — you just pay a few extra ms.

6. Choosing consistency per operation, not per system

A mature system picks consistency per call:

• Payments capture → linearisable (consensus).
• User profile read → strong leader-read.
• Friend count → eventual (off by 1 doesn’t matter).
• Notifications fanout → eventual + idempotent.
• Ledger writes → strong + transactional.
• Activity feed → eventual + bounded staleness.

DynamoDB, Cosmos, Cassandra all expose per-call consistency knobs precisely because one number for the whole system is wrong.

7. Sequence — a CP write with quorum

client ── write X ──▶ leader
                     leader ──── X ──▶ follower 1
                     leader ──── X ──▶ follower 2
                     leader ──── X ──▶ follower 3 (slow / partitioned)
                     leader waits for majority (2 of 3 followers) + self = 3 of 4
                     leader commits, replies OK

If two followers can't be reached: the leader stalls writes (CP). Cassandra at LOCAL_QUORUM would still serve if local DC quorum is up — same trade made differently.

8. Failures you should be ready to talk about

• Split brain — two nodes both think they're leader. Cured by Raft’s majority + term number.
• Stale reads after failover — if you read from a follower right after a leader change. Use read-index / leader-lease.
• Network partition — minority side stops serving writes (CP) or keeps serving and reconciles later (AP).
• Clock skew — breaks pseudo-orderings; Spanner uses TrueTime, CockroachDB uses hybrid logical clocks.
• Slow replica — raises tail latency in W=all setups. Mitigate with majority quorum + read-repair.

9. Hands-on (30 min)

Spin up a 3-node etcd cluster in Docker and play with it:

docker network create etcd
for i in 1 2 3; do
  docker run -d --network etcd --name etcd$i quay.io/coreos/etcd:v3.5 \
    /usr/local/bin/etcd \
    --name etcd$i \
    --initial-cluster etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380 \
    --initial-advertise-peer-urls http://etcd$i:2380 \
    --listen-peer-urls http://0.0.0.0:2380 \
    --advertise-client-urls http://etcd$i:2379 \
    --listen-client-urls http://0.0.0.0:2379
done

Now:

• etcdctl put /k v and read from any node — linearisable.
• docker pause etcd1 (kill the leader) and watch failover; writes resume in <1 s.
• Pause 2 of 3; writes stall (CP).

Feel the trade-off in real time. It’ll change how you talk about it in interviews.

10. What's next

This session is foundational — every later SYS session (chat, news feed, search, distributed job queue) will lean on these terms. From here you can read papers like Raft, Paxos Made Simple, Spanner, and Dynamo without bouncing off the abstract.

Reading material

Books:

• Designing Data-Intensive Applications — Martin Kleppmann (ch. 5: Replication, ch. 8: The Trouble with Distributed Systems, ch. 9: Consistency and Consensus — the entire core of this session)
• Database Internals — Alex Petrov (Part II on distributed systems)
• Distributed Systems: Principles and Paradigms — Tanenbaum, Van Steen (the textbook)

Papers:

• Brewer's Conjecture and the Feasibility of Consistent, Available, Partition-Tolerant Web Services (Gilbert & Lynch, 2002) — the proof of CAP.
• Problems with CAP, and Yahoo's Little Known NoSQL System (Daniel Abadi) — the original PACELC proposal.
• CAP Twelve Years Later: How the 'Rules' Have Changed (Eric Brewer, 2012) — Brewer's own retrospective.
• Spanner: Google's Globally-Distributed Database (OSDI 2012) — TrueTime, the case for CP at planet scale.

Official docs:

• DynamoDB — Consistency model
• Cassandra — Tunable consistency

Blog posts:

• Jepsen — distributed system safety analyses — Kyle Kingsbury's tests of every major system; the empirical view.
• You Can't Sacrifice Partition Tolerance — codahale.com — the classic essay arguing CAP is misunderstood.

In-depth research material

• jepsen — github.com/jepsen-io/jepsen — the test framework behind the Jepsen analyses.
• Distributed Systems Reading List — github.com/theanalyst/awesome-distributed-systems — ~10k ★, curated papers.
• CMU 15-440 Distributed Systems — course page — assignments + lecture notes.
• MIT 6.824 Distributed Systems — course page (Robert Morris) — the gold-standard university course; lectures on YouTube.
• Software Engineering Daily — Kleppmann episode
• Latent Space — Spanner deep dive

Videos

• Distributed Systems 1.1: Introduction — Martin Kleppmann (Cambridge) · 15 min — the start of the best distributed-systems lecture series on the internet, free.
• Distributed Systems Course — full 6-hour lecture series — Scientific Programming School · 6 h 23 min — Kleppmann's Cambridge series concatenated; jump to the consistency chapters.
• CAP Theorem Simplified — ByteByteGo · 6 min — the 5-min C/A/P refresher; Alex Xu's signature animation style.
• Designing Data-intensive Applications with Martin Kleppmann — The Pragmatic Engineer · 1 h 26 min — the man himself on the book; covers CAP, consensus, replication.
• The Two Generals' Problem — Tom Scott · 8 min — the only watchable explanation of why "reliable messaging over an unreliable network" is provably impossible. 7.8M views; mandatory.

LeetCode — Verifying An Alien Dictionary

• Link: https://leetcode.com/problems/verifying-an-alien-dictionary/
• Difficulty: Medium
• Why this problem: Build char→rank map from order; check each adjacent word pair.
• Time-box: 30 minutes. Look up the editorial only after.

Assignment / Deliverables

Give yourself a clean 2-hour window and complete all of these before moving on:

1. Read the deep-dive above end-to-end — no skimming. Take notes in your own words.
2. Solve the LeetCode problem below without help first. Only look at the hint after 15 focused minutes; only look at editorial after 30. Log your time.
3. Reproduce one code snippet locally. Pick the snippet that felt least obvious and get it running in a scratch file / notebook.
4. Draw the core diagram from memory. Paper, whiteboard, or tldraw — doesn't matter. If you can't, re-read section 2 and try again.
5. Write a 3-line takeaway in your prep journal: what surprised you, what you still don't understand, what you'd read next.
6. Skim one item from the Reading material section. Bookmark the rest for the weekend.
7. Commit any code + notes to your prep repo with message session-NN: <one-line summary>.

Stretch (optional, +30 min): explain today's topic to a rubber duck / a friend / a voice note. If you can't teach it in 5 minutes, you don't own it yet — flag it and revisit next weekend.

Post-session checklist

By the end of this session you should be able to:

• State CAP precisely — including the partition precondition.
• Explain why PACELC is a more useful framing day-to-day.
• Compute quorum requirements for given N, R, W.
• Draw the Raft leader-election + AppendEntries flow.
• Place 5 real systems on the CAP/PACELC map.
• Solve verifying-an-alien-dictionary — partial-order reasoning, same shape as CRDT merge ordering.

Generated from sessions_data.py + content_part*.py. To edit a video / leetcode / title, edit the data file and re-run write_sessions.py.

Part 2: Sharding & Replication — Partition Keys, Hot Spots, Multi-Region

Why this session matters

It builds on the rhythm of one focused topic, paced so you have time to actually absorb it rather than rush.

Agenda

• Vertical vs horizontal partitioning — when each makes sense
• Sharding strategies — range, hash, consistent hashing, directory
• Hot keys and hot shards — detection and mitigation
• Replication topologies — leader-follower, multi-leader, leaderless
• Multi-region — active/passive, active/active, conflict resolution

Pre-read (skim before the session)

• DDIA Ch. 6 — Partitioning
• Amazon Dynamo paper
• Consistent Hashing (Karger et al., 1997)
• Discord — Storing Billions of Messages

Deep dive

1. Why partition

When data outgrows one machine, you have three choices: bigger machine (vertical), more machines (horizontal), or less data (deletion / archival). Vertical scaling hits limits — fastest single-server SSD writes top out at ~3 GB/s, RAM at TBs but $$$ exponential.

Horizontal partitioning (= sharding) splits the data set across nodes. Each shard holds a subset; routing layer figures out which.

2. Range partitioning

Sort by key (timestamp, user_id), assign contiguous ranges to shards.

shard 1: user_id 0       .. 100_000
shard 2: user_id 100_001 .. 200_000
shard 3: user_id 200_001 .. 300_000

Pros: range scans are local. Great for time-series (logs, metrics).

Cons: hot spots if writes cluster in one range (recent timestamps → shard with now()). Mitigate by bucketing time + hash prefix.

Used by: HBase, Bigtable, MongoDB (default), CockroachDB.

3. Hash partitioning

Hash the key, mod by shard count.

shard = hash(user_id) % N

Pros: uniform distribution → no hot shards on uniform keys.

Cons: range scans now touch every shard. Resharding (changing N) reshuffles all data.

Used by: Cassandra (token rings), Redis Cluster.

4. Consistent hashing — the fix to "all data reshuffled"

Map both keys and nodes onto a ring of size 2^64. Each key belongs to the next node clockwise on the ring. Add a node → only the keys between it and its predecessor move. Remove a node → its keys go to its successor.

       N1
       │
N4 ────●──── N2
       │
       N3

key 0xABCD lands between N3 and N4 → owned by N4.

Add N5 between N3 and N4 → only some of N4's keys move to N5.

Virtual nodes — give each physical node 100–1000 virtual positions on the ring. Smooths load when nodes are heterogeneous and limits reshuffling on add/remove.

Used by: Cassandra, Dynamo, Riak, memcached client libs.

5. Directory-based — a lookup table

Maintain a service that maps key → shard. Total flexibility, can rebalance arbitrarily. Cost: extra hop, the directory itself must be HA.

Used by: Vitess (MySQL sharding), Citus (Postgres), HDFS NameNode.

6. Picking a partition key

The single most consequential schema decision. Optimise for:

1. Even load — avoid keys with skewed distribution (country, tenant_id with one huge tenant).
2. Locality of access — queries should touch one or few shards. Chat app: shard by chat_id, not user_id.
3. Cardinality — must have enough distinct values to spread.

Composite keys (e.g., (tenant_id, user_id) with hash on tenant_id, range on user_id) are common in multi-tenant systems.

7. Hot keys

The Justin Bieber problem — one celebrity follower count update triggers writes to one shard at 100K QPS.

Mitigations:

• Salting: write to key:0, key:1, …, key:99; reads aggregate. Trades read amplification for write distribution.
• Read-side caching: hot keys live in CDN/Redis; the DB sees << QPS.
• Write coalescing: batch many "+1" into a single "+N" every 100 ms.
• Dedicated shard: pin known hot keys to a dedicated, larger shard.

8. Hot shards (vs hot keys)

A hot shard gets disproportionate traffic even with no single hot key — e.g., partitioned by region and one region is 10× others. Mitigate by:

• Re-partition with a more uniform key.
• Subshard the hot shard.
• Use a routing layer that biases reads to replicas.

9. Replication topologies

Leader-follower (single-leader, master/slave):

• One node accepts writes; others copy.
• Failover requires election or manual promotion.
• Used by: Postgres, MySQL, MongoDB replica set, Kafka.

Multi-leader:

• Two+ leaders, each accepts writes; they replicate to each other.
• Write conflict resolution is the hard problem (LWW, CRDT, application code).
• Used by: rare in OLTP; common in cross-region BDR setups.

Leaderless:

• Any node accepts writes; clients write to W replicas, read from R, ensure W + R > N.
• Read repair + anti-entropy reconcile divergence.
• Used by: Dynamo, Cassandra, Riak.

10. Multi-region

Active/passive (DR site) — writes go to one region; passive replicates async. RPO = replication lag (seconds), RTO = failover time (minutes). Cheap, simple, lossy.

Active/active (multi-region writes) — writes accepted everywhere. Three flavours:

1. Geo-partitioned — each region owns a subset of keys (EU rows in EU, US rows in US). No conflicts. Latency local for owned keys, cross-region for others. CockroachDB, Spanner, Cosmos DB.
2. CRDT-based — eventual consistency with deterministic merge. Counter, set, JSON. Riak, Yjs, Automerge.
3. Last-Write-Wins on timestamps — simple, lossy on concurrent writes. Cassandra default.

11. Spanner / CockroachDB — the cheating answer

Synchronised clocks (TrueTime) or hybrid logical clocks let these systems give you strong consistency across regions at ~10–50 ms write latency. They use Paxos/Raft groups per partition, replicated across regions. Most fintech and any greenfield CP-needing system now starts here.

12. Operational realities

• Resharding is the worst day of the year. Plan capacity early; double when you cross 70%.
• Backfills — adding a column or migrating shards while writes continue. Always idempotent, always restartable.
• Read-your-writes — even on a single region, reading from a replica can return stale data. Use leader reads or "session consistency" tokens.
• Schema changes — Postgres-style online DDL has limits; in sharded systems, schema migrations need to be backwards/forwards compatible.

13. Sizing rule of thumb

Total disk / shard ≤ 60% of node disk          (room for compactions, backups)
RPS per shard ≤ 70% of saturation              (room for failures, hot spots)
Replication factor = 3                          (industry default)
Cross-region replicas = 1 per region            (active/passive minimum)

Reading material

Books:

• Designing Data-Intensive Applications — Martin Kleppmann (chs. 5–7: replication, partitioning, transactions — the canonical reference)
• Database Internals — Alex Petrov (Part II on distributed systems, leader election, replication logs)
• Building Microservices, 2nd ed. — Sam Newman (the chapter on data + ownership across regions)

Papers:

• Dynamo: Amazon's Highly Available Key-value Store — DeCandia et al. 2007 — the consistent-hashing, quorum, vector-clock origin story.
• Spanner: Google's Globally Distributed Database — Corbett et al. 2012 — TrueTime + externally consistent multi-region.
• Consistent Hashing and Random Trees — Karger et al. 1997 — the paper that named consistent hashing.
• Slicer: Auto-Sharding for Datacenter Applications — Adya et al. 2016 (Google) — Google's general-purpose sharding service.

Official docs:

• CockroachDB Architecture — ranges, leaseholders, raft groups.
• Vitess — Sharding — the YouTube/Slack MySQL sharding stack.
• DynamoDB — Partitioning and Distribution — hash + range partitions, hot partition mitigation.
• MongoDB Sharding — the chunk + balancer model.

Blog posts:

• How Discord Stores Trillions of Messages — Discord Engineering — sharding Cassandra/ScyllaDB by channel.
• Sharding & IDs at Instagram — Instagram Engineering — the Snowflake-style ID generator and Postgres sharding.
• Replication in Distributed Systems — Martin Kleppmann (blog post version of DDIA ch. 5)

In-depth research material

• Vitess — github.com/vitessio/vitess — ~18k ★, MySQL sharding used at YouTube, Slack, Square.
• TiDB — github.com/pingcap/tidb — ~37k ★, MySQL-compatible distributed SQL with auto-sharding.
• CockroachDB — github.com/cockroachdb/cockroach — ~29k ★, Spanner-style multi-region SQL.
• Discord — How Discord Stores Billions of Messages — the earlier Cassandra-era post; pairs with the trillions follow-up.
• Figma — How we built a 50x faster Postgres — horizontal sharding Postgres without going Spanner.
• Notion — The journey to sharding — Postgres sharding for a 200B+ block dataset.
• CRDT survey — Shapiro et al. 2011 — the convergent replicated data types canon for multi-master.
• Aurora paper — Verbitski et al. 2017 — storage-replication done at the page level.
• Jepsen — distributed systems safety analyses — every replication bug in production, dissected.
• Spanner: Becoming a SQL System — Bacon et al. 2017 — follow-up paper on Spanner's SQL layer.

Videos

• Consistent Hashing | Algorithms You Should Know — ByteByteGo · 7 min — the canonical animated explanation; watch first.
• How do databases scale? Sharding, Replication — ByteByteGo — ByteByteGo · 14 min — strategies and trade-offs in plain English.
• CMU 15-721 — Database Sharding & Partitioning — Andy Pavlo · 1 h 14 min — the deep academic treatment; partitioning schemes, query routing.
• How Google Spanner Works — ByteByteGo / Spanner authors — 16 min — TrueTime + global consistency walkthrough.
• Vitess: How we scale Slack's MySQL fleet — Slack/Vitess · 31 min — a real shop's sharding playbook.

LeetCode — Consistent Hashing Design

• Link: https://leetcode.com/problems/consistent-hashing-design/
• Difficulty: Medium
• Why this problem: Treat as design — a TreeMap of hash→node; lookup is upper-bound.
• Time-box: 30 minutes. Look up the editorial only after.

Assignment / Deliverables

Give yourself a clean 2-hour window and complete all of these before moving on:

1. Read the deep-dive above end-to-end — no skimming. Take notes in your own words.
2. Solve the LeetCode problem below without help first. Only look at the hint after 15 focused minutes; only look at editorial after 30. Log your time.
3. Reproduce one code snippet locally. Pick the snippet that felt least obvious and get it running in a scratch file / notebook.
4. Draw the core diagram from memory. Paper, whiteboard, or tldraw — doesn't matter. If you can't, re-read section 2 and try again.
5. Write a 3-line takeaway in your prep journal: what surprised you, what you still don't understand, what you'd read next.
6. Skim one item from the Reading material section. Bookmark the rest for the weekend.
7. Commit any code + notes to your prep repo with message session-NN: <one-line summary>.

Stretch (optional, +30 min): explain today's topic to a rubber duck / a friend / a voice note. If you can't teach it in 5 minutes, you don't own it yet — flag it and revisit next weekend.

Post-session checklist

By the end of this session you should be able to:

• Pick a partition strategy for a chat app, a time-series store, a multi-tenant SaaS.
• Diagram consistent hashing with virtual nodes; explain what happens on node add.
• List 3 mitigations for a hot key and 2 for a hot shard.
• Compare leader-follower, multi-leader, leaderless on conflict resolution.
• Explain why Spanner can be strongly consistent across regions.
• Solve consistent-hashing-design — TreeMap of hash → node; upper-bound lookup.

Generated from sessions_data.py + content_part*.py. To edit a video / leetcode / title, edit the data file and re-run write_sessions.py.