System Design interview questions: Replication

In single-leader (primary–secondary) replication, only the leader, also called the primary, accepts client write requests. The leader applies each write locally and then ships it to followers through the replication log. Followers serve reads and replay that log, but they do not accept writes directly, which prevents divergent copies and write conflicts.

The follower holds a copy of the data by applying the replication log sent by the leader, and it can serve client read requests. Clients send writes to the leader, not to the follower, so the leader remains the single source of truth for write ordering.

Synchronous replication means the primary waits for at least one replica to confirm it has received a write before reporting success to the client. This guarantees the write exists on a second node, improving durability, at the cost of higher write latency. Asynchronous replication does not wait and can lose acknowledged writes if the primary fails.

Complete the sentence. In asynchronous replication, the primary writes data locally and reports success to the client immediately, then propagates changes to replicas later. This reduces write latency but can leave the primary and replicas temporarily out of sync if the primary fails before replication completes.

A write is committed on the leader at 10:00:00. The follower receives and applies that same write at 10:00:03. The 3-second delay between the leader's commit and the follower's visibility of the write is called replication lag.

This violates read-after-write (read-your-writes) consistency. The user's write went to the leader, but the read was served by a follower that had not yet received the replicated update because of replication lag. A common fix is to route the user's own reads to the leader for a short window after their write (for example, one minute), or to use session-based sticky routing so the user's reads go to the leader until the followers catch up.

In single-leader replication, a slow or failed follower can block all writes system-wide when using synchronous replication for all followers.

Synchronous replication increases write latency: the leader must wait for at least one follower to confirm the write before acknowledging it to the client. This guarantees durability on multiple nodes but adds a network round-trip to every write — the standard consistency-vs-latency trade-off compared with asynchronous replication, which acknowledges immediately but risks data loss if the leader fails.

Route that one user's reads to the primary for a short window after their write. The symptom is replication lag — the async replica hasn't applied the write yet — so pinning just that user to the primary gives them read-your-writes consistency while everyone else keeps the replica offload. Sending all reads to the primary or forcing synchronous replication is far more expensive than the problem warrants.

True. Asynchronous replication acknowledges the write on the primary before the change has propagated, so a read routed to a lagging replica can miss the just-written value — the classic read-your-writes anomaly. Mitigations include pinning that user's reads to the primary for a short window after a write, using a synchronous or quorum write, or tracking replication lag.

Read replicas offload read traffic from the primary database by streaming changes asynchronously to one or more followers. They let you scale read throughput linearly (add a replica, add read capacity) without scaling the primary write path. The key limitation is replication lag: replicas trail the primary by milliseconds to seconds, so reads against a replica may return stale data — a write followed immediately by a read-your-own-write to a different replica can miss the write. Applications must direct writes (and reads-after-writes) to the primary and route read-heavy, stale-tolerant queries (reports, timelines, analytics) to replicas.

In a single-leader replication topology all writes go to one node that propagates changes to read-only followers, while a multi-leader topology accepts writes on more than one node and must resolve write conflicts.

Read-after-write consistency in a single-leader system with asynchronous followers is achieved by routing a user's reads to the leader for a short window after their write, or by reading from a follower only after it has replicated past that write. Ordered log application alone does not guarantee it, because followers can lag behind the leader and serve stale data.

Linearizability (strong consistency) is the guarantee this configuration breaks. With W=2, R=2 and N=5, R + W = 4 does not exceed N, so the write and read quorums can be disjoint and a read after an acknowledged write may return a stale value. Setting R + W > N restores quorum overlap.

Read-your-writes consistency is restored by routing the writing user's reads to the leader, or to a replica known to have applied their write, for a short window after the write. Other users' reads can still use any replica, so most read traffic keeps the scaling and latency benefits of asynchronous replication.

In an asynchronous single-leader replication system, the time delay between a write being committed on the leader and that write becoming visible on a follower replica is called replication lag. When this delay causes a client that has already read a newer value to subsequently read an older value from a different replica, appearing to go backward in time, the violated consistency guarantee is called monotonic reads consistency.

Shard users by user_id across N shards, and make each shard an independent 5-replica Raft group spread 2/2/1 across North America, Europe, and Asia. Prefer placing each shard's leader in the user's home region using leader priority or leadership transfer, which reduces write latency but is not guaranteed by Raft alone. A write is acknowledged only after a majority (at least 3 of 5) of replicas durably persist it. Any single-region outage removes at most 2 replicas, so 3 survive and can elect a new leader through Raft; because every committed entry sits on a majority, any elected leader holds all committed writes, so none are lost. Replica failure is absorbed by the remaining majority, and the failed replica catches up through log replication when it returns. Read-your-writes is handled by routing the publishing user's own reads to the shard leader for a fixed grace window (for example 30 seconds) after their last write. After the window, timeline and third-party reads go to the nearest replica whose log is within a configured lag threshold of the leader, and lagging replicas are removed from the read pool until they catch up. Users who relocate keep their data on the home-region shard, accepting higher write latency, and a background job can migrate permanently relocated users with an asynchronous copy followed by a single cutover.

Each partition runs single-leader, log-based replication (a Raft-style consensus log) with N=3, one replica per region. (a) Protocol and quorums: the leader appends each write to its WAL and sends the entry to both followers. A write is acknowledged as committed only after a majority of the replicas, the leader plus one follower (W=2 of N=3), have durably flushed it, so the committed write lives in two regions and survives the loss of any single region. Reads that tolerate staleness use R=1 from the nearest follower. (b) Read-your-writes: every commit response returns the LSN of the entry it produced, and the client session stores the highest LSN it has seen as a session token. A session-bound read is served by a local follower only if that follower's applied LSN is at least the session token; otherwise the read waits briefly for the follower to catch up or is routed to the leader. (c) Failover: heartbeats detect leader loss within about a second. Surviving replicas run a Raft-style election in which a candidate needs votes from a majority (2 of 3). Each leadership term carries a monotonically increasing epoch (fencing token); followers reject any append from a lower epoch, and a leader can commit only with acknowledgements from a majority within its own term, so a stale or partitioned old leader cannot commit or be accepted by followers. The winner is the candidate with the most up-to-date log (highest committed LSN), so no committed transaction is lost, and promotion fits within 30 seconds. (d) Lag monitoring: continuously track, per follower, the difference between the leader's committed LSN and the follower's applied LSN, along with the time lag. Alert when lag exceeds a threshold (e.g., 5 seconds). A lagging follower is removed from the synchronous acknowledgement set until it recovers. If too few replicas remain to form W=2, writes block rather than commit with fewer than two regions, and session reads on a lagging follower are rerouted to the leader, which bounds user-visible staleness.

Single-leader replication gives linearizable writes, but every write must travel to the one leader, adding cross-region latency. Multi-leader replication lowers local write latency but needs conflict resolution and loses linearizability. Leaderless Dynamo-style systems offer tunable quorum consistency but require reconciling concurrent writes.

I would use a tiered replication topology: a cross-region Raft consensus group with five voting members, one per region, for ride-state records, and asynchronous local streaming for driver-location telemetry. Each committed ride-state write must be acknowledged by a quorum of three of five members before the client sees success, so every committed record is durable in at least three regions and survives the loss of any single region. Local replicas serve ride-status lookups and nearby-driver queries, so reads stay within the region and meet the 100 ms P99 target; these local reads are eventually consistent, and any read that must be linearizable (for example final fare settlement) is routed through the Raft leader or served as a quorum read, accepting cross-region latency for that operation only. Telemetry uses plain asynchronous replication with no follower acknowledgment and last-writer-wins on timestamps, which is acceptable because a slightly stale position is still useful and telemetry carries no no-data-loss requirement. For conflicts, ride-state transitions are never merged: the Raft leader serializes all writes to a ride in log order, assigns a monotonic per-ride sequence number, and rejects any transition that violates the state machine (for example completed back to in_progress) before commit. For failover, when the leader's region fails, the surviving members detect the heartbeat timeout and run a Raft election in which only a member with an up-to-date log (higher last log term, then longer log) can win; because a committed entry exists on a quorum, the new leader is guaranteed to hold every committed ride-state record. A leader epoch (Raft term) acts as a fencing token attached to every write, so a partitioned old leader cannot commit conflicting entries. Client writes that were in flight when the region died are retried using idempotent, ride-id-keyed operations, and any write acknowledged to a client was already committed by quorum and therefore survives.

Requirements. Beyond upload/download, the product is sync: a change on one device must propagate to the others, efficiently. 'Efficiently' means delta sync — never re-send an unchanged file. We must handle conflicts (two offline edits) without data loss and keep version history. Metadata ops (browse folders) are latency-sensitive; bulk byte transfer is throughput-bound.

Chunking & dedup. Files are split into content-hashed chunks (~4 MB; a chunk's id is the hash of its bytes). Editing one paragraph of a 1 GB file changes only the chunk(s) covering it, so the client uploads just those — a few MB, not a GB. Because chunks are content-addressed, identical chunks (same file shared by many users, or unchanged blocks across versions) are stored once — large dedup savings.

Block vs metadata split. Chunk bytes go to an object store (S3-style): content-addressed, effectively infinite, optimised for large immutable blobs — this holds the exabytes. The metadata DB holds the file→chunk-list mapping, the folder tree, ownership, and version history — small records with fast point/range reads to meet the <100 ms browse budget. The two layers have opposite access patterns, so they're separate systems.

Sync protocol & conflicts. Each device keeps a local view and learns of remote changes via a notification service (long-poll/push): 'your namespace changed past version V'. The client then pulls the metadata diff since its last-known version and downloads only the new chunks. Concurrent offline edits are detected by version: the second writer's base version is stale, so rather than overwrite, the system keeps both as a conflict copy (or merges where it safely can) and surfaces it — never a silent loss.

Consistency. Per file/namespace there's a monotonic version; metadata operations are ordered by it so all devices converge on the same state. Upload is commit-after-chunks: the client first writes all chunks to the object store, then flips the metadata pointer to the new chunk list. A half-uploaded file's pointer never goes live, so a reader never sees a file referencing chunks that aren't fully stored.

Scaling. The metadata DB shards by user_id (a user's files and folder tree stay together). The object store scales horizontally on its own. Sync notifications fan out to each of a user's devices. Dedup + delta sync are the multipliers — they cut both stored bytes and transferred bytes by large factors, which is what makes exabyte-scale storage and 5,800 ops/sec affordable.

Requirements & freshness. The workload is wildly read-heavy (~6,000:1 at peak) and global, with a p99 < 150 ms budget everywhere. The unlock is freshness tolerance: most catalog fields (title, description, images) can be seconds stale, while price/availability must propagate fast and be correct at checkout. That split is what lets me cache and replicate aggressively for the bulk of reads while treating the few critical fields specially.

Caching. A multi-tier cache fronts the read path: a small in-process cache on each app server, a shared distributed cache (Redis) behind it, and a CDN/edge cache for whole responses where possible. Reads are cache-aside: check cache → on miss, read the store, populate cache with a TTL. Writes update the store and invalidate/refresh the affected keys. With sensible TTLs this absorbs the overwhelming majority of the 30M reads/sec so the database barely sees them.

Replication. The single primary can't serve reads at this scale, so I add read replicas: all writes go to the primary, reads fan out across replicas (and mostly never reach them thanks to cache). This removes the read load from the primary, which now only handles the ~5K writes/sec. The catch is replication lag — a read replica may briefly trail the primary.

Consistency. I run eventual consistency for general catalog reads, with staleness bounded by cache TTL and replica lag — fine for browsing. For correctness-critical reads (the price/availability shown at checkout), I read from the primary (or a freshly-revalidated cache entry) so a customer never buys at a stale price. This is a deliberate CAP/freshness trade-off per field, not one global setting.

Hot keys & thundering herd. A flash-sale item that's 40% of reads would crush whichever single node holds its cache entry, so I spread the hot key: replicate it across multiple cache nodes / serve it from many edge locations / pin it in every app server's local cache — turning one hotspot into many. For the stampede when a hot entry expires and thousands miss at once, I use request coalescing (single-flight) so only one request recomputes while others wait, plus staggered/jittered TTLs and serve-stale-while-revalidate so an expiry never lets a herd through to the database.

Multi-region & bottleneck. Globally, I put edge caches and regional read replicas on each continent so reads are served near the user within the 150 ms budget; the primary stays in one region (writes are rare). The dominant bottleneck is the single write primary plus cache misses on hot keys — scaled by adding replicas and edge capacity, the hot-key and herd mitigations above, and, only if writes themselves outgrow one primary, sharding the catalog by product id. A read-your-writes need (seller edits then reloads) is met by routing that session's reads to the primary or a sticky/fresh replica.

Resolve the tension by giving up a single global write path rather than weakening the guarantees. Topology: shard accounts by a home region assigned at registration. Each shard is an independent Raft consensus group with 3 replicas placed in 3 different regions, with its leader in the home region, so the system is N independent single-leader problems instead of one global multi-leader problem. Quorum: a write commits once a majority (2 of 3) has persisted it in the Raft log, so one replica or one region can fail without losing committed data. Linearizable home-region balance reads are served by the leader only after it confirms it still holds leadership (lease read or ReadIndex, which requires a majority heartbeat), so no separate R=2 read quorum is needed for correctness. Failover and split-brain: when the leader is lost, a new leader is elected by majority; each leader carries a monotonically increasing term (fencing token), and followers reject entries from older terms, so a partitioned minority cannot commit and at most one leader exists per term. Non-home-region writes: the request is forwarded to the home-region leader and committed there; cross-continent round trips plus a majority commit can exceed 500ms p99, so constraint (2) is explicitly relaxed for remote-region writes and stated as a trade-off, while home-region writes stay within the SLA. Optimistic local acceptance with asynchronous reconciliation is rejected because it breaks linearizability and risks divergence under partition. Exactly-once application: every client transaction carries a client-generated unique request id; the state machine stores a durable dedup table keyed by request id, updated in the same log entry as the balance change, so a retried request either is rejected as a duplicate or returns the originally stored result. Log entries are applied in strict log index order on every replica, and re-applying an already-applied index is a no-op, so replays after recovery cannot double-apply money movement.

The old leader is in a minority partition (1 of 5 nodes) and cannot commit any writes because it lacks a majority quorum (3 of 5). It may append entries to its local log, but those entries stay uncommitted. When the partition heals, the old leader sees a higher term from the majority-elected new leader and reverts to follower. Its uncommitted entries conflict with the new leader's log and are truncated and overwritten through Raft's log-repair mechanism. The term number acts as a fencing token: followers reject AppendEntries RPCs carrying a stale term, and the old leader can never gather the majority acknowledgments it would need to commit. No committed data is lost; only the old leader's uncommitted entries are discarded.

Yes. The write is guaranteed to be present in the new leader's log because of Raft's Leader Completeness property, which states that any entry committed in a given term appears in the logs of all leaders of later terms. The mechanism is the election restriction: a voter grants its vote only to a candidate whose log is at least as up-to-date as its own, comparing the last entry's term first and then log length, and a candidate needs votes from a majority to win. The committed entry sits on 3 of 5 nodes, and any electing majority of 3 must overlap that set in at least one node. That overlapping voter holds the entry and therefore refuses any candidate that lacks it, so the winning leader must already contain the committed write. The acknowledged write cannot be lost by the leader crash.