Hi everyone, welcome to the thirty-first issue of The Main Thread. In today’s article, we will discuss various ways to implement replication in distributed databases.
On the surface it feels like a storage problem but it’s more of a survival problem. A single database node is a single point of failure. It goes down, the application goes down. It gets overwhelmed, the application faces all sorts of latency issues. It is deployed in one datacenter, and users on the other side of the world feel every millisecond of that distance.
Replication solves all three: fault tolerance, read throughput, and geographic latency by maintaining copies of the data on multiple machines. But the moment we have more than one copy of the data, we have a new problem of keeping all copies consistent.
This is where real engineering happens. In the CAP theorem piece, we established that during a network partition, we must choose between consistency and availability. Replication is where that choice becomes concrete. Each topology provides a different answer to the question: “when two replicas disagree, which one is right?“
By the end of this article, we will have understood three major replication strategies - leader-follower, multi-leader, and leaderless - not just mechanically but as a set of deliberate tradeoffs between consistency, availability, and operational complexity.uy
TL;DR
Leader-follower is the default topology. It is simple, has strong consistency, a single write path. But the leader is the bottleneck and failover is dangerous.
Synchronous replication gives us durability guarantees at the cost of write latency. Asynchronous replication gives us speed at the cost of possible data loss on failover.
Multi-leader removes the write bottleneck but introduces conflict resolution as a first-class engineering problem.
Leaderless (Dynamo-style) maximizes availability with quorum-based consistency - but “quorum” is not a synonym for “consistent”.
Read replicas scale reads, but replication lag means follower reads can return stale data. Usually, people underestimate how often this matters.
Split-brain is the nightmare scenario when two nodes both believe they are the leader. Prevention requires fencing, not just detection.
Leader-Follower Replication
This is the most common topology. One node is designated the leader (also called primary or master). Writes exclusively go to the leader. The leader writes the change to its local storage and at the same time, sends a replication log to one or more followers (also called replicas, secondaries). Followers apply the log in order, maintaining an identical copy of the data.
Depending on the consistency requirements, reads can be served either by the leader or any of the followers.
What’s in the replication log?
Databases use different formats. Postgres uses a Write-Ahead-Log (WAL) which is a byte-level log of changes to storage pages. MySQL’s binlog records logical operations (which rows changed). The logical log format is more portable and useful for heterogeneous replication (replication to a data warehouse, for example). The physical format is faster but tightly coupled to the storage engine’s internals.
Synchronous vs. Asynchronous: The fundamental tradeoff
This is the most important decision in leader-follower replication.
Synchronous replication: The leader waits for the follower to confirm it has written the data before acknowledging the write to the client.
If the leader fails immediately after confirming a write, the follower has the data. No writes are lost. But every write waits on the slowest follower. A network hiccup on one replica stalls all writes. In practice, we can’t run all followers synchronously because one laggy replica will make our system unusable.Asynchronous replication: The leader acknowledges the write immediately and replicates in the background.
Writes are fast. But the cost is that if the leader fails before the replication log reaches the followers, those writes are permanently lost even though the client received a success acknowledgement. Data loss is not just a theoretical concern; it happens.Semi-synchronous: This is the practical middle ground. Here, one designated follower is synchronous, the rest are asynchronous. Postgres calls these synchronous standbys.
We get the durability of sync replication (at least two nodes always have the latest write) without the performance sensitivity of requiring all replicas to confirm.
Replication lag and its consequences
Even with async replication, followers usually catch up within milliseconds. But “usually” is not “always”. Under high load, the follower’s hardware which is also serving reads struggles and due to this replication lag grows. A follower might be seconds or even minutes behind the leader.
This creates subtle correctness problems that are much harder to debug than outright failures.
Read-your-write violations: A user updates their profile information. The write goes to the leader. The user then immediately refreshes the page but this read is served by a follower node which is not yet caught up, and returns stale or old profile information. The user thinks the update failed, clicks save again, and creates a duplicate.
The fix is to route reads for data that a user just wrote to the leader, for a window after the write. Or track the replication position of recent writes and wait for followers to catch up to that position before serving reads.
Monotonic reads violation: A user reads a comment thread. Request A goes to follower 1 (caught up). Request B goes to the follower 2 (two seconds behind). The user sees a reply before they see the original comment. Time appears to go backward.
The fix is to route a user’s reads to the same follower consistently (typically via consistent hashing on userID. Granted, lag is a problem but at least it’s monotonically moving forward.Causal ordering violation: User A posts a question. User B reads it and posts an answer. User C reads B’s answer but not A’s question because their read hits a lagging replica.
As a fix, causal consistency requires tracking causal dependencies, not just timestamps. This is hard to retrofit - design for it or accept it as a known limitation.
Failover and Split-Brain
When the leader fails, a follower must be promoted. This is failover, and this is where most leader-follower systems find their sharpest edges.
Detecting failure
This is the first problem because we can’t distinguish a dead leader from a leader that’s alive but unreachable due to a network partition. Most systems use heartbeat timeout: if a leader doesn’t respond within N seconds, assume it’s dead. But N is a guess. If it is too short, we will failover a healthy but temporarily slow leader. If it is too long, we will delay recovery after a real failure.
Choosing the new leader
This is the second problem. In an async system, followers may be at different positions in the replication log. The obvious choice is the follower with the most recent data - but that requires coordination, which requires systems to have enough nodes in consensus. This is why many production databases use Raft-based consensus for leader election. See my blog post and GitHub repo for the Raft implementation in Go from scratch.
Split-Brain
This is the nightmare. During a network partition, the old leader can’t see the new leader. Both believe they are primary. Both accept writes. Now, we have two authoritative but diverging copies of data.
When the partition heals, we have two conflicting histories. This is not recoverable without either discarding data or complex merge logic.
STONITH (shoot the other node in the head)
This is the operational solution: before the new leader accepts writes, it forces the old leader to shut down. Cloud providers implement this via control plane - they can forcibly power off an instance. On-premises, it’s done via IPMI/BMC hardware controllers. The principle is simple: it’s better to lose one node than have two authoritative leaders.
Fencing tokens
Each leader lease is associated with a monotonically increasing epoch. Writes from an old epoch are rejected by followers and storage systems. A node that wakes up from a pause and discovers it has an outdated epoch knows it must step down.
Multi-Leader Replication
What if you need writes in multiple locations simultaneously? A single leader system can’t accept writes in a Mumbai datacenter without routing them through a Virginia center and adding hundreds of milliseconds to every write.
Multi-leader replication allows multiple nodes to accept writes, each replicating to others.
The following are the use cases where multi-leader makes sense:
Multi-datacenter write availability: Users in each region write to a local leader; replication happens asynchronously across datacenters
Offline clients: Calendar apps that accept changes while the phone is offline, syncing when connectivity returns (the phone is effectively a leader)
Collaborative editing: Google Docs-style concurrent editing on a shared document
Having write conflicts is the price we pay for multi-leader. In a single-leader system, concurrent writes are serialized at the leader. In multi-leader systems, two writes to the same record can happen simultaneously on different leaders. When they replicate to each other, there is no single authority to decide which one wins.
Conflict resolution strategies
Last Write Wins (LWW): In this, we attach a timestamp to every write. The highest timestamp wins. This is simple to implement but terrible in practice. Clocks are not reliable across distributed nodes - clocks drift, NTP adjustments, and leap seconds mean timestamps from different nodes are not comparable. LWW silently discards writes. Dynamo DB and Cassandra use LWW as their default (it is acceptable only when data loss is tolerable and writes are designed to be idempotent).
Conflict-free Replicated Data Types (CRDTs): These are data structures designed such that concurrent updates always merge deterministically. A counter CRDT accepts increments from any node and merges by summing all increments. A G-Set (grow-only set) merges by union. CRDTs are mathematically correct but constrain our data model - we can’t always express our domain using CRDT primitives.
Operational Transformation (OT): This is used by Google Docs. This transforms concurrent operations relative to each other so that applying them in any order yields the same result. These are complex to implement correctly and are well-studied for collaborative text editing.
Application-level Resolution: In this, we propagate the conflict to the application, which applies domain-specific merge logic. A shopping cart conflict might merge by taking the union of items. A bank account conflict might reject one write with an error. These are correct for complex domains; requires every write path to handle conflict callbacks.
Let’s take an example of Riak style custom conflict resolution:
def resolve_conflict(siblings: list[dict]) -> dict:
"""
Riak-style conflict resolution: application provides a merge function.
siblings = list of conflicting versions for the same key.
"""
if not siblings:
return {}
# Domain-specific: for user profiles, take most recent update per field
merged = {}
for sibling in siblings:
for field, value in sibling.items():
if field not in merged:
merged[field] = value
elif value.get('updated_at', 0) > merged[field].get('updated_at', 0):
merged[field] = value # Later timestamp wins per field
return mergedThere is no universally correct conflict resolution strategy. The right strategy depends entirely on what the data represents.
Leaderless Replication
The Amazon Dynamo paper (2007) popularized a topology with no leader at all. Any node can accept writes. Reads are served by multiple nodes simultaneously. Consistency is maintained via quorums rather than a single authoritative node.
With N replicas, requiring W write confirmations and R read responses, we get overlap guarantees when R + W > N.
At N=3: W = 2, R = 2 → R + W = 4 > 3. At least one node in every read quorum must have seen the latest write.
Common configurations:
W=N, R=1: Every write reaches all nodes. Reads are fast and always current. Write availability suffers as any node failure blocks all writes.
W=1, R=N: Writes are fast. Reads require all nodes. Read availability suffers.
W=majority, R=majority: The balanced default. Tolerates minority failures on both read and write paths.
class LeaderlessClient:
def __init__(self, nodes: list, W: int, R: int):
self.nodes = nodes
self.W = W # Write quorum
self.R = R # Read quorum
def write(
self,
key: str,
value: str,
version: int) -> bool:
confirmations = 0
for node in self.nodes:
try:
node.put(key, value, version)
confirmations += 1
except NodeUnavailableError:
pass
# Write succeeds only if quorum met
return confirmations >= self.W
def read(self, key: str) -> str:
responses = []
for node in self.nodes:
try:
responses.append(node.get(key))
except NodeUnavailableError:
pass
if len(responses) < self.R:
raise QuorumNotMetError()
# Return highest-versioned response;
# trigger read repair for stale nodes
latest = max(responses, key=lambda r: r.version)
self._read_repair(key, latest, responses)
return latest.value
def _read_repair(self, key: str, latest, all_responses):
# Bring stale nodes up to date during the read path
for response in all_responses:
if response.version < latest.version:
response.node.put(
key,
latest.value,
latest.version
)Sloppy quorums and hinted-handoff
If a node in the quorum is unavailable, Dynamo allows a different node (outside the natural preference list) to accept the write temporarily, with a “hint” that it should forward the write to the intended node when it recovers. This improves write availability but weakens the consistency guarantee. A sloppy quorum is not a true quorum.
Anti-entropy
In this, background processes compare replicas and fix divergences. Dynamo uses a Merkle tree (a hash tree where each leaf is a hash of a data block, and each internal node is a hash of its children). Two nodes can efficiently identify which data ranges differ by comparing tree roots, then recursively narrowing it to the divergent trees.
Hard truth about quorums
Even with R + W > N, leaderless systems are not strongly consistent. We can get stale reads in edge cases - concurrent writes, network delays during quorum collection, sloppy quorums. These systems offer eventual consistency with tunable availability, not strong consistency. The CAP theorem doesn’t disappear because we have a quorum.
Read Scaling Patterns and When They Break
One argument in favour of replicas is: add followers to scale read throughput. Ten followers means 10x read capacity, right?
Not quite. Here’s where it breaks:
Replication lag limits the staleness we can tolerate. Read scaling only works for reads where slightly stale data is acceptable. If our application needs read-your-writes or monotonic reads semantics, a significant fraction of reads must go to the leader anyway. The more consistency requirements the application has, the smaller the fraction of reads we can safely route to followers.
Write load is still bottlenecked on the leader. Adding followers doesn't help write throughput. If we are write-heavy, read replicas don't solve our scaling problem. Sharding does, but that's a different architecture with its own set of tradeoffs.
Replica read capacity is not free. Each follower must apply the full replication log. Under heavy write load, followers fall further behind. A follower that's consuming CPU and I/O to apply writes has less capacity for reads. The faster the write rate, the more replication lag grows, the more stale the follower reads are - exactly when we need freshness the most.
The read scaling ceiling in practice: Leader-follower read scaling works well when the read:write ratio is high 10:1 or better), the application tolerates eventual consistency, and the write load is moderate enough that followers can keep up. Below those conditions, we add operational complexity for diminishing returns.
Choosing The Topology
Most systems should start with leader-follower. It is the best-understood topology, with the most operational tooling, the clearest failure modes, and strong enough consistency for most applications.
Move to multi-leader or leaderless when you have a specific, concrete requirement, geographic distribution requiring local writes, or availability requirements that preclude a single write path, not because the complexity sounds interesting.
Mental Model
Every replication topology answers the same question: how do we handle disagreement between copies?
Leader-follower avoids the question by ensuring there is only ever one authoritative source for writes. Multi-leader confronts disagreement head-on with conflict resolution. Leaderless uses probabilistic quorum overlap to make disagreement unlikely, and anti-entropy to repair it when it happens.
The tradeoffs cascade from that design choice. Single authoritative writer → simple consistency, constrained write availability. Multiple writers → high availability, conflict resolution complexity. Probabilistic overlap → tunable tradeoffs, no single bottleneck, eventual consistency.
Replication doesn't make the system simpler. It makes data more available and durable but at the cost of explicit decisions about what "correct" means when two nodes disagree. We must make those decisions deliberately, before production forces them on us.
That’s all for this post. I hope you enjoyed it. I would love to know your thoughts and battle stories.
Namaste!
— Anirudh
