There is a small category of infrastructure that is boring on the surface and terrifying underneath: the thing that every other service leans on to agree with itself. A distributed lock service is that thing. It answers questions that sound trivial and are not: who is the current leader of this shard, which one of these three replicas is allowed to write right now, has the config changed, is node 4,812 still alive. Every microservice in a 10,000-node cluster calls it, often on a hot path, and if it hands back a wrong answer even once - two leaders, two lock holders, a stale config - you get split brain and data corruption that surfaces days later.

So the bar is brutal. This system has to be more correct than anything that depends on it, which is nearly everything, and it has to stay correct across network partitions, node crashes, GC pauses, and clock skew. That is a very different design posture from “make it fast.” Here, safety is the product and performance is the constraint you fight to claw back afterward.

Let me do this the way I would in an interview: state the requirements, size it, sketch the architecture, then take the two genuinely hard parts - consensus and the locking semantics - from the naive version, break them, and evolve.

Functional Requirements (FR)

In scope:

  • Distributed locks. A client can acquire a named lock, hold it, and release it. At most one client holds a given lock at any real instant. This is mutual exclusion across machines.
  • Leader election. A set of processes can elect exactly one leader for a role (a shard master, a cron runner, a primary replica) and detect when that leader dies so a new one takes over.
  • Configuration coordination / metadata store. A small, strongly-consistent key-value store for things like “current sharding map”, “feature flag X”, “list of live brokers”. Clients can read keys and watch them for changes.
  • Ephemeral membership. Keys tied to a client session that vanish automatically when the client’s session dies (crash, partition). This is what makes both locks and leader election self-healing.
  • Watches / notifications. A client registers interest in a key or prefix and is told when it changes, so it does not poll.
  • Linearizable reads and writes. A read sees the latest committed write. No stale reads on the paths that matter.

Explicitly out of scope (say this so you control scope):

  • Bulk data storage. This is not a database. Values are small (bytes to a few KB), the total dataset fits in memory. If you are storing blobs or millions of large rows, you are using the wrong tool.
  • High write throughput per key. Every write goes through consensus. This is thousands to low tens of thousands of writes/sec for the whole cluster, not millions. It is a coordination layer, not a data plane.
  • Cross-datacenter strong consistency as the default. A single consensus group spanning continents pays a WAN round trip on every write. We will run a group per region and discuss the WAN case, but global linearizability is a deliberate, expensive opt-in.
  • General pub/sub / message queue. Watches are edge-triggered change notifications, not a durable event bus. Do not build a queue on top of it.

The single decision that drives everything: correctness comes from a consensus protocol, not from a clever lock table. Locks, leader election, and config are all thin veneers over one hard primitive - a replicated, linearizable log agreed on by a quorum. Get that primitive right and the features fall out. Get it wrong and no amount of API polish saves you.

Non-Functional Requirements (NFR)

  • Consistency: linearizable. This is the whole point. A successful write is visible to all subsequent reads; there is a single, total order of operations. We will accept lower throughput to get this. Anything weaker and locks stop being locks.
  • Availability: quorum-available. The service is up as long as a majority of the consensus nodes are reachable. With 5 nodes it tolerates 2 failures. It is deliberately CP in CAP terms - during a partition, the minority side stops serving writes rather than risk divergence. We would rather be unavailable to a minority than wrong.
  • Scale: a 10,000-node client fleet. Hundreds of thousands of open sessions, each heartbeating. Read-heavy: maybe 90% reads (config lookups, lock status) to 10% writes (acquire/release/elect). Target tens of thousands of writes/sec and hundreds of thousands of reads/sec cluster-wide.
  • Latency: writes bounded by one consensus round trip, target p99 under 10ms within a region. Reads served locally where safe, sub-millisecond; linearizable reads pay one round trip.
  • Durability: a committed write survives the loss of any minority of nodes. The log is fsynced to disk on a quorum before an operation is acknowledged. No acknowledged write is ever lost.
  • Fault model: crash-stop and network partitions, not Byzantine. Nodes may crash, pause (GC, VM migration), or be partitioned; they do not lie. This lets us use Raft/Paxos-family consensus rather than expensive BFT.

Back-of-the-Envelope Estimation (BoE)

Let me ground the numbers, because they justify why this is an in-memory, quorum-replicated design and not a database.

Client fleet and sessions:

Client nodes                 10,000
Sessions per node            ~1-3 (usually one shared client per host)
Open sessions (round)        ~20,000
Session heartbeat interval   ~3 sec (session timeout ~10s = 3x heartbeat)
Heartbeat traffic            20,000 / 3  ≈ 6,700 heartbeats/sec

Heartbeats alone are ~6.7K ops/sec of keep-alive traffic. These are cheap and can be handled by followers/observers without hitting the consensus log, but they are the constant background load, and they matter: 20,000 sessions each needing liveness tracking is the real scale pressure, not the lock count.

Operation mix (writes go through consensus, reads may not):

Assume each client host does ~5 coordination ops/sec at peak
Total ops               10,000 * 5      = 50,000 ops/sec
Read : write split      ~90 : 10
Reads                   ~45,000 reads/sec
Writes (consensus)      ~5,000 writes/sec

Why writes are the ceiling. Every write is a Raft/ZAB round: leader appends to its log, replicates to followers, waits for a quorum to fsync, then commits. Even at 5,000 writes/sec, each write touches a majority of nodes and a disk. A single consensus group tops out in the low tens of thousands of writes/sec regardless of hardware, because the bottleneck is the serialized log plus fsync plus network fan-out, not CPU. That number - “one consensus group does ~10-20K writes/sec, full stop” - is the most important figure in the whole design and drives the sharding decision later.

Data size (why it fits in memory):

Keys (locks + config + membership)   ~1,000,000
Avg key + value + metadata           ~256 bytes
Total dataset                        1,000,000 * 256B ≈ 256 MB

A quarter of a gigabyte. The entire keyspace lives in RAM on every node, with the log and periodic snapshots on disk for durability. There is no disk-seek read path; reads are memory lookups. This is why it is a coordination service, not a database: the whole point is that the dataset is tiny and the agreement is expensive.

Storage / durability sizing:

Write rate            5,000 writes/sec
Log entry size        ~256 bytes
Raw log growth        5,000 * 256B      ≈ 1.3 MB/sec  ≈ 110 GB/day

The log grows fast, but it is compacted: periodically the state is snapshotted and the log prefix before the snapshot is discarded. Steady-state on-disk footprint is a recent snapshot (~256 MB) plus a bounded log tail (minutes of writes), not 110 GB/day forever. Snapshotting is a first-class operation, not an afterthought.

Watch fan-out:

Popular key (e.g. sharding map)    watched by many clients
Watchers on a hot key              up to ~10,000 (whole fleet)
On change                          10,000 notifications to push

A single config change on a globally-watched key means a fan-out to potentially the entire fleet. Watch delivery, not lock throughput, is often the surprising scaling problem, and we handle it explicitly in the deep dive.

High-Level Design (HLD)

The core is a small ensemble of servers (odd number, typically 3 or 5) running a consensus protocol over a replicated log. One is the leader (handles all writes); the others are followers (replicate the log, can serve reads). Clients connect over long-lived sessions. A thin client library hides leader discovery, session heartbeats, watch re-registration, and retries.

        10,000 client hosts, each with the coordination client library
        (session + heartbeat + local watch cache)
                              |
        ┌─────────────────────┼─────────────────────┐
        │                     │                     │
        │   reads may hit any node; writes           │
        │   are forwarded to the leader              │
        ▼                     ▼                     ▼
 ┌────────────┐        ┌────────────┐        ┌────────────┐
 │ Follower A │        │  LEADER    │        │ Follower B │
 │            │◄──────►│            │◄──────►│            │
 │ log replica│  Raft  │ append log │  Raft  │ log replica│
 │ state (RAM)│ (ZAB)  │ state (RAM)│        │ state (RAM)│
 └─────┬──────┘        └─────┬──────┘        └─────┬──────┘
       │                     │                     │
   ┌───▼───┐             ┌───▼───┐             ┌───▼───┐
   │ WAL + │             │ WAL + │             │ WAL + │
   │ snap  │             │ snap  │             │ snap  │
   │ (disk)│             │ (disk)│             │ (disk)│
   └───────┘             └───────┘             └───────┘

        A write is acknowledged only after a QUORUM (3 of 5)
        has appended it to their WAL and fsynced to disk.
        Followers C, D omitted for space; 5-node ensemble.

Write flow (acquire a lock, set a config key, elect a leader - all the same underneath):

  1. Client sends the write to whatever node it is connected to.
  2. If that node is a follower, it forwards to the current leader (or the client library already knows the leader and goes direct).
  3. The leader assigns the operation the next slot in its log, appends it, and sends AppendEntries to all followers.
  4. Each follower appends to its own write-ahead log, fsyncs to disk, and acks.
  5. When a quorum (including the leader) has persisted the entry, the leader marks it committed and applies it to the in-memory state machine.
  6. The leader replies success to the client. Followers apply the committed entry to their own state as they learn the commit index.

The magic is step 5: a majority has durably agreed on this operation and its position in the total order before the client hears “yes.” That is what makes it linearizable and crash-safe.

Read flow:

  1. For a normal read, a client can read from a follower’s in-memory state - fast, but that follower might be slightly behind the leader.
  2. For a linearizable read (the default for lock status and anything correctness-sensitive), the read is served by the leader after it confirms it is still the leader via a quorum check (a ReadIndex heartbeat), guaranteeing it has seen every committed write. One round trip, no stale answer.

Session flow (the backbone of locks and elections):

  1. On connect, the client establishes a session with a timeout (say 10s).
  2. The client library heartbeats every ~3s to keep the session alive.
  3. Ephemeral keys (locks, election candidacy) are tagged with the session ID.
  4. If heartbeats stop for longer than the timeout, the leader declares the session dead and deletes all its ephemeral keys through the consensus log - which automatically releases that client’s locks and withdraws it from elections. Self-healing without a human.

Component Deep Dive

Four hard parts, each taken naive-first: (1) consensus itself - the beating heart; (2) lock semantics and the fencing token problem, which is where most “distributed lock” designs are quietly broken; (3) leader election built on the primitives; (4) watches and their fan-out. I will break each before fixing it.

1. Consensus: agreeing on a total order of operations

Everything reduces to one problem: get a majority of nodes to agree on an ordered log of operations, and keep agreeing even as nodes crash and rejoin.

Naive: a single primary with async replicas

Pick one node as the writer. It takes all writes, applies them, and lazily copies its state to backups. Reads can go anywhere. Simple, fast, familiar.

Where it breaks:

  • Split brain on partition. The primary gets partitioned from its backups. The backups, seeing the primary “gone”, promote a new primary. Now there are two primaries, both accepting writes, both handing out the same lock to different clients. This is the exact failure a lock service exists to prevent, and the naive design walks straight into it.
  • Data loss on failover. Replication is async, so an acknowledged write on the old primary may not have reached the backup that gets promoted. The write vanishes. For a lock service, a “vanished” release or acquire is corruption.
  • No agreement on order. Two writes that race on different nodes have no defined order. There is no single truth.

Async primary-backup is fine for a cache. For a lock service it is disqualifying, because it can produce two leaders and lose acknowledged writes - the two cardinal sins.

Better: quorum consensus (Raft / Multi-Paxos / ZAB)

Replace “one primary tells backups” with “a majority must agree before anything is committed.” This is the Raft/ZAB/Paxos family. The essential rules:

  • Elected leader with a term/epoch. Nodes elect a leader for a monotonically increasing term number. Only the leader proposes log entries. If a leader is partitioned, the majority side elects a new leader with a higher term.
  • Majority to commit. An entry is committed only when a quorum has persisted it. With 5 nodes, quorum is 3; you tolerate 2 failures.
  • Terms kill split brain. The old, partitioned leader still thinks it is leader, but it cannot get a quorum to accept its writes (it is in the minority), so it can commit nothing. When it rejoins and sees a higher term, it steps down. At most one leader can ever commit in a given term, and terms only increase. That is the mathematical guarantee against two effective leaders.
5-node ensemble, network splits into {A,B} and {C,D,E}:

 {A,B}  (minority)          {C,D,E}  (majority)
 old leader A               elect new leader C, term = 8
 tries to commit W ...      commits writes freely, has quorum (3/5)
 gets acks from only B      A's proposals: max 2 acks < 3 = NEVER commit
 -> A commits NOTHING       -> single source of truth continues on majority

 On heal: A sees term 8 > its term 7, steps down, catches up from log.
 No split brain. No lost committed write.

Why majority is the magic number. Any two majorities of the same set must overlap in at least one node. That overlapping node carries the truth forward across leader changes - it will have the latest committed entries and will reject a would-be leader that is missing them. This quorum-intersection property is the entire reason consensus works, and it is why the ensemble is an odd number: 5 nodes tolerate 2 failures, but so does 6, so 6 just costs you an extra machine for no extra fault tolerance.

Durability detail that people skip in interviews: the fsync in step 4 is not optional. If a follower acks before the entry hits stable storage and then crashes and restarts, it can “forget” an entry it acked, breaking the quorum-intersection argument. Consensus safety assumes acknowledged log entries survive a restart. That fsync is why write latency is a few milliseconds, not microseconds, and why you put the WAL on fast disks (NVMe) on a dedicated device.

ApproachTwo-leader safeLoses acked writesWrite latencyVerdict
Single async primaryNoYesVery lowDisqualified for locks
Sync primary-backup (2 nodes)No (no majority)NoLowDeadlocks on 1 failure
Raft / ZAB quorum (3 or 5)YesNo1 round trip + fsyncThe answer
BFT consensus (PBFT)YesNoHigher (more rounds)Overkill (no Byzantine faults)

We pick Raft (or ZAB, ZooKeeper’s protocol - same guarantees). It is understandable, has clear leader election and log-matching rules, and is exactly the crash-fault-tolerant, non-Byzantine model we specified.

2. Lock semantics: why “acquire a key” is not a correct lock

This is the part that is subtly wrong in most designs, including many production ones built on Redis. Consensus gives us a correct store. It does not by itself give us a correct lock, because of what happens on the client side after acquisition.

Naive: lock = “create a key if it does not exist”

To lock resource R, atomically create key /locks/R with your client ID. If it exists, you wait or fail. To unlock, delete the key. Because the store is linearizable, only one client can create the key - so only one client holds the lock. Looks airtight.

Where it breaks - the fundamental problem:

  • A held lock does not stay released when the holder dies. If the holder crashes without deleting the key, the lock is held forever. Fix: make it ephemeral, tied to the holder’s session, so it auto-releases when the session times out. Good - but this introduces the real bug.
  • The fatal race: process pause vs. session timeout. Client A acquires the lock. Then A hits a long stop-the-world GC pause (or a VM migration, or gets scheduled off CPU) that exceeds the session timeout. The lock service, seeing no heartbeat, declares A’s session dead and releases the lock. Client B acquires it and starts writing to R. Then A wakes up from its pause, still believing it holds the lock, and also writes to R. Two clients act as lock holders at the same real instant. The store did nothing wrong - the lock genuinely moved from A to B - but A did not know it lost it.

This is the canonical distributed locking failure. No amount of consensus fixes it, because the gap is between the store’s view and the client’s stale belief. A lock alone cannot protect a resource against a paused holder.

timeline:
 A: acquire lock  ──► [........ long GC pause ........] ──► write R (STALE!)
 service:              session times out, lock released
 B:                                    acquire lock ──► write R
                                                        ^ both A and B write R

The fix: fencing tokens

The lock service must hand out a monotonically increasing fencing token with every successful acquisition. This token is simply the log index / version at which the lock was granted - and because it comes from the consensus log, it strictly increases and can never repeat.

The rule: every write to the protected resource must include the fencing token, and the resource (or its storage layer) must reject any write carrying a token lower than the highest it has already seen.

 A acquires lock -> token = 33
 A pauses ...
 service releases A's lock (session timeout)
 B acquires lock -> token = 34   (log index moved on)

 B writes R with token 34   -> storage records "last token = 34", ACCEPTS
 A wakes, writes R with token 33 -> storage sees 33 < 34 -> REJECTS

 A's stale write is fenced off. Correctness restored.

The token turns “I think I hold the lock” into “I can prove I held a lock at least as recent as anyone else’s write.” The protected resource becomes the final arbiter. This is the single most important idea in the whole design and the thing that separates a real lock service from a SETNX that looks like one. The lock service’s job is not just mutual exclusion; it is issuing verifiable, ordered proof of exclusion that downstream storage can enforce.

Practical shape of a correct lock in this system (ZooKeeper-style, avoids the herd):

1. create EPHEMERAL SEQUENTIAL node under /locks/R/  ->  /locks/R/lock-000000034
   (sequence number = your position in line; also serves as fencing token)
2. list children of /locks/R/, sort by sequence
3. if you are the LOWEST sequence -> you hold the lock. Use token 34 on all writes.
4. else -> WATCH only the node immediately ahead of you, then wait.
5. when that one node disappears (release or session death), re-check step 2.
6. release = delete your node (or just let your session die).

Each waiter watches exactly one predecessor, so a release wakes exactly one client, not all of them (the “herd effect” fix). Session-death cleanup makes it self-healing. The sequence number is the fencing token.

3. Leader election: a lock with a name and a watch

Leader election is not a separate mechanism; it is the locking primitive pointed at a role instead of a resource.

Naive: everyone reads “is there a leader?” and one writes it

Each candidate reads /leader. If empty, it writes its own ID. Whoever wrote wins.

Where it breaks: the read-then-write is not atomic across candidates unless it goes through the store as a single conditional operation, and worse, when the leader dies there is no automatic detection - /leader still holds the dead node’s ID until someone notices. You would poll, which is slow and racy, and you would still need the fencing-token discipline so a deposed-but-paused leader cannot keep acting.

The correct pattern: ephemeral sequential + watch-predecessor

Identical machinery to the lock:

1. Each candidate creates an EPHEMERAL SEQUENTIAL node under /election/role-X/
2. The candidate with the LOWEST sequence number is the leader.
3. Every other candidate watches the node just below it.
4. When the leader's session dies, its ephemeral node vanishes,
   the next-lowest candidate's watch fires, and it becomes leader.
5. The new leader's fencing token is its (higher) sequence number.
  • Automatic failover. Leader dies, session expires, its node disappears, the successor is notified in one hop. No polling, no human.
  • Exactly one leader by construction. Sequence numbers are unique and totally ordered by the consensus log; “lowest wins” is unambiguous.
  • Fencing carries over. The new leader’s sequence number is strictly higher, so any action the new leader takes on shared state supersedes the old leader’s. A deposed leader that resumes after a pause is fenced exactly like the stale lock holder.

The crucial interview point: a newly elected leader must not assume it is the only leader that thinks it is leader. The old one may be paused and about to wake. Leadership grants the right to try, and the fencing token makes its attempts either win (if still current) or safely lose (if superseded). Election plus fencing, never election alone.

4. Watches and fan-out: telling 10,000 clients that one key changed

Clients must learn about changes (a new leader, a config flip, a lock release) without polling. That is a watch.

Naive: clients poll the key every second

10,000 clients each reading a hot config key once a second is 10,000 reads/sec on one key, most returning “no change.” It wastes the read path and adds seconds of latency to change propagation. It also does not scale with fleet size - double the fleet, double the useless load.

Better: server-side watches (edge-triggered)

A client registers a watch on a key or prefix. The server remembers “session S cares about key K.” On the next change to K, the server sends S a one-shot notification, and S must re-register if it wants further updates. One-shot keeps server memory bounded and avoids duplicate storms.

Where the naive watch design breaks - the thundering herd:

  • A change to a globally-watched key (the sharding map watched by all 10,000 clients) means the server must push 10,000 notifications at once, and then absorb 10,000 near-simultaneous re-read + re-watch requests. That burst can overwhelm the leader.

Fixes:

  • Serve watches and reads from followers/observers. Watch registration and delivery do not need the leader. Followers apply the committed log and can fire watches locally. Reads for the changed value can be served by followers too. This spreads the fan-out across all nodes instead of pinning it to the leader.
  • Add non-voting observers/learners. For a huge read/watch fan-out, add nodes that replicate the log but do not vote in consensus (ZooKeeper observers, etcd learners). They absorb read and watch load without slowing down writes (writes still only need the voting quorum). This is how you scale reads to hundreds of thousands/sec while keeping the write quorum small (5).
  • Client-side jitter on re-registration. Clients wait a small random delay before re-reading and re-watching after a notification, smearing the herd over time instead of a synchronized stampede.
  • Coalesce rapid changes. If a key changes 5 times in a millisecond, a one-shot watch naturally collapses that into “it changed, go look”, and the client reads the latest value once. Edge-triggered semantics are a feature here.

API Design & Data Schema

API

The data model is a hierarchical namespace of small nodes (like a filesystem), each optionally ephemeral and/or sequential. Locks, elections, and config are all just conventions over this.

# --- Sessions ---
POST /session/create
  body: { timeout_ms: 10000 }
  200:  { session_id: "0x3f2a...", timeout_ms: 10000 }

POST /session/heartbeat
  body: { session_id: "0x3f2a..." }
  200:  { alive: true }
  410:  session expired -> client must reconnect and re-establish state

# --- Nodes (the primitive under everything) ---
PUT /kv/{path}
  params: ephemeral=true|false, sequential=true|false,
          session_id=..., prev_version=<N|any|none>   # compare-and-set
  body:   { value: "<=1KB payload" }
  200:    { path: "/locks/R/lock-000000034", version: 34, created_index: 71183 }
  409:    precondition failed (prev_version mismatch, or exists when none required)

GET /kv/{path}?linearizable=true
  200:  { value: "...", version: 34, ephemeral: true, owner_session: "0x3f2a..." }

GET /kv/{prefix}/children
  200:  { children: ["lock-000000034", "lock-000000035"] }

DELETE /kv/{path}?prev_version=34
  200:  { deleted: true }

# --- Watches ---
POST /watch
  body: { path: "/config/sharding-map", recursive: false, session_id: "..." }
  200:  { watch_id: "w-991", current_version: 12 }
  # server later pushes over the session stream:
  #   { watch_id: "w-991", event: "CHANGED"|"DELETED", path: "...", version: 13 }

# --- Lock helper (thin wrapper over ephemeral+sequential above) ---
POST /lock/acquire
  body: { name: "R", session_id: "...", wait_ms: 5000 }
  200:  { held: true, fencing_token: 34 }    # USE THIS TOKEN ON EVERY PROTECTED WRITE
  409:  { held: false, position: 3 }         # queued behind 3 holders/waiters

POST /lock/release
  body: { name: "R", session_id: "...", fencing_token: 34 }
  200:  { released: true }

# --- Leader election helper ---
POST /election/join
  body: { role: "shard-7-master", session_id: "..." }
  200:  { is_leader: true|false, fencing_token: 61, watching: "role-000000060" }

The fencing_token field is the load-bearing part of the API. Any protected downstream write path must accept it and reject stale ones.

Data / storage schema

Two layers: the replicated state machine (in memory on every node, the source of truth) and the on-disk consensus artifacts (WAL + snapshots for durability).

In-memory state machine (the keyspace), one node record:

node {
  path            string     # "/locks/R/lock-000000034", primary key, unique
  value           bytes      # <= 1KB
  version         int64      # bumped on every write; the fencing token source
  created_index   int64      # consensus log index at creation (global order)
  modified_index  int64      # consensus log index at last modify
  ephemeral       bool       # deleted when owner_session dies
  owner_session   int64      # null unless ephemeral
  seq_counter     int64      # per-parent, for SEQUENTIAL children
}
# Index: primary hash/tree on `path`; secondary index on `owner_session`
#        so "delete all ephemerals for a dead session" is O(k), not O(all keys).

Session table (in-memory, replicated):

session {
  session_id      int64      # primary key
  timeout_ms      int32
  last_seen_index int64      # log index of last heartbeat processed
  owned_paths     set<path>  # ephemeral nodes to reap on expiry
}

On-disk, per node:

WAL (write-ahead log):  append-only sequence of committed log entries
   entry { index, term, op(PUT/DELETE/EXPIRE/...), path, value, session_id }
   fsynced before the entry is acked. This IS the durability guarantee.

Snapshot:  periodic full dump of the in-memory state machine + last_included_index/term
   taken every N entries; lets the WAL prefix be truncated (log compaction).

Store choice: a custom in-memory replicated state machine over a Raft/ZAB log, NOT SQL and NOT a general NoSQL store. Reasons:

  • We need linearizable, totally-ordered operations with quorum durability - that is precisely what a consensus log gives and what a single-writer SQL DB cannot without becoming the SPOF we are trying to eliminate.
  • We need ephemeral, session-bound keys and watches as first-class concepts. No off-the-shelf database has these; they are the defining features.
  • The dataset is tiny and hot (~256 MB, all reads from RAM). A disk-oriented database’s strengths (large data, complex queries, indexes over big tables) are irrelevant, and its weaknesses (no quorum linearizability, no ephemerality) are disqualifying.

This is exactly why ZooKeeper and etcd exist as their own category rather than being “just use Postgres.”

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each.

1. Write throughput on the single consensus group (breaks first). All writes serialize through one leader and one log, fsynced to a quorum. That is a hard ceiling around 10-20K writes/sec no matter the hardware, because the bottleneck is the serialized, fsynced, quorum-replicated log. Fix: (a) Batch and pipeline - the leader groups many client ops into one AppendEntries and one fsync, amortizing the per-op cost; this alone multiplies throughput several times. (b) Shard by namespace - run multiple independent consensus groups, each owning a slice of the keyspace (e.g. /locks/A-M on group 1, /locks/N-Z on group 2, /config/* on group 3). Route each key to its group by consistent hashing on the path. Locks and config for unrelated resources have no reason to share a log. This is how you go past one group’s ceiling. The trade-off: you lose cross-group atomicity (a transaction spanning two groups needs a 2-phase commit or must be avoided by design), so shard on natural correctness boundaries.

2. Read/watch fan-out overwhelming the voting nodes. Hundreds of thousands of reads/sec and herd-notifications on hot keys can bury the 5 voting members. Fix: add non-voting observers/learners that replicate the log and serve reads + watches but do not participate in the write quorum. Reads scale horizontally with observers while write latency stays tied to the small voting set. Serve non-linearizable reads from the nearest observer; reserve leader ReadIndex reads for the correctness-critical paths.

3. The leader is a single point for writes. If the leader dies, writes stall until a new one is elected. Fix: this is inherent and bounded, not fatal. Leader election completes in one election timeout (typically 150ms-1s, randomized to avoid split votes). During that window writes are unavailable but nothing is lost or corrupted - a deliberate, short CP pause. Keep the ensemble at 5 (not 3) so you survive 2 failures and re-elect quickly; keep election timeouts tuned to your network RTT.

4. Hot keys / hot locks. A single wildly-popular lock or a globally-watched config key concentrates load on whichever group owns it. Fix: for hot watched config, front it with observer replicas and client-side caching keyed by version (clients hold the value and only re-fetch on a change notification). For a hot lock, that is usually a design smell in the client - a single global lock is a serialization point by definition; push the contention down by sharding the resource itself (lock per partition, not one lock for everything) so the lock traffic spreads across keys and groups.

5. Session/heartbeat storm at 10K nodes. 20,000 sessions heartbeating every 3s is constant load, and a network blip can cause mass session expiry followed by a reconnect stampede. Fix: (a) let followers/observers terminate heartbeats so keep-alives never touch the leader or the log; only actual expiry (a write) goes through consensus. (b) Tune the session timeout well above the heartbeat interval (10s timeout, 3s heartbeat = tolerate 2 missed beats) so transient jitter does not trigger spurious expiry - a spurious expiry falsely releases live locks and triggers needless failovers. (c) Add jittered reconnect backoff in the client so a partition heal does not produce a synchronized reconnect flood.

6. Log growth and snapshot pauses. The WAL grows ~110 GB/day uncompacted; naive snapshotting stalls the node while it serializes 256 MB. Fix: incremental / copy-on-write snapshots taken in the background without blocking the apply loop, then truncate the WAL prefix. A node that falls too far behind (was down during compaction) is caught up by shipping it a snapshot plus the log tail rather than replaying from genesis.

7. Cross-datacenter / WAN writes. A consensus group spanning regions pays inter-region RTT (tens to hundreds of ms) on every write, because the quorum spans the WAN. Fix: one consensus group per region for region-local coordination (the common case). For the rare truly-global lock, use a single global group placed with quorum in the regions that matter and accept the WAN write latency, or layer a hierarchy (regional groups coordinated by a global group for the few keys that need it). Do not make every write pay the WAN tax for the 1% that need global scope.

8. Clock dependence. Note what we did not do: locks here are not time-leased by wall clock (unlike a naive Redis lease). Correctness comes from sessions + the consensus log + fencing tokens, not from synchronized clocks. Session timeouts use elapsed time only to detect probable death; the fencing token, not a clock, guarantees safety if that detection is wrong. This is deliberate - clock-based lock expiry is exactly how paused-holder bugs get in.

Wrap-Up

The trade-offs that define this design:

  • CP, not AP. We chose consistency and partition-tolerance over availability. During a partition the minority side refuses writes. For a lock service this is correct: an unavailable lock is a stall, a wrong lock is corruption. Availability is bounded by needing a quorum, and that is a feature.
  • Consensus is the product; locks/elections/config are veneers. One hard primitive - a linearizable, quorum-replicated log via Raft/ZAB - and everything else (ephemeral nodes, sequential IDs, watches) composes on top. Build the primitive right and the features are thin.
  • Fencing tokens over trust. The lock service does not just grant exclusion, it grants ordered, verifiable proof of exclusion that downstream storage enforces. This is what makes it correct despite GC pauses, VM migrations, and partitions that a lock-alone design cannot survive. If you remember one thing from this design, remember that a distributed lock without a fencing token is not safe.
  • Small voting quorum, wide observer fan-out. Keep the write quorum at 5 for fast, cheap consensus; scale reads and watches horizontally with non-voting observers. Writes stay bounded and safe; reads stay cheap and plentiful.
  • Shard the log to break the throughput ceiling. Multiple independent consensus groups partitioned by namespace push past one group’s ~10-20K writes/sec limit, at the cost of cross-group atomicity - so you shard on natural correctness boundaries.

One-line summary: a small odd-sized ensemble running Raft/ZAB over a fsynced, quorum-replicated in-memory log, exposing session-bound ephemeral sequential nodes and watches, from which locks and leader election are built - correct across partitions and paused clients because every acquisition carries a monotonic fencing token that downstream storage enforces, scaled by sharding the log across consensus groups and fanning reads out to non-voting observers.