Tag: leaderless replication

DDIA | Replication | Leaderless Replication | Sloppy Quorums | Hinted Handoff | Multi-datacenter operations

October 28th, 2023
In previous posts, we looked into leaderless replications and quorums. In this post we continue learning about them.

Quorums are not as fault-tolerant as they could be. A network interruption can easily cut off a client from a large number of database nodes.

Although those nodes are alive, and other clients may be able to connect to them, to a client that is cut off from the database nodes, they might as well be dead. In this situation, it’s likely that fewer than w or r reachable nodes remain, so the client can no longer reach a quorum.

In a large cluster (with significantly more than n nodes) it’s likely that the client can connect to some database nodes during the network interruption, just not to the nodes that it needs to assemble a quorum for a particular value. In that case, database designers face a trade-off:
- Is it better to return errors to all requests for which we cannot reach a quorum of w or r nodes?
- Or should we accept writes anyway, and write them to some nodes that are reachable but aren’t among the n nodes on which the value usually lives?
Sloppy Quorum – writes and reads still require w and r successful responses, but those may include nodes that are not among the designated n “home” nodes for a value.

Hinted Handoff – Once the network interruption is fixed, any writes that one node temporarily accepted on behalf of another node are sent to the appropriate “home” nodes. This is called hinted handoff.

Sloppy Quorums only assures that writes are successful given there are w nodes available somewhere. Reads could still be returning stale value until hinted handoff completes.

Sloppy quorums are optional in dynamo implementations, for Riak they are turned on.

Multi-datacenter operation

Leaderless replication is also suitable for multi-datacenter operation, since it is designed to tolerate conflicting concurrent writes, network interruptions, and latency spikes.

Cassandra and Voldemort – the number of replicas n includes nodes in all datacenters, and in the configuration you can specify how many of the n replicas you want to have in each datacenter. Each write from a client is sent to all replicas, regardless of datacenter, but the client usually only waits for acknowledgment from a quorum of nodes within its local datacenter so that it is unaffected by delays and interruptions on the cross-datacenter link. The higher-latency writes to other datacenters are often configured to happen asynchronously, although there is some flexibility in the configuration

Detecting Concurrent Writes

Dynamo-style databases allow several clients to concurrently write to the same key, which means that conflicts will occur even if strict quorums are used. Conflicts also occur due to read repair and hinted handoffs.

Application developers have to make sure these conflicts are resolved as databases don’t do that automatically.
1. Last write wins (discarding concurrent writes) – One approach for achieving eventual convergence is to declare that each replica need only store the most “recent” value and allow “older” values to be overwritten and discarded. For example, we can attach a timestamp to each write, pick the biggest timestamp as the most “recent,” and discard any writes with an earlier timestamp. This conflict resolution algorithm, called last write wins (LWW). LWW achieves the goal of eventual convergence, but at the cost of durability: if there are several concurrent writes to the same key, even if they were all reported as successful to the client (because they were written to w replicas), only one of the writes will survive and the others will be silently discarded.
2. The “happens-before” relationship and concurrency – whenever you have two operations A and B, there are three possibilities: either A happened before B, or B happened before A, or A and B are concurrent. What we need is an algorithm to tell us whether two operations are concurrent or not. If one operation happened before another, the later operation should overwrite the earlier operation, but if the operations are concurrent, we have a conflict that needs to be resolved.
  
  The algorithm works as follows:
  - The server maintains a version number for every key, increments the version number every time that key is written, and stores the new version number along with the value written.
  - When a client reads a key, the server returns all values that have not been overwritten, as well as the latest version number. A client must read a key before writing.
  - When a client writes a key, it must include the version number from the prior read, and it must merge together all values that it received in the prior read. (The response from a write request can be like a read, returning all current values, which allows us to chain several writes like in the shopping cart example.)
  - When the server receives a write with a particular version number, it can overwrite all values with that version number or below (since it knows that they have been merged into the new value), but it must keep all values with a higher version number (because those values are concurrent with the incoming write).
  - If you make a write without including a version number, it is concurrent with all other writes, so it will not overwrite anything—it will just be returned as one of the values on subsequent reads.
  - If several operations happen concurrently, clients have to clean up afterward by merging the concurrently written values, also called sibling values. Merging sibling values is essentially the same problem as conflict resolution in multi-leader replication. A simple approach is to just pick one of the values based on a version number or timestamp (last write wins), but that implies losing data. Another option is to have a union of the two values, but that is also error prone. We will discuss later approaches for handling these sibling values.
3. Version Vectors – When there are multiple replicas, we need to use a version number per replica as well as per key. Each replica increments its own version number when processing a write, and also keeps track of the version numbers it has seen from each of the other replicas. This information indicates which values to overwrite and which values to keep as siblings. The collection of version numbers from all the replicas is called a version vector.
Thanks for stopping by! Hope this gives you a brief overview in to sloppy quorums, hinted hand-offs, multi-data center operations and detecting concurrent writes when we are using leaderless replication. Eager to hear your thoughts and chat, please leave comments below and we can discuss.
DDIA | Replication | Leaderless Replication

October 21st, 2023
So far in previous posts we have read about leader based replication. There is another way of handling replication and that is leaderless. It didn’t become popular until Amazon used it for DynamoDb.

As the name suggest there is no leader which accepts writes, instead client can send write requests to multiple replicas. In some leaderless implementations, the client directly sends its writes to several replicas, while in others, a coordinator node does this on behalf of the client.

Not only writes, but the read requests are also send to all replicas in parallel. The client may get different responses from different nodes; i.e., the up-to-date value from one node and a stale value from another. Version numbers are used to determine which value is newer .

Let’s say we have three nodes, and one node went down. Meanwhile we get a write request, this requests gets written to the two healthy replicas. Now the down replica comes back. How do we ensure that it gets the latest writes from other nodes?

Two mechanisms are often used in Dynamo-style datastores:
1. Read repair – When a client makes a read from several nodes in parallel, it can detect any stale responses. And then can write the new data to the node from which it gets stale response. This approach works well for values that are frequently read.
2. Anti-entropy process – In addition, some datastores have a background process that constantly looks for differences in the data between replicas and copies any missing data from one replica to another. Unlike the replication log in leader-based replication, this anti-entropy process does not copy writes in any particular order, and there may be a significant delay before data is copied.
Quorums for reading and writing

If there are n replicas, every write must be confirmed by w nodes to be considered successful, and we must query at least r nodes for each read. As long as w + r > n, we expect to get an up-to-date value when reading, because at least one of the r nodes we’re reading from must be up to date.

In Dynamo-style databases, the parameters n, w, and r are typically configurable. A common choice is to make n an odd number (typically 3 or 5) and to set w = r = (n + 1) / 2 (rounded up).

The quorum condition, w + r > n, allows the system to tolerate unavailable nodes as follows:
- - If w < n, we can still process writes if a node is unavailable.
  - If r < n, we can still process reads if a node is unavailable.
  - With n = 3, w = 2, r = 2 we can tolerate one unavailable node.
  - With n = 5, w = 3, r = 3 we can tolerate two unavailable nodes.
  - Normally, reads and writes are always sent to all n replicas in parallel. The parameters w and r determine how many nodes we wait for—i.e., how many of the n nodes need to report success before we consider the read or write to be successful.
Limitations –

With a smaller w and r you are more likely to read stale values, because it’s more likely that your read didn’t include the node with the latest value.

However, even with w + r > n, there are likely to be edge cases where stale values are returned. These depend on the implementation, but possible scenarios include:
- If a sloppy quorum is used, the w writes may end up on different nodes than the r reads, so there is no longer a guaranteed overlap between the r nodes and the w nodes.
- If two writes occur concurrently, it is not clear which one happened first. In this case, the only safe solution is to merge the concurrent writes. If a winner is picked based on a timestamp (last write wins), writes can be lost due to clock skew.
- If a write happens concurrently with a read, the write may be reflected on only some of the replicas. In this case, it’s undetermined whether the read returns the old or the new value.
- If a write succeeded on some replicas but failed on others, and overall succeeded on fewer than w replicas, it is not rolled back on the replicas where it succeeded. This means that if a write was reported as failed, subsequent reads may or may not return the value from that write.
- If a node carrying a new value fails, and its data is restored from a replica carrying an old value, the number of replicas storing the new value may fall below w, breaking the quorum condition.
- Even if everything is working correctly, there are edge cases in which you can get unlucky with the timing.
Thanks for stopping by! Hope this gives you a brief overview in to leaderless replication. Eager to hear your thoughts and chat, please leave comments below and we can discuss.

Tag: leaderless replication

DDIA | Replication | Leaderless Replication | Sloppy Quorums | Hinted Handoff | Multi-datacenter operations

Multi-datacenter operation

Detecting Concurrent Writes

DDIA | Replication | Leaderless Replication

Quorums for reading and writing