Key-Value Store: System Design

A key-value store is the simplest database that is of any practical use. It offers 2 operations, to store a value under a key and to return the value that was stored under a key, and it makes no attempt to understand what the values mean. Simple as it is, it lies underneath a remarkable number of other systems. A web application keeps its users’ sessions in one. A cluster of machines keeps its configuration in one. A file system uses one for its directory of files, and an object store for its index of objects. Most of the larger storage systems in use today contain a key-value store as their innermost component, and the designs in the later chapters of this book rely on one repeatedly.

On a single machine, a key-value store is an afternoon’s work: a table in memory and a file on disk. The difficulty, and the subject of this chapter and the 5 that follow, begins when the store has to outgrow 1 machine and to outlive it. The data may be larger than any 1 disk. The requests may be more than any 1 processor can answer. And the machine will certainly fail at some point, by losing power, or its disk, or its connection to the network, and a store on which other systems depend cannot be permitted to lose its data or to stop answering on that account. Meeting these demands means spreading the data across many machines and keeping several copies of every part of it, and the moment there are several copies, the questions arise that define the whole field: which copy is right when they differ, who decides, and what happens when the machines cannot reach each other to agree.

This chapter designs such a store from its requirements. It is the system design, and it stays at the level of components and the messages between them. Chapters 2 to 6 then build each component in Rust and measure it.

Why such a store is needed

Consider a shop whose web site runs on 3 web servers behind a load balancer. A customer’s requests are distributed among the 3, so that the request which adds an item to the basket may be handled by 1 server and the request which displays the basket by another. The basket therefore cannot be kept in the memory of any 1 web server. It has to be held by some service that all 3 can reach.

This arrangement is typical, and it divides the system into parts of 2 very different characters. The web servers hold nothing that matters. Any of them can be restarted, replaced or duplicated at any moment without loss, and a component of this kind is called stateless. The shared store is the opposite. Everything of value is in it, a failure of it is a failure of every web server at once, and it cannot be restarted on a fresh machine, since a fresh machine would not have the data. The simplicity of the stateless part of a system is paid for by the difficulty of the part that holds the state, and this chapter is about that part.

What is demanded of the store follows directly from its position. Each demand has a consequence for the design, and each consequence is the subject of a later section.

Use cases

The store is used by 3 kinds of party. An application is any service that keeps its data in the store, such as the web servers of the shop. An operator is whoever looks after the machines, whether a person or an automated system. And monitoring is the system that watches the health of every machine and raises an alarm when something is wrong.

Functional requirements

The functional requirements state what the store does. Each is derived from a use case and is specific enough that a test could establish whether it is met.

Non-functional requirements

The non-functional requirements state how well the store must do these things: how much data, how many requests, how quickly, and through which failures. It is these, far more than the functional requirements, that determine the architecture. A program that kept 1 file on 1 machine would satisfy functional requirements 0 to 6 completely, and would satisfy almost none of what follows.

Durability and availability are easily confused and should be kept apart. Durability concerns data that has already been written, and requires that none of it be lost. Availability concerns requests that are yet to come, and requires that they be answered. Data can be perfectly durable and entirely unavailable, as when every copy of a key is on a machine that has been switched off for an hour and will return intact. The design provides for the 2 properties by different mechanisms, and they sometimes pull in opposite directions.

Which kind of consistency

The word “consistency” has to be given an exact meaning before anything can be designed to provide it, since storage systems offer 3 quite different things under that name. They are best distinguished by a single situation: a value has just been written, 1 of the copies of the data has not yet received the change, and a reader happens to be answered by that copy.

Under strong consistency, the reader receives the new value. The system behaves, as far as any client can tell, as though there were only 1 copy of the data, and a write that has been acknowledged is visible to every subsequent read. The precise form of this property is called linearizability. Its cost is that reads and writes must involve a leader or a majority of the copies, which makes them slower and means that they fail when a majority cannot be reached.

Under eventual consistency, the reader may receive the old value. The system promises only that, once writes cease, all copies will in time come to hold the same value. Any copy can answer any request, which makes the system fast and keeps it available through almost any failure, and the burden passes to the applications, which have to cope with stale values and with conflicting ones.

Tunable consistency lets each request choose. With N copies, a write waits for W of them to confirm and a read consults R of them. Whenever R + W is greater than N, the copies consulted by any read must include at least 1 that took part in the latest write, and the read sees that write. With 3 copies, the choice of W = 2 and R = 2 behaves strongly, and W = 1 and R = 1 is fast and only eventually consistent. The flexibility is real, and so is its cost, which is that every caller must choose correctly for every request.

This design chooses strong consistency, and the reason lies in the fourth use case. “Add an item to the basket, unless somebody else has changed the basket in the meantime” works only if the version that a client reads is the latest. If reads could be stale, 2 web servers might each read version 1, each conclude that nobody had changed the basket, and each write, with 1 of the 2 items lost. This is precisely the lost update that conditional writes exist to prevent, and a conditional write over stale reads prevents nothing. What strong consistency costs is set out in the section on guarantees: during a split of the network, the smaller side stops answering.

Core entities

The core entities are the things that the system stores, together with the things it stores them on. Naming them at this stage, with their fields, provides a vocabulary that the API, the protocol and the code of later chapters all share.

The first 2 entities are 2 representations of the same data, and the distinction between them runs through every chapter that follows. The records are a history: “set user-1 to ada”, “set user-1 to ada lovelace”, “delete user-2”. The entries are the state that results from applying that history in order. Both are kept, because they serve different purposes. Records are what is written to disk and copied to other nodes, since a sequence of changes is easy to append to, to transmit and to compare. Entries are what reads are answered from, since nobody wishes to replay a history in order to learn the present value. The version of an entry is the sequence number of the last record that changed it, which fastens every entry to a definite point in the history.

Capacity estimate

Arithmetic ahead. This estimate turns non-functional requirement 4 into a number of machines. The result is 24 storage nodes, each holding 138 GB, and nothing else from this section is needed later.

Skip to the high-level design

Show the calculation

The estimate depends on 1 quantity that has to be measured and cannot be assumed: the time a disk takes to store a write durably. Chapter 2 measures it with a program called synccost. On an NVMe disk, appending a record and waiting for the disk to confirm took 5.4 ms, so that a node which waits once for each record stores 185 records per second. Readers can run synccost on their own hardware and repeat the calculation with their own figure.

Storage. An entry consists of a key of about 32 bytes, a value of 1024 bytes and about 44 bytes of bookkeeping, 1100 bytes in all.

Quantity	Calculation	Result
Size of 1 copy of the data	1000000000 × 1100 bytes	1.1 TB
Size of 3 copies, as non-functional requirement 0 demands	1.1 TB × 3	3.3 TB

Writes. Every write is stored on 3 nodes, so that 20000 writes per second become 60000 records per second across the cluster. At 185 records per second for each node, 325 nodes would be needed. The time the disk takes, however, is nearly the same for 1 record as for 20, and a node can therefore gather the records that arrive during 1 wait and store them together. Storing several records with a single wait is called group commit, and it is the subject of chapter 6. With 20 records for each wait on average, a node stores 185 × 20 = 3700 records per second.

Quantity	Calculation	Result
Records to store per second	20000 writes × 3 copies	60000
Records per node per second, with 1 wait for each record	1 ÷ 0.0054 s	185
Records per node per second, with 20 records for each wait	185 × 20	3700
Nodes required	60000 ÷ 3700	16.2, rounded up to 17
Nodes chosen		24
Capacity of 24 nodes	24 × 3700	88800 records per second
Data on each node	3.3 TB ÷ 24	138 GB

We choose 24 nodes. At the required rate of writes, a cluster of that size works at 68% of its capacity, which leaves the surviving nodes room to carry the load of one that has failed.

Reads. Reads are answered from memory and do not wait for the disk. Spread over 24 nodes, 100000 reads per second come to about 4200 for each node, far below what a node can answer from memory. The rate of reads therefore has no influence on the number of nodes.

The API

Applications reach the store over HTTP. The API follows the style known as REST, in which the things a service holds are presented as resources, each identified by a URL, and the standard methods of HTTP are the operations upon them. Adhering to the standard meanings of the methods and status codes has a practical reward. Every existing HTTP client, proxy and cache already understands them, and so handles the API correctly without knowing anything about key-value stores.

The rules the API follows

Reference

Each operation is specified below by its request and its possible responses. store.example stands for the address of the load balancer.

Set the value of a key

PUT /v1/keys/user-1 HTTP/1.1
Host: store.example
Content-Type: application/octet-stream
Content-Length: 3

ada

HTTP/1.1 201 Created
ETag: "1"
Content-Length: 0

Read the value of a key

GET /v1/keys/user-1 HTTP/1.1
Host: store.example

HTTP/1.1 200 OK
ETag: "1"
Content-Type: application/octet-stream
Content-Length: 3

ada

Delete a key

DELETE /v1/keys/user-2 HTTP/1.1
Host: store.example
If-Match: "3"

HTTP/1.1 204 No Content

List keys

GET /v1/keys?prefix=user-&limit=2 HTTP/1.1
Host: store.example

HTTP/1.1 200 OK
Content-Type: application/json

{
  "keys": [
    { "key": "user-1", "version": 2, "size": 12 },
    { "key": "user-2", "version": 3, "size": 5 }
  ],
  "next_after": "user-2"
}

Errors

Every error response has the content type application/problem+json and a body with these fields.

HTTP/1.1 412 Precondition Failed
Content-Type: application/problem+json

{
  "type": "/problems/condition-failed",
  "title": "The condition on the key's version does not hold",
  "status": 412,
  "detail": "the key is at version 2"
}

Health of a node

HTTP/1.1 200 OK
Content-Type: application/json

{ "status": "ok", "keys": 2, "last_sequence": 5 }

The sections that follow take up the parts of this reference that embody a decision, and show each exchange as it was actually performed by the server of chapter 2.

Writing and reading a key

PUT stores the body of the request as the value of the key. Since values are opaque bytes, the body is sent without any encoding. The first write of a key is answered with 201 Created. All the responses in this section were produced by the server of chapter 2, with its date header removed.

$ curl -i -X PUT http://127.0.0.1:7001/v1/keys/user-1 --data-binary 'ada'
HTTP/1.1 201 Created
etag: "1"
content-length: 0

The version of an entry is conveyed in the etag header. In HTTP, an entity tag, or ETag, identifies 1 particular state of a resource, which is exactly what a version is. GET returns the value together with its tag.

$ curl -i http://127.0.0.1:7001/v1/keys/user-1
HTTP/1.1 200 OK
content-type: application/octet-stream
etag: "1"
content-length: 3

ada

Conditional writes

Functional requirement 3 is directed at a problem called the lost update, which arises when 2 clients read a value, each modify it independently, and each write it back.

Both writes succeed, and milk is missing from the final value. Nothing failed, and no error was reported to anyone. The write at step 4 was computed from what B read at step 2, which by then was no longer current, and the store, having been told only to set the value, set it.

The remedy is to let a write state which version it was based on. HTTP provides the If-Match header for this purpose: the server performs the write only if the present version of the resource is the version named in the header.

$ curl -i -X PUT http://127.0.0.1:7001/v1/keys/user-1 -H 'If-Match: "1"' --data-binary 'ada lovelace'
HTTP/1.1 204 No Content
etag: "2"

The key was at version 1, the condition held, and version 2 resulted. The same request sent a second time is refused, since version 1 is no longer current.

$ curl -i -X PUT http://127.0.0.1:7001/v1/keys/user-1 -H 'If-Match: "1"' --data-binary 'ada byron'
HTTP/1.1 412 Precondition Failed
content-type: application/problem+json
content-length: 144

{"type":"/problems/condition-failed","title":"The condition on the key's version does not hold","status":412,"detail":"the key is at version 2"}

The status 412 Precondition Failed informs web server B that its copy is out of date. B reads again, obtains tea, milk at version 2, adds the sugar, and writes with If-Match: "2". The procedure of reading, modifying and writing conditionally, repeated until the write succeeds, is the general method by which clients of a store cooperate without locks, and it recurs throughout the later chapters under the name of compare-and-set.

The header If-None-Match: * expresses the complementary condition, that the write is to be performed only if the key does not exist. With it, several clients may all attempt to create the same key, and exactly 1 succeeds.

$ curl -i -X PUT http://127.0.0.1:7001/v1/keys/user-1 -H 'If-None-Match: *' --data-binary 'someone else'
HTTP/1.1 412 Precondition Failed
content-type: application/problem+json
content-length: 144

{"type":"/problems/condition-failed","title":"The condition on the key's version does not hold","status":412,"detail":"the key is at version 2"}

On a GET, If-None-Match with a version asks for the value only if it has changed since that version. If it has not, the server answers 304 Not Modified with no body, which spares the transfer of data that the client already holds.

$ curl -i http://127.0.0.1:7001/v1/keys/user-1 -H 'If-None-Match: "2"'
HTTP/1.1 304 Not Modified
etag: "2"

Listing keys

GET /v1/keys returns the keys that begin with prefix, in sorted order, and no more than limit of them. When more remain, the response carries next_after, and the client obtains the next page by passing that value as after.

$ curl -i 'http://127.0.0.1:7001/v1/keys?prefix=user-&limit=1'
HTTP/1.1 200 OK
content-type: application/json
content-length: 71

{"keys":[{"key":"user-1","version":2,"size":12}],"next_after":"user-1"}

$ curl -i 'http://127.0.0.1:7001/v1/keys?prefix=user-&limit=1&after=user-1'
HTTP/1.1 200 OK
content-type: application/json
content-length: 66

{"keys":[{"key":"user-2","version":3,"size":5}],"next_after":null}

The position in the listing is thus expressed as the last key of the previous page. A page number would not serve, for a reason that applies to every listing of data that may change. If a key is inserted ahead of the reader’s position between 2 requests, every later key moves down by 1 place, and a request for “page 2” returns a key that the reader has already received. A request for the keys after user-1 is unaffected by insertions and deletions elsewhere. Chapter 15 meets the same problem in a news feed and solves it in the same way.

Deleting a key

$ curl -i -X DELETE http://127.0.0.1:7001/v1/keys/user-2
HTTP/1.1 204 No Content

Errors

A failed request is answered with a status code in the 400s or the 500s and a body in the format of RFC 9457. The type field identifies the category of the error and is what a client program should test. The title is a fixed description of the category, and detail, when present, describes the particular occurrence.

$ curl -i http://127.0.0.1:7001/v1/keys/user-2
HTTP/1.1 404 Not Found
content-type: application/problem+json
content-length: 80

{"type":"/problems/key-not-found","title":"The key does not exist","status":404}

The 2 ranges of status codes call for different behaviour on the part of the client. A code in the 400s signifies a fault in the request, and sending the same request again will produce the same error. A code in the 500s signifies that the server was unable to carry out a valid request, which may well succeed if repeated.

Retries

Suppose that a client sends a request and loses its connection before any response arrives. It has no means of knowing whether the operation was performed. This situation, the lost response, is the most characteristic difficulty of distributed systems and reappears in nearly every chapter of this book. The API meets it by making every operation idempotent, so that repeating a request is safe whichever was the case.

The protocol between nodes

The API described so far serves applications. The storage nodes also communicate with one another. A node sends the records it has written to the nodes that hold the other copies, and every node obtains the current ring from the coordinator. This internal traffic has different needs from a public API and uses a protocol of its own.

Protocol Buffers is a format in which the types of messages are declared in a schema file and encoded in a compact binary form. gRPC is a protocol for remote procedure calls that carries such messages over HTTP/2. The schema file is the specification of the internal protocol, and it is written before any implementation.

The Record message corresponds to the Record entity.

Listing 1.1 The Record message. proto/store.proto

// A change to one key, as the leader of a shard wrote it to its log.
message Record {
  uint64 sequence = 1;
  oneof change {
    Put put = 2;
    Delete delete = 3;
  }
}

message Put {
  string key = 1;
  bytes value = 2;
}

message Delete {
  string key = 1;
}

Every field of a message has a number, and it is by the number that the binary encoding identifies the field. Names occur only in the schema. This arrangement is what allows nodes running different versions of the software to communicate, which they must do whenever a cluster is upgraded 1 node after another without stopping. A newer version may add a field under a number not previously used. An older node disregards numbers it does not recognise, and a newer node supplies a default value where a field is missing. The single rule that makes this work is that a field number, once used, keeps its meaning for ever.

A leader sends records to its followers by means of the Append call. The terms leader, follower and term are defined in the section on replication.

Listing 1.2 The replication service. proto/store.proto

// Sent by the leader of a shard to each follower.
service Replication {
  // Appends records to the follower's log. A call with no records is a
  // heartbeat: it tells the follower that the leader is alive.
  rpc Append(AppendRequest) returns (AppendResponse);
}

message AppendRequest {
  // The shard, named by the last position it owns.
  fixed64 shard = 1;
  // The term of the sender. A follower rejects a leader of an older term.
  uint64 term = 2;
  // The sequence of the record that precedes `records` in the leader's log.
  uint64 previous_sequence = 3;
  repeated Record records = 4;
  // Every record up to this sequence is on a majority of the replicas.
  uint64 committed_sequence = 5;
  // Added for leader election. The term of the record at previous_sequence,
  // and the id of the sender.
  uint64 previous_term = 6;
  uint32 leader = 7;
}

message AppendResponse {
  uint64 term = 1;
  // False when the follower's log does not end at `previous_sequence`.
  bool appended = 2;
  // The sequence of the last record in the follower's log.
  uint64 last_sequence = 3;
}

The field previous_sequence is what keeps a follower’s log identical to its leader’s. A follower appends the records it is sent only if its own log at present ends at previous_sequence. If it does not, the follower answers appended = false together with the actual last sequence number of its log, and the leader sends again from that point. No follower’s log can therefore contain a gap or hold records in a different order from the leader’s.

The coordinator stores the ring as a Ring message.

Listing 1.3 The ring, as the coordinator stores it. proto/store.proto

// The ring, as the coordinator stores it. Routers and nodes read it and watch it
// for changes.
message Ring {
  uint64 version = 1;
  repeated Node nodes = 2;
  repeated Shard shards = 3;
}

message Node {
  string id = 1;
  string address = 2;
  // The positions this node owns on the ring.
  repeated fixed64 tokens = 3;
}

// A range of the ring and the nodes that hold its keys.
message Shard {
  // The shard owns the positions from first_token to last_token, both included.
  fixed64 first_token = 1;
  fixed64 last_token = 2;
  uint64 term = 3;
  string leader = 4;
  repeated string followers = 5;
}

Routers and storage nodes read the ring when they start and thereafter watch it. A watch is a request that the coordinator holds open and answers whenever the stored value changes, so that every component learns of a new version of the ring within milliseconds and none of them has to ask repeatedly.

High-level design

The high-level design names the components of the system, states the role of each, and traces the path of a request through them. Figure 1.7 shows the whole. The numbers in the figure are the steps of a write, and they appear again in its legend and in the text below.

The coordinator is a small replicated store for configuration. It runs on 3 machines, keeps a complete copy of its data on each, and remains available as long as a majority of them is running. It may seem odd that a key-value store should depend on a second, smaller store, and the reason is a circularity that cannot otherwise be escaped. The ring is what tells every component where a key is to be found, and the store cannot keep the ring within itself, since the ring would be needed in order to find it. The ring is therefore kept in a separate system at a fixed address that every component knows. Chapter 4 shows how a group of 3 machines agrees on a value, which is what the coordinator does.

A write to the key user-1 passes through the following steps.

0. Before any request arrives, every router holds a copy of the ring, and the coordinator informs the routers of every change to it.
1. An application sends PUT /v1/keys/user-1 to the load balancer.
2. The load balancer forwards the request to a router that passed its most recent health check.
3. The router hashes user-1, looks up the resulting position in its copy of the ring, finds that the leader of the shard is node-10, and forwards the request there.
4. node-10 appends a record to its log and waits for its disk to confirm.
5. node-10 sends the record to both followers, node-1 and node-6.
6. As soon as 1 follower has confirmed, node-10 applies the record and answers 201 Created.
7. The response returns to the application by way of the router and the load balancer.

A read follows steps 1 to 3 and is then answered by node-10 from memory, with steps 4 to 6 omitted.

2 features of this arrangement deserve to be drawn out, since they are deliberate.

The routers are stateless. A component without state can be run in any number of instances, added and removed at will, and replaced after a failure without any procedure of recovery. The design confines all lasting state to the storage nodes and the coordinator, so that those are the only components which require replication and the careful treatment of failure that occupies the rest of this chapter.

And, as step 0 indicates, the coordinator takes no part in the handling of requests. If it becomes unreachable, reads and writes continue undisturbed on the copies of the ring that the routers already hold. Only changes to the ring are suspended: a change of leader cannot be recorded and a node cannot be added until the coordinator returns. A component that every request had to consult would be both a bottleneck and a single point of failure, and keeping the coordinator off the path of requests avoids both.

Sharding

Sharding is the division of the data into parts that are stored on different nodes, each part being a shard. With sharding, the total volume of data can exceed the capacity of any 1 node, and the load of requests is distributed among all of them. The central question is by what function a key is assigned to a node, and the obvious answer turns out to have a serious flaw.

Hashing modulo the number of nodes

The obvious method hashes the key and takes the remainder on division by the number of nodes, node = hash(key) mod N. Provided that the hash function distributes its output evenly, every node receives an equal share of the keys, and any component that knows N can locate any key by a single calculation.

The flaw appears when N changes. A key stays where it is only if its hash leaves the same remainder under the old number of nodes and the new. The following table shows, for a few hash values, what happens when N increases from 4 to 5.

Of any 20 consecutive hash values, only 4 leave the same remainder modulo 4 and modulo 5. The addition of a fifth node therefore relocates 80% of all keys, although the new node has need of only 20% of them, and the other 60% merely change places among the existing nodes to nobody’s advantage. During such a relocation the cluster expends a large part of its capacity on copying data, at exactly the time when a node was added because capacity was short.

Consistent hashing

Consistent hashing is a method of assignment under which a change in the set of nodes relocates only the keys that have to move. The range of the hash function is regarded as a circle, called the ring, on which the largest value is followed by the smallest. Every node is given a position on the ring, called its token. A key is hashed to a position by the same function, and it belongs to the node whose token is the first at or after that position, proceeding clockwise.

In figure 1.8 the ring has only 100 positions, to keep the arithmetic easy. Node B has the token 37, and the token before it is 12, so that B owns the positions 13 to 37. A key that hashes to position 20 belongs to B. A key that hashes to 90 lies beyond the last token, passes the end of the range, continues from 0, and belongs to A at position 12.

The gain of the method is seen when a node is added.

Node E is given the token 50, which lies between the token of B and that of C. E takes over the positions 38 to 50, which formerly belonged to C, and every other position keeps its owner. The addition has relocated 13% of the positions, and every relocated key has gone to E. The removal of a node has the reverse effect: its positions pass to its clockwise neighbour, and no other key is touched.

The last row is the price. Under the first method a component needs to know 1 number, and under the second it needs the whole ring. How every component comes to have the ring, and what happens when its copy is out of date, is taken up below.

Virtual nodes

With a single token for each node, the size of a node’s range depends on where its own token and its predecessor’s happen to fall, and the ranges can differ enormously. Furthermore, when a node is removed its entire range passes to 1 neighbour, which then carries a double load.

Both defects are removed by giving every node many tokens. The tokens of a node are called its virtual nodes. Each node then owns many small ranges scattered around the ring. Its total share is the sum of many random range sizes, which varies far less than any single one of them, and the ranges of a removed node pass to many different neighbours.

The program ringsim measures the effect. It places 100000 keys on a ring of 24 nodes, with 1, 8, 64 and 256 tokens for each node, and reports the smallest and the largest share of keys held by any node. It then adds a node and counts the keys whose owner has changed.

$ cargo run --release -p ringsim
100000 keys on 24 nodes, ideal share 4.17%
     1 tokens per node: smallest share  0.16%, largest share 11.26%
     8 tokens per node: smallest share  1.42%, largest share  7.83%
    64 tokens per node: smallest share  3.35%, largest share  5.07%
   256 tokens per node: smallest share  3.86%, largest share  4.73%
adding node 25 to a ring with 256 tokens per node, ideal movement 4.00%
  on the ring: 4332 keys changed owner (4.33%), 4332 of them to the new node
  hash mod N : 96001 keys changed owner (96.00%)
replicas of user-1: node-10, node-1, node-6
position of user-1: 0x7c7615c11438e5c9

We use 256 tokens for each node. With 24 nodes, the ring contains 6144 tokens, and each range between adjacent tokens holds about 180 MB on average. A larger number of tokens would improve the balance further, at the cost of a larger ring to store and to distribute.

The ring uses a 64-bit hash function, xxHash3, for tokens and keys alike. The token of a virtual node is the hash of the name of its node followed by a number, as in node-7#12. Since the tokens are derived from the names, every router and node that knows the names of all the nodes computes an identical ring, and no table of tokens has to be transmitted.

The tokens are kept in a sorted map. To find the owner of a key, a lookup takes the tokens from the key’s position to the end of the map, followed by those from its beginning, which is clockwise order on the ring. The cost is logarithmic in the number of tokens.

Shards on the ring

Each range between adjacent tokens is a shard. The replicas of a shard are placed on the first 3 distinct nodes encountered going clockwise from its range. The node that owns the range is the preferred leader of the shard, and the other 2 are its followers. For the key user-1, ringsim reports node-10, node-1 and node-6, and these 3 nodes serve as the example in every chapter that follows.

The word “distinct” matters. Consecutive tokens on the ring may belong to the same physical node, since each node has 256 of them, and 2 replicas placed on 1 machine would give no protection against the loss of that machine. Tokens of a node already chosen are therefore passed over.

The ring in a running system

The ring has so far been described as a data structure. It remains to describe how the components of a running cluster obtain it, use it and change it, and it is here that most of the practical difficulty lies.

Where the ring is kept. There is a single authoritative copy, the Ring message held by the coordinator. It contains the name, address and tokens of every node, and the present leader of every shard. Every change increases its version number by 1. Routers and storage nodes hold a copy in memory and keep it current by means of a watch.

How a request is routed. Routing requires no communication with the coordinator or with other routers. A router hashes the key and performs a binary search in its sorted list of tokens, which with 6144 tokens takes 13 comparisons.

What happens when a ring is out of date. A router may forward a request an instant before a new version of the ring reaches it, and the request then arrives at a node that is no longer responsible for the key. If that node were to answer, it might return a stale value or accept a write that belongs elsewhere, and strong consistency would be lost through nothing more than a delayed notification. Routers therefore attach the version of their ring to every forwarded request, in a Ring-Version header. A node that holds a newer ring and is not responsible for the key refuses the request with the status 421 Misdirected Request, which HTTP defines for a request sent to a server that cannot answer for the resource. On receiving it, the router waits for the newer ring and forwards again, and the client observes nothing but a slightly slower response. The principle, that a component acting on possibly stale information should declare which version it is acting on so that the recipient can check, reappears in this book wherever stale knowledge is dangerous.

How a node joins. A new node cannot assume ownership of its ranges at once, since it does not yet hold their data. Ownership is therefore transferred range by range, each after its data has been copied.

How a node leaves. When an operator removes a node, the procedure runs in reverse. The departing node copies each of its ranges to the node that will own it, and the coordinator reassigns a range when its copy is complete. When a node fails without warning, there is no opportunity to copy. Its ranges are still held by the other 2 replicas of each shard, the shards it led elect new leaders, and a replacement node is supplied with the data by the surviving replicas.

Gossip as an alternative. Some systems distribute the ring without any coordinator. Each node holds a copy, and the nodes exchange their copies in pairs, each repeatedly choosing another at random and comparing versions, so that a change spreads through the cluster as a rumour does through a crowd.

A gossip protocol removes the dependence on a coordinator and scales to very large clusters, since no node has to communicate with more than a few others. Its disadvantage is that for some seconds after a change, different nodes hold different versions of the ring and may direct the same key to different nodes. Designs founded on gossip have to tolerate such disagreement, and accordingly tend to be those that offer eventual consistency. We use a coordinator, since a single authoritative ring is far easier to reason about when strong consistency is the aim, and a cluster of 24 nodes has no need of the scalability of gossip.

The cost of sharding

An operation on a single key involves a single shard. An operation on many keys may involve many. Listing keys by prefix, functional requirement 4, is of the second kind, because hashing deliberately destroys the order of keys: keys with a common prefix hash to unrelated positions and are scattered over all the nodes. A router has to send such a request to every node and merge their sorted results, so that a listing takes time proportional to the number of nodes.

This is the general price of hashing, and systems whose work consists largely of range queries choose differently. They divide the key space into contiguous ranges of keys, without hashing, so that neighbouring keys remain on the same node and a range query touches few nodes. In exchange they must actively move the boundaries of the ranges in order to keep the load even, since nothing in an unhashed key space guarantees that keys are evenly spread.

Replication

Replication is the keeping of copies of the same data on several nodes, each copy being a replica. Non-functional requirement 0 lays down that an acknowledged write survives the loss of any 1 machine, from which it follows that a write must be stored on at least 2 machines before it is acknowledged.

Replication with a leader

If every replica accepted writes independently, 2 clients might write different values for the same key to 2 replicas at the same moment, and the replicas would then hold conflicting data, to be reconciled afterwards by some rule that is bound to discard one client’s work. This design avoids such conflicts by appointing 1 replica of every shard as its leader. All writes to the shard go to the leader, which assigns each the next sequence number, so that the writes have a single agreed order, and sends the resulting record to the other replicas, the followers. A follower applies the records in the order of their sequence numbers and thereby arrives at the leader’s state.

Acknowledgement by a majority

The leader acknowledges a write as soon as its record is stored in 2 of the 3 logs, its own and that of 1 follower. It does not wait for the second follower.

The rule has 2 consequences, both of them wanted. The record is on 2 machines at the moment the client receives its acknowledgement, as non-functional requirement 0 demands. And the response time of a write is governed by the faster of the 2 followers, so that a follower which is slow, restarting or unreachable delays nothing, which is what keeps the p99 latency within the bound of non-functional requirement 3.

Any set containing more than half of the replicas is called a majority. With 3 replicas, a majority consists of 2. The whole importance of majorities derives from a single property: any 2 majorities of the same shard have at least 1 node in common.

To see what this property accomplishes, consider what happens when the leader of a shard fails. Its followers elect a new leader, and a candidate requires the votes of a majority. The majority that stored record 42 and the majority that elects the new leader share at least 1 node, and that node holds record 42. By the rule of the election, a node votes only for a candidate whose log is at least as complete as its own. The node holding record 42 therefore refuses its vote to any candidate that lacks it, and such a candidate cannot assemble a majority. Every newly elected leader consequently holds every acknowledged record, and no acknowledged write is lost in a change of leader. Chapter 4 implements this argument and tests it.

The table shows that an even number of replicas tolerates no more failures than the odd number below it, while costing more. We therefore use 3 replicas. A deployment that must survive 2 simultaneous failures uses 5.

Asynchronous replication

In asynchronous replication, the leader acknowledges a write as soon as its own log holds the record and sends it to the followers afterwards. Writes are faster, since they wait for 1 disk and for no network round trip. But if the leader’s machine is lost immediately after an acknowledgement, no other machine has the record, and a write that was acknowledged is gone. Asynchronous replication is therefore incompatible with non-functional requirement 0, and this design uses acknowledgement by a majority. Later chapters adopt the asynchronous form where its risk is acceptable, as in the cache of chapter 9, and reject it where it is not, as in the lock service of chapter 10.

Failure handling

A distributed system is defined as much by what it does when its parts fail as by what it does when they work. This section takes each kind of failure in turn and states how it is detected, what clients observe, and how the system recovers.

Failure of a leader

A leader sends an Append call to each of its followers every 100 ms. When there are no new records, the call is empty and is known as a heartbeat, a periodic message whose arrival shows that its sender is running. A follower that receives no Append for 1 second concludes that its leader has failed.

Every election opens a new term, identified by a number that increases by 1. Terms are how nodes tell the present leader from a former one. In figure 1.16, node-10 still regards itself as the leader of term 4 when it returns. The first message it receives carries term 5. Since that is higher than its own, node-10 gives up its leadership and continues as a follower. Every Append call carries the term of its sender so that the receiver can make this comparison, and a returning leader that has been superseded is thereby prevented from doing any harm.

Network partition

A network partition is a failure of the network that divides the nodes into groups which cannot communicate with each other, while every node continues to run. Seen from inside one group, a partition is indistinguishable from the failure of all the nodes in the other, and no node can discover which has occurred.

In figure 1.17, node-10 is isolated and still regards itself as the leader. It accepts a write and appends a record to its own log, but it cannot send the record to a follower. Since no majority holds the record, node-10 does not acknowledge, and its client receives an error. Meanwhile 2 of the 3 replicas are in the other group, which elects a new leader and continues to process writes.

The rule of the majority thus resolves, without any further mechanism, the danger that a partition presents, namely 2 leaders each accepting writes that the other knows nothing of. At most 1 group can contain a majority of a shard’s replicas, so at most 1 leader can acknowledge writes, and no node is required to determine why it has lost contact with the others.

Summary of failures

The last row marks the limit of the design. With 3 replicas, the store tolerates the loss of any 1 machine and remains consistent under any partition of the network. A shard that has lost a majority of its replicas is unavailable, and has to be so, since the single surviving replica cannot know whether it holds every acknowledged write.

Guarantees and trade-offs

During a network partition, a distributed store cannot offer both consistency and availability. It must either refuse some requests in order to keep all its answers consistent, or accept all requests and allow the 2 groups to diverge. This result is known as the CAP theorem, and it is less a theorem to be proved than an observation that follows directly from the previous section: a node that cannot reach the others must choose between answering from what it has, which may be out of date, and declining to answer. This design gives priority to consistency.

The availability allowed by non-functional requirement 1 can be related to elections by a short calculation. An allowance of 21.6 minutes a month is 1296 seconds. If an election takes 5 seconds and the failed node led 1/24 of all the shards, the election causes 1/24 of all writes to fail for 5 seconds and consumes about 0.2 seconds of the allowance. Several thousand such failures a month would be within it. In practice, therefore, the requirement is not threatened by the failures of individual machines, which this design absorbs, and is limited by events that affect many nodes together, such as a faulty release of the software or the loss of a segment of the network.

The design, deployed

The figure of the high-level design showed 1 box for each kind of component. A running system has many of each, and their number and placement are part of the design. Applications reach the store through a pair of load balancers, of which 1 is active and the other takes over its address if it fails. Behind them are 12 routers, which can be any number, since they are stateless. The 24 storage nodes are divided equally among 3 availability zones, which are separate buildings or halls with their own power and network, and the ring is constructed so that the 3 replicas of every shard fall in 3 different zones. The loss of an entire zone then removes 1 replica of each shard and leaves a majority of every one. The coordinator has 1 machine in each zone for the same reason.

How real systems do it

The design of this chapter is not the only way of building a key-value store, and it is instructive to place it among the systems that are in use.

Its 2 ancestors are quite different from each other. Dynamo, described by Amazon in 2007, introduced to a wide audience the combination of consistent hashing, virtual nodes and replication to the next nodes on the ring that this chapter uses for placement. In every other respect Dynamo chose the opposite course. It has no leaders. Any replica accepts a write, a request succeeds when W of N replicas have answered, conflicting versions are detected with vector clocks and handed back to the application to be reconciled, and membership spreads by gossip. It is the model of eventual and tunable consistency described earlier, and Cassandra and Riak descend from it. The other ancestor is the line of consensus protocols, Paxos and later Raft, in which a leader orders the writes and a majority must store each one. The coordinator of this chapter is a system of that kind, as etcd and ZooKeeper are.

The design here combines the 2: placement by a hash ring, as in Dynamo, with a consensus group for each shard, as in Raft. Systems that are built this way in practice, among them CockroachDB, TiKV and YugabyteDB, differ from our design in 1 important respect. They do not hash keys. They divide the ordered key space into contiguous ranges, each replicated by its own Raft group, and they split a range when it grows beyond a set size and merge ranges that shrink. This keeps neighbouring keys together, so that a listing by prefix reads 1 or 2 ranges where ours must ask every node, and it lets the system move load by moving the boundaries. The price is the machinery for splitting, merging and rebalancing, which a hash ring does not need. Google’s Bigtable and Spanner also partition by range.

Several things that a production store requires are absent from this chapter and from the code that follows. A log that grows for ever must be cut short by taking snapshots of the state. Changing the set of members of a shard while it runs needs a protocol of its own, which Raft calls joint consensus. Reads served by a leader on the strength of a recent heartbeat depend on clocks behaving reasonably, and stricter systems confirm leadership with a majority for each read. And none of this has been tested here in the way that such systems are tested in earnest, by a tool such as Jepsen, which partitions a real cluster at random and checks every response against the definition of linearizability.

Mistakes to avoid

A first look at the low-level design

Every guarantee stated above rests on an assumption that has so far been taken for granted: that a record which a node has reported as stored is still in that node’s log after the node has crashed and restarted. If a node could lose such a record, a record counted as stored on 2 of 3 nodes might in truth be stored on 1, and the whole argument from majorities would be without foundation. A storage node justifies the assumption by observing 2 rules.

1. Every change is appended to a log on disk, and the node waits for the disk to confirm, before the change is applied in memory and before any response is sent.
2. On starting, a node rebuilds its state in memory by reading its log from beginning to end.

Chapter 2 implements a storage node on these 2 rules: a format for records on disk, a log with its procedure of recovery, and an HTTP server that provides the API of this chapter. It includes a simulation that cuts a log short at every possible byte and verifies that recovery returns exactly the records that were complete. The entities of this chapter appear in its code under the same names.

The simulation of the ring

References

• G. DeCandia and others, “Dynamo: Amazon’s Highly Available Key-value Store”, SOSP 2007.
• D. Ongaro and J. Ousterhout, “In Search of an Understandable Consensus Algorithm”, USENIX ATC 2014.
• D. Karger and others, “Consistent Hashing and Random Trees”, STOC 1997.
• F. Chang and others, “Bigtable: A Distributed Storage System for Structured Data”, OSDI 2006.
• J. Corbett and others, “Spanner: Google’s Globally-Distributed Database”, OSDI 2012.
• M. Herlihy and J. Wing, “Linearizability: A Correctness Condition for Concurrent Objects”, ACM TOPLAS, 1990.
• S. Gilbert and N. Lynch, “Brewer’s Conjecture and the Feasibility of Consistent, Available, Partition-Tolerant Web Services”, ACM SIGACT News, 2002.
• M. Kleppmann, Designing Data-Intensive Applications, O’Reilly, 2017.

Terms introduced in this chapter