Replication and Consensus in Distributed Systems

If all you need is to scale your application to higher load, the simplest approach is to buy a more powerful machine (AKA vertical scaling or scaling up). One approach is called shared-memory architecture, where many CPUs, many RAM chips, and many disks can be joined together under one OS, and a fast interconnect allows any CPU to access any part of the memory or disk.

The problem with a shared-memory approach is that the cost grows rapidly: a machine with twice as many CPUs, RAM, and disk capacity typically costs significantly more than twice as much. And such a machine can’t necessarily handle twice the load.

Another approach is the shared-disk architecture, which uses several machines which are connected via a fast network with independent CPUs and RAM, but stores data on an array of shared disks. This architecture is used for some data warehousing workloads, but contention and overhead limits their scalability.

Shared-nothing architecture

In contrast, in a shared-nothing architectures (AKA called horizontal scaling or scaling out), each VM running the database software is called a node. Each node uses its CPUs, RAM, and disks independently. Any coordination between nodes is done at the software level using a conventional network.

No special hardware is required by a shared-nothing system, so you can use whatever machines have the best price/performance ratio. You can potentially distribute data across multiple geographic regions, and thus reduce latency for users and potentially be able to survive the loss of an entire datacenter.

While a distributed shared-nothing architecture has many advantages, it usually also incurs additional complexity for applications and sometimes limits the expressiveness of the data models you can use. There are two common ways data is distributed across multiple nodes:

1. Replication
2. Partitioning

Replication

Replication means keeping a copy of the same data on multiple machines that are connected via a network. There are several reasons why you would do this:

• To keep data geographically close to your users (to reduce latency)
• To allow the system to continue working even if some of its parts have failed (increase availability)
• To scale out the number of machines that can serve read queries (increase read throughput)

All of the difficulty in replication lies in handling changes to replicated data. There are 3 popular algorithms for replicating changes between nodes:

1. Single-leader
2. Multi-leader
3. Leaderless

Each node that stores a copy of the database is called a replica. The obvious question is: how do we ensure that all the data ends up on all the replicas? Every write to the database would need to be processed by every replica. The most common solution for this is called leader-based replication.

When clients want to write to the database, they must send their requests to the leader, which first writes the new data to its local storage. Then it sends the data change to all of its followers as part of a replication log or change stream. Each follower takes the log and updates its local copy of the database accordingly.

When a client wants to read from the database, it can query either the leader or any of the followers. However, writes are only accepted on the leader.

Synchronous vs Async Replication

Should replication happen synchronously or asynchronously? Consider the example, where 1 follower is synchronous and another is async:

The leader waits until follower 1 has confirmed that it received the write before reporting success to the user, and before making the write visible to other clients. But it doesn’t wait for a response from follower 2. There is a substantial delay before follower 2 processes the message.

The advantage of async replication is that the follower is guaranteed to have an up-to-date copy of the data that is consistent with the leader. If the leader suddenly fails, we can be sure that the data is still available on the follower. The disadvantage is that if the synchronous follower doesn’t respond (because it crashed), the write cannot be processed. The leader must block all writes and wait until the synchronous replica is available.

For that reason, in practice, only one of the followers is made synchronous, and the others are async. If the synchronous follower becomes unavailable or slow, one of the async followers is made synchronous. This guarantees that you have an up-to-date copy of the data on at least 2 nodes (this configuration is AKA semi-synchronous).

However, leader-based replication is often configured to be completely async. In this case, if the leader fails and is not recoverable, any writes that have not yet been replicated to followers are lost.

Adding new followers

From time to time, you need to set up new followers (to increase the no. of replicas, or to replace failed nodes). How do you ensure that the new follower has an accurate copy of the leader’s data?

Simply copying data files from the leader is typically not sufficient: clients are constantly writing to the database. Fortunately, setting up a follower can usually be done without downtime by locking the leader:

1. Take a snapshot of the leader’s database at some point in time
2. Copy the snapshot to the new follower node
3. The follower connects to the leader and requests all the data changes that have happened since the snapshot was taken. The snapshot is associated with an exact position in the leader’s replication log.
4. When the follower has processed the backlog of data changes since the snapshot, we say it has caught up.

Node Outages

Follower failure: Catch-up recovery

On its local disk, each follower keeps a log of the data changes it has received from the leader. If a follower crashes, it can recover easily: using that log, it knows the last transaction that was processed before the fault occurred. So, the follower can connect to the leader and request all the data changes that occurred during the time when the follower was disconnected.

Leader failure: Failover

Failover is trickier: one of the followers needs to be promoted to be the new leader, clients need to be reconfigured to send their writes to the new leader, and the other followers need to start consuming data changes from the new leader.

Failover has many challenges:

1. If async replication is used, the new leader may not have received all the writes from the old leader before it failed. If the former leader rejoins the cluster as a follower, what should happen to those writes? The new leader may have received conflicting writes in the meantime. The most common solution is for the old leader’s unreplicated writes to simply be discarded, which is dangerous as well.
2. In certain fault scenarios, it could happen that 2 nodes both believe that they are the leader. This situation is called split brain.
3. What is the right timeout before the leader is declared dead? A longer timeout means a longer time to recovery. But if the timeout is too short, there could be unnecessary failovers.

For workloads that consist of mostly reads and only a small percentage of writes, there is an attractive option: create many followers, and distribute the read requests across those followers. This removes load from the leader and allows read requests to be served by nearby replicas. In this read-scaling architecture, you can increase the capacity for serving read-only requests simply by adding more followers.

However, this approach only realistically works with async replication — if you tried to synchronously replicate to all followers, a single node failure or network outage would make the entire system unavailable for writing. And the more nodes you have, the likelier it is that one will be down. Unfortunately, if an application reads from an async follower, it may see outdated information if the follower has fallen behind.

This inconsistency is just a temporary state — if you stop writing to the database and wait a while, the followers will eventually catch up and become consistent with the leader. For that reason, this effect is known as eventual consistency. We call the delay between a write on the leader and being reflected on a follower — the replication lag.

Consensus

Consensus is one of the most fundamental problems in distributed computing. There are a no. of situations in which it is important for nodes to agree:

• Leader Election: In a database with single-leader replication, all nodes need to agree on which node is the leader. The leader position might become contested if some nodes can’t communicate with others due to a network fault. Here, consensus is important to avoid a failover, resulting in a split brain.
• Atomic commit: In a database that supports transactions spanning several nodes or partitions, a transaction may fail on some nodes but succeed on others. All nodes in a distributed transaction must either commit or abort.

On the surface, it seems simple: the goal is simply to get several nodes to agree on something. You might think that this should be simple. Unfortunately not.

The FLP impossibility result, a foundational theorem in distributed computing, states that no deterministic algorithm can guarantee consensus in a fully asynchronous system if even a single process may fail.

Single Node Transactions

For transactions that execute at a single database node, atomicity is commonly implemented by the storage engine. When the client asks the database node to commit the transaction, the database makes the transaction’s writes durable and then appends a commit record to the log on disk.

If the database crashes in the middle of this process, the transaction is recovered from the log when the node restarts: if the commit record was successfully written to disk before the crash, the transaction is considered committed; if not, any writes from that transaction are rolled back. Thus, on a single node, transaction commitment crucially depends on the order in which data is durably written to disk: first the data, then the commit record.

The key deciding moment for whether the transaction commits or aborts is the moment at which the disk finishes writing the commit record: before that moment, it is still possible to abort (due to a crash), but after that moment, the transaction is committed (even if the database crashes). Thus, it is a single device (the controller of one particular disk drive, attached to one particular node) that makes the commit atomic.

However, what if multiple nodes are involved in a transaction? In these cases, it is not sufficient to simply send a commit request to all of the nodes and independently commit the transaction on each one. In doing so, it could easily happen that the commit succeeds on some nodes and fails on other nodes, which would violate the atomicity guarantee. For this reason, a node must only commit once it is certain that all other nodes in the transaction are also going to commit.

Two-Phase Commit (2PC)

2PC is a classic algorithm for achieving atomic transaction commit across multiple nodes. The basic flow of 2PC is illustrated below:

Instead of a single commit request, as with a single-node transaction, the commit/abort process in 2PC is split into two phases. A 2PC transaction begins with the application reading and writing data on multiple database nodes, as normal. We call these database nodes participants in the transaction.

When the application is ready to commit, a coordinator (AKA transaction manager) begins phase 1: send a prepare request to each of the nodes, asking them whether they are able to commit. The coordinator then tracks the responses from the participants:

• If all participants reply “yes,” then the coordinator sends out a commit request in phase 2, and the commit actually takes place.
• If any of the participants replies “no,” the coordinator sends an abort request to all nodes in phase 2.

Fault tolerance Consensus

A consensus algorithm must satisfy the following properties:

• Uniform agreement: No two nodes decide differently.
• Integrity: No node decides twice.
• Validity: If a node decides value v, then v was proposed by some node.
• Termination: Every node that does not crash eventually decides some value.

If you don’t care about fault tolerance, then satisfying the first 3 properties is easy: you can just hardcode one node to be the “dictator,” and let that node make all of the decisions. However, if that one node fails, then the system can no longer make any decisions.

Consensus assumes that when a node “crashes,” it suddenly disappears and never comes back. Any consensus algorithm requires at least a majority of nodes to be functioning correctly in order to assure termination. That majority can safely form a quorum. To configure a quorum-based system, 3 parameters are defined:

• N: The total number of nodes (replicas) in the system.
• W (Write Quorum): The minimum no. of nodes that must acknowledge a write before it is confirmed to the client.
• R (Read Quorum): The minimum no. of nodes that must respond to a read request to ensure the latest data is retrieved.

To guarantee strong consistency (ensuring a read always sees the latest successful write), the following mathematical condition must be met:

R + W > N

System designers adjust R and W to prioritize either speed or reliability based on the CAP Theorem:

• High Write Performance: Set a low W. This increases write speed and availability but requires a high R to maintain consistency, slowing down reads.
• High Read Performance: Set a low R. This is common in read-heavy applications but requires, meaning a single down node can block all writes.
• Eventual Consistency: If R + W ≤ N, the system cannot guarantee that a read will see the latest write, leading to eventual consistency rather than strong consistency.

Consensus algorithms

The best-known fault-tolerant consensus algorithms are Raft, VSR, Paxos, and Zab. Raft prioritizes understandability by:

• Leader Election: Initially all nodes are followers. If a follower doesn’t hear from a leader, it becomes a candidate and sends RequestVote RPCs to other nodes. If the cluster elects it as leader, it manages all client requests and log replication.
• Log Replication: The leader receives commands from clients, writes them to its log, and replicates these entries across follower nodes. A command is considered committed once a majority of nodes acknowledge it.

Most of these algorithms are total order broadcast algorithms which requires messages to be delivered exactly once, in the same order, to all nodes.

In single-leader replication, it takes all the writes to the leader and applies them to the followers in the same order, thus keeping replicas up to date — essentially total order broadcast. But why don’t we worry about consensus there? Because the leader is manually chosen and configured by the humans in your operations team — you essentially have a “consensus algorithm” of the dictatorial variety.

Remember, that this system does not satisfy the termination property of consensus

Multi-Leader Replication

As we discussed, single-leader replication has one major downside. However, it rarely makes sense to use a multi-leader setup within a single datacenter, because the benefits rarely outweigh the added complexity. Here are some use cases:

1. Multi-datacenter operation: Each datacenter has one leader
2. Clients with offline operation: An application that needs to continue to write to database while it is disconnected from the network.
3. Collaborative editing: The application must obtain a lock on the document before another user can edit it.

Handling Write Conflicts

The biggest problem with multi-leader replication is that write conflicts can occur. In a single-leader database, the second writer will either block and wait for the first write to complete, or abort the second write transaction, forcing the user to retry the write.

In a multi-leader setup, if both writes are successful, the conflict is only detected asynchronously at some later point in time. Then, it may be too late to ask the user to resolve the conflict.

In principle, you could make the conflict detection synchronous — i.e., wait for the write to be replicated to all replicas before telling the user that the write was successful. However, by doing so, you would lose the main advantage of multi-leader replication. If you want synchronous conflict detection, you might as well just use single-leader replication.

Operational transformation

This is the conflict resolution algorithm behind collaborative editing applications such as Google Docs. It was designed particularly for concurrent editing of an ordered list of items, such as the list of characters that constitute a text document.

The algorithm relies on 3 main components:

• Operations: Each user edit (e.g., insert “a” at position 5, delete character at position 8) is represented as a distinct operation.
• Transformation Functions: These are rules that adjust an incoming operation so it can be correctly applied to the current local document state, which may have changed since the incoming operation was generated. For example, if User A inserts a character at position 0, User B’s subsequent delete operation at position 2 must be transformed to target position 3 to delete the correct character.
• Control Algorithm: This part determines which operations need to be transformed and the specific order of transformations, managing the flow of operations between clients and the server to ensure consistency properties are met.

Conflict-free replicated datatypes (CRDTs)

CRDTs are a family of data structures for sets, maps, ordered lists, counters, etc. that can be concurrently edited by multiple users, and automatically resolve conflicts in sensible ways. CRDTs are generally categorized into 2 main approaches:

• State-based CRDTs (CvRDTs): Replicas exchange their full local state. A merge function, combines the incoming state with the local one. This approach is simpler to implement but can involve high network overhead as the entire state is sent.
• Operation-based CRDTs (CmRDTs): Replicas broadcast the update operations themselves. These operations must be commutative and delivered reliably in causal order. This method is more bandwidth-efficient but requires a more sophisticated communication infrastructure to ensure delivery guarantees.

Thanks for reading 🎉

References: Designing Data Intensive Applications

References: Designing Data Intensive Applications