Distributed Systems Fundamentals: Consensus, Clocks, and Failure

A distributed system is one where the failure of a computer you did not even know existed can render your own computer unusable. That quote, from Leslie Lamport, captures the essential challenge: in a distributed system, you cannot trust the network, you cannot trust the clocks, and you cannot trust other nodes to be alive. Understanding these failure modes is the foundation of distributed systems design.

What Makes Distributed Systems Hard

On a single machine, operations happen in a well-defined order. Memory is shared. If a function call fails, you get an exception immediately. None of this is true in a distributed system.

Network Partitions

The network can fail in ways that are worse than “everything is down.” A network partition means some nodes can talk to each other but not to other nodes. You end up with two groups that each think the other is dead.

Normal state:
  Node A ←→ Node B ←→ Node C

Network partition:
  Node A ←→ Node B    ✗    Node C
  (A and B can talk)       (C is isolated)
  
Both sides think they are the "real" cluster.

This is not theoretical. Network partitions happen in production due to switch failures, misconfigured firewalls, overloaded network links, and cloud provider issues. The CAP theorem tells us that during a partition, you must choose between consistency (all nodes see the same data) and availability (all nodes can respond to requests). You cannot have both.

Clock Skew

On a single machine, time.now() gives you a monotonically increasing value. Across machines, clocks drift. Even with NTP synchronization, clocks across data centers can differ by tens of milliseconds, and in pathological cases, by seconds.

This means you cannot use wall-clock timestamps to determine the order of events across machines. If Node A writes at timestamp 100 and Node B writes at timestamp 99, that does not mean A’s write happened after B’s — B’s clock might just be running slow.

Node A's clock:  100.000s
Node B's clock:   99.985s  (15ms behind)
Node C's clock:  100.023s  (23ms ahead)

Event at Node A at local time 100.000
Event at Node B at local time 100.010
Which happened first? You cannot tell from timestamps alone.

Google’s TrueTime (used by Spanner) solves this with atomic clocks and GPS receivers in every data center, giving a bounded uncertainty interval. Most of us do not have that luxury.

Partial Failures

In a distributed system, some components can fail while others continue working. A request might succeed on two out of three nodes. A network cable might drop 5% of packets. A node might respond, but so slowly that it might as well be dead.

This ambiguity is the hardest part. When you send a message and get no response, you do not know if:

• The message was lost
• The remote node crashed
• The remote node processed the message but the response was lost
• The remote node is processing the message very slowly

Each possibility requires a different response, but you cannot distinguish between them.

Consensus: Getting Nodes to Agree

Consensus is the problem of getting multiple nodes to agree on a single value, even when some nodes fail or messages are lost. This is the foundation of leader election, distributed transactions, and replicated state machines.

Paxos: The Original (and Confusing) Algorithm

Paxos, invented by Leslie Lamport in 1989, was the first proven consensus algorithm. It is notoriously difficult to understand — Lamport himself noted that most people who claim to understand Paxos have not actually implemented it correctly.

The basic idea: a proposer suggests a value, and a majority of acceptors must agree. The algorithm runs in two phases:

Phase 1 — Prepare. The proposer picks a unique proposal number n and sends a PREPARE(n) message to all acceptors. Each acceptor promises not to accept any proposal numbered less than n and returns any value it has already accepted.

Phase 2 — Accept. If the proposer gets promises from a majority, it sends ACCEPT(n, value) to all acceptors. If an acceptor has not promised to a higher-numbered proposal, it accepts.

Proposer               Acceptor 1    Acceptor 2    Acceptor 3
   │                       │              │              │
   │── PREPARE(1) ────────▶│              │              │
   │── PREPARE(1) ───────────────────────▶│              │
   │── PREPARE(1) ──────────────────────────────────────▶│
   │                       │              │              │
   │◀── PROMISE(1) ───────│              │              │
   │◀── PROMISE(1) ──────────────────────│              │
   │     (majority!)       │              │              │
   │                       │              │              │
   │── ACCEPT(1,"v") ─────▶│              │              │
   │── ACCEPT(1,"v") ────────────────────▶│              │
   │── ACCEPT(1,"v") ─────────────────────────────────▶│
   │                       │              │              │
   │◀── ACCEPTED ─────────│              │              │
   │◀── ACCEPTED ─────────────────────────────────────│
   │     Value "v" is chosen (majority accepted)       │

Paxos guarantees safety (nodes never disagree on the chosen value) but not liveness (the algorithm can stall if multiple proposers keep preempting each other). In practice, Multi-Paxos optimizes for the common case by electing a stable leader who acts as the sole proposer.

Raft: Consensus Made Understandable

Raft, published in 2014 by Diego Ongaro, was explicitly designed to be easier to understand than Paxos while providing the same guarantees. It has become the dominant consensus algorithm in practice, used by etcd, Consul, CockroachDB, and TiKV.

Raft divides consensus into three sub-problems:

Leader election. At any time, one node is the leader and the rest are followers. The leader handles all client requests. If the leader fails, a new election happens.

Log replication. The leader receives commands from clients, appends them to its log, and replicates the log to followers. A command is committed once a majority of nodes have it.

Safety. Raft guarantees that if a command is committed, it will be present in the logs of all future leaders.

Term 1: Node A is leader
  Client ──cmd──▶ A (leader) ──replicate──▶ B, C, D, E
                  A commits after majority (3/5) acknowledge

Term 2: A dies, B becomes leader (elected by C, D, E)
  Client ──cmd──▶ B (leader) ──replicate──▶ C, D, E
                  B has all previously committed entries

The key insight in Raft is the concept of terms — monotonically increasing integers that act as logical clocks. Each term has at most one leader. If a node receives a message from an older term, it rejects it. This prevents stale leaders from causing confusion.

# Simplified Raft node state
class RaftNode:
    def __init__(self, node_id, peers):
        self.node_id = node_id
        self.peers = peers
        self.current_term = 0
        self.voted_for = None
        self.log = []
        self.commit_index = 0
        self.state = 'follower'
        self.election_timeout = random.uniform(150, 300)  # ms

    def on_election_timeout(self):
        # No heartbeat from leader — start election
        self.state = 'candidate'
        self.current_term += 1
        self.voted_for = self.node_id
        votes = 1  # vote for self
        
        for peer in self.peers:
            response = peer.request_vote(
                term=self.current_term,
                candidate_id=self.node_id,
                last_log_index=len(self.log) - 1,
                last_log_term=self.log[-1].term if self.log else 0
            )
            if response.vote_granted:
                votes += 1
        
        if votes > len(self.peers) // 2:
            self.state = 'leader'
            self.start_heartbeats()

Leader Election Patterns

Leader election is a specific application of consensus: nodes must agree on which single node is the current leader.

Election via Consensus

The most robust approach: use a consensus algorithm (Raft, Paxos) to elect a leader. This is what etcd and ZooKeeper do. The leader holds a “lease” that must be periodically renewed.

Election via Distributed Lock

A simpler approach: nodes try to acquire a lock in a shared store (Redis, ZooKeeper). The node that gets the lock is the leader. It must periodically renew the lock, and if it fails to renew, another node acquires it.

class LeaderElection:
    def __init__(self, redis_client, node_id, lock_key="leader"):
        self.redis = redis_client
        self.node_id = node_id
        self.lock_key = lock_key
        self.lease_time = 10  # seconds

    def try_become_leader(self):
        # SET NX: only set if key does not exist
        acquired = self.redis.set(
            self.lock_key, self.node_id,
            nx=True, ex=self.lease_time
        )
        return acquired

    def renew_lease(self):
        # Only renew if we are still the leader
        pipe = self.redis.pipeline()
        current = self.redis.get(self.lock_key)
        if current == self.node_id:
            self.redis.expire(self.lock_key, self.lease_time)
            return True
        return False

Distributed Locking: Harder Than You Think

Distributed locking seems straightforward — acquire a lock, do work, release the lock — but the failure modes are subtle and dangerous.

The fundamental problem: a lock must be both safe (only one client holds it at a time) and live (it is eventually released even if the holder crashes). These properties are in tension.

Why a Simple Redis Lock Is Not Enough

Consider this scenario with a Redis-based lock:

1. Client A acquires the lock with a 10-second timeout.
2. Client A starts processing but hits a long garbage collection pause.
3. The lock expires after 10 seconds.
4. Client B acquires the lock and starts processing.
5. Client A’s GC pause ends — it still thinks it holds the lock.
6. Both clients are now operating “under the lock” simultaneously.

Timeline:
Client A: [acquire lock] [processing...] [GC PAUSE............] [still thinks it has lock!]
Client B:                                 [acquire lock] [processing...]
Lock:     [A holds] [A holds] [expired] [B holds] [B holds]

Danger zone: both A and B think they hold the lock

Fencing Tokens

The solution is fencing tokens: each lock acquisition gets a monotonically increasing token. The protected resource rejects any operation with a token older than the most recent one it has seen.

# Lock service returns incrementing fencing tokens
token_42 = lock_service.acquire()  # Client A gets token 42
token_43 = lock_service.acquire()  # Client B gets token 43

# Storage service checks the token
def write(data, fencing_token):
    if fencing_token < self.last_seen_token:
        raise StaleLockError("Token expired")
    self.last_seen_token = fencing_token
    self.storage.write(data)

Even if Client A wakes up from its GC pause and tries to write with token 42, the storage rejects it because it has already seen token 43 from Client B.

Vector Clocks and Logical Timestamps

Since physical clocks cannot be trusted to order events across machines, distributed systems use logical clocks.

Lamport Timestamps

Each node maintains a counter. On every local event, increment the counter. On every sent message, include the counter. On every received message, set your counter to max(local_counter, received_counter) + 1.

Lamport timestamps give you a partial ordering: if event A happened before event B, then timestamp(A) < timestamp(B). But the reverse is not true — two events with different timestamps might be concurrent.

Vector Clocks

Vector clocks give you a complete picture of causality. Each node maintains a vector of counters — one per node in the system.

Three nodes: A, B, C
Each maintains a vector [a, b, c]

Node A sends message to B:
  A's clock: [2, 0, 0] → sends with message
  B receives: B's clock was [0, 1, 0]
  B updates: [max(0,2), max(1,0)+1, max(0,0)] = [2, 2, 0]

Comparison rules:
  [2, 3, 1] happened before [3, 3, 2]  (all components ≤, at least one <)
  [2, 3, 1] is concurrent with [1, 4, 1]  (2>1 but 3<4)

class VectorClock:
    def __init__(self, node_id, num_nodes):
        self.node_id = node_id
        self.clock = [0] * num_nodes

    def increment(self):
        self.clock[self.node_id] += 1

    def merge(self, other_clock):
        for i in range(len(self.clock)):
            self.clock[i] = max(self.clock[i], other_clock[i])
        self.clock[self.node_id] += 1

    def happened_before(self, other):
        return (all(a <= b for a, b in zip(self.clock, other.clock))
                and any(a < b for a, b in zip(self.clock, other.clock)))

    def is_concurrent(self, other):
        return not self.happened_before(other) and not other.happened_before(self)

Vector clocks are used by Dynamo-style databases (Riak, DynamoDB) to detect conflicting writes. When two writes are concurrent (neither happened before the other), the system knows it has a conflict that needs resolution.

Gossip Protocols: How Nodes Discover Each Other

Gossip protocols (also called epidemic protocols) are a decentralized way for nodes to share information. Each node periodically picks a random peer and exchanges state. Information spreads exponentially, like a rumor through a social network.

Round 1: Node A knows about new member X
  A tells B → now A, B know

Round 2: A tells C, B tells D
  → A, B, C, D know

Round 3: A tells E, B tells F, C tells G, D tells H
  → 8 nodes know

After O(log N) rounds, all N nodes know.

Gossip protocols are remarkably robust. They tolerate node failures, network partitions, and message loss. Even if 30% of messages are lost, the information still propagates — just slightly slower.

Membership protocols use gossip for failure detection. Each node periodically gossips its list of known members and their heartbeat counters. If a node’s heartbeat counter has not increased for a threshold period, it is suspected dead.

class GossipNode:
    def __init__(self, node_id, peers):
        self.node_id = node_id
        self.peers = peers
        self.member_list = {node_id: {'heartbeat': 0, 'timestamp': time.time()}}

    def heartbeat(self):
        self.member_list[self.node_id]['heartbeat'] += 1
        self.member_list[self.node_id]['timestamp'] = time.time()

    def gossip(self):
        target = random.choice(self.peers)
        target.receive_gossip(self.member_list)

    def receive_gossip(self, remote_list):
        for node_id, info in remote_list.items():
            if node_id not in self.member_list:
                self.member_list[node_id] = info
            elif info['heartbeat'] > self.member_list[node_id]['heartbeat']:
                self.member_list[node_id] = info

    def detect_failures(self):
        now = time.time()
        for node_id, info in self.member_list.items():
            if now - info['timestamp'] > FAILURE_THRESHOLD:
                self.mark_suspected(node_id)

Cassandra, Consul, and Serf all use gossip protocols for cluster membership and failure detection.

The Split-Brain Problem

Split-brain occurs when a network partition divides a cluster into two (or more) groups, each believing it is the authoritative cluster. Both groups continue accepting writes, causing data divergence that is extremely difficult to resolve.

Before partition:
  [A] ←→ [B] ←→ [C] ←→ [D] ←→ [E]
  Leader: A

After partition:
  [A] ←→ [B]          [C] ←→ [D] ←→ [E]
  Group 1: "A is leader"    Group 2: "Elect C as new leader"
  Both accept writes → data divergence!

Prevention Strategies

Quorum-based decisions. Only allow a group to operate if it has a majority of nodes. In a 5-node cluster, a group needs at least 3 nodes. This guarantees at most one partition can operate.

STONITH (Shoot The Other Node In The Head). When a node suspects split-brain, it proactively shuts down the other node using out-of-band mechanisms (IPMI, power fencing). Brutal but effective.

Witness nodes. Place a lightweight witness node in a third network zone that breaks ties. The group that can reach the witness wins.

# Quorum check before accepting writes
class ClusterNode:
    def can_accept_writes(self):
        reachable = self.count_reachable_peers()
        total = self.total_cluster_size
        
        if reachable + 1 > total // 2:  # +1 for self
            return True  # We have majority
        else:
            # Refuse writes — we might be in a minority partition
            return False

How CockroachDB Achieves Distributed SQL

CockroachDB is a practical example of many concepts from this article combined into a production system. It provides a PostgreSQL-compatible SQL database that is distributed, consistent, and fault-tolerant.

Raft for consensus. Data is divided into ranges (similar to shards), and each range is replicated using Raft consensus. Writes are committed only when a majority of replicas acknowledge — guaranteeing strong consistency.

Hybrid logical clocks. CockroachDB uses a combination of physical timestamps and logical counters to order events. This avoids the need for specialized hardware (like Google’s TrueTime) while still providing serializable isolation.

Distributed transactions. CockroachDB supports multi-range transactions using a protocol similar to two-phase commit, but with optimizations for the common case where ranges are co-located. The transaction coordinator uses a parallel commit protocol that reduces latency.

Automatic rebalancing. When nodes are added or removed, CockroachDB automatically moves ranges to maintain an even distribution. This is transparent to the application — no manual resharding required.

The result is a system where you write standard SQL, but your data is automatically distributed, replicated, and fault-tolerant across multiple machines (or data centers). The trade-off is latency: consensus requires network round trips, so writes are slower than a single-node database. CockroachDB optimizes for correctness over raw speed.

Wrapping Up

Distributed systems are hard because they must operate in an environment where networks partition, clocks drift, and nodes fail at arbitrary times. Consensus algorithms (Raft, Paxos) give us a foundation for agreement despite these failures. Logical clocks let us reason about ordering without trusting physical time. Gossip protocols let nodes discover and monitor each other without centralized coordination.

The theme across all of these topics is managing uncertainty. You cannot eliminate the possibility of failure — you can only design systems that continue operating correctly despite it. Start by understanding the failure modes, then choose the tools and patterns that match your specific requirements for consistency, availability, and partition tolerance.