Develop a distributed consensus algorithm for reliable data replication.

Question

Anonymous · Accepted Answer

I'll lay out a complete, implementable design in the Raft/Multi-Paxos family — a leader-based replicated log. I'm assuming the common case: crash-recovery faults (not Byzantine), 3–7 replicas in one or two regions, and a requirement for linearizable reads and writes. I'll note at the end what changes if those assumptions don't hold. MODEL AND GUARANTEES Assumptions. Asynchronous network — messages may be dropped, delayed, reordered, or duplicated, but not undetectably corrupted (checksum every record). Nodes fail by crashing and may recover with their durable state intact. Clocks are not assumed synchronized for safety, only for liveness. Tolerance. N = 2f+1 replicas survive f failures. Safety holds always; liveness holds once the network becomes reliably timely (partial synchrony) and a majority is alive. Guarantee. Every replica applies the same sequence of commands in the same order to a deterministic state machine, and once a write is acknowledged it survives any f crashes. Externally, the system is linearizable. The entire protocol rests on one property: any two majorities intersect in at least one node. Every rule below exists to make that intersection carry information forward.Here [http://forward.Here]'s the write path — every acknowledged write must be durable on a majority before it's committed:## Replica state Persistent (fsync before responding to any RPC): currentTerm, votedFor, log[], and snapshot metadata. Volatile: commitIndex, lastApplied. Leader-only volatile: nextIndex[], matchIndex[]. A term is a monotonically increasing logical clock; each term has at most one leader. Every RPC carries a term, and a node seeing a higher term immediately steps down. LEADER ELECTION Elections use randomized timeouts (e.g. 150–300 ms, or 10× your p99 RTT) so candidates rarely collide. Two additions to the naive scheme matter in production: pre-vote, which stops a node returning from a partition with an inflated term from deposing a healthy leader, and CheckQuorum, which makes followers refuse votes while they still hear from a leader. on RequestVote(term, candidateId, lastLogIndex, lastLogTerm, isPreVote): if term < currentTerm: return (currentTerm, false) if heardFromLeaderRecently(): return (currentTerm, false) # CheckQuorum upToDate = (lastLogTerm, lastLogIndex) >= (log.lastTerm(), log.lastIndex()) if isPreVote: return (currentTerm, term > currentTerm and upToDate) # no state change if term > currentTerm: currentTerm = term; votedFor = nil; role = Follower if votedFor in {nil, candidateId} and upToDate: votedFor = candidateId; fsync(); resetElectionTimer() return (currentTerm, true) return (currentTerm, false) The upToDate check is the first of three rules that carry safety. Because any two majorities intersect, and a candidate needs a majority of votes, a candidate whose log is missing a committed entry cannot win — some voter in every majority has that entry and will refuse. LOG REPLICATION on AppendEntries(term, leaderId, prevIndex, prevTerm, entries, leaderCommit): if term < currentTerm: return (currentTerm, false) if term > currentTerm: currentTerm = term; votedFor = nil role = Follower; resetElectionTimer() if prevIndex > log.lastIndex(): return (currentTerm, false, hint = log.lastIndex() + 1) if log.termAt(prevIndex) != prevTerm: t = log.termAt(prevIndex) return (currentTerm, false, hint = firstIndexOfTerm(t)) # skip whole term for (i, e) in entries: idx = prevIndex + 1 + i if log.has(idx) and log.termAt(idx) != e.term: log.truncateFrom(idx) # truncate ONLY on a real conflict log.append(e) fsync() # durable before acking commitIndex = min(leaderCommit, prevIndex + len(entries)) return (currentTerm, true, matchIndex = prevIndex + len(entries)) The prevIndex/prevTerm check gives the Log Matching property: if two logs contain an entry with the same index and term, all preceding entries are identical. It holds by induction — the check verifies the predecessor, which was itself verified when it was written. The hint lets a lagging follower be caught up in O(number of divergent terms) round trips instead of O(entries). Never truncate on a stale or duplicated AppendEntries that carries no conflicting entry; that's a classic way to lose committed data. THE COMMIT RULE on ack(follower f, matchIndex m): matchIndex[f] = max(matchIndex[f], m) nextIndex[f] = matchIndex[f] + 1 N = median(matchIndex ∪ {log.lastDurableIndex()}) # majority-replicated index if N > commitIndex and log.termAt(N) == currentTerm: commitIndex = N apply(lastApplied+1 .. commitIndex); replyToClients() The log.termAt(N) == currentTerm condition is subtle and non-negotiable. An entry from a previous term that is present on a majority is not necessarily committed — a later leader could still overwrite it. A new leader must therefore append a no-op entry in its own term immediately on election; committing that no-op transitively commits everything before it. Omitting this is the single most common way real implementations lose acknowledged writes. Together these three rules yield Leader Completeness: any entry committed in term T is present in the log of every leader of every term > T. The proof is by contradiction — the committing majority and the electing majority of the next term share a node, which either voted no or already had the entry. LINEARIZABLE READS Reads must not be served from a leader's local state without proof it's still leader — a partitioned leader can serve arbitrarily stale data. on ClientRead(q): if role != Leader: redirect(leaderId) wait until noOpOfCurrentTerm is committed idx = commitIndex confirm leadership via one heartbeat round to a majority wait until lastApplied >= idx return stateMachine.query(q) This is ReadIndex: no disk write, one round of heartbeats, and concurrent reads batch into a single round. If you're willing to assume bounded clock drift, a leader lease removes even that round trip, at the cost of a safety assumption about clocks — I'd default to ReadIndex and add leases only under measured load pressure. Followers can serve reads by fetching a readIndex from the leader and waiting for their own apply index to catch up, which offloads read traffic without weakening linearizability. MEMBERSHIP CHANGES Add or remove one voter at a time; consecutive single-member configurations always have intersecting majorities, so no joint-consensus phase is needed. Three constraints: * A new node joins first as a non-voting learner, streaming the log until it's within a bounded number of rounds of the leader. Only then is the config change proposed. This prevents a slow new member from stalling commits. * A config change takes effect the moment it is appended, not when committed, and a leader must have committed an entry in its current term before proposing one — otherwise an uncommitted config change can be rolled back into a conflicting one. * A leader that removes itself continues to serve until the change commits, then steps down. LOG COMPACTION Uncapped logs are unrecoverable in practice. Each replica independently snapshots the state machine at lastApplied, storing lastIncludedIndex, lastIncludedTerm, the active configuration, and the client session table, then truncates the prefix. A leader whose nextIndex[f] has fallen below its snapshot boundary sends InstallSnapshot in chunks instead of log entries. The state machine needs a consistent point-in-time snapshot — copy-on-write structures or an LSM tree make this cheap; a naive stop-the-world serialize will show up as latency spikes. EXACTLY-ONCE CLIENT SEMANTICS Consensus alone gives at-least-once: a client that times out and retries can apply the same command twice. Give each client a unique ID and a monotonic sequence number, and have the state machine keep a per-client table of the highest applied sequence number and its response. Duplicate commands return the cached response instead of re-applying. The session table is part of the state machine, so it's replicated and snapshotted with it. Expire sessions via a committed entry, never a local timer, or replicas will diverge. PERFORMANCE Batch many client commands into one AppendEntries and one fsync — group commit is the dominant throughput lever. Pipeline by advancing nextIndex optimistically rather than waiting for each ack. Issue the leader's own fsync in parallel with the broadcast, counting the leader toward the quorum only once its own write is durable. Under WAN deployments, consider flexible quorums: election and replication quorums only need to intersect each other, so you can shrink replication quorums at the cost of larger election quorums. FAILURE MODES WORTH DESIGNING FOR EXPLICITLY * Disk loss. A node that loses its persistent state must never rejoin under the same identity — with votedFor erased it can vote twice in one term and elect two leaders. Wipe and rejoin as a new member, or persist an incarnation token that peers validate. * Silent corruption. Checksum every log record and snapshot; on a checksum failure, truncate at that point and refuse to serve until re-caught-up from the leader. * Clock jumps. Nothing in the safety argument may depend on wall-clock time. Use monotonic clocks for timers; if you adopt leases, bound them by measured drift with a wide margin. * fsync lies. Verify that your storage stack actually honors fsync (write barriers, disabled volatile write caches). A great deal of "impossible" consensus data loss is a lying disk. VERIFICATION Specify the protocol in TLA+ and model-check the safety invariants (election safety, log matching, leader completeness, state machine safety) for small cluster sizes — this catches design errors cheaply. Then build the implementation on a deterministic simulation harness where the network, clocks, and disks are all pluggable and seeded, so you can run millions of randomized partition/crash/reorder schedules and replay any failure exactly. Finish with black-box linearizability checking (Jepsen with Elle) against the real binary under injected faults. IF YOUR ASSUMPTIONS DIFFER If nodes can be malicious rather than merely crash, you need 3f+1 replicas and signed messages — HotStuff or PBFT. If replicas are spread across high-latency regions and leader bottlenecking hurts, leaderless protocols like EPaxos commit non-conflicting commands in one round trip at the cost of far more complex conflict handling. If your data exceeds one machine, shard it and run an independent consensus group per shard, layering two-phase commit over the groups for cross-shard transactions. I'd be glad to go deeper on any piece — the TLA+ spec, a reference implementation of the leader loop in a specific language, or the ReadIndex and lease trade-off under your actual latency budget.

Develop a distributed consensus algorithm for reliable data replication.

Practice this question with AI

Go Premium

Community Answers

Unlock Full Access

Practice More Questions