Skip to main content

Leaders and Followers

Adapted from: Martin Kleppmann, Designing Data-Intensive Applications (O'Reilly, 2017), Chapter 5 — rewritten and condensed for teaching.

Replication means keeping the same data on multiple networked machines. There are three good reasons to do it: keeping data physically close to users (lower latency), staying up even when parts of the system fail (higher availability), and spreading read traffic across more machines (higher read throughput). This lesson — and the rest of this section — assumes your dataset comfortably fits on one machine, so every replica holds a full copy; a later lesson relaxes that and covers partitioning datasets too big for any single machine to hold.

If data never changed, replication would be trivial: copy it once, done. Every genuine difficulty in replication comes from handling changes to already-replicated data — and that's the entire subject of this lesson and the ones that follow it.

Leader-Based Replication​

Each node holding a copy of the data is a replica. With several replicas, one question dominates everything else: how do we make sure every write reaches every replica? The dominant answer is leader-based replication (also called active/passive or master–slave):

  1. One replica is the leader (primary/master). Every write must go through it — it writes the change to its own local storage first.
  2. Every other replica is a follower (read replica/secondary). The leader records each write into a replication log and streams it to every follower, which applies the changes in the exact order the leader processed them.
  3. Reads can go to the leader or any follower — but writes are only ever accepted by the leader.

This is a built-in feature of PostgreSQL, MySQL, Oracle, and SQL Server, plus several non-relational databases (MongoDB, RethinkDB). It isn't limited to databases either — Kafka and RabbitMQ's highly-available queues use the exact same leader/follower shape internally.

Synchronous vs. Asynchronous Replication​

Whether a follower confirms a write before the leader reports success to the client is the single most consequential replication setting there is.

  • Synchronous — the leader waits for a follower's confirmation before telling the client the write succeeded. Guarantees that follower has an up-to-date copy — genuinely useful if the leader then crashes. The cost: if that follower is slow, unreachable, or crashed, every write blocks until it recovers.
  • Asynchronous — the leader fires the change at followers and doesn't wait. Writes stay fast and available even if every follower has fallen behind — at the cost that an unrecoverable leader crash can silently lose any writes it hadn't yet replicated, even ones already confirmed to a client.

Making every follower synchronous is impractical — a single flaky node would stall the entire system. The common middle ground is semi-synchronous: exactly one follower is synchronous at a time (guaranteeing the data lives on at least two nodes — leader plus that one follower), and if it becomes slow or unreachable, a different asynchronous follower is promoted to take its place. Many real deployments skip synchronous replication entirely and run fully asynchronous — a real durability trade-off, but one that's especially common once you have many followers, or followers spread across distant datacenters where waiting on a round-trip would tank write latency.

Chain replication and consensus

Losing data on leader failure is serious enough that research keeps chipping away at it — chain replication (used in production by Microsoft Azure Storage) is a synchronous variant designed to avoid the "any one slow follower blocks everything" problem. There's also a deep connection between replication consistency and consensus (getting distributed nodes to agree on a value) — a topic this series returns to in a later lesson.

Check: leader-based and sync/async replication​

Quiz

Leader-based replication and sync vs. async

1 / 3

In leader-based replication, which nodes are allowed to accept write requests directly from clients?

Pick an answer to see if you are right.

Setting Up New Followers​

Adding a follower — to scale out or replace a failed node — can't just mean copying data files, since writes never stop arriving; a naive file copy would capture different parts of the database at different moments in time, producing something that makes no coherent sense. The real process, done without downtime:

  1. Take a consistent snapshot of the leader's data, ideally without locking the whole database (most databases already support this, since backups need the same capability).
  2. Copy that snapshot to the new follower.
  3. The follower connects to the leader and asks for every change since the exact log position the snapshot corresponds to (PostgreSQL calls this a log sequence number; MySQL calls it binlog coordinates).
  4. Once the follower has replayed that entire backlog, it's caught up and can keep consuming the live stream like any other follower.

Handling Node Outages​

Individual nodes go down — sometimes from a genuine fault, sometimes from routine maintenance like a kernel patch reboot. The goal is always to keep the system running, and to shrink the blast radius of any one node's outage as much as possible.

Follower Failure: Catch-Up Recovery​

The easy case. Every follower keeps its own log of changes received from the leader, so after a crash or a network blip it always knows the last change it fully processed. On reconnecting, it simply asks the leader for everything since that point — the exact same mechanism used to bootstrap a brand-new follower.

Practice: catch-up recovery​

Leader Failure: Failover​

Trickier. A follower must be promoted, clients must be redirected to it, and the remaining followers must start consuming from the new leader. This whole process is called failover, and it can run manually (an operator makes the call) or automatically:

  1. Detect the failure. There's no foolproof way to know why a node stopped responding — crash, power loss, network partition, or something else entirely. Most systems fall back to a timeout: no response for N seconds (30 is typical) means "assume dead."
  2. Elect a new leader. Either the remaining replicas vote (a majority-based election), or a separate controller node appoints one. The best candidate is normally whichever replica has the most up-to-date data from the old leader, to minimize lost writes. (Getting distributed nodes to agree on this is itself a consensus problem — a later lesson goes deeper.)
  3. Reconfigure the system. Clients start sending writes to the new leader. If the old leader eventually rejoins, it needs to recognize it's no longer in charge and demote itself to a follower — otherwise you can end up with two nodes both convinced they're the leader.

Failover has an unusually large number of ways to go wrong:

  • Lost writes. Under asynchronous replication, the new leader may be missing writes the old leader had already confirmed to clients — and if the old leader later rejoins, those unreplicated writes are typically just discarded, silently violating whatever durability the client thought it had.
  • Cross-system inconsistency. GitHub once promoted a follower whose auto-increment counter lagged behind the old leader's — the new leader reused primary keys the old leader had already assigned. Those same keys were also referenced in a separate Redis store, and the collision leaked private data to the wrong users. The lesson: discarding writes can ripple into other systems that assumed those writes were final.
  • Split brain. Certain fault patterns can leave two nodes simultaneously convinced they're the leader. If both accept writes with no conflict-resolution process, data corruption or loss is close to guaranteed. Some systems defend against this with fencing (bluntly: STONITH — "shoot the other node in the head," forcibly shutting one of them down) — but a poorly designed fencing mechanism can misfire and shut down both nodes.
  • The timeout dilemma. Too long a timeout means a longer outage before failover kicks in. Too short, and an ordinary load spike or transient network hiccup triggers an unnecessary failover — which, if the system was already under strain, tends to make things measurably worse, not better.

There's no clean, universal fix for any of this — which is exactly why some operations teams deliberately keep failover manual, even when their database software fully supports doing it automatically.

Check: node outages and failover​

Quiz

Handling outages and leader failover

1 / 3

Why is recovering from a FOLLOWER failure generally much simpler than recovering from a LEADER failure?

Pick an answer to see if you are right.

Implementation of Replication Logs​

What actually travels from leader to follower? Four real approaches, in practice:

MethodHow it worksMain weakness
Statement-basedForward the literal INSERT/UPDATE/DELETE statements; followers re-execute themBreaks on nondeterministic functions (NOW(), RAND()), autoincrement columns, or statements depending on existing data — unless executed in exactly the same order everywhere
Write-ahead log (WAL) shippingShip the storage engine's own low-level WAL (used in B-Trees and Other Indexing Structures for crash recovery) byte-for-byteTightly coupled to the storage engine's on-disk format — leader and followers usually can't run different database software versions, which can force downtime during upgrades
Logical (row-based) logA separate, storage-engine-independent format describing row-level changes (inserted row's new values, deleted row's key, updated row's new values)None of WAL shipping's version-coupling — and it's also easy for external systems to parse, which is exactly what powers change data capture
Trigger-basedDatabase triggers fire custom application code on every write, which then handles replication itselfHigher overhead and more bug-prone than engine-native replication, but the only option offering full flexibility — partial replication, cross-database-type replication, or custom conflict logic

Statement-based replication was MySQL's default before 5.1, and VoltDB still uses it — but only by requiring transactions to be fully deterministic, which sidesteps its usual failure mode entirely. WAL shipping is how PostgreSQL and Oracle implement built-in replication. Logical log replication is MySQL's modern default (binlog in row-based mode), and it's the technique underneath most real-world change data capture pipelines — streaming a database's changes out to a search index, a cache, or a data warehouse. Trigger-based tools like Bucardo (Postgres) or Oracle GoldenGate exist specifically for the cases the built-in engine replication can't handle: partial replication, cross-database-type replication, or genuinely custom conflict-resolution logic.

Check: replication log implementations​

Quiz

Statement-based, WAL, logical, and trigger-based replication

1 / 3

Why can statement-based replication produce different results on different replicas, even though every replica executes the exact same SQL statement?

Pick an answer to see if you are right.

Key takeaways
  • Leader-based replication funnels every write through one node, which streams an ordered replication log to followers — reads can go anywhere, writes only to the leader.
  • Synchronous replication trades write availability for durability guarantees; asynchronous replication trades durability for speed and availability — most real systems land on a semi-synchronous compromise, or go fully asynchronous.
  • New followers bootstrap from a consistent snapshot plus the leader's exact log position at that snapshot, then replay forward until caught up — the identical mechanism a reconnecting follower uses after any outage.
  • Leader failover — detecting the failure via timeout, electing a replacement, and reconfiguring clients — is where most of replication's genuinely hard problems live: lost writes, cross-system inconsistency, split brain, and an unavoidable tension in choosing the failure-detection timeout.
  • Replication logs come in four practical flavors — statement-based (fragile under nondeterminism), WAL shipping (fast but version-coupled), logical/row-based (decoupled, and the basis of change data capture), and trigger-based (slow but maximally flexible).

Further Reading​