Distributed key-value store

01 · CONTEXT

What a distributed key-value store needs to decide

A key-value store holds key-value pairs and answers two operations: write and read by key. In an interview, the starting point is gathering requirements: how much data fits on a single machine, how available the service needs to be, and what is worse for the product, returning a stale value or refusing the operation? That last question defines the whole design.

It is the choice of the CAP theorem. A consistency-oriented (CP) store gives up availability during a network partition; an availability-oriented (AP) store gives up consistency and keeps serving, even if that returns old data.

The case studied here follows the line of systems like Dynamo and Cassandra, which adopt eventual consistency: strong consistency would require blocking new operations until all replicas agree, which does not go with high availability. The design is decentralized, with no single point of failure, because every node has the same responsibilities.

1. The client talks to a coordinator, which acts as an intermediary between it and the storage nodes.

2. The storage nodes are peers: none is a leader, all have the same responsibilities, so there is no single node whose failure takes the service down.

02 · DATA

Partition and replicate

A data set that does not fit on one machine is split into partitions spread across several servers, in the spirit of sharding. There are two challenges: distributing the data evenly and moving as little as possible when nodes join or leave. Consistent hashing is the technique pointed to for both. Here it appears as a ring: servers and keys are mapped to positions on the same ring, and each key belongs to the server it meets when walking the ring.

For high availability, each piece of data is replicated asynchronously on N servers, with N a configurable parameter. After the key is mapped to a position on the ring, you walk clockwise from it and pick the first N servers to hold the copies. Replication brings the opposite problem: the copies can diverge, and the rest of the design exists to manage that.

1. The coordinator maps the key to a position on the ring and identifies the server responsible for it.

2. Walking clockwise from the key's position, the coordinator picks the first servers on the ring as replicas; the fourth node stays outside this set.

03 · QUORUM

Read and write quorum

Replicas let you choose how many of them must respond. With N replicas, W is the write quorum and R the read quorum: a write is only considered successful after W acknowledgments, and a read waits for R responses. If W plus R is greater than N, strong consistency is guaranteed, because there is always at least one node in common between whoever received the write and whoever answers the read, and that node has the most recent data. This connects to what consistency patterns e consenso discuss in general.

The configuration is a trade between latency and consistency. Requiring only one response on a read, with the write confirmed by all replicas, optimizes the read; the reverse optimizes the write. With a small quorum the operation returns quickly, since the coordinator does not wait for the slowest replica, but strong consistency is no longer guaranteed when the sum of the quorums does not exceed N.

04 · CONFLICTS

Versions and vector clocks

Replicating gives availability, but causes inconsistency between replicas: two concurrent writes can reach different nodes and produce conflicting values. The solution is to treat each modification as a new immutable version of the data and use a mechanism capable of detecting and reconciling conflicts. The vector clock is the common technique: a server-version pair attached to the data, which makes it possible to check whether a version precedes, succeeds, or conflicts with another.

05 · FAILURES

Detecting and tolerating failures

To know who has gone down without depending on a central node, each server keeps a membership list with heartbeat counters and increments them periodically. It also periodically sends its heartbeats to a set of random nodes, which pass them on to others: this is the gossip protocol. If a member's counter does not increase for a predefined period, it is considered down.

Once the failure is detected, the system needs to stay available, which ties in with the availability patterns. In a sloppy quorum, instead of requiring the original replicas, the system uses the first healthy servers on the ring and ignores the ones that are down. Another server temporarily stands in for the one that went down and, when it comes back, the changes are handed back to it: this is hinted handoff.

Hinted handoff covers temporary failures. For a replica that is permanently unavailable, an anti-entropy protocol is applied that compares the replicas' data and updates each one to the newest version. A Merkle tree is used to detect the inconsistency and reduce the volume of data transferred.

1. One of the replica nodes stops sending heartbeats; the others, exchanging membership lists by gossip, come to consider it down.

2. The coordinator ignores the failed node and uses the next healthy server on the ring, which temporarily stands in for it.

3. When the node comes back, the stand-in hands back to it the changes it received while it was away.

4. For a permanent failure, the remaining replicas compare their data using Merkle trees and synchronize only the parts that diverge.

06 · NODE

Write and read paths inside a node

Inside each node, the write design follows Cassandra's architecture. The request is first persisted to a commit log file; then the data goes to an in-memory cache (see cache fundamentals); when that cache fills up or reaches a predefined limit, the data is flushed to an SSTable on disk.

On a read, the node checks memory first. If the data is not there, it is fetched from disk, and an efficient way is needed to find out which SSTable contains the key. The Bloom filter is commonly used for this, avoiding opening files that certainly do not have the data. The choice of storage engine is part of database selection.

1. On a write, the node records the request in the commit log and writes the data to the in-memory cache.

2. When the in-memory cache fills up or reaches a limit, the contents are flushed to sorted files on disk.

3. On a read, the node looks at memory first; if the data is there, the response goes out without touching the disk.

4. Without the data in memory, the node goes to disk and uses a filter to find out which files may contain the key.