Distributed Locks, Leases, Leader Election and Consensus

Pixly sends every user a weekly digest email on Sunday morning. The job was written when there was one server. Now there are eighteen, the job runs on each of them, and two million users receive eighteen copies of the same email. The obvious fix, "make sure only one server runs the job", turns out to be one of the trickier problems in distributed systems.

This article is about one-at-a-time behaviour across many machines: distributed locks, leases, leader election, and the consensus algorithms underneath them. Its most useful lesson comes first: you often do not need a lock at all.

• How to avoid needing a distributed lock.
• The difference between locks for efficiency and locks for correctness.
• Leases, renewal and expiry, and why a paused process breaks naive locks.
• Fencing tokens, with a runnable demonstration.
• Database-based and Redis-based locks, and their limits.
• Leader election patterns.
• Consensus: what Raft does, why majorities matter, and what ZooKeeper and etcd provide.

A distributed lock adds a dependency, latency and subtle failure modes. Many "we need a lock" problems have simpler answers you have already met:

Notice the pattern: let the data store be the referee. A database can enforce "only one" for a specific row far more reliably than application servers can coordinate among themselves.

Martin Kleppmann's analysis of distributed locking draws a distinction that clarifies every later decision (How to do distributed locking):

• A lock for efficiency prevents doing the same work twice when doing it twice is merely wasteful: recomputing a cache entry, sending a non-critical report twice. If the lock occasionally fails and two nodes do the work, the cost is small.
• A lock for correctness prevents two nodes from acting when both acting would corrupt data or cause harm: two leaders writing to the same file, two processes moving the same money.

For efficiency, a simple, fast lock (for example a single Redis key with a timeout) is fine. For correctness, a lock alone is not enough, for reasons the next section explains, and you need fencing.

A lock held across a network cannot be "held until released", because the holder might crash and never release it. So distributed locks are leases: the lock is granted for a limited time (say 30 seconds). The holder must renew it before it expires, or lose it. If the holder dies, the lease expires and another node can take over.

Leases solve the crash problem and create a new one. From article 18: a process can pause (a long garbage-collection pause, a VM pause, heavy swapping) without knowing it. The timeline:

t=0   Client A acquires the lease (30 s)t=5   A starts a long garbage-collection pauset=30  The lease expires. The lock service gives it to client B.t=31  B writes to the shared resourcet=45  A wakes up, still believing it holds the lock, and writes too      -> two writers; A's stale write may overwrite B's

Checking "do I still hold the lock?" right before writing does not fix it, because the pause can happen between the check and the write. And the lease's timing assumes clocks on the lock service and the client run at similar rates.

The fix moves the check to the resource being protected. Each time the lock service grants the lock, it returns a fencing token: a number that increases with every grant. The client sends the token with every write. The resource (database, storage service) remembers the highest token it has accepted and rejects any write carrying a lower one. Kleppmann's article describes exactly this mechanism, and the Redis documentation on distributed locks recommends implementing fencing tokens when correctness matters (Distributed Locks with Redis).

This Java program scripts the timeline above twice: once with storage that accepts any write, once with storage that checks fencing tokens.

// File: FencingDemo.javapublic class FencingDemo {    static long nextToken = 33;          // the lock service hands out increasing tokens    static String storedValue;    static long highestTokenSeen;    static long acquireLock(String client) {        long token = nextToken++;        System.out.println("  " + client + " acquires the lock, token " + token);        return token;    }    // Storage without fencing accepts any write.    static void writeUnfenced(String client, String value) {        storedValue = value;        System.out.println("  " + client + " writes: value = " + storedValue);    }    // Storage with fencing rejects a token older than one it has already accepted.    static void writeFenced(String client, long token, String value) {        if (token < highestTokenSeen) {            System.out.println("  " + client + " write REJECTED: token " + token + " < " + highestTokenSeen);            return;        }        highestTokenSeen = token;        storedValue = value;        System.out.println("  " + client + " writes with token " + token + ": value = " + storedValue);    }    public static void main(String[] args) {        System.out.println("Without fencing:");        storedValue = "initial";        acquireLock("A");        System.out.println("  A pauses for 40 s (garbage collection); its 30 s lease expires");        acquireLock("B");        writeUnfenced("B", "B's update");        writeUnfenced("A", "A's stale update");        System.out.println("  final value: " + storedValue);        System.out.println("With fencing:");        storedValue = "initial";        highestTokenSeen = 0;        long tokenA = acquireLock("A");        System.out.println("  A pauses for 40 s (garbage collection); its 30 s lease expires");        long tokenB = acquireLock("B");        writeFenced("B", tokenB, "B's update");        writeFenced("A", tokenA, "A's stale update");        System.out.println("  final value: " + storedValue);    }}

Without fencing:  A acquires the lock, token 33  A pauses for 40 s (garbage collection); its 30 s lease expires  B acquires the lock, token 34  B writes: value = B's update  A writes: value = A's stale update  final value: A's stale updateWith fencing:  A acquires the lock, token 35  A pauses for 40 s (garbage collection); its 30 s lease expires  B acquires the lock, token 36  B writes with token 36: value = B's update  A write REJECTED: token 35 < 36  final value: B's update

Fencing requires the protected resource to cooperate. A database can do it with a conditional write (WHERE fencing_token < :token, much like optimistic concurrency in article 14). If the resource cannot check tokens (for example an external email provider), you cannot make the lock safe for correctness, and you should fall back to idempotency on that resource instead.

If the work already touches a relational database, it is often the simplest coordination service you have:

• Row locks (SELECT ... FOR UPDATE) for short critical sections inside one transaction (article 14).
• Advisory locks: PostgreSQL offers application-defined locks identified by a number, held per session or per transaction (PostgreSQL: Explicit Locking). Useful for "only one migration runner at a time".
• A lease table for longer leadership, with a version column that doubles as a fencing token:

CREATE TABLE leases (    name        VARCHAR(100) PRIMARY KEY,   -- for example 'weekly-digest-scheduler'    owner       VARCHAR(100),    expires_at  TIMESTAMPTZ NOT NULL,    token       BIGINT NOT NULL DEFAULT 0);-- Acquire or renew: succeeds if the lease is free, expired, or already mine.UPDATE leasesSET owner = :me, expires_at = now() + interval '30 seconds', token = token + 1WHERE name = 'weekly-digest-scheduler'  AND (expires_at < now() OR owner = :me)RETURNING token;

Using the database's clock (now()) for expiry avoids depending on each client's clock. The returned token is the fencing token for writes.

The basic single-instance recipe in the Redis documentation is SET resource_name my_random_value NX PX 30000: set the key only if it does not exist (NX), with a 30-second expiry (PX), and a random value unique to this client. Release it only if the value still matches, using an atomic compare-and-delete (newer Redis versions provide a command for this; older ones use a short Lua script), so a client never deletes a lock that has since passed to someone else (Distributed Locks with Redis).

Its limits, also described on that page:

• With a single Redis primary and asynchronous replication, a failover can lose a just-acquired lock: client A locks on the primary, the primary crashes before replicating, the replica is promoted, and client B acquires the same lock.
• The documentation describes Redlock, which acquires the lock on a majority of several independent Redis instances, and links to Kleppmann's critique and the Redis author's response. Its own disclaimer advises implementing fencing tokens for correctness and notes that Redis does not use a monotonic clock for key expiry.

A balanced conclusion: a Redis lock is a good efficiency lock. For correctness, use fencing with a resource that checks tokens, or a coordination service built on consensus.

Many systems need exactly one node to act as leader: the scheduler that enqueues jobs, the primary database, the node that assigns partitions. Leader election chooses one node among several and replaces it when it fails. Microsoft's Leader Election pattern describes the idea: elect one instance to coordinate others, so tasks are not duplicated and conflicts are avoided (Leader Election pattern). The Amazon Builders' Library discusses how leader election is used inside AWS systems and its pitfalls, including the need for leaders to stop acting promptly when they lose leadership (Leader election in distributed systems).

The usual implementation is a lease on a well-known name in a strongly consistent store:

1. Every candidate tries to acquire the lease (leader/weekly-digest).
2. The winner is leader; it renews the lease regularly (for example every 10 seconds for a 30-second lease).
3. Others watch the lease and try again when it expires.
4. The leader stops leader work if it fails to renew in time, and uses the lease's fencing token on every protected write.

Important properties:

• At most one effective leader is guaranteed only with fencing; without it, a paused old leader can still act briefly.
• Failover time is roughly the lease duration plus detection. Shorter leases mean faster failover and more renewal traffic and false failovers.
• The leader is a bottleneck if all work goes through it. Prefer leaders that coordinate (assign work, hold a schedule) over leaders that do all the work.

How does a "strongly consistent store" agree on who holds a lease, even when some of its own nodes fail? With a consensus algorithm: a protocol by which a group of nodes agrees on a value (or an ordered log of values) despite failures, as long as a majority of them are working and can communicate.

Raft was designed to be understandable, and its authors' site collects the paper and visualisations (The Raft Consensus Algorithm). At interview depth:

1. Terms and elections. Time is divided into numbered terms. If a follower hears nothing from a leader for a randomised timeout, it becomes a candidate, increments the term and asks the others for votes. A node votes for at most one candidate per term, and a candidate needs a majority of votes to become leader. At most one leader can win in a term.
2. Log replication. Clients send commands to the leader, which appends them to its log and replicates them to followers. An entry is committed once a majority has stored it; committed entries are applied in order by every node.
3. Safety. A candidate cannot win an election unless its log is at least as up to date as a majority's, so committed entries are never lost when leadership changes.

Paxos is the older, more famous family of consensus algorithms; many systems use variants of it or Raft. You do not need to implement either. You need to know what they give you and what they cost.

A majority of 2f + 1 nodes is f + 1. Any two majorities overlap in at least one node, so two partitions can never both elect a leader or both commit conflicting entries. That is what prevents split brain (article 15).

Even-sized clusters add cost without adding tolerance, which is why consensus clusters are usually 3 or 5 nodes. Every committed write needs a round trip to a majority, so spreading a consensus cluster across distant regions makes writes slow.

You rarely run consensus yourself. You use a service built on it:

• Apache ZooKeeper, described in its documentation as a centralised service for maintaining configuration information, naming and distributed synchronisation, offering features such as ephemeral nodes that disappear when a client's session ends (ZooKeeper overview).
• etcd, a strongly consistent, distributed key-value store built on Raft, used for configuration, service discovery and coordination, and known as Kubernetes' backing store (Why etcd).

Both provide leases, watches (be notified when a key changes) and revision numbers that can serve as fencing tokens. Keep them for small, critical metadata: who is leader, which node owns which partition, configuration. They are not general-purpose databases.

Bad: a cron entry on every server.

Better, with a lock: every server tries to acquire a lease row in PostgreSQL at 06:00 on Sunday; the winner sends all 2 million emails. Problems: one server does all the work (slow), and if it pauses or dies half-way, the next lease holder starts over and resends.

Best, with no lock in the hot path:

1. A single scheduler (elected with a lease, or a managed scheduler service) enqueues one send-digest job per user (or per batch of users) into a queue, keyed by (user_id, week).
2. Many workers process jobs in parallel (competing consumers).
3. Each worker records digest_sent (user_id, week) with a unique constraint in the same transaction as queuing the email for sending, so a redelivered job does nothing.
4. If the scheduler fails half-way, the new leader re-enqueues the week's jobs; the unique constraint makes duplicates harmless.

The "only one" requirement is now enforced per user by the database, and leadership is needed only for the cheap act of scheduling. Article 37 designs a general job scheduler on these ideas.

• Reaching for a distributed lock where a unique constraint would do.
• Using a lock for correctness without fencing.
• Releasing a lock with a plain delete, removing someone else's lock.
• Locks without expiry, which stay held forever after a crash.
• Running consensus clusters with an even number of nodes, or stretched across distant regions without considering latency.
• Building your own election protocol on heartbeats and timeouts.

• When "only one node should do X" comes up, first propose a lock-free design.
• If a lock is truly needed, call it a lease, give its duration and renewal, and add fencing tokens if correctness depends on it.
• For leader election, name a consensus-backed store (etcd, ZooKeeper) or the database, and say the leader stops when it cannot renew.
• If asked how consensus works, give the Raft summary: terms, majority votes, log replication, commit on majority.

acquire():  SET job-lock "locked"           (no NX, no expiry)release():  DEL job-lock

acquire():  token = random unique value            SET job-lock token NX PX 30000        -> succeeds only if free; expires in 30 s            if not set: someone else holds it; retry later with jitterwork:       renew before expiry if the work is long (extend only if the value is still mine)release():  atomically delete job-lock only if its value equals token

• Next: Fault Tolerance: Timeouts, Circuit Breakers, Bulkheads, Backpressure and Graceful Degradation starts Module 7.
• Later: the job scheduler (article 37) and ride matching (article 41) apply leases and leader election in full designs.