System Design: [Coordination service name]

Reference design for a distributed lock and coordination service: a small, strongly consistent, replicated store that other systems use for leader election, locks with leases, fencing tokens, configuration, membership and change notification. It covers consensus with Raft (or Multi-Paxos), sessions and keep-alives, failover grace periods, watches, read scalability and the ways teams misuse such a service. Concepts draw on the Chubby lock service (Burrows, OSDI 2006), the Raft paper (Ongaro and Ousterhout, USENIX ATC 2014), ZooKeeper (Hunt et al., USENIX ATC 2010) and the etcd v3.6 API documentation; the text and the worked design are original. It follows the structure of the System Design Document template. Replace every [bracketed] value and every row marked Example.

Document control

Approvals

Revision history

Contents

1. Overview
2. Requirements
3. Capacity estimation
4. API design
5. Data model
6. High-level design
7. Detailed design
8. Scalability and performance
9. Reliability, security and operations
10. Trade-offs and alternatives
11. Delivery plan and risks
12. Appendices

1. Overview

1.1 Problem statement

[Example: Teams implement leader election with database row locks, cache-based locks with TTLs, and hand-written heartbeats. Last quarter two scheduler instances both believed they were leader after a long garbage collection pause and ran the same billing job twice. Configuration is pushed by redeploying, and service discovery lags by minutes.]

1.2 Proposed solution

[Example: A coordination service runs as cells of five replicas across three zones, replicated with Raft. It exposes a small hierarchical key-value store with revisions, sessions held alive by keep-alives, locks and elections built on sessions, and watches that stream changes from a given revision. Every lock acquisition returns a fencing token (the lock key's revision) that protected resources must check. A stateless proxy tier absorbs keep-alives and fans out watch notifications. Values are capped at Example 64 KB and the service is explicitly not a general data store.]

1.3 Decisions requested from reviewers

1.4 45-minute interview path

2. Requirements

2.1 Core functional requirements

2.2 Non-functional requirements

2.3 Below the line (out of scope)

• [Example: Storing application data, queues or large blobs (see system-design/design-distributed-kv-store and system-design/design-message-queue)]
• [Example: Fine-grained locks held for milliseconds per request]
• [Example: Cross-region linearizability within one cell; each region runs its own cell, with one global cell for rare global decisions]
• [Example: Rate limiting and counters at request rates]
• [Item deferred, with the release that will take it]

3. Capacity estimation

3.1 Assumptions

3.2 Keep-alives

Keep-alives refresh leases, and lease refreshes are handled by the leader. Consequence: keep-alives alone are five times the write rate, so a proxy tier terminates client sessions and refreshes them upstream in batches; the cell sees one stream per proxy instead of one per client.

3.3 Writes

Every write is appended to the replicated log and fsynced by a quorum. One fsync per write would cap throughput near $1 / f = 500$ writes/s per replica; group commit batches entries:

Consequence: batching makes 2 000 writes/s easy, but all writes go through one leader, so adding replicas does not add write capacity (it adds latency). Write growth is handled by more cells partitioned by tenant, not bigger cells.

3.4 Reads and watches

Reads at 50 000/s are 25 times the write rate. Consequence: reads must be served by followers and proxies, using read-index confirmation for linearizable reads and local state for serializable ones (7.1, 7.4).

Watches: $S \times w = 500\,000$ watches; at about 200 B each that is about 100 MB of memory, which is small. Fan-out is the risk: a change to a key watched by all 100 000 sessions creates 100 000 notifications, which take one server $100\,000 / 20\,000 = 5$ s to send. Consequence: watches register at proxies, so the cell sends one notification per proxy and proxies fan out in parallel; clients add jitter before re-reading so the change does not create a read herd.

3.5 Data size and snapshots

$K \times v = 10^{6} \times 1$ KB $= 1$ GB, kept in memory with the log on SSD. Consequence: cap the keyspace at Example 2 GB; a new or lagging replica installs a snapshot of that size in about 20 s at 100 MB/s, which keeps replacement of a failed replica quick. Enforce value size limits and per-tenant quotas to hold the cap.

3.6 Cell size

A cell of $2F + 1$ voting replicas tolerates $F$ failures. Five replicas across three zones (2 + 2 + 1) survive the loss of any one zone or of any two replicas. Three voters in three zones also survive a zone loss, but only one replica failure. Consequence: five voters, not three, so that one replica can be down for a planned upgrade while another fails unexpectedly.

3.7 Summary

4. API design

4.1 Operations

4.2 Lock acquisition as a transaction

A lock is a key created only if it does not exist, attached to the caller's session so it disappears if the session ends:

Txn(
  compare: create_revision("locks/billing-run") == 0
  then:    put("locks/billing-run", owner="sched-7", session=S123)
  else:    get("locks/billing-run")
)
-> succeeded: token = header.revision (for example 482113)
-> failed:    watch "locks/billing-run" for delete, then retry

For fair queuing, each waiter instead creates locks/billing-run/{session} and the holder is the key with the lowest create_revision; each waiter watches only the key just before it, so a release wakes one waiter, not all.

4.3 Protected resource check

The resource, not the lock service, enforces exclusion. It stores the highest token it has accepted per protected object and rejects anything lower:

WriteJobResult(job_id, result, fencing_token = 482113)
  if fencing_token < last_seen_token[job_id]: reject (stale holder)
  else: last_seen_token[job_id] = fencing_token; apply

4.4 Errors

5. Data model

5.1 Replicated state

All state is one replicated state machine per cell, built from the Raft log:

The cell keeps a single 64-bit revision counter that increases with every committed change; it orders all changes and serves as the fencing token source. Watches live in proxy and server memory, not in the replicated state.

5.2 Storage choices

5.3 Access patterns

6. High-level design

7. Detailed design

7.1 Deep dive: consensus and linearizable reads

Choice: C, using a mature implementation rather than writing one. Reason: the team must operate and debug the protocol during incidents; Raft's structure (leader election, log replication, safety) is easier to reason about, and its membership-change procedure supports replacing replicas without downtime.

Linearizable reads have three options:

Choice: read index by default, leader lease off. Reason: the extended Raft paper describes confirming leadership with a majority before answering a read and notes that relying on a lease makes safety depend on timing; for a lock service, safety wins over a fraction of a millisecond. A new leader commits a no-op entry at the start of its term before serving reads, so it knows which entries are committed.

7.2 Deep dive: sessions, keep-alives and failover

Sessions tie ephemeral keys and locks to a live client. A session ends when no keep-alive arrives before its TTL; then its ephemeral keys and locks are deleted in one committed transaction, so watchers see a consistent change.

During leader failover no keep-alive is acknowledged. Two measures keep healthy sessions alive:

1. A new leader does not expire any session until it has given every session at least one full TTL from its election (session expiry is extended on leadership change).
2. The client library, on losing contact, enters a "jeopardy" state: it tells the application that its locks may be at risk, pauses work that needs them, and keeps trying other replicas for the grace period. If contact returns and the cell confirms the session is still alive, work resumes; if the session expired or the grace period ends, the application treats its locks as lost.

The application must also stop acting on a lock when its own process paused (garbage collection, VM migration). Timeouts on the client cannot detect that reliably, which is why fencing tokens (7.3) exist.

7.3 Deep dive: locks, leader election and fencing tokens

A lock gives mutual exclusion in the coordination service, not at the resource. A holder can be paused after it last checked its session, wake up after the session expired and another client took the lock, and still send a write.

Choice: D for every lock that guards a side effect, with C as an extra margin for resources that cannot check tokens yet. Reason: only the resource can reject a stale write after the fact. Teams that cannot add a token check must design the work to be idempotent instead.

Leader election uses the same recipe: each candidate creates an ephemeral key under elections/{name}/; the lowest create_revision leads and its revision is the leader epoch, used as the fencing token for everything the leader writes. Followers watch only their predecessor.

7.4 Deep dive: watches, caching and read scalability

Choice: C, served by followers and proxies, plus small proxy caches for serializable reads. Reason: read-then-watch-from-revision gives clients a gap-free view; proxies merge identical watches so the cell sees one per proxy. Keep history for Example 1 hour; a watcher that falls further behind gets compacted and must re-read.

Read scaling ladder, in order:

1. Serializable reads from any follower for discovery and config that tolerate staleness.
2. Read-index linearizable reads on followers.
3. Proxy caches kept current by watches.
4. Non-voting replicas (learners) that replicate the log and serve reads without slowing commits.
5. More cells, partitioned by tenant.

7.5 Misuse pitfalls and usage policy

7.6 How real systems do it

How real systems do it (paraphrased, not copied). Burrows describes Google's Chubby in "The Chubby Lock Service for Loosely-Coupled Distributed Systems" (OSDI 2006). Chubby is aimed at coarse-grained locking and small files, with cells of typically five replicas that use Paxos to elect a master holding a master lease. Clients keep sessions alive with keep-alives; when a client loses contact, its library signals jeopardy and waits through a grace period (45 s by default in the paper) so sessions survive master failover. Lock holders can obtain a sequencer that servers check before acting, with a lock-delay as a weaker fallback for servers that do not check. Clients cache data and the master invalidates caches before changes complete; proxies can absorb keep-alives and reads. Separately, Ongaro and Ousterhout's "In Search of an Understandable Consensus Algorithm" (USENIX ATC 2014) defines Raft, decomposing consensus into leader election, log replication and safety; its extended version covers linearizable read handling. This template reuses the ideas, not the text or figures.

8. Scalability and performance

Write latency budget (NFR-02, Example): client to proxy 0.5 ms, proxy to leader 0.5 ms, leader append and parallel replication to the fastest two followers across zones 2 ms plus fsync 2 ms, apply 0.2 ms, response path 1 ms; about 6 ms typical, with headroom for cross-zone variance up to the 25 ms p99.

9. Reliability, security and operations

9.1 Failure-mode table

9.2 Security

9.3 Observability

10. Trade-offs and alternatives

11. Delivery plan and risks

11.1 Milestones

11.2 Rollout and rollback

[Example: Each consumer migrates behind a flag that selects the old or new election mechanism; both run in observe-only mode for a week and must agree on the leader. Rollback flips the flag; the old mechanism stays deployed until the migration is complete for 30 days.]

11.3 Risks

12. Appendices

12.1 Open questions

12.2 Glossary

12.3 Related templates

• Distributed Key-Value Store System Design (system-design/design-distributed-kv-store): for data that does not belong here.
• Job Scheduler System Design (system-design/design-job-scheduler): main consumer of leader election and fencing.
• Message Queue System Design (system-design/design-message-queue): controller election and partition metadata.
• Stream Processing Engine System Design (system-design/design-stream-processing-engine): coordinator leader and checkpoint pointer.
• Distributed Cache System Design (system-design/design-distributed-cache): slot map storage.

12.4 References

Standard and licence note

Based on the publicly described designs and documentation cited in 12.4 (Chubby, Raft, ZooKeeper, etcd v3.6); no standard governs coordination services. This is a reference design to adapt, with illustrative numbers marked Example. Original text written for this collection; sources were used as inspiration only and no text, figures or diagrams were copied. Check the current documentation of any implementation you adopt, because API details change between releases. Template licence: MIT.