Distributed Systems Fundamentals Questions

Direct answer

A pure Lamport clock gives causal ordering: if event a happened-before event b, its counter guarantees a's value is less than b's. But the numbers carry no relationship to real elapsed time, so you can't use them to answer "how stale is this" or "did this happen a minute ago or an hour ago." An NTP-synchronized wall clock gives a number that approximates real time, but NTP only bounds error statistically and gives zero causal guarantee: two events with a genuine happens-before relationship on different nodes can still get timestamps in the wrong order if the nodes' clocks are skewed. A hybrid logical clock (HLC) combines a physical-time component with a logical counter so it inherits the causal correctness of a Lamport clock while staying close to real time, which is why production distributed databases (CockroachDB is the standard example) use it instead of relying on either mechanism alone.

Happens-before, briefly

Event a happens-before event b if a causes b directly (same process, sequential order), or a is a message send and b is its corresponding receive, or the relation holds transitively through a chain of such steps. Two events with no such chain between them are concurrent, and no clock discussed here can tell you which one "really" happened first, because there isn't a meaningful answer to that question for concurrent events.

Pure Lamport clocks

Each process keeps an integer counter C. On a local event, C = C + 1. When sending a message, the current C is attached to it. On receiving a message carrying Cm, the process sets C = max(C, Cm) + 1. This guarantees: if a happened-before b, then C(a) < C(b). It does not guarantee the converse: C(a) < C(b) does not mean a happened-before b, since a and b could be concurrent and simply landed on different counter values by coincidence of ordering. A Lamport timestamp also has no connection to real time at all: a value of "3" tells you nothing about whether the event happened a millisecond or a year ago.

NTP-synchronized wall clocks

NTP periodically corrects a machine's local clock against a reference, keeping it within a bounded error window that is typically small on a well-run local network and considerably looser across the public internet. Two problems follow for ordering purposes: first, if two nodes' clocks are off from each other by even a small nonzero amount, a later event on a fast node can get a smaller raw timestamp than an earlier, causally-prior event on a slow node, so ordering purely by wall-clock timestamp can violate causality. Second, an NTP correction can step a machine's clock backward when it's found to be running ahead, so consecutive readings from the very same machine aren't even guaranteed to be increasing, which breaks any protocol that assumes "later timestamp means later real event" on a single node, let alone across two.

Hybrid logical clocks (HLC)

An HLC keeps a pair (pt, l): a physical-time estimate pt and a logical counter l. Update rule:

• On a local event: pt' = max(local physical clock now, pt). If pt' changed (moved forward from a fresh physical reading), reset l = 0; otherwise l = l + 1. Set (pt, l) = (pt', l).
• On sending a message: attach the current (pt, l).
• On receiving a message with (pt_m, l_m): pt' = max(local physical clock now, pt, pt_m). If pt' equals both the local pt and the message's pt_m, set l' = max(l, l_m) + 1; if it equals only one of them, take that side's l and add 1; if it equals neither (a fresh physical reading is ahead of both), reset l' = 0.
• Compare two HLC values by pt first, breaking ties with l.

Because pt is always a running maximum, it can never move backward, even if the underlying physical clock is momentarily behind or gets stepped back by an NTP correction. And because a message receive always folds in the sender's pt_m and bumps l accordingly, HLC preserves the same causal guarantee Lamport clocks give: a send always compares less than its corresponding receive.

Worked example: two nodes with skewed clocks

Node A's physical clock and node B's physical clock disagree by a few units (B is running behind A). Initialize both HLC as (pt=0, l=0).

Pure Lamport clock trace:

1. A does local event e1: Lamport(A) = 1.
2. A sends message m to B carrying Lamport = 1.
3. B does its own local event e2, concurrently, before m arrives: Lamport(B) = 1.
4. B receives m: Lamport(B) = max(1, 1) + 1 = 2. The receive event correctly shows Lamport = 2, after both e1 and e2 (each at 1). But e1 and e2 both show 1 and are genuinely concurrent; the equal values don't mean "simultaneous," they mean "no causal relationship was ever established between them," and the values give no hint whether e1 happened at real time 100 or real time 100,000.

Raw NTP-timestamp trace, same events:

1. e1 on A is stamped with A's clock reading: 100.
2. A sends m, timestamped 101 by A's clock (one tick later).
3. B's local event e2 is stamped with B's clock reading: 99 (B is running 3 units behind A for this illustration).
4. B receives m; its clock reads 100 at that moment, so the receive event is stamped 100.

Comparing raw timestamps: m was sent at A-time 101 but appears received at B-time 100, a smaller number, even though the receive must happen after the send. Ordering purely by these NTP timestamps would incorrectly conclude the receive happened before the send.

HLC trace, same events, applying the update rule above:

1. e1 on A: physical clock reads 100. pt' = max(100, 0) = 100, changed, so l = 0. HLC(e1) = (100, 0).
2. A sends m; A's clock has ticked to 101 by send time: pt' = max(101, 100) = 101, changed, l = 0. HLC(send) = (101, 0); message m carries (101, 0).
3. B's local event e2: B's clock reads 99. pt' = max(99, 0) = 99, changed, l = 0. HLC(e2) = (99, 0).
4. B receives m=(101,0); B's own physical clock reads 100 at that moment. Apply the receive rule: pt' = max(100, 99, 101) = 101. Since pt' equals the message's pt_m (101) but not B's own pt (99) or the local physical reading (100), l' = l_m + 1 = 0 + 1 = 1. HLC(receive) = (101, 1).

Result: HLC(receive) = (101, 1) compares greater than HLC(send) = (101, 0), correctly ordering receive after send, even though B's own physical clock (100) was behind A's send-time clock (101). The HLC borrowed A's physical time component through the max() step specifically to preserve this ordering, which the raw NTP-timestamp comparison above could not do. And the HLC's physical component, 101, stays in the same real-time neighborhood as both clocks (99-101), unlike a pure Lamport counter, which would have produced something like "3" with no relationship to real time at all.

What HLC gives you that neither gives alone

1. Monotonicity even when the underlying physical clock runs behind or is stepped backward by an NTP correction, because the pt component is a running maximum that only ever advances.
2. The same happens-before guarantee Lamport clocks give, since a message receive always compares greater than its corresponding send.
3. A value that stays close to real wall-clock time, which lets you do things a pure Lamport counter cannot: read "as of physical time roughly T", size a lease's expiry in real seconds, or meaningfully compare timestamps from two nodes that never directly exchanged a message. Two Lamport clocks that never communicated are not comparable in any useful sense, since Lamport's ordering guarantee only holds across a causal chain; two HLCs that never communicated still each track real elapsed time, so they remain approximately comparable.

Leases and distributed coordination

A lease grants a component an exclusive right to act until an expiry, and that expiry check has to survive the granting node's physical clock moving. If a lease's expiry were checked against a raw NTP-disciplined wall clock, a backward correction on that clock could make an already-expired lease look valid again. Because HLC's physical component only advances, timestamps derived from it never regress even if the underlying physical clock momentarily does, which is why HLC (or something with the same monotonicity property) is the safer building block underneath lease-based coordination than a raw wall clock.

Trade-offs & pitfalls

A common trap is reading Lamport(a) < Lamport(b) as "a happened first" when a and b are actually concurrent; the clock never promised that. Another is assuming HLC removes the need for a synchronized physical clock entirely: HLC's pt component is still fed by the same underlying, NTP-disciplined system clock, so gross clock error still degrades how close HLC stays to real time. HLC protects the ordering and monotonicity guarantees from that error; it doesn't eliminate the error itself.

Linearizability and serializability sound similar but answer different questions. Linearizability is about a single object: every operation on it must appear to happen instantaneously at some point between when it was invoked and when it returned, and that ordering must match real time. Serializability is about multiple objects touched by a transaction: the outcome of running several transactions concurrently must be equivalent to running them in some serial order, but that order does not have to match real time or even the order the transactions actually started in. A system can have one property without the other, and assuming the wrong one silently breaks a different class of guarantee.

Two mechanisms that actually enforce serializability

• Two-phase locking (2PL): a transaction acquires every lock it needs before releasing any of them, and once it starts releasing locks it may acquire no more. This physically prevents conflicting concurrent access, at the cost of blocking and potential deadlock.
• Optimistic concurrency control (OCC): transactions proceed without locking, then get validated at commit time; if another transaction's concurrent writes conflict with what this one read, it aborts and retries. This avoids blocking under low contention but wastes work under high contention.

When you need one but not the other

Consider a key-value store advertising single-copy semantics: every replica must behave as if there is exactly one physical copy of the data, so any client reading a key right after a write, from any client, on any replica, sees that write or a later one, never a stale value. The same requirement shows up as a highly available configuration service needing linearizable reads: if a client reads a feature flag or a routing rule right after it changed, it must get the new value, since acting on a stale one applies the wrong policy. Neither of these needs serializability: there is no multi-key transaction to isolate, just one key's recency.

The mirror case: a reporting system running multi-row aggregate queries across many tables needs those queries to see an internally consistent snapshot (serializability, or at least snapshot isolation), but does not need that snapshot to be the absolute latest possible instant in real time. A report built from data a few hundred milliseconds behind the live system is fine, as long as every row it reads is mutually consistent with every other row it reads.

Worked example: what breaks if you assume the wrong one

Linearizable but not serializable, no cross-key transaction: a key-value store gives linearizable single-key reads and writes but has no multi-key transactions. A funds transfer moves 30 units from account A (currently 100) to account B (currently 50) as two separate linearizable writes: write A=70, then write B=80. A concurrent reader can land exactly between the two writes and read A=70 and B=50. Both individual reads are linearizable, each reflects the latest write to that specific key at the moment it was read, but the reader just observed a total of 70+50=120, when the true, fully-settled total is 70+80=150: 30 units appear to have vanished mid-transfer. That is the anomaly linearizability alone does not prevent, because it says nothing about atomicity across two different keys.

Serializable but write-skew possible, snapshot isolation only: a hospital scheduling system enforces one invariant, that at least one doctor remains on call.

$\text{doctors on call} \ge 1$

Two doctors, Alice and Bob, are both currently on call, so the on-call count is 2. Both, concurrently, read a snapshot showing 2 doctors on call and each independently decide it is safe to go off-call, and both commit that decision under snapshot isolation, since neither transaction's write conflicts with what the other actually wrote (each only writes their own on-call flag). The result: 0 doctors on call, violating the invariant, even though each transaction, viewed alone against its own snapshot, looks perfectly valid. Full serializability, not just snapshot isolation, would detect that these two transactions' reads and writes interfere and force one to abort; snapshot isolation's weaker check does not.

Trade-offs & pitfalls

• Common wrong turn: treating serializable as automatically meaning fresh or linearizable. It is not: transactions can be serialized in an order that does not match when they actually ran.
• Common wrong turn: treating a single-key linearizable store as if it gives transactional safety across several keys. It does not, by itself, unless the store also offers multi-key transactions on top.
• Snapshot isolation is cheaper than full serializability, since it does not need to detect every possible interleaving, only genuine write-write conflicts, and is what most production databases default to, which is exactly why the write-skew anomaly above shows up in practice more often than people expect.

Read-your-writes, monotonic reads, and monotonic writes are per-client session guarantees layered on top of a weaker (often eventually consistent) store, without paying for full linearizability across the whole system. Read-your-writes (RYW) promises a client that any read after its own write reflects that write. Monotonic reads promises a client's successive reads never go backwards in time. Monotonic writes promises a client's own writes are applied in the order it issued them. Each is enforced with a small piece of per-session state, not by coordinating the whole cluster.

What each guarantee promises, and how it breaks

Worked example: a monotonic-reads violation and its fix

Consider a likes counter on a post, replicated across replica X and replica Y with asynchronous replication.

Without the guarantee:

1. Client issues Read1, routed to replica X, which has applied all writes up to likes = 42. Client sees 42.
2. Client issues Read2 moments later. This time the request lands on replica Y, which has only applied writes up to likes = 40 (Y is lagging behind X). Client sees 40, a value older than what it already observed.

With the mechanism:
3. The client library remembers the highest value it has seen, last_seen_version = 42.
4. Read2 is sent with that version attached. Replica Y checks its own applied version (40) against the required version (42), sees it has not caught up, and either forwards the read to a replica that has (X, or any replica at version 42 or later) or holds the request briefly until its own replication catches up. The client never observes a regression.

Trade-offs and pitfalls

Sticky routing is the cheapest fix but weakens load balancing and complicates failover: if the pinned replica dies, the client loses its anchor, and if the guarantee is tied to a session cookie rather than the account, two devices logged into the same account (a phone and a laptop) are different sessions from the server's point of view and do not automatically share the guarantee with each other. Version tokens are more portable across devices and do not concentrate load onto one replica, but they add a token to every request and response, and the server still has to decide what a replica does when it cannot yet satisfy the token: wait, redirect, or refuse, which is itself a small latency-versus-freshness decision. Finally, these guarantees compose per client but do not add up to global consistency: a system can offer all three to every client individually and still allow two different clients to observe each other's writes in different orders. They make one user's own experience feel correct; they are not a substitute for linearizability when multiple clients must agree on a single order of events.

Linearizability, sequential, causal, and eventual consistency are four progressively weaker guarantees about the order in which operations on shared data appear to happen. Linearizability makes every operation look instantaneous and match real, wall-clock time. Sequential consistency drops the real-time requirement but still gives every observer the same single global order. Causal consistency only orders operations that are actually cause-and-effect related, letting unrelated operations be seen in different orders on different replicas. Eventual consistency drops ordering guarantees almost entirely and only promises that replicas converge once writes stop. Each weaker model permits more anomalies than the one above it.

Worked example: why causal consistency prevents an anomaly eventual consistency allows

Consider a social feed. Two events happen, in this order, involving the same user's friend:

1. Event P: a user publishes Post P.
2. Event C: after reading Post P, the user's friend writes Comment C, which references Post P.

Because the friend read P before writing C, C causally depends on P: P happened-before C.

• Under causal consistency, any replica that delivers C to a reader must already have delivered P to that same reader. There is no way for a client to see Comment C replying to Post P without also being able to see Post P: the system enforces the happened-before relationship on delivery.
• Under eventual consistency alone, P and C might replicate along different paths (different shards, different network routes) with no ordering guarantee between them. A reader on a lagging replica could receive C's replication packet before P's, and briefly render a comment that references a post the reader's own client cannot find yet, an orphaned reply. That is exactly the anomaly eventual consistency does not rule out and causal consistency does.

Because eventual consistency only promises the base table converges, a secondary index or materialized view (for example, a 'comments by post' index used to render the feed) can lag the base write for an unbounded window: the index might still return zero comments for Post P for some time after Comment C has already durably landed on a majority of the base replicas, since building the index from the base table's write stream is itself an eventually-consistent process, not an atomic one.

Trade-offs & pitfalls

• Common wrong turn: treating eventual consistency as one well-defined guarantee. It is really the absence of a guarantee during the convergence window, so two systems both labeled eventually consistent can behave very differently depending on how long that window typically is, and what session-level guarantees (read-your-writes, monotonic reads) are layered on top.
• Sequential consistency is rarely offered as a named product feature; it mostly shows up as an accidental byproduct of serving reads from any replica of a system that internally agrees on a single write order, without adding a real-time freshness check on the read path.
• Causal consistency requires tracking dependencies, commonly via vector clocks or similar metadata, which costs storage and complicates garbage collection, the same trade-off logical clocks introduce elsewhere in this material.
• Senior answers name the actual anomaly each model still allows, not just that it is looser. An answer that only says eventual is looser than causal, without naming a concrete permitted anomaly, is incomplete.

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