Distributed Systems Fundamentals Questions

Direct answer

These three terms describe how many times a record's effect can show up downstream, not how many times it crosses the wire. At-least-once guarantees nothing is silently dropped but tolerates re-delivery, so duplicates are possible. At-most-once guarantees no duplicates but tolerates silent loss. Exactly-once means the record's effect appears exactly once, even though delivery itself is usually still at-least-once under the hood; the "exactly" part is enforced by deduplication or a transactional write, not by literally never redelivering anything.

At-least-once

Mechanism: the consumer commits its read offset only after it has finished processing a record.

Scenario producing a duplicate: a consumer reads offset 100 from a partition and processes it, say incrementing an inventory counter, but crashes before committing that offset. On restart it resumes from the last committed offset, 99, and re-reads and reprocesses offset 100, incrementing the counter a second time. The counter's effect happened twice for one logical record.

At-most-once

Mechanism: the consumer commits its read offset before processing the record.

Scenario producing a loss: a consumer commits offset 100 immediately on receipt, then crashes while processing that record. On restart it resumes from offset 100 onward, since that offset is already committed, so the record at offset 100 is never processed. This is the mirror image of the at-least-once bug: it's the same commit-versus-process ordering, flipped.

Exactly-once: what it actually takes at the consumer

• Idempotent processing: design the effect so applying it twice produces the same result as applying it once, for example replacing "increment counter by 1" with "set counter to max(current, computed value)", or recording each processed key in a table with a uniqueness constraint so a repeat attempt is rejected rather than reapplied.
• Dedup window: a bounded structure, such as a set of recently-processed record ids, that the consumer checks before applying an effect. It must be sized to cover the maximum plausible redelivery delay; if consumer restarts are the main cause of redelivery and offsets commit every 30 seconds, a window covering the last few minutes of ids is enough, but an outage longer than the window falls back on whatever uniqueness constraint the underlying store provides.
• Transactional writes: commit the output write and the offset advance as a single atomic operation, so a crash between "wrote output" and "committed offset" cannot happen. Kafka's transactional producer does this across topic-partitions; a database sink can do the same by writing the output row and a processed-offsets row inside one database transaction.

Worked example: exactly-once via a transactional write

Consumer is at offset 100. It opens a transaction, upserts the inventory row (an idempotent write), writes offset=101 to an offsets table, and commits the transaction atomically.

• If the consumer crashes before the commit: nothing happened; the output was never applied and the offset was never advanced. Re-reading offset 100 next time reproduces the exact same atomic attempt, with no partial effect ever visible in between.
• If it crashes after the commit: offset 101 is already recorded, so the consumer will not re-read offset 100 on restart.

There is no window in which the output exists but the offset doesn't, or vice versa, because both changes are part of one transaction.

Practical guidance

Trade-offs & pitfalls

Exactly-once adds real coordination cost (transactions, dedup lookups), so its throughput and latency are worse than plain at-least-once, which is why it's reserved for cases where correctness genuinely requires it rather than applied everywhere by default. A dedup window sized too small silently degrades to at-least-once during a long outage, without any error being raised. A common interview trap is conflating a broker-level exactly-once guarantee (Kafka's own transactions between its topics) with true end-to-end exactly-once: Kafka's guarantee stops at the Kafka cluster boundary, so if the final effect lands somewhere outside it, an external database or a third-party API, that external system still has to be transactional or idempotent for the guarantee to actually hold all the way through.

Direct Answer

Paxos and Raft are both protocols that let a cluster of nodes agree on a value, or, in the log-replication case most real systems use, an ordered sequence of values, despite crashes and message delays, using a majority quorum so that any two decisions are guaranteed to have at least one node in common. Raft reaches the same safety guarantee as Paxos but organizes the protocol into named, sequential subproblems, mainly a single strong leader that serializes all writes during its term, which most engineers find much easier to implement correctly than Paxos's more general and symmetric design.

How Each Actually Reaches Agreement

Paxos (Multi-Paxos in practice). A proposer picks a proposal number and sends a Prepare message to the acceptors; each acceptor promises not to accept any proposal numbered lower and reports back the highest-numbered proposal it has already accepted, if any. Once the proposer hears back from a majority, it sends an Accept message carrying the value from the highest-numbered already-accepted proposal it was told about, not necessarily its own original value, and the value is chosen once a majority of acceptors accept it. Multi-Paxos elects a stable leader so that steady-state operation can skip repeating the Prepare phase for every new value.

Raft. Raft splits the same problem into leader election, where nodes agree on a single leader for a numbered term using randomized timeouts and majority votes, log replication, where the leader appends client commands to its own log and replicates them to followers, treating an entry as committed once a majority of nodes have stored it, and a safety rule that a candidate can only win an election if its log is at least as up to date as a majority of the cluster, which prevents a new leader from ever overwriting an already-committed entry.

Why Raft Is Easier to Implement and Reason About

Raft's decomposition gives each subproblem, who's the leader, how entries get replicated, how membership changes safely, its own explicit invariant, so an implementer can reason about one piece at a time. Paxos's proposer and acceptor roles are more general and symmetric, any node can propose at any time, which is elegant but produces more possible interleavings of concurrent proposals to reason about, especially once you move from the single-value textbook description to a real, steady-state, multi-value system, which is where most of the genuinely tricky Paxos engineering, such as stable leader election, log compaction, and membership changes, actually lives, and where the original paper says relatively little.

Comparison Table

A Worked Trace: Why a Competing Proposer Can't Just Overwrite the Value

Three acceptors, A1, A2, A3. Proposer P1 sends Prepare(1) to all three; none has accepted anything yet, so all three promise and report nothing. P1 gets a majority, 3 of 3, so it sends Accept(1, X). A1 and A2 accept proposal (1, X) before a second proposer, P2, starts a competing round. P2 sends Prepare(2) to A2 and A3; it doesn't reach A1. A2 has already accepted (1, X), so it promises not to accept below 2 and reports that it already accepted (1, X). A3 has accepted nothing, so it promises and reports nothing. P2 now has a majority of promises, A2 and A3, but because A2 reported an already-accepted value, the protocol requires P2 to propose that same value, X, rather than whatever value P2 originally intended. P2 sends Accept(2, X), not Accept(2, Y). Even though P2 won the second round, the value that gets chosen is still X. This is exactly the mechanism that keeps Paxos safe under concurrent proposers: a later round can change who proposes, but it cannot change what gets chosen once a value has reached a majority.

When to Still Reach for Paxos

Reach for Paxos, or a variant, instead of Raft when you're extending or must interoperate with an existing Paxos-based system where a rewrite isn't worth the risk, or when you need the extra flexibility Paxos's more general, symmetric design supports, such as non-majority or weighted quorum arrangements tuned for a specific wide-area latency profile, or when your team already has deep, tested Paxos expertise and infrastructure that a switch to Raft wouldn't meaningfully improve on.

When to Rely on an External Coordination Service Instead of Embedding Consensus Yourself

Most application teams don't actually need to choose between Raft and Paxos at all; they need a small set of coordination primitives, such as leader election for their own service, a shared lock, a bit of shared configuration, or service discovery, and implementing a correct Raft or Paxos group from scratch means owning a lot of subtle correctness surface, including log compaction, snapshotting, membership changes, and safe leadership transfer, for something an existing, battle-tested coordination service already does well. In that case, point your application at etcd, ZooKeeper, or Consul rather than embedding a consensus implementation. The case for building your own is when the consensus group needs to sit directly in your own data path for latency or throughput reasons, for example when you're building a replicated database yourself and every write needs to go through your own consensus group rather than round-tripping to an external service.

Trade-offs and Pitfalls

• It's a common misreading to treat Paxos as worse than Raft; it's a general, provably minimal algorithm. What actually makes it hard is that the original description covers a single value, and turning that into a real, steady-state, multi-value system requires additional engineering, such as a stable leader, log compaction, and membership changes, that Raft specifies as part of its core design instead of leaving as an exercise.
• Don't conflate which consensus algorithm to use with the more common real decision, which is whether to implement any consensus algorithm yourself at all; for most teams, depending on an existing coordination service is the right default, and only teams actually building infrastructure-level replicated systems typically end up choosing between Raft and Paxos directly.

A text editor where users type and delete concurrently without a lock needs a sequence CRDT (a Conflict-free Replicated Data Type: a data structure whose merge operation is commutative, associative, and idempotent, so concurrent replicas always converge to the same state without coordination). Each character gets a globally unique, immutable identifier, deletions mark a tombstone instead of physically removing the character, and a deterministic tie-break rule orders characters that were inserted concurrently at the same position. The design pays for this with permanent per-character metadata and a garbage-collection problem for tombstones.

Data structure

• Represent the document as an ordered sequence of atoms: { id: (site, counter), char, prev_id, tombstone }.
• id is globally unique because it pairs a site identifier (one per editing client) with that site's own monotonically increasing counter.
• prev_id records which atom this one was inserted immediately after, at the moment of insertion. That is what preserves each user's actual intent (insert right after the character I was looking at), even if other edits land nearby before this one is delivered.

Concurrent insert resolution

• Insert(id, prev_id, char) is broadcast to every replica.
• When two inserts share the same prev_id, both intended the same insertion point, that shared prev_id is exactly what concurrency looks like here. A deterministic comparator orders competing children of the same prev_id by site identifier (higher-precedence site placed first), so every replica, regardless of arrival order, produces the same final sequence.

Deletion and tombstones

• Delete(id) sets tombstone=true on that atom. The atom stays in the structure, because its ordering role must persist: a later, concurrently-arriving insert may still reference it as prev_id and needs it to resolve correctly.
• A tombstoned atom cannot be physically removed until every replica has definitely applied it.

Garbage collection of tombstones

• Replicas exchange a version vector (one counter per site, the highest counter from that site each replica has durably applied) during anti-entropy (a periodic background exchange where replicas compare state and reconcile any differences, rather than waiting for a live update to arrive).
• The element-wise minimum across all replicas' version vectors is the causal stability frontier: any tombstone at or below that frontier has been seen everywhere and can be physically purged.
• Cost: computing and propagating that frontier needs periodic all-to-all or gossip-based exchange, and a replica that stays offline indefinitely blocks compaction for everyone unless it is explicitly evicted from the version-vector set.

Worked example: two concurrent inserts at the same position

Both replicas have already converged on a one-character document containing 'X', a single atom with id (A,1). Two users, on different replicas, both position their cursor right after 'X' at the same time:

• Site A's user types 'p': op1 = Insert(id=(A,2), prev=(A,1), char='p')
• Site B's user types 'q': op2 = Insert(id=(B,1), prev=(A,1), char='q')

Both operations reference prev=(A,1): that shared reference is the concurrency. The merge rule for atoms competing for the same prev is order competing children by site identifier, descending, so a child from site B is placed before a child from site A when both point at the same prev.

• At replica A: op1 applies locally first, giving 'Xp'. When op2 arrives, it is spliced into the children of (A,1) and re-sorted by the rule above, giving order [op2, op1], so the document becomes 'Xqp'.
• At replica B: op2 applies locally first, giving 'Xq'. When op1 arrives, the same re-sort rule applies, again giving [op2, op1], so the document becomes 'Xqp'.

Both replicas land on 'Xqp' even though they applied the two operations in opposite order. That is what merging deterministically means in practice: the comparator, not arrival order, decides the final sequence.

graph LR
    X["X (id A,1)"] --> Q["q (id B,1)"]
    Q --> P["p (id A,2)"]

Now say 'q' is later deleted by its author. Atom (B,1) is not removed, only flagged tombstone=true, so the document reads 'Xp', but the underlying structure still holds three atoms, two live and one tombstone, until compaction runs.

Trade-offs & pitfalls

• Metadata growth: every character carries an id pair, a prev pointer, and (once deleted) a tombstone bit; on a long-lived, heavily-edited document the tombstone count can exceed the live character count, so uncompacted storage is proportional to live plus deleted atoms rather than just live ones.
• The same observed-remove technique (adding creates a fresh tag, removing targets only the tags actually observed) applies directly to a shared wishlist service: adding an item is an add-tag, removing it is a remove-tag scoped to the tags a client has actually seen, and the tombstone-compaction and consistent-read story are identical to the text editor's, just at item granularity instead of character granularity.
• Formatting spans (bold, italic) and structural moves (relocate a paragraph) do not compose as cleanly as single-character insert and delete; a naive extension can lose the same intention-preservation guarantee, which is why production collaborative-editing CRDTs spend real engineering effort specifically on this.
• Common wrong turn: implementing deletion by removing the atom outright instead of tombstoning it. That breaks any concurrent insert whose prev_id pointed at the now-missing atom, since there is nothing left to splice after.

Raft's leader election works by giving every follower a randomized election timeout; whichever follower's timeout fires first becomes a candidate, bumps the term number, votes for itself, and asks every other node for its vote. Randomizing the timeout is what keeps split votes rare: it makes it unlikely that two followers time out at nearly the same instant and both start competing candidacies in the same term. A short timeout detects a dead leader faster but risks more split votes and RPC churn; a long timeout is calmer but leaves the cluster leaderless longer after a real crash.

Approach

Implement the state machine directly: a follower's only action is to reset its timer whenever it hears from a leader or grants a vote. When the timer fires with no such contact, it becomes a candidate, increments its term, votes for itself, and fires RequestVote RPCs at every peer in parallel. It counts granted votes as they arrive; on a majority, it becomes leader. Any RPC response or incoming request carrying a higher term makes it step down to follower immediately; this term-monotonicity is what prevents two nodes from believing they are both leader for the same term.

type Server struct {
    mu            sync.Mutex
    id            int
    peers         []int
    currentTerm   int
    votedFor      *int
    state         string // "follower", "candidate", "leader"
    votes         int
    electionTimer *Timer
}

func (s *Server) onElectionTimeout() {
    s.mu.Lock()
    defer s.mu.Unlock()
    if s.state == "leader" {
        return // leaders don't run elections; they send heartbeats instead
    }
    s.startElectionLocked()
}

func (s *Server) startElectionLocked() {
    s.state = "candidate"
    s.currentTerm++
    s.votedFor = &s.id
    s.votes = 1 // vote for self
    term := s.currentTerm
    s.resetElectionTimer()

    for _, peer := range s.peers {
        go func(peer int) {
            resp := sendRequestVote(peer, s.id, term, s.lastLogIndex(), s.lastLogTerm())
            s.mu.Lock()
            defer s.mu.Unlock()
            if resp.Term > s.currentTerm {
                s.currentTerm = resp.Term
                s.becomeFollowerLocked()
                return
            }
            if s.state != "candidate" || term != s.currentTerm {
                return // stale response from a prior term, or we already moved on
            }
            if resp.VoteGranted {
                s.votes++
                if s.votes > (len(s.peers)+1)/2 {
                    s.becomeLeaderLocked()
                }
            }
        }(peer)
    }
}

func (s *Server) resetElectionTimer() {
    d := randomDuration(electionTimeoutMin, electionTimeoutMax)
    s.electionTimer.Reset(d)
}

func (s *Server) onAppendEntries(args AppendEntriesArgs) AppendEntriesReply {
    s.mu.Lock()
    defer s.mu.Unlock()
    if args.Term >= s.currentTerm {
        s.currentTerm = args.Term
        s.becomeFollowerLocked()
        s.resetElectionTimer() // heard from a valid leader, don't start a new election
    }
    // log-matching and append logic omitted here
    return AppendEntriesReply{Term: s.currentTerm}
}

Key points

• Persist currentTerm and votedFor to stable storage before replying to any RPC. A node that crashes and restarts without doing this could grant a second vote in a term it already voted in, which is a real safety violation, not just a liveness inconvenience.
• A candidate only votes for itself and only requests votes once it becomes a candidate; a follower that already voted for someone else this term must refuse any other RequestVote for that same term.
• Majority is computed against the full cluster size n, including the leader itself, not just the peers list:

$\text{majority} = \lfloor n/2 \rfloor + 1$

Complexity

• Message complexity: one election attempt sends O(n) RequestVote RPCs, one per peer, linear in cluster size per attempt.
• State per node is O(1) beyond the log itself: a handful of integers and a timer.
• The number of election attempts before success is not bounded in the worst case, repeated collisions are theoretically possible forever, but the randomized timeout range makes the expected number of retries small in practice, since each retry independently redraws a new random timeout, making the same two nodes colliding again increasingly unlikely round over round.

Worked example: split vote and recovery, printed trace

Three nodes N1, N2, N3, currently term 0, no leader. Each node draws its election timeout independently from the same range.

Case A, no collision, the common case: N2 draws the shortest timeout and fires first. It becomes a candidate for term 1, votes for itself, and sends RequestVote(term=1) to N1 and N3. Neither has timed out or voted this term, so both grant their vote. N2 now has 3 of 3 votes, itself plus two grants, exceeding the majority threshold of 2, so N2 becomes leader for term 1. N1 and N3 reset their timers on granting the vote, so neither starts its own candidacy.

Case B, a genuine split vote: suppose instead N1, N2, and N3 all draw timeouts close enough together that all three fire before any RequestVote arrives. Each becomes a candidate for term 1, votes for itself, and requests votes from the other two. Because each node already voted for itself in term 1 before any request from a peer arrived, each rejects the other two's requests, having already voted this term. No candidate reaches a majority: N1 has 1 vote, N2 has 1 vote, N3 has 1 vote. All three time out again and start a new election for term 2, each independently drawing a fresh random timeout. Because these new timeouts are drawn independently, the chance that all three collide again is much lower than the first time, and typically one of them fires meaningfully before the others and wins term 2 uncontested.

Edge cases

• A network partition splits the cluster into a majority side and a minority side: the minority side's nodes keep timing out and calling elections forever, incrementing the term each time, but can never reach a majority, so they correctly never elect a leader. This is deliberate: it is what prevents split-brain, at the cost of the minority side being unavailable for writes until the partition heals.
• A node that crashes and restarts must reload currentTerm and votedFor from stable storage before participating again; skipping this can let it vote twice in a term it already voted in before crashing.
• A slow node, a garbage-collection pause or a disk stall, is indistinguishable from a crashed node from its peers' point of view; both simply stop sending heartbeats, and the same election timeout mechanism reacts to both identically, which can trigger an unnecessary election when a node was merely paused, not dead.
• Election timeouts are purely local, measured intervals, a node's own clock counting elapsed time, so cross-machine clock skew, which matters a great deal for wall-clock timestamp ordering elsewhere in this topic, does not matter here.

Trade-off: short vs long election timeout

• A short timeout detects a genuinely dead leader faster, a shorter unavailability window, but increases both the chance of a split vote and the RPC and CPU overhead from repeated election attempts under network jitter.
• A long timeout keeps the cluster calmer, fewer accidental elections during transient network hiccups, at the direct cost of a longer window with no leader after a real crash, since followers simply wait longer before noticing.
• In practice the timeout range is chosen relative to the network's expected round-trip time: long enough that a heartbeat reliably beats the timeout under normal jitter, short enough that a real leader loss is noticed promptly. There is no universal constant; it has to be tuned to the deployment's actual network characteristics.

stateDiagram-v2
    Follower --> Candidate: election timeout fires
    Candidate --> Candidate: split vote, retry next term
    Candidate --> Leader: receives majority of votes
    Candidate --> Follower: sees higher term or valid leader
    Leader --> Follower: discovers higher term

With 5 replicas and R + W > 5 for every configuration worth choosing, all three quorum configurations give the same consistency guarantee: an acknowledged read is guaranteed to overlap an acknowledged write in at least one replica. The choice is really about where you want the availability and latency cost to land, not about correctness. Read-optimized (R=2, W=4) makes reads cheap and highly available at the cost of write availability and latency; write-optimized (R=4, W=2) is the mirror image; majority (R=3, W=3) splits the cost evenly. The size of that trade-off is a binomial-tail calculation, worth doing rather than eyeballing.

The model

Let $a$ be the probability a single replica is both alive and reachable from the coordinator for a given operation, independent and identically distributed across the 5 replicas. The probability that at least $k$ of $n$ replicas are available is the binomial upper tail:

Read availability for a configuration is $A(R, 5, a)$; write availability is $A(W, 5, a)$.

Consistency: for all three configurations, $R + W = 6 > n = 5$, so a successful read set and a successful write set are guaranteed to share at least one replica; that shared replica is what lets an acknowledged read observe the most recent acknowledged write, assuming the coordinator correctly resolves conflicting versions on read (by timestamp or vector clock).

Worked computation

Pin $a = 0.95$: a single replica has an independent 5% chance of being down or unreachable for a given request. Then $A(2,5,0.95)$ derives as:

The other five values in the table below follow the same expansion with different bounds:

At this per-replica reliability, majority sits in the middle at about 0.12% unavailability on both sides. Read-optimized cuts read unavailability by roughly 40x versus majority (0.0030% vs 0.1158%) but write unavailability rises nearly 20x (2.2593% vs 0.1158%); write-optimized is the exact mirror. For a read-heavy workload like a product catalog or a social feed, read-optimized buys a meaningfully more available read path at a write-availability cost that matters far less because writes are rare; for a write-heavy workload like event ingestion, write-optimized does the same in reverse.

R and W also set how many replicas an operation must wait for, not just how many must be reachable: R=2 only needs the two fastest responses, so read tail latency tracks the second-fastest replica, while R=4 tracks the fourth-fastest, which is usually a much bigger latency cost than the availability arithmetic alone suggests.

Trade-offs and pitfalls

This model assumes independent, identically distributed replica failures, which is the least realistic part of it: a rack or region outage takes down multiple replicas at once, and if replicas are spread out to make that unlikely, cross-region network latency then dominates real-world quorum latency far more than the independent-availability math predicts. Correlated failure modes shrink the effective advantage of any of these configurations relative to what the independent-failure formula suggests, sometimes sharply. Mean availability also hides tail latency: a configuration with high average availability can still have poor P99 (99th percentile) read latency if the quorum happens to include a consistently slow replica, something the availability formula does not capture at all. Finally, R + W > n only buys the overlap guarantee for a single key's operations as seen by a correct coordinator; it says nothing about cross-key transactions, and it assumes the coordinator resolves conflicting versions correctly in the first place.