Design a distributed key-value store

01Problem and scope

A distributed key-value store maps a key within one tenant's namespace to value bytes and a version; the store does not interpret the value's contents. Before choosing replication or partitioning, specify which updates must succeed together, what a later read must see and which failures an acknowledged write must survive. This design supports exact-key lookup, replacement and deletion, with conditional updates that reject an obsolete expected version. Joins, arbitrary search and transactions across unrelated keys require different contracts. Cart-42 at version 7 illustrates concurrent updates; the store does not interpret its application-level contents.

Interviewer: “Make that store available everywhere.” Candidate: “Must two devices updating cart-42 agree immediately, or may they accept competing versions and merge later?” Interviewer: “Prevent one device silently overwriting a completed edit. Start within one region.” We require each key's successful updates to have one agreed order; we are not offering every operation of a general-purpose database.

Version checks prevent concurrent edits from silently replacing each other. Client A adds an item while client B removes an item, both starting from cart-42 version 7. The store cannot decide the correct shopping meaning of these competing edits, but it can prevent both callers from believing they replaced the same version. The rejected caller rereads the cart, and the application decides how to combine the edits.

We exclude cross-key transactions, arbitrary search, global range scans and active-active cross-region writes. A tenant can have many keys, but one atomic operation addresses one key plus the store’s internal request-result metadata. This scope lets us show which replica group may update a key and when an update is safe to acknowledge. We will use tested consensus and storage engines; an interview sketch describes their required behavior rather than pretending a new database implementation is a weekend project.

02Functional requirements

1. Write a value. Create or replace a tenant-scoped value up to 1 MiB. The response returns a version token identifying the committed state.
2. Read an exact key. Read one exact key and receive its bytes and version, or an authoritative not-found result. A not-found response must follow the same consistency rule as a returned value.
3. Conditionally replace or delete. Replace or delete only if the supplied version matches. A mismatch returns conflict and the current version under the same authority, allowing client A to reconsider the requested edit.
4. Retry a mutation. Retry a mutation with the same request ID and canonical payload within the documented retry horizon. The caller receives the original outcome even when the first response disappeared.
5. Expand and move partitions. Increase cluster capacity and move partitions without losing committed records or allowing two independent owners to accept conflicting writes.
6. Recover replicas and backups. Recover a failed replica and restore historical backups without exposing partially restored state as authoritative.

Acceptance boundaries

Unconditional PUT is allowed only when the application deliberately accepts replacement semantics. It is not a hidden merge operation. DELETE creates a tombstone, a stored deletion marker ordered with updates; recreation receives a fresh version so an old expected-version token cannot accidentally match a new incarnation. Administrative scans for backup and repair are separate privileged interfaces, not an accidental promise of a public global range query.

03Non-functional requirements

1. Regional latency. Target p95 of 20 ms per operation under normal conditions. Global low-latency writes are outside this initial promise.
2. Availability. Assume 99.95% eligible-operation availability over a month, with brief election pauses and minority-partition unavailability explicitly allowed.
3. Durability. Acknowledged mutations survive one replica-node or zone failure using three appropriately placed replicas. A complete regional loss needs a separately defined backup/replication recovery plan.
4. Consistency. Provide linearizable operations within each key in its home region: a later read sees a completed write or something newer. Only a group able to establish the required majority accepts these operations.
5. Workload and isolation. Assume 100,000 reads/s and 20,000 writes/s, 1 KiB average values and a 1 MiB maximum. Enforce value-size and per-tenant byte quotas so an abusive writer cannot exhaust a partition.
6. Retention. Keep committed live values until deletion, retry results for an assumed one-hour horizon, and backup history for an assumed thirty-day retention policy.

Correctness takes priority

1. One mutation order. All successful conditional mutations of a key fit one order; at most one succeeds against a particular current version.
2. Real-time reads. A linearizable read respects completed operations. An isolated former leader must refuse strong reads even if its disk looks healthy.
3. Explicit degradation. Minority-isolated clients receive unavailable, not misleading success. A separately named stale-read API may exist, but cannot silently replace GET. The availability percentage never authorizes dropping successful writes.

04Capacity estimates

Disk capacity depends on retained live values; write bandwidth depends on every replacement, its replicated copies and storage maintenance. Keeping those quantities separate prevents a busy store from looking small merely because it repeatedly updates the same keys.

Assume 100,000 reads/s, 20,000 writes/s and 1 KiB average values.

Capacity is not placement

Many logical partitions share nodes. Throughput, failure reserve, distinct-zone replicas and uneven traffic may require more than 68 node equivalents. A p95 latency target is not an average; the 20 ms average in the concurrency calculation is a separate workload assumption.

Retries add load without adding successful writes

Failed conditional results can also need retention. A ten-minute outage does not stop client retries, so attempted ingress may greatly exceed successful mutation rate. Include admission control and client backoff. Retained live bytes and daily rewritten bytes are separate measurements.

05APIs and contracts

The API turns the version rule into a client workflow: read a value and its version, submit the intended change with that version, then handle a conflict or recover the outcome of a timed-out request. The request ID identifies the attempted change; the version identifies the state it expects to replace.

Every request is scoped by authenticated tenant, not a caller-selected namespace alone. A mutation includes a requestId scoped to the authenticated tenant and key. Within that scope, its fingerprint identifies the supplied input: expected version, operation type and value. Reusing the ID with different input is rejected. The same requestId on another key is a separate operation, because those keys may have independent partition authorities. Return 409 for a version conflict, 413 for the size limit, 429 for tenant admission limits and a retryable unavailable result when the required owner cannot be reached. A timeout leaves the outcome unknown.

Versions are opaque persistent tokens, not timestamps supplied by client A. A recreated cart must not reuse the deleted cart’s version. For this design, use a partition incarnation plus ordered mutation revision, carrying that identity through migration. The implementation can choose another proven nonrepeating representation.

There is no public listing pagination because range scans are excluded. Administrative snapshots expose a snapshot identifier and continuation cursor under a separate consistency contract. The mutation retry horizon is one hour in this exercise; clients older than that must use an explicit status/reconciliation path or reread before forming a new conditional intent. They cannot expect deduplication records to exist forever.

06Data model and access patterns

The versioned API needs more than stored values: it must remember retry outcomes, route each key to its current owner and recover the agreed update history. A replicated log stores commands in that agreed order; a state machine applies them using fixed rules to produce the next state and result.

The metadata names positions in this process. An owner epoch identifies a placement generation, a log term identifies a leadership period, and a log index identifies a command position. The applied index records how far a replica has installed committed commands into its local state. A snapshot saves state at such a boundary so recovery knows where replay must resume.

A storage engine can keep recent ordered entries in a memtable, an in-memory sorted structure backed by durable recovery data, and flush immutable sorted files to disk. A log-structured merge design later combines files through compaction, removing obsolete versions where retention and replication safety permit. Sequential writes are efficient, but reads may inspect several files and compaction rewrites bytes. Bloom filters can cheaply rule out files that definitely lack a key; a positive result is only a possibility.

The replicated consensus log records the commands agreed by the replica group; the storage engine's write-ahead log lets a node recover its local updates after a crash. An implementation may integrate them or avoid redundant logging with a carefully justified protocol; “we have two logs” is not itself a guarantee. Measure write amplification, read amplification, disk space, and fsync latency.

Apply a committed command as one atomic local engine batch: update the value or tombstone, record the request result, and advance applied-index metadata together. On restart, replay only according to the engine’s recovery contract, never expose a value without its matching deduplication result. Snapshotting includes those records and the applied boundary, rather than copying arbitrary files at unrelated moments.

A tombstone suppresses old values still present in immutable files or stale replicas. Retiring it requires proof that the relevant older state cannot reappear under supported repair and snapshot rules. User deletion and physical erasure from backups are distinct policies. The directory owns placement metadata; it does not contain the user value and cannot decide whether client A’s conditional replacement succeeded.

07Basic working design

Begin with one server, a key index, a tested local storage engine and a durable recovery log. An in-memory dictionary alone would lose cart-42 on restart. The API authenticates client A, reads version 7, and accepts req-a-9 with the replacement value only inside the storage engine’s serialized mutation path.

The owner checks the request-result table first. If req-a-9 is new, it compares the expected version with the current cart, constructs version 8 and atomically records both value and result under its durable-write policy. Only then does it reply. A crash after the log becomes durable but before the reply can be recovered: replay reconstructs both records, and a retry returns version 8.

GET consults this owner and the same ordered state; DELETE installs a new tombstone version through the same path. This baseline is already correct for concurrent requests on one machine if its serialization and recovery protocol are correct. It does not survive loss of the only disk or serve traffic during machine repair.

We now have a concrete benchmark target: exact-key read latency, conditional-write latency including log synchronization, live dataset capacity and write amplification under steady-state compaction. Measuring a memory-map microbenchmark would omit the very work that supports the promised acknowledgment.

08Find the baseline flaws

The live dataset is 11.24 TB before copies. It exceeds the assumed 500 GB usable live budget by more than twentyfold, so one storage server cannot hold it. Rewriting existing keys generates log and compaction traffic even when the number of live keys stays constant. A design counting only retained values can run out of write bandwidth first.

An asynchronous second copy is not enough for the durability contract. At t0, A logs version 8 and acknowledges client A; at t1, A’s disk is destroyed before B receives it. Promoting B loses an acknowledged write. Waiting for a second durable copy improves this interval, but an election protocol must also prevent promoting a history that omits committed entries.

A naive read-then-write conditional check fails independently: client A and client B both read version 7, both compare outside the serialized path, and both write a replacement. The last writer wins while both callers heard success. The check and update must be one ordered state-machine action, not two HTTP calls.

Finally, load-balancing reads across stale copies breaks the selected GET contract. Client A completes version 8, then reads version 7 from B. A replica must also confirm that it can serve current reads; copying writes alone is insufficient. These counterexamples explain why the next changes address ordering, placement and storage behavior separately.

09Improve the design, step by step

Change one: replicate one ordered decision stream. A single-disk loss motivates three replicas across failure zones, with a tested leader-based consensus protocol. The leader proposes commands, waits for the protocol’s durable commit condition and applies them before success. A valid replacement leader preserves committed history. This changes disk-loss recovery from “restore yesterday’s cart” to continuing from committed state. It costs inter-replica bandwidth, synchronization latency and temporary unavailability without a majority. The new danger is a stale leader answering strong reads; read authority must be confirmed. Asynchronous replicas are simpler and may improve availability for a weaker contract, but they are rejected for acknowledged-loss protection here.

Change two: split many keys across independent authorities. The 11.24 TB dataset and measured per-owner throughput trigger hash-based logical partitions with separate replica groups. The router resolves (tenant,cart-42) to partition 18. Moving a small logical partition changes fewer placements than replacing one giant physical-node modulo map. Benefits are aggregate capacity and parallelism across keys. Costs include a replicated directory, more consensus groups and coordinated migrations. A router may use an old map, so storage owners check the placement epoch and redirect requests sent to the wrong owner. Range placement would be preferable for ordered scans, but our exact-key API does not need them. A single very hot key still cannot be split without changing its semantics.

Change three: budget the engine’s deferred work. Steady-state random updates and disk pressure motivate an LSM-style engine with sorted files, Bloom filters and managed compaction. Batching improves sustained ingestion; file filters avoid some absent-key reads. It costs background CPU, rewritten bytes and temporary space. Compaction debt can stall foreground writes, so limit ingestion when maintenance cannot keep up. A B-tree engine remains a reasonable alternative for the measured read/update mix; benchmark both rather than calling an LSM universally faster. Large values may require separate blob placement, but that adds garbage-collection and publication boundaries and is deferred until the 1 MiB workload demonstrates a need.

Change four: isolate operational work. Replica catch-up, backup and tenant bursts can consume the same I/O as client A’s request. Reserve bandwidth and concurrency for each class; throttle migrations and apply per-tenant byte limits before queues grow indefinitely. This protects the 20 ms objective at the cost of slower administrative progress and explicit 429/unavailable responses. Unlimited buffering is rejected because it converts overload into latency and memory exhaustion. If a workload truly needs long asynchronous ingestion, expose a different admission and completion contract rather than silently weakening PUT.

Each step keeps the per-key decision at one authority. Scaling the cluster never changes a successful expected-version check into a best-effort suggestion.

10Detailed architecture

Authenticated routing

Client libraries contact an authenticated gateway or route directly through an equivalent authenticated protocol. The router caches a versioned partition map from a durable metadata quorum. Hashing locates a logical range; metadata identifies its replica group and current routing epoch. The router does not pick a random replica for a strong operation.

Partition replication

Partition 18 has leader A and followers B/C in separate configured failure domains. Each node holds the replicated log and its local state engine. The engine is an implementation boundary inside a storage node, not a fourth independent copy. Leaders order mutations, apply committed commands, and perform a safe read protocol. Followers replicate and catch up; they are eligible for leadership only under the consensus election rules.

Movement and background work

Each request waits for authentication, routing, consensus or a strong-read check, and the response. Compaction, repair, migration, metrics and backup run in the background with limits on their resource use. The final diagram makes these distinctions visible so an arrow to a directory or a backup cannot be mistaken for a committed user-data write.

Implementation option and limits

A coherent implementation uses a proven Raft library for each logical partition and RocksDB for local ordered state, with a small replicated metadata service for placement. RocksDB is an embedded storage engine, not a distributed database; it does not supply ownership, consensus or the retry protocol. An existing distributed database is preferable when its documented operations meet the contract. A small control-plane store such as etcd can hold placement metadata; the ten-billion-key payload estimate is not a recommendation to place the entire dataset in etcd.

11Write path and acknowledgement

A mutation succeeds only after its ordered command is durably committed and applied together with its request result. The following trace tests two updates against the same version.

1. The router hashes (tenantA,cart-42) and finds partition 18, ownership epoch 6, led by node A with followers B and C. Routing metadata is cached, but a stale epoch receives a redirect or rejection.
2. A receives request req-a-9 and proposes a command containing the expected version and new value. The replicated log orders it with other commands for partition 18.
3. The command is durably replicated and committed according to the consensus protocol. When applied in log order, it checks version 7, writes version 8, and records the request result. A competing request based on version 7 cannot also replace version 8.
4. A replies with version 8 only after commit and application. Client A's next strong read goes through a leader that confirms its current authority and has applied the necessary committed index; a former isolated leader must not answer stale data as current.
5. If the reply is lost, retrying req-a-9 returns version 8. If a different request tries expected version 7, it receives a conflict and must reread before deciding how to merge application data.

The actual acknowledgment includes the result identity, not just a generic 200. If the version check fails when its command is applied, the failure result is also associated with req-a-9 so a repeated request does not change meaning after another cart update. The protocol may optimize known duplicates, but correctness cannot rely on an unreplicated memory cache of request IDs.

Deletes follow the same ordered path and install a fresh tombstone version. A successful deletion does not authorize an old replica to resurrect version 7. During migration, clients may repeat the request through a new owner, so request-result state must move with the key or remain accessible through the owner’s supported retry protocol. Copying only user values would reopen the lost-response ambiguity.

12Read and delivery path

Before serving a strong read, the leader must confirm that it still leads and has applied the required committed commands. Its label alone proves neither.

A block cache inside the engine speeds access without inventing a second authority: cached blocks are interpreted through the current engine state. An application-side value cache would need a separate validated freshness protocol to serve strong GET. We do not quietly add such a cache just to hit a latency target. Clients wanting low-latency stale snapshots can opt into an explicitly weaker operation.

13Correctness deep dive

The consensus protocol orders commands; the deterministic state machine decides their meaning. Suppose committed log positions 101 and 102 contain client A’s add-pen request and client B’s remove-book request, both expecting version 7.

apply(command, committedIndex):
  saved = result(command.tenant, command.key, command.requestId)
  if saved exists:
      outcome = original result if fingerprint matches else invalid-reuse
  else:
      current = value(command.tenant, command.key)
      if current.version != command.expectedVersion:
          outcome = conflict(current.version)
      else:
          outcome = success(freshVersion(partitionIncarnation, committedIndex))
          prepare replacement bytes or tombstone with that version
  atomic engine batch:
      install replacement only for a new successful request
      store fingerprint and outcome only when no prior result exists
      advance applied index, including duplicate and rejected commands

The displayed version 8 is shorthand for the opaque nonrepeating token. The key point is that client B’s check happens after client A’s applied change in the agreed order. Another write cannot run between the version check and its update.

14Failure and recovery

Recovery must preserve the same per-key history even when the process, replica group or storage medium changes. For each failure below, identify which owner can still prove that history before allowing more strong reads or writes.

15Operations, security, and cost

Authenticate node-to-node replication and administrative control, encrypt tenant data under the required threat model, and authorize tenant-scoped keys at every serving path. A direct follower endpoint must not bypass namespace checks. Rate-limit bytes as well as operation count: one 1 MiB mutation costs about a thousand average 1 KiB mutations in replication payload.

Monitor p95 and p99 committed-operation latency, unavailable partitions, quorum loss, leader churn, disk synchronization time, compaction debt and hot-key concentration. A low global average can hide one inaccessible customer partition. Distinguish client attempts, proposals, committed mutations and conflicts to diagnose a retry storm accurately.

The 33.72 TB three-copy live estimate excludes transient compaction output and recovery reserve. If operational policy permits only 70% steady storage occupancy, the corresponding capacity budget is about 48.2 TB before additional index/log overhead. This is a planning assumption, not a universal engine threshold. Increasing replica count from three to five raises live copy bytes by two thirds while changing the tolerated failure/latency tradeoff; it does not increase per-key write concurrency proportionally.

Roll out engine formats and protocols with mixed-version compatibility. Exercise leader loss after commit, stale-leader reads, duplicate request IDs, deleted-key recreation and migration during writes. Verify both returned histories and durable state with fault injection. A successful backup command or a green cluster membership page does not prove that the promised conditional operation survives those interleavings.

16Decision ledger and limitations

The central choice was a linearizable, single-key API. The tables relate that promise to its replication, storage and routing costs, and identify the requirements that would justify a different contract.

An availability-first multi-writer store can accept updates in more isolated locations, but returns concurrent versions or applies a merge rule. A shopping application may prefer merging independent add-item operations to rejecting a temporary partition; that is a different API and conflict model. It cannot be substituted under the existing expected-version promise without explaining changed outcomes.

The first scaling limit may be one hot cart, not total data size. A storage partition can move to a larger owner, but it cannot create parallel successful mutations against the same prior version. Further scaling requires the application to accept independently updated subkeys or a weaker consistency rule.

17Interview closing

“I start with one durable owner and make value, version and retry outcome one recoverable state transition. The single machine fails our storage and failure targets, so I split many keys into logical partitions and replicate each partition through a tested majority protocol.

“Clients route through versioned placement metadata. The leader commits and applies a mutation before acknowledging success, and confirms current authority before a strong read. Conditional mutations with the same expected version are serialized: once one advances the version, the other conflicts. If a reply is lost, retrying the original request identity recovers its recorded result rather than applying another write.

“I pay for replica bytes, log synchronization, compaction and minority-side unavailability. I protect foreground work from migrations and tenant bursts, and test safe membership changes. The next measurements are hot-key concentration, steady-state write amplification and tail latency during one-node failure.”

Interviewer: “Now allow writes in two disconnected regions.” Candidate: “I cannot keep the same immediate conditional-update promise on both sides. I would discuss routing each key to one authority or introducing application-mergeable operations and explicit conflicts. The user-visible semantics must change before the topology does.”

Recall the key ideas

Answer from memory before opening each card. Explain why the choice works and what it costs. Revisit missed cards tomorrow.