Amazon Dynamo and Cassandra: the leaderless KV store

01 · CONTEXT

The problem: never reject a write

The platform described in the paper is an online store with hundreds of services, running on tens of thousands of servers spread across several data centers. At this scale, there are always server and network components failing, and the design treats failure as the normal case, not the exception. The example that guides everything is the shopping cart: the service must always be able to write and read the cart, even with disks failing or a data center down.

Most of these services only write and fetch data by primary key and do not need the complex queries of a relational database. For them, the relational database was an inefficient solution: the extra functionality requires expensive hardware and highly skilled staff, the available replication technologies tend to choose consistency over availability and scaling out with smart partitioning was not simple. The paper adds that, in the company's experience, stores with ACID guarantees tend to have poor availability.

The design choice follows from this. In the face of network failures, strong consistency and high availability cannot be achieved at the same time. Traditional stores may reject writes when they cannot reach all replicas or a majority of them. Dynamo, by contrast, chooses to be eventually consistent: updates reach all replicas, but only over time. It wants to be always writable, so it pushes conflict resolution to read time, so that no write is rejected. This raises two questions, when to resolve and who resolves, which sections 04 onward answer.

The contract is deliberately narrow. The interface is by primary key only, objects are small (generally under 1 MB) and stored as blobs, there are no isolation guarantees, and only one key is updated at a time. The environment is assumed to be trusted, with no authentication or authorization, and each service runs its own instance, with an initial design for up to hundreds of hosts.

Latency is also part of the contract. Instead of the mean or median, Amazon's service level agreements are measured at the 99.9th percentile, because customers with more history require more processing and sit at the high end of the distribution, which an average would hide. A store page may depend on more than 150 services, so each dependency has to meet its limit. A typical SLA for Dynamo users is that 99.9% of reads and writes finish within 300 ms.

02 · PARTITIONING AND REPLICATION

The ring and the preference list

The first technique is particionamento via consistent hashing. The output of the hash function is treated as a fixed circle, the ring, in which the largest value wraps around to the smallest. Dynamo applies MD5 to the key, obtaining a 128-bit identifier, and each node is assigned a position in that space. To find a key's owner, you compute its position and walk clockwise to the first node with a larger position. This way, each node is responsible for the region between itself and its predecessor, and a node joining or leaving affects only its immediate neighbors.

Basic consistent hashing has two problems: each node's random position produces uneven distribution of data and load, and the algorithm ignores performance differences between machines. Dynamo answers with virtual nodes: instead of one point on the circle, each server gets several, called tokens. If a node goes down or goes into maintenance, its load spreads evenly across the others; when it comes back, it takes over a roughly equal share from each of the others. And the number of virtual nodes for each machine can follow its capacity. This is what allows growing one node at a time.

For durability, each item is copied to N nodes, with N configured per instance. The node that owns the key, the coordinator, writes locally and replicates to the N-1 clockwise successors. The list of nodes responsible for a key is the preference list. It has more than N nodes, to absorb failures, and is built by skipping ring positions until it contains only distinct physical machines, since a node with several tokens could occupy more than one position. Any node can compute the list for any key, and the list is built to spread the replicas across several data centers.

How does the request reach the coordinator? The client can go through a generic load balancer that picks a node based on load information, or use a client library that knows the partitions and talks directly to the coordinator. In the first case, the request can land on any node in the ring; if that node is not among the first in the key's preference list, it forwards the request to the first of them. To keep latency low at the high percentiles, Dynamo avoids multi-hop routing: each node holds enough information to route directly, which the paper calls a zero-hop DHT.

1. The cart service calls put; the load balancer routes the request to any node in the ring, here node A.

2. The key's hash becomes a position on the ring. Walking clockwise, the first node with a larger position is B, the key's owner.

3. Node A is not among the first in this key's preference list, so it forwards the request to the coordinator B.

4. B writes the key locally and replicates it to the clockwise successors, C and D, forming the preference list of the size N chosen for the instance.

03 · TUNABLE CONSISTENCY AND TEMPORARY FAILURES

R, W, and the sloppy quorum

To keep replicas coherent, Dynamo uses a protocol similar to that of quorum systems with two configurable values: R is the minimum number of nodes that must participate in a successful read and W is the minimum for a write. With R + W > N, you get a quorum-like system. An operation's latency is dictated by the slowest of the R (or W) replicas consulted; that is why R and W are usually smaller than N, in exchange for better latency.

On the write path, the coordinator generates the version clock, writes the new version locally, and sends it to the N highest-ranked reachable nodes in the list; if at least W-1 respond, the write is considered successful. On a read, the coordinator requests all existing versions from those nodes, waits for R responses, and returns the result; if there are versions with no causal relationship, it returns all of them.

If it used a traditional quorum, Dynamo would become unavailable during server failures and network partitions. That is why it adopts a sloppy quorum: reads and writes use the first N healthy nodes in the preference list, which may not be the first N found walking the ring. If a preferred node is down, the replica that would go to it goes to another node and carries a hint in its metadata about the original recipient. Nodes that receive hinted replicas keep them in a separate local database, scanned periodically; when they detect that the recipient has recovered, they try to deliver the replica and, if they succeed, can delete the copy without reducing the total number of replicas.

1. During a write, node C is unreachable. With a traditional quorum the operation would be unavailable; Dynamo does not require all the preferred nodes to be reachable.

2. B uses the first healthy nodes in the preference list, skipping the one that is down; the replica that would go to C goes to E, the next in the list after D.

3. The replica delivered to E carries a hint in its metadata that the original recipient was C. E keeps it in a separate local database, scanned periodically.

4. When E detects that C has recovered, it tries to deliver the replica to C. Once delivery completes, E can delete the copy without reducing the total number of replicas.

There are caveats. Hinted handoff works best when membership churn is low and failures are transient; hinted replicas can be lost before returning to the owner, and background synchronization exists for that, covered in section 05. And, in practice, most services in production use W greater than 1, to achieve the desired durability. The common (N, R, W) configuration across several instances is (3, 2, 2).

04 · VERSIONS AND CONFLICTS

Vector clocks and the cart

Since writes are never rejected, Dynamo treats each modification as a new immutable version of the data and allows several versions to exist at the same time. Most of the time the new version supersedes the previous ones and the system itself decides which one is authoritative, which the paper calls syntactic reconciliation. When failures combine with concurrent updates, versions branch and the system cannot collapse them; the client then has to do semantic reconciliation. The typical example is merging cart versions: that way, an add-to-cart is never lost, but removed items may reappear.

To tell whether two versions are on parallel branches or have a causal order, Dynamo uses vector clocks. A vector clock is a list of (node, counter) pairs, associated with each version of each object. If the counters of the first clock are less than or equal to those of the second, the first is an ancestor of the second and can be forgotten; otherwise, the changes conflict and require reconciliation. When updating an object, the client specifies which version it is updating, passing along the context obtained from a previous read.

The paper's example shows the mechanism. A node Sx handles the first write and creates version D1; the second, handled by the same node, creates D2. Then the same client updates the object and another server, Sy, handles the write, creating D3. Next, a different client reads D2 and a third server, Sz, writes D4. A node that knows D3 and receives D4 sees there is no causal relationship between them, and both versions must be kept and presented to the client, on a read, for reconciliation.

When a read finds branches that cannot be reconciled syntactically, the coordinator returns the leaf objects, each with its version information in the context. An update made with that context is taken to be the reconciliation, and the branches merge into a new version. If the latest version of the cart is not available, the item is applied to the oldest version and the divergent versions are reconciled later.

1. The cart service calls get. The coordinator requests all existing versions from the reachable nodes in the preference list.

2. The coordinator waits for only R responses. Replicas X and Y return versions with no causal relationship, and it returns both to the service, with the context.

3. The service merges the versions, which for the cart means taking the union of the items, and writes back passing the context; the write becomes a new version that joins the branches.

4. After responding, the coordinator waits a short time for the pending responses. If one brought an old version, such as replica Z, it updates it: this is read repair.

5. In the background, replicas compare Merkle trees per key range and synchronize only the keys that are out of sync.

Who resolves conflicts can be the store or the application. The store only has simple policies, such as last write wins; the application knows the schema and chooses the method that best serves the client, such as merging carts. The paper describes usage patterns: business-logic-specific reconciliation, which is a popular case, as in the cart; timestamp-based reconciliation, where the object with the highest physical timestamp wins, used by the session service; and a high-performance read engine, where R is typically 1 and W equals N.

How often do versions diverge? In the cart service, over 24 hours measured in production, 99.94% of requests saw exactly one version; 0.00057% saw two, 0.00047% saw three, and 0.00009% saw four. Experience shows that the increase in divergent versions comes from an increase in concurrent writers, generally robots and rarely people, and not from failures.

05 · SYNCHRONIZATION AND MEMBERSHIP

Repair, anti-entropy, and gossip

There are two ways to fix stale replicas. The first is read repair: after responding, the coordinator waits a bit for the pending responses and, if any came back with an old version, updates those nodes with the latest one, which relieves the anti-entropy of that task. The second is anti-entropy, a replica synchronization protocol that covers durability threats such as hinted replicas that were lost before returning to the owner.

To find divergences quickly and transfer little data, anti-entropy uses Merkle trees: a hash tree in which the leaves are hashes of individual key values and each parent node is the hash of its children. Each branch can be checked independently, without downloading the tree or the entire dataset. Each node keeps one tree per key range it hosts; two nodes exchange the root of their common ranges and, if the roots are equal, nothing needs to be synchronized. If they differ, they descend exchanging hashes of the children down to the leaves and identify the keys that are out of sync. The cost is that, when a node joins or leaves, many ranges change and the trees have to be recomputed.

Ring membership is not detected automatically. A node going down rarely means a permanent departure and should not cause rebalancing or repair of the replicas, and a manual error could start new nodes by mistake. That is why joining and leaving nodes are explicit operations, done by an administrator. A gossip protocol propagates the changes and maintains an eventually consistent view of membership: every second, each node contacts a randomly chosen peer and the two reconcile their histories. Token mappings travel over the same channel, so each node knows its peers' ranges and routes requests directly. To avoid logical partitions of the ring, some nodes act as seeds, known to everyone.

Failure detection is local: a node considers another failed if it does not respond to its messages, even if it responds to others. With constant request traffic, the node quickly notices the failure, uses alternate nodes for the suspect's partitions, and retries periodically. Early versions used a decentralized detector to maintain a global view of failure state, but it was concluded that the explicit join and leave methods make that view unnecessary.

The partitioning scheme evolved in production. The first strategy, the one initially deployed, gave T random tokens to each node and, since they were chosen at random, the ranges varied in size. It brought problems: during peak season, bootstrapping a node could take almost a day; when a node joins or leaves, the Merkle trees of the affected ranges have to be recomputed; and there was no easy way to take a snapshot of the key space, which complicated archiving.

The second strategy divided the hash space into Q equal partitions, keeping T random tokens per node, and the tokens came to serve only to map positions to node lists. The advantage was decoupling partitioning from partition placement. It was only a provisional configuration in the migration from the first to the third.

The third strategy also uses equal-size partitions, but with Q/S tokens per node, where S is the number of nodes. Compared with the first, it had better load-balancing efficiency and reduced by three orders of magnitude the metadata kept at each node. Since the ranges are fixed, a partition can be moved as a unit, by transferring the file, which simplifies bootstrapping and recovery. The downside is that changing membership requires coordination to preserve the properties of the assignment.

06 · RESULTS, LIMITS, AND CASSANDRA

What was learned and who inherited the idea

In production use, the cart service served tens of millions of requests, with well over 3 million checkouts in a single day, and the platform scaled to extreme peak loads without downtime during the holiday season. At the time of the paper, several internal services had been using Dynamo for about two years; applications received successful responses, with no timeouts, for 99.9995% of requests, and no data loss event had occurred.

Performance was also addressed. Latency at the 99.9th percentile is around 200 ms, an order of magnitude above the averages, and depends on load, object sizes, and locality. For services that needed more, each node can keep an in-memory object buffer, periodically written out by a thread. Even with a buffer of a thousand objects, 99.9th-percentile latency dropped by a factor of 5 at peak. The scheme trades durability for performance: a server crash can lose writes still in the buffer. To reduce the risk, the coordinator picks one of the N replicas to perform a durable write, and this does not affect performance because it waits for only W responses.

Another optimization picks as write coordinator the node that replied fastest to the previous read, which reduces variability and improves the 99.9th percentile. And there is an alternative to server-side coordination: move the state machine to the clients. The client periodically downloads the membership view from a random node, and in the 24-hour measurement this cut 99.9th-percentile latency by at least 30 ms and the average by 3 to 4 ms. The cost is the freshness of that view: clients poll a random node every 10 seconds and, in the worst case, see stale membership for that long.

The paper also points out limits. Dynamo adopts a full-membership model, in which every node knows the data hosted by its peers. This works for a few hundred nodes, but scaling to tens of thousands is not trivial, because the cost of maintaining the routing table grows with the size of the system; hierarchical extensions could overcome the limitation. For new applications, it is necessary to analyze early which conflict-resolution mechanism fits the business case.

Facebook's Cassandra is described in another paper, which presents it as a synthesis of known techniques and borrows the term preference list from Dynamo. It was designed for inbox search, a problem with very high write throughput, on the order of billions of writes per day, in which replicating across data centers was essential to keep search latency low. It inherits the ring but diverges on important points, summarized below.