1. The problem: why keep more than one copy?

Imagine Instagram's user database lived on exactly one PostgreSQL server. Four things go wrong as the product grows:

• Availability. Servers die: disks fail, kernels panic, a power supply pops, someone runs the wrong command, a whole data center loses power. With one copy, every login, profile view and follow fails until the machine is repaired — possibly hours. A company promising "four nines" (99.99%) gets ~52 minutes of downtime per year.
• Read scaling. Instagram is massively read-heavy — people view profiles, posts and follower lists far more often than they change them (assume ~100 reads per write). One server can serve perhaps tens of thousands of queries per second; we need millions.
• Latency. A user in Mumbai querying a database in Virginia pays ~200+ ms per round trip. A copy in Asia answers in a few ms.
• Durability. A committed write fsync'd to one disk is safe from crashes, not from that disk (or building) being destroyed. Real durability means "stored on several machines, ideally in several places".

Replication — keeping a copy of the same data on multiple machines (replicas) connected by a network — addresses all four. If the data never changed, replication would be trivial: copy once. All the difficulty, and this whole lesson, is about handling changes: how writes reach every replica, how long that takes, and what users see in the meantime.

2. Vocabulary you'll need

There are three fundamental architectures — single-leader, multi-leader and leaderless — and each can be combined with sync or async propagation. We'll go through them in that order.

3. Single-leader (leader–follower) replication

The most common design, used by PostgreSQL, MySQL, SQL Server, MongoDB replica sets, Redis, and Kafka partitions:

1. One replica is designated the leader. All writes from clients go to it.
2. The leader writes the change to its local storage and also appends it to the replication log.
3. Each follower pulls (or is pushed) the log and applies the changes in the same order, so it ends up with an identical copy.
4. Reads can go to the leader or any follower.

3.1 Synchronous vs asynchronous vs semi-synchronous

The key question: when does the leader tell the client "your write succeeded"?

 SYNCHRONOUS                                   ASYNCHRONOUS
 Client     Leader      Follower               Client     Leader      Follower
   | write    |             |                    | write    |             |
   |--------> | write local |                    |--------> | write local |
   |          |--- log ---->|                    | <-- OK --|             |   ← client is done
   |          |             | apply + fsync      |          |--- log ---->|
   |          |<---- ACK ---|                    |          |             | apply (ms…s later)
   | <-- OK --|             |                    |
 Latency = local write + 1 round trip           Latency = local write only
 Leader dies after OK? data is on follower ✔     Leader dies before shipping? write LOST ✘

In practice, "fully synchronous to every follower" is almost never used — one sick replica would freeze the system. The common production setup is: one synchronous (or semi-sync) follower in the same region for zero-data-loss failover, plus asynchronous followers in the same and other regions for read scaling and disaster recovery. Consensus-based systems (Raft/Paxos: etcd, CockroachDB, Spanner) generalise semi-sync: a write commits once a majority has it.

3.2 Adding a new follower without downtime

1. Take a consistent snapshot of the leader (e.g. pg_basebackup), noting its exact log position (LSN / binlog coordinates / GTID).
2. Copy the snapshot to the new machine.
3. The follower connects to the leader and requests all changes since that position.
4. Once it has caught up (lag ≈ 0), add it to the read pool.

The same procedure recovers a follower that crashed: it knows the last log position it applied, so it simply asks for everything after it (if the leader still retains that much log — otherwise it needs a fresh snapshot).

4. Replication lag and the anomalies it causes

With asynchronous followers, a read from a follower may return stale data. If writes stop, followers eventually catch up — that's eventual consistency. Normally lag is a few milliseconds, but it can grow to seconds or minutes when a follower is overloaded, when a big migration or batch job floods the log, or across a congested inter-region link. Three user-visible anomalies follow, each with a named guarantee that fixes it.

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

Problem: Alice edits her bio, the write goes to the leader, the app immediately reloads her profile from a lagging follower — and shows her old bio. She thinks the save failed and edits again.

 Alice's app            Leader                     Follower (200 ms behind)
    | UPDATE bio='🌴'      |                              |
    |------------------>   | commit                       |
    | <------ OK -------   |----- WAL (in flight) ----.   |
    | GET /profile/alice                               \  |
    |------------------------------------------------> |  |  (hasn't applied it yet)
    | <--------------- bio = old value  ✘ ------------ |  |
                                                        `-> applied 200 ms later

Guarantee: a user always sees their own writes (other users may still see stale data briefly). Techniques:

• Read your own data from the leader. "Things the user can edit" (own profile, own settings) are read from the leader; everything else from followers. Works when only a small fraction of reads are "own data".
• Time window. For N seconds after a user's last write (e.g. 10 s — longer than the p99 lag), route that user's reads to the leader. The app server remembers last_write_at in the session or a short-lived cache entry.
• Log-position tracking. After a write, return the leader's log position (LSN) to the client. Later reads carry it; the read is served by a follower only if that follower has replayed past that LSN (PostgreSQL: compare with pg_last_wal_replay_lsn()), otherwise wait briefly or go to the leader. This is precise and scales well.
• Optimistic UI. The client shows its own write locally without re-reading — great UX, but it hides rather than solves the problem (other devices of the same user still need one of the above; "cross-device read-your-writes" requires server-side tracking keyed by user, not by device).

4.2 Monotonic reads

Problem: Bob refreshes a post's comments twice. The first request hits a fresh follower and shows Carol's new comment; the second hits a more lagged follower and the comment disappears. Time appears to go backwards.

 Bob refresh #1 ──► Follower A (lag 10 ms)   → sees comments [c1, c2, c3]
 Bob refresh #2 ──► Follower B (lag 3 s)     → sees comments [c1, c2]        ✘ c3 vanished

Guarantee: once a user has seen data at some point in time, they never later see an older state. Techniques: sticky routing — always send a given user's reads to the same replica, e.g. replica = hash(user_id) % replicas (re-route only if it fails); or track the highest log position the user has seen and only read from replicas at least that fresh.

4.3 Consistent prefix reads

Problem: causally related writes become visible out of order. In comments: Alice asks "Where was this taken?" and Bob replies "Bali!". If the two writes are stored in different partitions that replicate with different lag, a viewer may see Bob's answer before Alice's question.

 Partition P1 (fast follower):  [12:00:05] Bob:   "Bali!"
 Partition P2 (slow follower):  [12:00:01] Alice: "Where was this taken?"   ← not replicated yet
 Viewer sees:  "Bali!"  … and only later the question it answers.   ✘

Guarantee: if a sequence of writes happens in a certain order, anyone reading them sees them in that order (a prefix of the history, never a jumbled subset). Techniques: keep causally related data in the same partition (e.g. all comments of a post partitioned by post_id, so one replication stream preserves order — exactly how a Cassandra comments_by_post table is laid out); attach causal metadata (a reply references its parent's ID, and the client hides a reply until its parent is visible); or use systems offering causal consistency.

5. Failover: when the leader dies

Followers failing is easy (route reads elsewhere, catch up later). The leader failing is hard: someone must become the new leader, clients must start sending writes to it, and the old leader must not come back thinking it's still in charge. This process is failover. It can be manual (an on-call engineer runs a promotion script) or automatic.

5.1 The steps of automatic failover

1. Detect the failure. There's no reliable way to know a machine is dead — only that it's not answering. Nodes exchange heartbeats; if the leader misses them for a timeout (e.g. 10–30 s), it's declared dead. Too short a timeout → needless failovers during a GC pause or load spike (which can make an overload worse); too long → longer outage (higher RTO).
2. Choose a new leader. Usually the follower with the most up-to-date log (minimising data loss), chosen by a consensus system (etcd/ZooKeeper/Consul via a tool like Patroni or Orchestrator) or by a majority vote among replicas (Raft-style election). Prefer the synchronous follower if one exists — it has every acknowledged write.
3. Reconfigure the system. Clients must send writes to the new leader: update service discovery / DNS / a virtual IP, or let a proxy (PgBouncer, HAProxy, ProxySQL) or the load balancer follow the leader. Other followers re-point to replicate from the new leader.
4. Deal with the old leader. When it comes back it must step down and become a follower — and any writes it accepted that were never replicated must be discarded or reconciled.

5.2 What can go wrong

• Lost writes (async replication). The old leader acknowledged writes the new leader never received. When it rejoins, those writes are thrown away — RPO > 0. Worse, they can clash with new data: a real-world GitHub incident (2012) involved a promoted MySQL follower that was behind; auto-increment IDs got reused, and those IDs were also used as keys in Redis, so some private data was shown to the wrong users. Lesson: with async replication, data loss on failover is normal, and other systems must be designed to tolerate it.
• Split brain. A network partition isolates the old leader, the rest of the cluster elects a new one, but the old leader is alive and still accepting writes from clients that can reach it. Now two leaders diverge — the most dangerous failure mode in replication.
• Flapping. A leader that is slow but not dead gets failed over repeatedly, each failover itself causing disruption.

5.3 Preventing split brain: quorums, leases and fencing tokens

• Majority quorum for leadership. A node may act as leader only while it can reach a majority of the cluster (e.g. 2 of 3). In a partition, at most one side has a majority, so at most one side can elect/keep a leader. Use an odd number of voters (3 or 5), ideally across three availability zones.
• Leases. Leadership is a time-limited lease (e.g. 10 s) granted by the consensus store and renewed continuously. A leader that can't renew must stop accepting writes before the lease expires; only after expiry can a new leader be chosen. (Relies on reasonably bounded clock drift.)
• STONITH ("shoot the other node in the head") — forcibly power off or network-isolate the old leader before promoting a new one.
• Fencing tokens — the most robust technique, because it doesn't trust the old leader to behave. Every time leadership is granted, the lock service hands out a monotonically increasing token (epoch/term number). Every write to storage carries the token, and the storage rejects any token lower than the highest it has seen.

 Lock service           Old leader L1                 New leader L2           Storage
   | grant lease, token=33 |                              |                      |
   |---------------------->|                              |                      |
   |                       |  … long GC pause / partition …                      |
   | lease expired; grant token=34                        |                      |
   |----------------------------------------------------->|                      |
   |                       |                              | write(x, token=34)   |
   |                       |                              |--------------------->| ok (max=34)
   |                       | wakes up, still thinks it leads                     |
   |                       | write(y, token=33)                                  |
   |                       |------------------------------------------------->| REJECTED: 33 < 34 ✔

Raft and Paxos have this built in: every message carries the term, and nodes reject messages from stale terms. That's why consensus-based databases (CockroachDB, Spanner, etcd) and tools like Patroni (which stores the leader key in etcd with a TTL) are trusted for automatic failover.

6. Multi-leader replication

In multi-leader (master–master, active–active) replication, more than one node accepts writes; each leader forwards its changes to all the others. Within one data center this rarely pays off, but it's compelling in three situations:

• Multi-region writes. A leader in each region means users write to a nearby leader with low latency, and each region keeps working if inter-region links fail. (With single-leader, every write from Asia must cross the world to the US leader.)
• Offline-capable clients. Each phone's local database is effectively a "leader" that syncs later — e.g. drafting a caption or queuing a DM in airplane mode.
• Collaborative editing (Google Docs-style), where each user's local copy accepts edits immediately.

6.1 Topologies

Leaders can replicate all-to-all (most robust; but messages can arrive in different orders at different nodes), in a circle or star (fewer links, but one failed node can break the propagation path). Each change is tagged with the IDs of nodes it has passed through to avoid infinite loops.

6.2 Conflicts and how to handle them

The big problem: two leaders can accept conflicting writes to the same record concurrently, and neither knows until replication brings them together — by then both users were told "saved". Options, from best to most complex:

1. Conflict avoidance (most common). Make all writes for a given record go through the same leader. E.g. each user has a home region; all writes to Alice's profile go to Alice's home leader, even if she's travelling. Conflicts become impossible unless the home region changes (failover).
2. Last-write-wins (LWW). Attach a timestamp to each write; the highest timestamp wins everywhere. Simple and convergent, but it silently discards the other write, and clock skew between servers can make an earlier write win. Acceptable for data where losing one of two near-simultaneous updates is fine (a profile's "last active" time); dangerous for anything additive.
3. Merge the values. Keep both and combine — e.g. concatenate, take the union of sets, or keep siblings and let the application or the user pick (Amazon's Dynamo shopping cart merged carts; Riak keeps "siblings").
4. CRDTs (Conflict-free Replicated Data Types). Data structures mathematically designed so that concurrent updates always merge to the same result regardless of order — no coordination needed.

6.3 CRDTs in one example: a like counter

Suppose likes on a post are counted in three regions. If each region stored a single number and replicated it with LWW, concurrent increments would be lost (US says 101, EU says 101 — the truth is 102). A G-Counter (grow-only counter) CRDT instead keeps one slot per region; each region increments only its own slot, and merging takes the per-slot maximum:

US state:   {US: 60, EU: 40, ASIA: 0}   (US got a like: US 59 → 60)
EU state:   {US: 59, EU: 41, ASIA: 0}   (EU got a like: EU 40 → 41)
merge = per-slot max → {US: 60, EU: 41, ASIA: 0}  → total = 101 ✔ (both likes counted)

Unlikes? Use a PN-Counter: two G-Counters (P for increments, N for decrements); value = sum(P) − sum(N).
Sets (e.g. "users who liked")? Use an OR-Set (observed-remove set): adds and removes carry unique tags,
so a concurrent add and remove resolve deterministically.

CRDTs are used in Riak, Redis Enterprise active-active, Azure Cosmos DB, collaborative editors (Yjs, Automerge) and some messaging systems. They're powerful but only exist for specific data shapes; arbitrary business invariants ("balance must not go negative") can't be guaranteed without coordination.

7. Leaderless (Dynamo-style) replication

Amazon's 2007 Dynamo paper popularised a design with no leader at all: the client (or a coordinator node acting for it) sends each write to several replicas in parallel and reads from several replicas in parallel. Cassandra, ScyllaDB, Riak and Voldemort work this way (DynamoDB, despite the name, uses leaders per partition internally).

7.1 N, W, R and the quorum condition

• N — the replication factor: how many nodes store each key (chosen by consistent hashing: the next N nodes on the ring).
• W — how many replicas must acknowledge a write before it's reported successful.
• R — how many replicas must respond to a read; the client takes the value with the newest version/timestamp.

If R + W > N, the set of nodes you read from and the set that acknowledged the latest write must overlap in at least one node (pigeonhole principle), so the read sees the latest successful write. That's a quorum.

7.2 Worked quorum examples

General rule: a write tolerates N − W unavailable replicas and a read tolerates N − R. Latency is set by the W-th (or R-th) fastest replica, so quorums also cut tail latency: a slow node is simply not waited for.

Numeric example: N = 3, W = 2, R = 2, replica latencies 2 ms, 3 ms and 40 ms (one is in a GC pause). A quorum write completes at 3 ms (second fastest), not 40 ms. With W = 3 it would take 40 ms — and fail entirely if that node were down.

Even with R + W > N, edge cases remain: concurrent writes resolved by LWW can drop one; a write that succeeded on fewer than W nodes is reported as failed but isn't rolled back on the nodes that took it; and sloppy quorums (next) break the overlap guarantee. Quorums give strong-ish per-key behaviour, not linearizability or transactions.

7.3 Cassandra consistency levels

7.4 Sloppy quorums and hinted handoff

Suppose N = 3 and two of a key's three "home" replicas are unreachable. A strict quorum write (W = 2) fails. A sloppy quorum instead accepts the write on other healthy nodes that aren't among the key's N home nodes, so the system stays writable. Each stand-in node stores the write with a hint — "this belongs to node C" — and when C comes back, it forwards the data: hinted handoff. This raises write availability, but a read quorum from the home nodes may not see the write until handoff completes, so R + W > N no longer guarantees freshness. (Riak and Dynamo use sloppy quorums; Cassandra uses hinted handoff for temporarily down replicas, but hints don't count toward the consistency level except with ANY.)

7.5 Read repair and anti-entropy

Replicas that missed writes (down, overloaded, dropped hints) must eventually be fixed. Two mechanisms:

• Read repair. During a read, the coordinator notices a replica returned an older version (Replica C's v1 in the diagram) and writes the newest version back to it. Cheap and automatic, but only repairs keys that are actually read.
• Anti-entropy repair. A background process compares replicas and copies missing data — covering rarely-read keys too. Comparing billions of keys naively would mean shipping all data; instead replicas compare Merkle trees.

A Merkle tree (hash tree) has a hash of each key range at the leaves, and each parent is the hash of its children. Two replicas compare root hashes: if equal, the ranges are identical and they're done. If not, they descend only into children whose hashes differ — so they find the few differing ranges with a handful of hash exchanges instead of comparing every key.

          Replica A                                  Replica B
           root: 9f3c                                 root: 71ab        ← differ → descend
          /          \                               /          \
     h(L): 4e1d    h(R): 88a0                   h(L): 4e1d    h(R): 2c77  ← left equal: skip whole half
      /    \        /     \                      /    \        /     \
   r1:1a  r2:7b  r3:c0  r4:5e                 r1:1a  r2:7b  r3:c0  r4:d9  ← only range r4 differs
                                                                          → stream just r4's rows
 With 2^20 leaf ranges, finding one bad range takes ~20 levels of comparisons, not 1M.

Cassandra's nodetool repair builds Merkle trees per token range; operators run it regularly (more often than gc_grace_seconds, so deletions propagate before tombstones are purged). The same idea appears in Git, blockchains, and DynamoDB/Riak anti-entropy.

8. How the replication log is implemented

"The leader sends its changes to followers" hides an important choice: what exactly is sent?

8.1 Change Data Capture in the Instagram architecture

CDC treats the database's own replication log as the single source of truth for "what changed". When Alice renames herself, the row change flows to the search indexer (update her document in Elasticsearch), a cache invalidator (delete her cached profile in Redis) and the data warehouse — all without the User Service writing to four systems and risking partial failure. See Message Queues & Kafka.

9. How Instagram uses replication

9.1 Putting it together: a profile edit and a profile view

1. Alice (home shard's primary in US-East) edits her bio in Singapore. The User Service in the Asia region forwards the write to the shard's primary in US-East (~200 ms round trip; edits are rare, so this is acceptable).
2. The primary commits and ships WAL to its same-region synchronous standby (zero data loss on failover) and asynchronously to replicas in Europe and Asia.
3. The service records "Alice wrote at LSN X" (or simply "Alice wrote at time T"). For the next few seconds, Alice's own profile reads go to the primary or to an Asia replica that has replayed past LSN X — read-your-writes.
4. Everyone else viewing Alice's profile in Asia reads from the local replica (~2 ms). If it's 300 ms behind, they see the old bio for 300 ms — nobody notices.
5. A CDC/event consumer updates the search index and invalidates the cached profile in every region.

9.2 Cassandra writes for likes and feed entries

A like is written to Cassandra in the user's local region with LOCAL_QUORUM (2 of 3 local replicas, a few ms), and Cassandra replicates it asynchronously to the other regions' 3 replicas each. Reads of like lists use LOCAL_ONE or LOCAL_QUORUM depending on how fresh they must be. Conflicts are rare because likes are insert-only keyed by (post_id, user_id) — inserting the same like twice is idempotent, so last-write-wins is harmless here.

Rough numbers (assumptions)

Assume a User DB shard: 20,000 reads/s and 200 writes/s at peak.
One server comfortably handles ~5,000 reads/s for this query mix  → need ≥ 4 read replicas + headroom
→ primary + 1 sync standby (US-East) + 2 async replicas per region × 3 regions.

Cassandra storage with RF = 3 in each of 3 regions = 9 copies:
  100 TB of unique data → 900 TB of disk (before compaction overhead).
  Durability is excellent, but replication factor is a direct cost multiplier.

WAL volume: 200 writes/s × ~2 KB of WAL (row + indexes) ≈ 400 KB/s per shard
  → trivial bandwidth, but a 50 GB backfill job generates 50 GB of WAL → minutes of replica lag.

10. Trade-offs & common pitfalls

Common pitfalls

• Treating replication as a backup. A DROP TABLE or a buggy migration replicates to every replica within milliseconds. You still need point-in-time backups (and maybe a delayed replica).
• Reading your own writes from a lagging replica — the "my edit didn't save" bug.
• Unmonitored lag — a replica hours behind still serving traffic.
• Fully synchronous replication to a remote region — every write pays the cross-ocean RTT, and the remote region's outage stops all writes.
• Automatic failover without fencing — two leaders accept writes and diverge.
• Over-aggressive failure detection — a 2-second timeout fails over on every GC pause.
• LWW on additive data — counters or lists silently lose concurrent updates. Use CRDTs or conflict avoidance.
• Forgetting anti-entropy — never running Cassandra repair lets replicas drift and deleted data resurrect.

11. Real-world technologies

12. Interview corner

13. Check your understanding

Related lessons

• Sharding — replication copies data; sharding splits it (you need both)
• CAP, PACELC & Consistency Models — the theory behind sync vs async and quorums
• Consistent Hashing — how leaderless systems pick the N replicas for a key
• Multi-Region Architecture & DR — replication across regions, RPO/RTO, failover drills
• Databases — PostgreSQL vs Cassandra and why each replicates differently
• Message Queues & Kafka — CDC and replicated logs
• Resilience Patterns — timeouts and health checks used in failure detection