Design a Distributed
Key-Value Store.
SAY THIS: “Before proposing a topology for Design a Distributed Key-Value Store, I’ll clarify the core functional requirements, burst characteristics, data volume, and availability SLOs to define rigid system boundaries.”
What the system does
- put(key, value): store or update key-value pair.
- get(key): retrieve value associated with key.
- High availability and fault tolerance: survive node failures without data loss.
- Horizontal scalability: scale storage and throughput by adding nodes.
- Tunable consistency: support configurable R (read quorum) and W (write quorum).
What to agree on
- Key size: up to 256 bytes; value size: up to 1 MB.
- Low latency: p99 read and write under 10 ms within same datacenter.
- Handle network partitions (CAP theorem AP system with eventual consistency).
SAY THIS: “Let’s establish order-of-magnitude sizing for peak throughput, memory footprint, and network ingress to size the critical latency path before making technology choices.”
At an illustrative 200 bytes per record/session, high-volume state requires dedicated primary in-memory partitions. Add replicas, allocator overhead, policy indexes, and headroom.
Always validate throughput and p99 latency targets under skewed traffic distributions and replica failovers.
Keep state shared and partitioned.
SAY THIS: “I’ll separate low-latency ephemeral cache state from authoritative durable storage, ensuring strict consistency where needed and partitioning for scale.”
Cache / Ephemeral state
state:{partitionKey}:activeWindow
- In-memory key-value cache (Redis Cluster / Memcached).
- TTL expiration matching window validity bounds.
- Hash tags to co-locate keys for atomic Lua scripts or transactions.
Durable / Authoritative store
- Distributed NoSQL / relational database (DynamoDB, PostgreSQL).
- Durable write-ahead log with replica failover.
- Cache-aside pattern: read misses warm cache from durable store.
- Asynchronous CDC / event streaming to telemetry sinks.
Local-only node state is not a shared cluster quota: N nodes could admit roughly N × the configured limits. Ensure state co-location and transactional boundaries align with partition keys.
SAY THIS: “The gateway tier authenticates incoming traffic, validates payloads, and routes requests to decoupled services backed by partitioned caching and authoritative storage tiers.”
Use tight timeouts, circuit breakers, and rate-limiting middleware at ingress to protect downstream services from cascading collapse.
Define the contract.
SAY THIS: “The API contract establishes idempotent mutation semantics, clear error taxonomy, and explicit status codes so upstream clients can safely retry.”
Success Path
Forward the request or return HTTP 200/201. Include standard tracking and rate/quota telemetry headers in the response.
Failure & Fallback
Use explicit HTTP status codes: 400 for invalid input, 429 for quota violations, and 503 with Retry-After during maintenance or partitions.
View example API payload contract
HTTP/1.1 200 OK
Content-Type: application/json
Cache-Control: max-age=300
{
"status": "success",
"data": { "id": "res_98a7f", "processedAt": "2026-11-11T12:00:00Z" }
}Include idempotency keys on write endpoints to guarantee at-most-once processing across network retries.
SAY THIS: “For the critical execution path, we enforce idempotency and concurrency controls using atomic primitives, avoiding lock contention and split-brain states.”
- Consistent Hash Ring: Map key to ring position using hash(key). The first N distinct physical nodes clockwise from the key position store replicas.
- Virtual Nodes (tokens): Each physical node owns V virtual positions on the ring to distribute data uniformly.
- Write Path: Coordinator sends write to all N replicas. Waits for W ACKs before returning success. On local node: write to Commit Log (WAL) -> MemTable in RAM -> flush to SSTables on disk (LSM-Tree).
- Read Path: Coordinator requests data from R replicas. Compares version timestamps/vector clocks. Returns most recent version and triggers read repair asynchronously.
| STRATEGY / PATTERN | WHY CHOOSE IT? | TRADE-OFF / COST |
|---|---|---|
| Master-Slave (Leader-Follower) | Single leader coordinates all writes. Simple consistency, but leader failover causes write downtime. | Coordinate partition dependencies |
| B-Tree storage engine | in-place disk updates cause random I/O | LSM-Trees turn random writes into sequential writes, making LSM superior for high write throughput. |
View reference production code implementation
// Tunable Quorum Consistency Evaluation (R + W > N)
export interface QuorumConfig {
N: number; // Total replicas across cluster
W: number; // Write quorum size
R: number; // Read quorum size
}
export function evaluateQuorum(config: QuorumConfig): { isStronglyConsistent: boolean; model: string } {
const { N, W, R } = config;
if (R + W > N) {
return {
isStronglyConsistent: true,
model: `Strong Consistency: R(${R}) + W(${W}) > N(${N}) guarantees read/write quorum overlap.`
};
} else {
return {
isStronglyConsistent: false,
model: `Eventual Consistency: R(${R}) + W(${W}) <= N(${N}) allows stale reads for high availability.`
};
}
}Avoid client-side read-modify-write loops without concurrency control. Use atomic scripts, row-level locks, or conditional mutations at the storage tier to guarantee linearizability.
View sequence trace & failure interaction flow
[ Client Read ]
|
v
[ Coordinator Node ] ---> (Queries R=2 Replicas in parallel)
|
+---> Node A: { value: "v2", timestamp: 1042 } <-- Newest vector timestamp chosen!
+---> Node B: { value: "v1", timestamp: 998 } <-- Asynchronous Read Repair triggered!SAY THIS: “The biggest operational risks involve node failover, partition skew, network isolation, and cascading downstream backpressure. Here is how we mitigate each.”
At least one node in read quorum overlaps with write quorum, guaranteeing the latest value is read.
Low latency and high availability, but high risk of reading stale data.
LWW uses wall-clock timestamps (simple, but clock drift can overwrite valid concurrent writes). Vector Clocks preserve concurrent versions for client reconciliation, but add metadata size.
During node outage, neighboring nodes store writes tagged with a hint. When the node recovers, neighbors deliver the hinted writes.
Nodes compare hierarchical hash trees (Merkle Trees) of their key ranges. Only branches with differences are synchronized across the network, minimizing transfer bandwidth.
Nodes exchange heartbeats randomly with neighbors. If a node fails to heartbeat within timeout, consensus marks it down without central health monitor.
Close with the trade-offs.
SAY THIS: “In closing, we optimized for sub-millisecond p99 latency and high availability under partition. We trade strict distributed linearizability for predictable throughput and operational simplicity.”
If the interviewer asks…
- R + W > N (Strong Consistency): At least one node in read quorum overlaps with write quorum, guaranteeing the latest value is read.
- W=1, R=1, N=3 (Eventual Consistency / Fast): Low latency and high availability, but high risk of reading stale data.
Keep the conclusion focused
- What guarantees are strictly enforced at runtime?
- What happens during network partitions or node failures?
- Which CAP trade-off was intentionally chosen and why?
Design a distributed, highly available key-value store similar to DynamoDB or Cassandra. The system must support put(key, value) and get(key) across multiple storage nodes with configurable consistency, automatic data partitioning, and failure recovery.
“Partition data with consistent hashing and virtual nodes, replicate across N nodes with tunable quorum (R + W > N for strong consistency), and use an LSM-Tree (WAL → MemTable → SSTable) for write-optimized local storage. Gossip protocol handles failure detection; Merkle trees drive efficient anti-entropy reconciliation.”
Shard Key Hotspot Prevention
Why is selecting a low-cardinality column like 'country_code' as a primary database shard key considered a dangerous architectural pitfall?