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.

Direct Answer

A distributed lock service has to guarantee that, at any real instant, at most one live client believes it safely holds the lock, and it has to keep making that guarantee even when a client crashes mid-hold or the network delays messages. The safest way to build one is on top of something that already solves distributed agreement (a consensus protocol with a majority quorum), because a lock without a canonical ordering source can end up with two clients each convinced they're the holder.

Comparing Three Implementations

Fencing Tokens: Why a Lease Alone Isn't Safe

A lease with a time-to-live tells you how long to wait before you're allowed to consider a holder dead, but it can't stop a holder that is still alive, just delayed, from acting after its lease has actually expired. A process pause (for example, a stop-the-world garbage collection pause) or a delayed network write can make a client believe it still holds the lock long after the lock service has moved on.

Fencing tokens close this gap. Every successful lock grant returns a monotonically increasing number (the fencing token). The client includes that token with every operation it performs against the protected resource. The resource itself, not just the lock service, tracks the highest token it has ever seen and rejects any operation carrying a lower one.

sequenceDiagram
    participant A as Client A
    participant B as Client B
    participant L as Lock Service
    participant R as Protected Resource
    A->>L: Acquire lock
    L-->>A: Granted, token = 34
    Note over A: Long garbage collection pause, exceeds the lease TTL
    L->>L: Lease expires, lock released
    B->>L: Acquire lock
    L-->>B: Granted, token = 35
    B->>R: Write(token = 35)
    R-->>B: Accepted, 35 is higher than any token seen so far
    A->>R: Write(token = 34)
    R-->>A: Rejected, 34 is lower than the last accepted token 35

Concretely: suppose the lease time-to-live is configured to 10 seconds. Client A acquires the lock and gets token 34, then falls into a garbage collection pause longer than that window. The lock service, seeing no renewal, expires A's lease and grants the lock to client B with token 35. B writes to the resource with token 35; the resource records 35 as the highest token seen. A wakes up, still believing it holds the lock since it has no way to know time passed, and sends its own write carrying its old token, 34. Because 34 is lower than the 35 the resource already recorded, the resource rejects A's write. The stale holder is blocked from doing damage even though the lock service itself never directly told A it lost the lock.

Lease-Based Locking vs. Optimistic Concurrency

Not every coordination problem needs an actual lock. A lease is a pessimistic mechanism: it reserves exclusive access for a time window whether or not anyone else is actually trying to touch the resource, and it inherits all the lease-renewal and clock-skew risk described above. An alternative for many single-resource coordination problems is optimistic concurrency control: read the resource's current version, compute the intended change, and write it back with a compare-and-swap that only succeeds if the version hasn't changed since the read. This sidesteps lease renewal and clock skew entirely, since there's no time window to defend, but it means contending clients retry instead of queueing, which gets expensive under high contention.

As a rule of thumb: prefer optimistic compare-and-swap when you're coordinating a single write to a single record and can express the whole operation as one atomic check-and-set. Reach for a real lock or lease when the critical section spans multiple steps or has side effects outside the data store itself, such as kicking off a batch job or calling an external system, that can't be rolled into one atomic compare-and-swap.

Trade-offs and Pitfalls

• A lock service alone, without fencing enforced at the resource, is not actually safe: the resource has to participate in validating tokens, or a stale holder's writes go through unchallenged.
• The well-known Redlock (multi-instance Redis locking) design has been criticized for relying on synchronized clocks and not accounting for pauses like the one in the worked example; that critique is really an argument for fencing tokens as the real safety mechanism, with the lease itself only providing liveness (bounding how long you wait before treating a holder as dead), not safety.
• A lease time-to-live that's too short causes unnecessary churn under normal jitter; one that's too long makes real crashes take longer to recover from. Neither extreme removes the need for fencing tokens at the resource.
• Single-primary database row locks scale the worst across regions and don't survive a failover cleanly unless the failover procedure is specifically built to carry the lock state or invalidate it.

Direct Answer

Eventual consistency means that if a piece of data stops changing, every copy of it, spread across different machines, will eventually show the same value, but there's no promise about how quickly that happens. Right after something changes, different copies can briefly disagree, so different people, or even the same person on different devices, can see different things for a short window.

Three Concrete Things Users Will Notice

1. A change that looks like it disappeared or reverted. You update something, say your profile bio, and it saves fine, but a moment later, on a different device or after a refresh, you briefly see the old version again. This happens because that device happened to read from a copy of the data that hadn't caught up yet, not because your change was lost. The same effect shows up in less obviously social products too: right after a recommendation or personalization model is updated, some requests can still be served by a copy of the system using the old values for a short window, so two people who do the exact same thing a minute apart can get visibly different recommendations, purely because of which copy answered them.

2. Actions that look duplicated. If a user doesn't get quick feedback that their action went through (a like, a form submission), they often retry it. If the retry and the original attempt both eventually land, the user can end up seeing what looks like two of the same action. This isn't really an eventual-consistency artifact on its own; it becomes a real duplicate unless the system also deduplicates the underlying writes, not just the on-screen display.

3. Optimistic updates that hide the delay, until they don't. Many products make the delay invisible to the person taking the action by updating their own screen immediately, before the write has actually finished spreading to other copies. For example, when you post a comment, it appears in your own feed the instant you hit submit, even though the write is still propagating to the copies that other users' feeds are reading from. This makes the product feel instant for the person who acted, but it means other people may not see that comment for a moment, and if the underlying write ultimately fails, the app has to quietly roll back the comment it optimistically showed you.

A Concrete Trace

Say a comment-posting service has two copies of the feed data, one near user A and one near user B. User A posts "Great point!". Step 1: A's client shows the comment in A's own feed immediately, the optimistic update, while the actual write is sent to A's nearby copy. Step 2: User B, served by their own nearby copy, refreshes their feed before the write has replicated over to B's copy; B does not see the comment yet. Step 3: once the write has replicated to B's copy, B's next refresh does show the comment. Nothing was lost; B was simply reading from a copy that hadn't caught up at step 2.

Trade-offs and What to Plan For

• Eventual consistency is a deliberate trade for availability and responsiveness, not a bug, but it is the wrong choice for data where a stale answer is actively harmful, such as an account balance, the last unit of inventory, or a security permission change. Those flows are usually worth paying for stronger consistency even if it's slower.
• A common and cheap mitigation for the "did my own change disappear" complaint is guaranteeing read-your-writes (RYW): making sure the person who just made a change always sees their own latest write, typically by routing their own subsequent reads back to the copy that has it, even while other users' view of that same data is still catching up.
• A common wrong turn is treating optimistic UI as if it solves eventual consistency; it only hides the delay from the person who acted. It doesn't change how long the write actually takes to reach everyone else, and it adds its own failure case, rolling back a shown-then-failed action, that the product needs to handle gracefully.

A CRDT (Conflict-free Replicated Data Type) is a data structure whose update and merge operations are mathematically guaranteed to make every replica converge to the same value, with no locking, coordination, or central authority, as long as every update eventually reaches every replica. State-based CRDTs ship the whole replica state and merge it with a commutative, associative, idempotent join function; operation-based CRDTs ship individual operations that must themselves be commutative and be delivered with causal ordering. A grow-only counter (G-Counter) and an observed-remove set (OR-Set) are the two simplest concrete examples of this guarantee in action.

Why convergence is guaranteed

Convergence works because the merge operation is commutative (order doesn't matter), associative (grouping doesn't matter), and idempotent (merging a state with itself changes nothing), so applying merges in any order, any number of times, produces the same final state. Formally, this kind of merge is called a join, and a replica's state is modeled as an element of a join-semilattice: a partially ordered set where the join always computes the least upper bound of two states. That is the actual property behind convergence without coordination: it is not that conflicts never happen, it is that the merge function is defined so a conflict has exactly one well-defined resolution no matter how or when it gets computed.

G-Counter (grow-only counter)

• State: a vector with one non-negative integer slot per replica, c[i].
• Local update: a replica only ever increments its own slot.
• Merge: element-wise maximum across the two vectors.

$c'[i] = \max(c_1[i], c_2[i])$

• Read: sum across all slots.

$\text{total} = \sum_{i} c[i]$

• Because each slot only ever grows for its own replica, taking the max per slot can never lose an increment either side already recorded.

OR-Set (observed-remove set)

• State: a set of (element, unique tag) pairs, split into an add-set and a remove-set of tags.
• Add(e): mint a fresh tag, insert (e, tag) into the add-set.
• Remove(e): copy every tag currently observed for e into the remove-set; it removes only tags this replica has actually seen, never tags added elsewhere that haven't arrived yet.
• Merge: union the add-sets, union the remove-sets.
• An element counts as present if it has at least one tag in the add-set that is not in the remove-set.

Naming the comparators explicitly

Worked example: G-Counter convergence

Three replicas A, B, C start at (0,0,0):

• Replica A processes 2 local increments: its state becomes (2,0,0).
• Replica B processes 3 local increments, concurrently, before hearing from A: (0,3,0).
• Replica C stays idle: (0,0,0).

A and B exchange state and merge (element-wise max): merge((2,0,0),(0,3,0)) = (2,3,0). Read = 2+3+0 = 5. C later merges with that result: merge((0,0,0),(2,3,0)) = (2,3,0). Read = 5. Whichever order the three replicas merge in, the final vector is (2,3,0) and the read is 5, exactly matching the 2+3=5 real increments actually performed. No increment is lost and none is double-counted.

Worked example: OR-Set add and remove race

Replicas R1 and R2 have already converged on a set containing 'milk' with tag t1. The two replicas are then partitioned from each other:

• R1's user removes 'milk': remove-set gains {t1}, the only tag R1 has ever observed for 'milk'.
• R2's user, unaware of the removal, re-adds 'milk': add-set gains a brand-new tag {t2}, so the add-set is now {t1, t2}.

On merge: add-set = {t1, t2} (union), remove-set = {t1} (union). 'milk' is present because t2 is in the add-set and not in the remove-set. This is the correct outcome: R2's re-add introduced a tag the remover never saw, so it survives, exactly the observed-remove semantics the name describes.

Trade-offs & pitfalls

• Storage and bandwidth: every element needs extra metadata (a vector slot per replica for counters, a unique tag per add for sets), and removed elements don't disappear until a garbage-collection pass establishes causal stability across replicas.
• The edge case that catches teams out: CRDTs don't compose across non-commutative operations. A G-Counter or OR-Set is safe because the operations that define it (increment, tagged add and remove) are commutative by construction. But if you build an append-only log CRDT and then bolt on an application-level 'delete the last 3 entries' operation defined by position, that composition is not well-defined under concurrency: 'last 3' means something different on each replica depending on how many entries have been concurrently appended there at the time the delete runs, so two replicas can end up deleting different entries even though each individually applied a correct-looking CRDT merge. The fix is the principle OR-Set already uses: target deletions by a stable element identifier, never by position or count.
• When to avoid: anywhere a global invariant spans multiple keys (uniqueness, a balance that must never go negative), or the business logic genuinely isn't commutative. CRDTs solve convergence, not arbitrary correctness.

Raft replicates a log by having the leader append every client command locally, then replicate it to followers via AppendEntries RPCs; once a majority, including the leader, has stored an entry, the leader marks it committed and applies it to its state machine, and followers catch up the same way once they learn the new commit index. A slow or long-disconnected follower gets caught up either by incrementally backfilling AppendEntries, or, if it has fallen too far behind, by receiving a full InstallSnapshot of compacted state. The reason two leaders can never commit conflicting entries at the same log position is not a single rule, it is two rules working together: how logs are kept consistent, the log-matching property, and who is even allowed to become leader, the election restriction.

Terms and leader election, brief recap

Each term has at most one leader, elected by majority vote (covered in depth elsewhere in this material); what matters for replication is just that every entry in the log is tagged with the term of the leader that created it.

AppendEntries and the commit rule

• The leader sends AppendEntries(prevLogIndex, prevLogTerm, entries, leaderCommit) to each follower. A follower only accepts the new entries if its own log already has an entry at prevLogIndex with term prevLogTerm, matching exactly. If it does not match, the follower rejects, and the leader backs off to an earlier index and retries; this is what repairs a follower whose log has diverged.
• An entry counts as committed once it, and transitively every entry before it, is stored on a majority of servers, and that entry was created in the leader's own current term. That second condition, from the leader's own term, is the subtle part; see the worked example below for why it is necessary.
• Once committed, the leader applies the entry to its state machine and advances leaderCommit, which followers learn on the next AppendEntries, including empty heartbeats, and apply locally too.

Log compaction via snapshotting

• When a follower has fallen far enough behind that the leader has already discarded, or compacted, the log entries it needs, the leader sends InstallSnapshot instead: a full compacted copy of state plus the index and term of the last entry included in it.
• The follower replaces its own log prefix with the snapshot and adjusts its own next-expected index accordingly, then resumes normal AppendEntries catch-up from there.

Why two leaders can never commit conflicting entries at the same position, in plain terms

Two facts, combined, make this impossible:

1. Log matching: if two servers' logs both contain an entry at the same index with the same term, every entry up to and including that index is identical on both. This holds because a leader never reorders or overwrites its own log during its term, it only appends, and every follower's AppendEntries acceptance check requires the previous entry to match exactly before accepting anything new.
2. Election restriction: a candidate can only win a vote from a node whose log is at least as up to date as the candidate's own, compared by last entry's term, then index. Since winning requires a majority of votes, and any already-committed entry was, by definition, stored on a majority, at least one voter in any winning majority must already hold that committed entry. Any two majorities out of the same n servers are guaranteed to overlap in at least one server:

$|Q_1| + |Q_2| > n \implies Q_1 \cap Q_2 \neq \emptyset$

That overlap is exactly why whoever wins an election has a log at least as up to date as a node that already holds any previously committed entry, so the new leader must already have it too. A new leader can extend the log with new entries, but it can never come into office missing, and therefore able to overwrite, something the cluster already agreed on.

The genuinely tricky part, and the reason committed equals majority replicated is not the whole rule, is entries a leader replicated to a majority before it finished committing them in its own term and then crashed. A later leader, elected without ever having seen that specific entry, could otherwise legally overwrite it, which is why Raft only counts an entry as truly committed once a majority holds an entry from the current leader's own term: committing that later entry transitively commits everything before it, by log matching, without ever needing to separately reason about whether an older, not-yet-current-term entry was safe on its own.

Worked example: why from the leader's own term matters, printed trace

Five servers, S1 through S5.

• Term 2: S1 is leader. It appends an entry at log index 2, call its value 'x', and replicates it to S1 and S2 only, 2 of 5 servers, not a majority, before S1 crashes.
• Term 3: S5 becomes leader; S5's log does not have 'x' yet, and that is fine, since 'x' was never committed and was never required for S5's election vote to be up to date. S5 overwrites index 2 with its own entry, 'y', replicates it to S5 and S3, 2 of 5, still not a majority, then S5 also crashes before committing it.
• Term 4: S1 restarts and is re-elected leader; its log, containing the old 'x' at index 2, is still eligible since 'y' was never committed to a majority either. S1 now replicates its old index-2 entry 'x' to a majority: S1, S2, and S3, 3 of 5. Index 2 now genuinely has 'x' on a majority of servers.

Here is the crux: if Raft's rule were simply committed equals majority replicated, index 2 (holding 'x') would now count as committed, purely because a majority happens to hold it, even though it was replicated under a different leader's term, term 2, than the one currently in power, term 4. Raft explicitly does not commit it on that basis alone. Only once S1, as term-4 leader, replicates a new entry from term 4 to a majority does index 2 become safely committed, as a side effect of committing that later entry, since log matching guarantees everything before it, including 'x' at index 2, is identical across that same majority too. Without this extra rule, a hypothetical term-5 leader that had never seen 'x' could still legally overwrite index 2 yet again, since nothing had actually locked it in, and that overwrite would silently undo something an operator might already have treated as durably committed.

sequenceDiagram
    participant L as Leader (term 4)
    participant F1 as Follower S2
    participant F2 as Follower S3
    L->>F1: AppendEntries(index=3, term=4)
    L->>F2: AppendEntries(index=3, term=4)
    F1-->>L: success
    F2-->>L: success
    Note over L: majority (3 of 5 incl. leader) now hold index 3 from term 4, so index 3 commits, and index 2 commits transitively

Trade-offs & pitfalls

• Common wrong turn: explaining commit as majority replication, full stop. That is necessary but not sufficient; the current-term condition above is exactly what closes the gap, and is the part people skip when reciting Raft from memory.
• Followers that reject an AppendEntries because prevLogIndex or prevLogTerm do not match force the leader to step backward and retry; a purely one-at-a-time backoff is simple but slow for a follower that has fallen far behind, which is part of why snapshotting exists as an alternative catch-up path rather than only incremental backfill.
• Operational note: watching how often followers need a full InstallSnapshot, versus incremental catch-up, is a useful signal that a follower, or the network path to it, is chronically slow. This applies just as well to a Raft-backed machine learning model-metadata store as to any other Raft-based service; the underlying signal is the same regardless of what the log stores.