Replication

Why replicate: keep data geographically close to users (lower latency), keep working despite failures (availability + durability), and scale read throughput (more machines serving reads).

Backups are still needed. Replicas reflect writes quickly, including accidental deletions, which propagate everywhere. Backups store old snapshots so you can go back in time. They're complementary (a snapshot often bootstraps a new replica).

Fullscreen

---

1. Single-Leader Replication

Also called leader-based, primary-backup, or active/passive. (Avoid the old "master-slave" term.)

Fullscreen

1. One replica is the leader; all writes go to it first (written to its local storage).
2. The leader sends every change to followers via a replication log / change stream; each follower applies writes in the same order.
3. Reads can go to the leader or any follower; writes only to the leader.

Used everywhere: PostgreSQL, MySQL, Oracle, SQL Server Always On, MongoDB, DynamoDB, Kafka, and consensus algorithms like Raft (which auto-elects a new leader). If sharded, each shard has its own leader (Chapter 7).

Synchronous vs asynchronous replication

Fullscreen

• Synchronous: leader waits for the follower's confirmation before reporting success. Pro: follower is guaranteed current. Con: if the sync follower is down/slow, all writes block.
• Asynchronous: leader doesn't wait. Pro: leader keeps working even if followers lag. Con: a confirmed write can be lost if the leader fails before replicating (durability not guaranteed).
• Semisynchronous: one follower synchronous, the rest async (guarantees an up-to-date copy on ≥2 nodes; promote another follower to sync if the sync one fails).
• Quorum: a majority (e.g., 3 of 5) updated synchronously, used in eventually consistent / consensus systems.
• Fully async is common despite weaker durability, especially with many or geo-distributed followers.

Setting up new followers (without downtime)

A plain file copy is inconsistent (data changes during the copy). Instead:

1. Take a consistent snapshot of the leader (ideally without locking, needed for backups anyway).
2. Copy the snapshot to the new follower.
3. Follower requests all changes since the snapshot's exact log position (PostgreSQL: log sequence number; MySQL: binlog coordinates / GTIDs).
4. When it has processed the backlog, it has caught up and streams changes live.

Object storage / zero-disk architecture: archiving the replication log + snapshots to S3-style storage (WAL-G, Litestream) enables backups and easy follower setup. Some databases serve live queries from object storage (cheap, multi-zone durable, conditional-write/CAS for transactions and leader election). Trade-offs: high latency, per-API fees (forces batching), immutable objects. Zero-disk architecture (ZDA) persists everything to object storage with disks/RAM only as cache (WarpStream, Bufstream, Redpanda Serverless, most cloud warehouses).

Handling node outages

• Follower failure (catch-up recovery): the follower knows its last processed transaction from its local log; on reconnect it requests the changes it missed. Simple, but a long absence means a heavy catch-up backlog (load on both nodes). The leader can't delete log entries an unavailable follower hasn't acknowledged (risk: running out of disk, or forcing the follower to restore from backup).
• Leader failure (failover): promote a follower, reroute writes, point other followers at the new leader.

Fullscreen

Failover dangers in detail:

• Lost writes (async): the new leader may lack the old leader's latest writes; rejoining old leader's writes are usually discarded, so "committed" writes weren't durable.
• Dangerous discards across systems: GitHub incident, a stale MySQL follower was promoted; its autoincrement counter reused primary keys that Redis still referenced, leaking private data to wrong users.
• Split brain: two leaders both accept writes → corruption. Fencing limits/shuts down the old leader (must be careful not to shut down both).
• Timeout tuning is genuinely hard (load spikes / network glitches can trigger needless failovers that worsen things).

→ Many teams prefer manual failover. Always promote the most up-to-date follower. These are fundamental distributed-systems problems (Chapters 9-10).

Implementation of replication logs

Fullscreen

---

2. Problems with Replication Lag (eventual consistency)

Read scaling (many async followers serve reads) works only with async replication, but a lagging follower returns stale data. If you stop writing, followers eventually catch up: eventual consistency. "Eventually" is deliberately vague, lag is usually sub-second but can grow to minutes under load/network problems. Three concrete anomalies:

Fullscreen

Read-your-writes (read-after-write) consistency

You submit data, then read from a stale follower → it looks like your write was lost. Guarantee: you always see your own updates (says nothing about other users). Techniques:

• Read things the user may have edited from the leader (e.g., always read your own profile from the leader, others' from followers).
• For ~1 minute after a user's last write, read from the leader; or avoid followers lagging > 1 minute.
• Client remembers the timestamp of its last write; serve reads only from replicas current to that timestamp (logical or system-clock timestamp).
• Cross-device read-after-write needs centralized metadata and routing all of a user's devices to the same region.

Regions vs availability zones: a region = datacenters in one geographic location; each datacenter is a zone. Zones in a region have very low latency (treat as one). Multi-zone survives a zonal outage; only multi-region survives a regional outage (at higher latency/cost).

Monotonic reads

Reading from different followers, a user can see a value, then a later read (from a more-lagging follower) makes it disappear, time goes backward. Guarantee (monotonic reads): once you've seen newer data, you won't later see older data. Stronger than eventual, weaker than strong consistency. Achieve it by routing each user to the same replica (e.g., hash of user ID).

Consistent prefix reads

If writes happen in a causal order (a question, then its answer), an observer must see them in that order, otherwise the answer appears before the question (causality violation). Especially a problem in sharded databases where shards replicate independently with no global ordering. Fixes: keep causally related writes on the same shard, or track causal dependencies (happens-before, later in chapter).

Solutions for replication lag

Think about behavior when lag hits minutes/hours. Pretending async replication is synchronous is a recipe for bugs. Handling these in app code is complex and error-prone, the simplest model is a database offering strong consistency (linearizability, Chapter 10) and ACID transactions (Chapter 8), treating it like a single node. The NewSQL trend offers strong consistency and distributed scalability/availability. Still, weaker-consistency replication has real advantages: better resilience to network interruptions and lower overhead.

---

3. Multi-Leader Replication

Single-leader's flaw: all writes funnel through one node, lose the leader (or its network) and you can't write. Multi-leader (active/active, bidirectional) lets multiple nodes accept writes, each forwarding to the others; each leader is also a follower of the others. Synchronous multi-leader ≈ single-leader, so the focus is asynchronous multi-leader.

Geographically distributed operation

Fullscreen

Within a region: normal leader-follower. Between regions: each region's leader replicates to the others. Compared with single-leader across regions:

Multi-leader is less common, often a retrofitted feature with subtle pitfalls (autoincrement keys, triggers, constraints), frequently considered "dangerous territory."

Topologies

Fullscreen

• All-to-all is most fault-tolerant (multiple paths), but messages can overtake each other → causality bugs (an update arrives before its insert). Fixing this needs version vectors, not timestamps (clocks can't be trusted).
• Circular / star need each node to forward writes (tagged with node IDs to prevent infinite loops); a single node failure interrupts the flow.

Sync engines and local-first software

Multi-leader is ideal for apps that must work offline: each device is a replica/leader; an async sync engine reconciles them when online (lag can be hours/days). Same shape as cross-region multi-leader, taken to the extreme.

• Real-time collaboration (Google Docs, Figma, Linear) is also multi-leader: each open tab is a replica; edits replicate asynchronously and show with low latency.
• Offline-first = keeps working offline. Local-first = keeps working even if the vendor shuts down (open sync protocol; Git is a local-first example).
• Sync-engine pros: instant UI (no network round-trip), offline support, simpler declarative programming (local reads/writes rarely fail), reactive updates. Best when all needed data fits on the client (not for huge catalogs). Examples: Firestore, Realm, Ditto (proprietary); PouchDB/CouchDB, Automerge, Yjs (open). Games' equivalent is netcode.

Dealing with conflicting writes

Concurrent writes on different leaders can conflict (two users rename the same wiki page to B and C). Concurrent means neither write was aware of the other (not necessarily the same wall-clock time).

Fullscreen

• Conflict avoidance: route all writes for a record to the same leader (e.g., a user's data always to their home region). Breaks down if the designated leader changes. (ID trick: one leader makes odd IDs, another even.)
• Last write wins (LWW): keep the value with the greatest timestamp, discard others. Simple, but silently loses data for concurrent writes (and "latest" is undefined for concurrent writes → effectively random; clock-sensitive, use a logical clock).
• Manual resolution: store all concurrent values as siblings, return them on read, resolve in app/UI. Problems: the API changes (string → set), user effort, careless auto-merge bugs (Amazon's cart anomaly, deleted items reappear when merging by set union).
• Automatic resolution: algorithms that merge concurrent writes so all replicas converge = strong eventual consistency. Examples by data type: text (preserve all inserts/deletes), collections (track deletions to avoid the cart bug), counters (sum increments/decrements correctly), key-value maps (merge per key). Limits exist (can't enforce "≤5 items" under concurrency without dropping some).
• CRDTs vs OT: two algorithm families for automatic merge.
  • OT (operational transformation): record insert/delete by index, then transform indexes to account for concurrent ops (used in Google Docs).
  • CRDT (conflict-free replicated datatype): give each element a unique immutable ID, order by IDs (no transformation needed; used in Redis Enterprise, Riak, Cosmos DB; sync engines Automerge/Yjs).
• Subtle conflicts: e.g., two overlapping meeting-room bookings created near-simultaneously, both saw the room free before inserting. (More in Chapters 8 and 13.)

---

4. Leaderless Replication (Dynamo-style)

No leader, any replica accepts writes directly (or via a coordinator that doesn't enforce ordering). Forgotten during the relational era, revived by Amazon's Dynamo (2007); open-source: Riak, Cassandra, ScyllaDB. (Note: the newer cloud DynamoDB is actually single-leader/Multi-Paxos, different from Dynamo.)

Writing when a node is down

No failover, the client sends writes to all replicas in parallel and accepts success once enough (e.g., 2 of 3) acknowledge. A recovered node has missed writes → returns stale reads. So clients read from several replicas in parallel too, and pick the value with the greatest version number/timestamp.

Catching up on missed writes:

• Read repair: on a multi-replica read, the client detects stale replicas and writes the newer value back. (Good for frequently-read data.)
• Hinted handoff: another replica temporarily stores writes ("hints") for an unavailable one, forwarding them on recovery. (Covers rarely-read data.)
• Anti-entropy: a background process compares replicas and copies missing data (no ordering, may lag).

Quorums for reading and writing

Fullscreen

With n replicas, a write needs w acks and a read queries r nodes. If w + r > n, the read and write sets overlap → at least one read sees the latest write. Common: n odd (3 or 5), w = r = (n+1)/2. Tolerance: w < n → tolerate write-time node loss; r < n → tolerate read-time loss. (n=3,w=2,r=2 tolerates 1 down; n=5,w=3,r=3 tolerates 2.) Quorums needn't be majorities, only the overlap matters. Smaller w/r (w+r ≤ n) → more stale reads but lower latency / higher availability.

Limitations of quorum consistency

Even with w + r > n, edge cases break the guarantee:

• A new-value node fails and is restored from an old-value replica → fewer than w copies of the new value.
• During rebalancing, read/write sets may stop overlapping.
• A read concurrent with a write may or may not see the new value (and a later read might see the old one).
• A write that succeeded on fewer than w replicas isn't rolled back on the ones where it succeeded.
• Clock-based LWW can silently drop writes.
• Concurrent writes create conflicts (resolved later).

→ Dynamo-style DBs are tuned for eventual consistency; w/r adjust the probability of staleness, not a hard guarantee. Monitor staleness (harder than leader-based lag, since there's no fixed write order).

Single-leader vs leaderless performance

• Reading from a single leader for freshness limits read throughput, is sensitive to leader slowness, and pauses during failover.
• Leaderless is more resilient: no failover, requests go to multiple replicas anyway, so one slow/unavailable replica barely matters. Request hedging (use the fastest responses) cuts tail latency, and it handles gray failures (degraded-but-not-down nodes) gracefully.
• But: detecting unavailable replicas + hinted handoff adds load; bigger r/w means waiting on more (possibly slow) replicas (quorums rarely exceed 4/7 or 5/9); a large network partition can prevent forming a quorum → some systems allow a sloppy quorum (Cassandra/ScyllaDB: consistency level ANY) where any reachable replica accepts writes (no guarantee later reads see them, but better than failing).

Multi-region (leaderless)

Suited to multi-region (tolerates conflicts, interruptions, latency spikes). A client writes to a local coordinator node, which forwards to local replicas + one replica per other region. Consistency levels let you require a quorum across all regions, per-region, or local-only (local quorum = lower latency, more stale).

Detecting concurrent writes

Because writes arrive in different orders at different nodes (variable delays, partial failures), replicas can permanently diverge unless conflicts are resolved. The central concept here is the happens-before relation:

• A happens before B if B knows about / depends on / builds upon A (B is causally dependent on A).
• Two operations are concurrent if neither happens before the other, regardless of wall-clock time (concurrency = mutual unawareness; analogy to relativity).

Capturing happens-before (single replica):

• Server keeps a version number per key, incremented on each write.
• On read, the server returns all siblings (non-overwritten values) + the latest version.
• A client must read before writing, then include the prior version number and merge the values it saw.
• A write with version V overwrites everything ≤ V (already merged) but keeps higher-versioned values as concurrent siblings.

This way the server detects concurrency by version numbers alone (no need to interpret values). Worked example: two clients adding to a shared shopping cart end up with merged siblings, no writes lost.

Version vectors (multiple replicas): a single version number isn't enough with multiple replicas accepting writes. Use a version number per replica per key, the collection is a version vector (Riak's "causal context"; the dotted version vector variant is used in Riak 2.0). Sent to clients on read, returned on write; lets the DB distinguish overwrites from concurrent writes and makes it safe to read from one replica and write to another.

Version vector vs vector clock: related but not identical; for comparing replica states, version vectors are the correct structure.

---

Real-world examples and analogies

• Single vs multi vs leaderless, one cashier, several cashiers, or self-service. Single-leader: one cashier rings up every sale (consistent, but a bottleneck and a single point of failure). Multi-leader: several cashiers each take sales and reconcile their tills later (fast, but two might sell the last ticket). Leaderless: self-checkout where you scan at several machines and the system reconciles by majority vote.
• Sync vs async replication, certified mail vs regular mail. Synchronous is certified mail: you wait for the signed receipt before moving on (safe, slow, blocks if the recipient is out). Asynchronous is dropping it in the mailbox and walking away (fast, but it might never arrive if you vanish first).
• Failover split brain, two acting captains. The captain goes silent; the first mate takes command. Then the captain reappears, still issuing orders. Now two people command the ship and crash it, unless there's a rule (fencing) that forcibly relieves the old captain.
• Replication lag / read-your-writes, posting then refreshing. You post a comment and immediately refresh, but the refresh hits a stale server that hasn't seen your post, it looks deleted. Read-your-writes is the promise that your own post always shows up for you.
• Monotonic reads, a clock that ticks backward. You see a new message, refresh, and it's gone because the next server lags. Monotonic reads guarantee the clock only moves forward for any one reader.
• Consistent prefix, overhearing a reply before the question. Through laggy channels you hear "About 10 seconds, Mr. Poons" before "How far can you see?", the answer precedes the question. Consistent prefix keeps causally ordered writes in order.
• LWW, two edits, one survives by coin flip. Two people rename a file at "the same time"; the system keeps whichever timestamp is larger and silently throws the other away, effectively a coin flip, and someone's work is gone.
• CRDT shopping cart, tracking removals, not just contents. Instead of merging two carts by union (which resurrects deleted items), the CRDT remembers "Book was removed," so the merge respects the deletion.
• Quorum (w + r > n), overlapping shifts. If every fact is written to 2 of 3 guards and you ask 2 of 3 guards, the two groups must share at least one guard who knows the latest, so you can't miss a recent update.
• Request hedging, asking three friends and using the first reply. You text three friends the same question and act on whoever answers first, instead of waiting on the slow one. That's how leaderless replication shrinks tail latency.

---

Cheat-sheet flashcards

Cover the answer and recall it.

• Three reasons to replicate? → Latency (proximity), availability/durability, read throughput.
• Do you still need backups with replication? → Yes, backups go back in time; replication propagates deletions too.
• Three replication families? → Single-leader, multi-leader, leaderless.
• In single-leader, where do writes go? → Only to the leader; reads can go to any replica.
• Synchronous vs asynchronous replication? → Wait for follower ack vs send-and-forget.
• What is semisynchronous? → One sync follower, the rest async.
• Risk of fully async replication? → Confirmed writes can be lost if the leader fails.
• Steps to set up a new follower? → Snapshot, copy, request changes since the snapshot's log position, catch up.
• Follower recovery method? → Catch-up: request missed changes from its last logged position.
• Three steps of failover? → Detect failure, choose new leader, reconfigure clients.
• What is split brain? → Two nodes both believe they're leader.
• What is fencing? → Shutting down/limiting an old leader to prevent split brain.
• Why pick the most up-to-date follower in failover? → Minimize lost writes.
• Three replication-log methods? → Statement-based, WAL shipping, logical (row-based) log.
• Why does WAL shipping hurt upgrades? → It's storage-coupled; leader/followers can't differ in version.
• What does a logical log enable? → Cross-version upgrades and change data capture (CDC).
• What is eventual consistency? → Replicas converge if writes stop; reads may be temporarily stale.
• Read-your-writes consistency? → You always see your own updates.
• Monotonic reads? → You never see data go backward in time.
• Consistent prefix reads? → Causally ordered writes are seen in order.
• One way to get monotonic reads? → Route each user to the same replica.
• Multi-leader's main use case? → Geo-distributed / offline / collaborative apps.
• Multi-leader's biggest downside? → Weak consistency; can't enforce cross-leader constraints (uniqueness, non-negative balance).
• Three multi-leader topologies? → All-to-all, circular, star.
• All-to-all topology's risk? → Writes overtaking each other (causality bugs).
• What is a sync engine? → A library that reconciles local replicas with the server (offline + collaboration).
• Offline-first vs local-first? → Works offline vs works even if the vendor shuts down.
• Four conflict strategies? → Avoidance, LWW, manual (siblings), automatic merge.
• Why is LWW lossy? → It discards all but one concurrent write.
• What is strong eventual consistency? → Eventual consistency + guaranteed convergence via automatic merge.
• CRDT vs OT? → Unique element IDs (no transform) vs index-based ops that are transformed.
• Leaderless: what handles missed writes? → Read repair, hinted handoff, anti-entropy.
• The quorum condition? → w + r > n (read and write sets overlap).
• Typical quorum settings? → n odd; w = r = (n+1)/2.
• Does w + r > n guarantee fresh reads? → Usually, but several edge cases break it.
• What is a sloppy quorum? → Letting any reachable replica accept writes during a partition.
• What is request hedging? → Using the fastest of parallel replica responses to cut tail latency.
• When are two operations concurrent? → Neither happens-before the other (mutual unawareness).
• What is a version vector? → A per-replica version number per key, to detect concurrency across replicas.
• Version vector vs vector clock? → Related; version vectors are correct for comparing replica states.

---

Common interview questions

1. Compare single-leader, multi-leader, and leaderless replication. (Who accepts writes; consistency vs availability/latency trade-offs; typical systems.)
2. Synchronous vs asynchronous replication, what does each guarantee and risk? (Up-to-date follower vs blocking; lost writes vs leader keeps working; semisynchronous/quorum middle ground.)
3. Walk through leader failover and the things that can go wrong. (Detect/elect/reconfigure; lost writes, split brain + fencing, timeout tuning; GitHub incident.)
4. Compare the three replication-log methods. (Statement-based nondeterminism; WAL coupling/upgrades; logical log decoupling + CDC.)
5. What is eventual consistency, and what three anomalies does replication lag cause? (Read-your-writes, monotonic reads, consistent prefix, with fixes.)
6. How do you implement read-your-writes consistency? (Read editable data from the leader; timestamp-based reads; route by user; cross-device caveats.)
7. When would you use multi-leader replication, and what are the risks? (Geo/offline/collaboration; weak consistency, conflicts, retrofit pitfalls.)
8. How do you resolve write conflicts in multi-leader/leaderless systems? (Avoidance, LWW (lossy), siblings, automatic merge; CRDT vs OT; strong eventual consistency.)
9. Explain quorums and the w + r > n condition. Does it guarantee fresh reads? (Overlap argument; tolerance math; edge cases that break it.)
10. What are read repair, hinted handoff, and anti-entropy? (Mechanisms for leaderless replicas to catch up.)
11. Why can leaderless replication be more resilient than single-leader? (No failover, parallel requests, request hedging, gray-failure tolerance, plus its costs.)
12. What does "concurrent" mean in distributed systems, and how do version vectors detect it? (Happens-before / mutual unawareness; per-replica version numbers; siblings + merge.)

---

Key terms glossary

---

Forward links (where these ideas return)

• Sharding/partitioning of large datasets → Chapter 7
• Request routing to the right leader/shard → Chapter 7
• Transactions and constraints → Chapter 8
• Unreliable clocks, network faults, gray failures → Chapter 9
• Linearizability, consensus, leader election, shared logs → Chapter 10
• Change data capture and stream processing → Chapter 12
• Scalable conflict detection/resolution → Chapter 13