Source: lessons/day-11-replication.md

Day 11 — Replication: Leader-Follower, Multi-Leader, Leaderless

Time budget: ~60 min · Level: The most consequential distributed-systems topic before CAP theorem. Nearly every non-trivial system uses replication, and nearly every production incident touches it.

Replication is the mechanism that makes data survive machine failures, scale reads geographically, and reduce read latency by putting copies closer to users. It's also the source of the subtlest bugs you'll encounter: stale reads, lost writes on failover, split-brain, replication lag, and conflicts.

Today: the map. Tomorrow's topics (sharding, consistent hashing, CAP) all assume you understand this cold.

Why replicate — the four reasons

  1. High availability. If one machine dies, another has the data.
  2. Read scaling. N replicas ≈ N× read throughput.
  3. Geographic locality. A replica in the user's region cuts read latency from 150ms → 5ms.
  4. Disaster recovery. Replicas in a different data center / region survive DC-level failures.

Every design that involves durable state involves at least one of these. Understanding which problem replication is solving in a given design tells you which topology to pick.

The three replication topologies

1. Leader-Follower (Single-Leader)

Model: One node is the leader (aka primary, master). All writes go to the leader. Writes are then replicated to followers (aka replicas, secondaries, slaves). Reads can go to leader or followers.

                  ┌──────────┐
                  │  LEADER  │
                  │  (writes)│
                  └────┬─────┘
                       │  WAL / binlog stream
        ┌──────────────┼──────────────┐
        ▼              ▼              ▼
  ┌──────────┐  ┌──────────┐  ┌──────────┐
  │ FOLLOWER │  │ FOLLOWER │  │ FOLLOWER │
  │  (reads) │  │  (reads) │  │  (reads) │
  └──────────┘  └──────────┘  └──────────┘

Real-world uses: PostgreSQL streaming replication, MySQL binlog replication, MongoDB replica sets (with automatic election), Redis Sentinel, Kafka partitions.

Sync vs Async: - Synchronous: leader waits for at least one follower to acknowledge before returning success. Zero data loss on leader failure, but a slow/dead follower blocks writes. - Asynchronous: leader returns as soon as it writes locally. Fast, but if leader dies before replication catches up, data is lost. - Semi-synchronous (production standard): wait for one follower to ack, but move on if slow. Balance of durability and availability.

Strengths: - Simple mental model: one place to write. - No write conflicts (there's only one writer). - Easy to reason about consistency.

Weaknesses: - Write throughput bounded by the single leader's capacity. - Failover complexity — when the leader dies, someone must promote a follower. Automatic failover mechanisms (Patroni, MHA, MongoDB elections) exist but are non-trivial to operate. - Split-brain risk during failover (see later section). - Replication lag — followers lag behind leader; reads from followers may be stale.

2. Multi-Leader (Master-Master)

Model: Multiple nodes accept writes. Each leader replicates its writes to the other leaders. Common in multi-region deployments where each region has its own leader.

       ┌──────────┐         ┌──────────┐
       │ LEADER-A │ ◄────►  │ LEADER-B │
       │ (us-east)│         │ (eu-west)│
       └──────────┘         └──────────┘
             ▲                    ▲
             │                    │
       (US writes)          (EU writes)

Real-world uses: MySQL circular replication, PostgreSQL BDR / EDB, Cassandra (arguably — but leaderless is a more accurate label), CouchDB, DynamoDB Global Tables.

Strengths: - Write scaling — each region handles writes locally. - Geographic write latency — Sydney users write to Sydney leader (5ms), not London leader (170ms). - Regional failure tolerance — losing a region doesn't stop writes elsewhere.

Weaknesses (this is where it gets nasty): - Write conflicts. If Alice updates her profile in us-east while a concurrent write updates it in eu-west, both writes must be reconciled somehow. Options: - Last-Write-Wins (LWW): simple, but silently discards data. - Vector clocks: track causal history; still requires human reconciliation for concurrent updates. - CRDTs (Conflict-free Replicated Data Types): mathematically-designed types (counters, sets, maps) that merge deterministically. Google Docs and Redis Enterprise Active-Active use CRDT-based approaches. - Application-level merge functions. - Consistency is at best eventual. Not suitable for money, inventory, or anything with cross-record invariants. - Higher operational complexity than leader-follower.

Rule of thumb: avoid multi-leader unless you have a specific need (geographic write locality, offline-first mobile apps like Notion/Linear). Most teams pick leader-follower + read replicas and stay much healthier.

3. Leaderless (Dynamo-style)

Model: No designated leader. Every node can accept reads AND writes. Coordination happens via quorums at read/write time.

                  ┌───────┐
                  │ NODE-A│
                  └───────┘
                     ▲
                     │  writes go to N nodes;
                     │  reads consult R nodes;
       ┌───────┐     │     ┌───────┐
       │ NODE-B│ ◄───┼───► │ NODE-C│
       └───────┘     │     └───────┘
                     │
                     ▼
                  ┌───────┐
                  │ NODE-D│
                  └───────┘

Real-world uses: Amazon Dynamo (the paper), Cassandra, ScyllaDB, Riak, Voldemort.

Key parameters: - N — total number of replicas per key (typically 3). - W — number of replicas that must acknowledge a write. - R — number of replicas that must respond to a read. - Golden rule: if W + R > N, at least one replica in every read overlaps with the writes, guaranteeing read-your-writes consistency and preventing stale reads.

Common configurations (N=3): - W=1, R=1 — fastest, most tolerant. Weakest consistency (W+R=2, not >N=3). - W=2, R=2 — balanced. W+R=4>3 — strong consistency. - W=3, R=1 — write-slow, read-fast. All replicas must ack writes. Any read is guaranteed fresh. - W=1, R=3 — write-fast, read-slow. Read must consult all replicas.

Handling failures: - Read repair: if a read discovers a stale replica, update it inline. - Hinted handoff: if a target replica is down, a nearby node stores a hint and replays when it comes back. - Anti-entropy (Merkle-tree sync): periodic background comparison to catch drift.

Strengths: - No single point of failure — any node can serve any request. - Linear scale-out — add nodes for capacity. - Tunable consistency per query (ONE, QUORUM, ALL).

Weaknesses: - Weaker consistency semantics — even with W+R>N, some anomalies (like lost updates on concurrent writes to the same key) require additional mechanisms (LWT/Paxos in Cassandra). - Read amplification — R replicas queried per read. - Application must reason about quorums.

The bomb everyone must understand: replication lag

Even with sync replication, there's always a moment where a write has landed on the leader but not yet on the followers. This is replication lag, and it's the source of the most-reported "the app is buggy" issues.

The user story

  1. Alice updates her profile picture (write to leader).
  2. Alice reloads the profile page (read hits a follower — lag = 200ms).
  3. Alice sees the OLD picture.
  4. Alice thinks the update failed and updates again.
  5. Repeat.

Mitigations

  • Read-your-writes consistency: route reads from the same user session to the leader for a few seconds after a write. Or track the write's LSN/timestamp and only read from a replica caught up to that point.
  • Sticky sessions: the same user's reads always go to the same region/replica (Day 4).
  • Monotonic reads: each subsequent read must return a version at least as fresh as the previous read (typically enforced via session tokens).
  • In-app "wait for replication" flag: if the previous write was very recent, force the next read to hit the leader.

Replication lag SLI — what to alert on

  • Lag in seconds (Postgres: pg_stat_replication.replay_lag).
  • Lag in bytes (pg_current_wal_lsn() - replay_lsn).
  • Alert threshold typically 1s for OLTP replicas, 10s for reporting/analytics.

Failover — where careers die

Scenario: Leader crashes. What happens?

Automatic failover steps

  1. Detection. Some watchdog (Patroni, Sentinel, MongoDB primary election) notices the leader is unreachable.
  2. Election. A quorum of remaining nodes elects a new leader (usually the follower with the least replication lag).
  3. Reconfiguration. Clients are redirected to the new leader (via DNS update, virtual IP failover, or client-side driver discovery).
  4. Old leader recovery. If the old leader comes back, it must be reconfigured as a follower (or fenced off entirely to prevent split-brain).

Split-brain — the classic failure mode

Two nodes both believe they are the leader. Both accept writes. When they reconcile, one set of writes must be dropped. Real data loss.

How split-brain happens: - Network partition between old and new leaders — both see the other as "dead." - Failover fires, but old leader is not actually dead — it comes back thinking it's still leader. - DNS TTL causes some clients to still write to the old leader.

Preventing split-brain: - Quorum-based election: need > N/2 nodes to agree on new leader. If old leader is on the minority side of a partition, it can't have a quorum, so it must stop accepting writes. - STONITH (Shoot The Other Node In The Head): the new leader physically or logically kills the old leader (e.g., revokes its cloud VM's network access). - Fencing tokens: each new leader gets an incrementing "epoch" number. All writes carry this token. Old leader's writes are rejected because their epoch is stale.

Practical: MongoDB replica sets, Redis Sentinel, Patroni for Postgres — all implement quorum-based election + fencing. Roll-your-own failover is a bad idea.

WAL / redo log shipping — how replication physically works

Most leader-follower systems ship the transaction log (WAL in Postgres, binlog in MySQL, oplog in MongoDB) to followers. The follower replays the log to reproduce the leader's state.

Three flavors: - Statement-based: ship the SQL statements. Simple but breaks on non-determinism (NOW(), RAND(), non-deterministic functions). - Row-based: ship the result of each modification (which rows changed, to what values). Deterministic, larger log volume. Modern default. - Physical (page-level): ship the actual disk-page changes. Fastest, but replicas must be byte-identical to the primary (same DB version, same OS, etc.). Postgres streaming replication.

Why this matters for you: - Statement-based replication can silently diverge — beware. - Physical replication in Postgres is the fastest but restricts you to matching binaries. - Logical replication (Postgres 10+) is row-based and enables cross-version upgrades and selective table replication.

Chain replication — the "exotic" alternative

Writes go to one end of a chain (head), reads from the other (tail). Each node passes the write to the next.

Strengths: simple correctness proofs, strong consistency, low overhead. Weaknesses: high write latency (must traverse whole chain), tail node is the read bottleneck.

Real use: internally in some object stores (Google's original Chubby, some Kubernetes etcd optimizations). Not commonly seen in application-level DBs.

Read replicas — where staleness bites

The most common replication use in practice: read replicas for scaling.

Bug patterns: 1. Read your own write returns nothing. User creates a record, immediately searches for it, hits a replica, sees empty result. 2. Analytics counters going backward. Two consecutive reads from different replicas can appear to move backward if one is more caught up than the other. 3. Session store on a replica. Reading auth session from a lagging replica → user gets logged out mid-session.

Golden rules: - Never put session data on read replicas without read-your-writes safeguards. - Analytics is OK to serve from replicas if you tolerate 1-10s staleness (usually fine). - Reports are OK. Live user data is NOT unless the design accounts for lag.

Multi-leader conflict resolution — when concurrent writes collide

If leader A does SET x=5 while leader B does SET x=7 concurrently, what wins after replication converges?

Strategies

  1. Last-Write-Wins (LWW). Attach a timestamp; latest timestamp wins. Simple but silently drops the losing write. Also vulnerable to clock skew.

  2. Vector clocks. Each node maintains a counter incremented per write. Two writes with concurrent versions (neither strictly before the other) are surfaced to the application to resolve.

  3. CRDTs. Data structures that provably converge: - G-Counter: grow-only counter (each node keeps its own count, total = sum). - PN-Counter: two G-counters (positive and negative), can decrement. - G-Set, 2P-Set, OR-Set: add-only, add-and-remove, add-with-tombstone. - LWW-Register with vector clocks. - RGA: replicated growable array (for text/CRDTs in Google Docs, Notion).

  4. Application-level merge. If two shopping carts diverge, merge them (union of items).

Modern reality: multi-leader is niche. Most teams that think they need it actually just need read replicas + write to the primary region.

Cassandra as a concrete leaderless case study

Cassandra combines: - Consistent hashing (Day 13) to determine which nodes own which keys. - Replication factor N=3 by default; keys stored on 3 consecutive nodes on the ring. - Configurable consistency per query: ONE, QUORUM (=2 for N=3), ALL, LOCAL_QUORUM (quorum within a DC). - Hinted handoff: if a target is down at write time, another node stores a hint and replays when it recovers. - Read repair: if a read discovers stale data, background-update the lagging replica. - Anti-entropy: periodic Merkle-tree comparisons.

For write-heavy workloads with tunable consistency, Cassandra is the reference. For strict consistency, choose Postgres + read replicas or Spanner/CockroachDB (SQL-with-Paxos).

Interview angles

"Design a global user-profile service with 50ms P99 read latency worldwide."

Answer skeleton: 1. Reads must be local → replicas in each region. 2. Writes are rare (~1% of ops) → single leader is fine. 3. Pick leader in us-east; async replicas in eu-west and ap-south. 4. Read from local replica, with staleness tolerance of ~1s (acceptable for profiles). 5. Failover: use managed failover (RDS Multi-AZ + read replicas or equivalent). 6. Capacity: 100M users × ~1KB = 100GB — single-region primary + 2 replicas fits easily. 7. Alert on replication lag > 5s.

"What happens if your Postgres primary dies during a payment write?"

Complete answer: 1. If synchronous replication is on and one replica had acked, we're safe — that replica has the write; promote it. 2. If we were on async, the last few writes may be lost when the replica is promoted. Application should have idempotency keys so retries don't cause double-writes. 3. Split-brain risk: the promoted replica must fence the old primary via STONITH. 4. Cost of sync replication: increased write latency (need to wait for one replica). Enterprise DBs typically use semi-sync. 5. For payments, we'd also have an audit log in a separate Kafka topic; even if the DB write is lost, we can replay from the log.

Spaced-repetition callback in today's quiz

  • Slow-start applies with every algorithm (Day 4 Q4 — you missed this) — surfacing in Q4 today as T/F, in the context of read-replica warm-up after failover. Same concept, new domain.
  • Capacity math habit — Q8 has multi-region replication traffic math to keep the habit sharp.
  • Post-mortem structure (Days 3/5/6/7/8/9/10 Q9) — Q9 today is a replica-lag incident.