Distributed System - Replication(Single Leader Replication)

This is a skimmed version of Chapter 5 ‘Replication’ from the book design_data_intensive_applications. I would recommend you to first read the whole chapter and use these notes for revision purpose only.

INTRODUCTION

Replication means keeping a copy of the same data on multiple machines that are connected via a network. Why?

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

In this blog, we will read about two popular algorithms for replicating data:

• Leader Based replication(also known as active/passive or master–slave replication)
  • Single Leader replication
  • Multi Leader replication
• Leaderless replication

SINGLE LEADER REPLICATION

• Each node that stores a copy of the database is called a replica.
• With multiple replicas, a question inevitably arises: how do we ensure that all the data ends up on all the replicas?. It works as follows

1. One of the replicas is designated the leader (also known as master or primary). 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.
2. The other replicas are known as followers (read replicas, slaves, secondaries, or hot standbys).Whenever the leader writes new data to its local storage, it also sends the data change to all of its followers as part of a replication log or change stream. Each follower takes the log from the leader and updates its local copy of the database accordingly, by applying all writes in the same order as they were processed on the leader.
3. 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 (the followers are read-only from the client’s point of view).

Synchronous Versus Asynchronous Replication

In the example, the replication to follower 1 is synchronous: 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. The replication to follower 2 is asynchronous: the leader sends the message, but doesn’t wait for a response from the follower.

Advantages/Disadvantages Of Synchronus/Asynchronus Replication

• The advantage of synchronous 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 has crashed, or there is a network fault, or for any other reason), the write cannot be processed.
  • The leader must block all writes and wait until the synchronous replica is available again.

For that reason, it is impractical for all followers to be synchronous: any one node outage would cause the whole system to grind to a halt. In practice, if you enable synchronous replication on a database, it usually means that one of the followers is synchronous, and the others are asynchronous.

If the synchronous follower becomes unavailable or slow, one of the asynchronous followers is made synchronous. This guarantees that you have an up-to-date copy of the data on at least two nodes: the leader and one synchronous follower.This configuration is sometimes also called semi-synchronous.

Setting Up New Followers

How do you ensure that the new follower has an accurate copy of the leader’s data?

Simply copying data files from one node to another is typically not sufficient: clients are constantly writing to the database, and the data is always in flux, so a standard file copy would see different parts of the database at different points in time. The result might not make any sense.

You could make the files on disk consistent by locking the database (making it unavailable for writes), but that would go against our goal of high availability.

You can set up a new follower by following these steps:

1. Take a consistent snapshot of the leader’s database at some point in time—if possible, without taking a lock on the entire database. Most databases have this feature, as it is also required for backups. In some cases, third-party tools are needed, such as innobackupex for MySQL.
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. This requires that the snapshot is associated with an exact position in the leader’s replication log. That position has various names: for example, PostgreSQL calls it the log sequence number, and MySQL calls it the binlog coordinates.
4. When the follower has processed the backlog of data changes since the snapshot, we say it has caught up. It can now continue to process data changes from the leader as they happen.

Handling 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 and is restarted, or if the network between the leader and the follower is temporarily interrupted, the follower can recover quite easily: from its log, it knows the last transaction that was processed before the fault occurred.
• Thus, the follower can connect to the leader and request all the data changes that occurred during the time when the follower was disconnected.
• When it has applied these changes, it has caught up to the leader and can continue receiving a stream of data changes as before.

Leader failure: Failover

Handling a failure of the leader 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. This process is called failover.

Failover can happen manually (an administrator is notified that the leader has failed and takes the necessary steps to make a new leader) or automatically. An automatic failover process usually consists of the following steps:

1. Determining that the leader has failed.(Can use heartbeat functionality to find this)

2. Choosing a new leader. This could be done through an election process (where the leader is chosen by a majority of the remaining replicas), or a new leader could be appointed by a previously elected controller node. Getting all the nodes to agree on a new leader is a consensus problem

3. Reconfiguring the system to use the new leader. Clients now need to send their write requests to the new leader. The system needs to ensure that the old leader(if become active again) becomes a follower and recognizes the new leader.

Potential Problems with Failover

1. If asynchronous replication is used, the new leader may not have received all the writes from the old leader before it failed. (Undurability).
2. In certain fault scenarios, it could happen that two nodes both believe that they are the leader. This situation is called split brain, and it is dangerous: if both leaders accept writes, and there is no process for resolving conflicts, data is likely to be lost or corrupted.
3. What is the right timeout before the leader is declared dead?

• A longer timeout means a longer time to recovery in the case where the leader fails. However, if the timeout is too short, there could be unnecessary failovers. For example, a temporary load spike could cause a node’s response time to increase above the timeout, or a network glitch could cause delayed packets. If the system is already struggling with high load or network problems, an unnecessary failover is likely to make the situation worse, not better.

Implementation of Replication Logs

STATEMENT-BASED REPLICATION

• The leader logs every write request (statement) that it executes and sends that statement log to its followers.
  • For a relational database, this means that every INSERT, UPDATE, or DELETE statement is forwarded to followers, and each follower parses and executes that SQL statement as if it had been received from a client.
• Disadvantages
  • Any statement that calls a nondeterministic function, such as NOW() to get the current date and time or RAND() to get a random number, is likely to generate a different value on each replica.
  • If statements use an autoincrementing column, or if they depend on the existing data in the database (e.g., UPDATE … WHERE ), they must be executed in exactly the same order on each replica, or else they may have a different effect. This can be limiting when there are multiple concurrently executing transactions.
• Statements that have side effects (e.g., triggers, stored procedures, user-defined functions) may result in different side effects occurring on each replica.

Statement-based replication was used in MySQL before version 5.1. It is still some‐ times used today, as it is quite compact, but by default MySQL now switches to row-based replication (discussed shortly) if there is any nondeterminism in a statement. VoltDB uses statement-based replication, and makes it safe by requiring transactions to be deterministic.

WRITE-AHEAD LOG (WAL) SHIPPING

• In the case of a B-tree , which overwrites individual disk blocks, every modification is first written to a write-ahead log so that the index can be restored to a consistent state after a crash.
• The log is an append-only sequence of bytes containing all writes to the database. We can use the exact same log to build a replica on another node: besides writing the log to disk, the leader also sends it across the network to its followers.
• When the follower processes this log, it builds a copy of the exact same data structures as found on the leader. This method of replication is used in PostgreSQL and Oracle, among others.
• Disadvantages
  • The main disadvantage is that the log describes the data on a very low level: a WAL contains details of which bytes were changed in which disk blocks.
  • This makes replication closely coupled to the storage engine. If the database changes its storage format from one version to another, it is typically not possible to run different versions of the database software on the leader and the followers.

That may seem like a minor implementation detail, but it can have a big operational impact. If the replication protocol allows the follower to use a newer software version than the leader, you can perform a zero-downtime upgrade of the database software by first upgrading the followers and then performing a failover to make one of the upgraded nodes the new leader.If the replication protocol does not allow this version mismatch, as is often the case with WAL shipping, such upgrades require downtime.

LOGICAL (ROW-BASED) LOG REPLICATION

To allow the replication log to be decoupled from the storage engine internals. This kind of replication log is called a logical log, to distinguish it from the storage engine’s (physical) data representation.

• A logical log for a relational database is usually a sequence of records describing writes to database tables at the granularity of a row:
  • For an inserted row, the log contains the new values of all columns.
  • For a deleted row, the log contains enough information to uniquely identify the row that was deleted. Typically this would be the primary key, but if there is no primary key on the table, the old values of all columns need to be logged.
  • For an updated row, the log contains enough information to uniquely identify the updated row, and the new values of all columns (or at least the new values of all columns that changed).
• A transaction that modifies several rows generates several such log records, followed by a record indicating that the transaction was committed. MySQL’s binlog (when configured to use row-based replication) uses this approach.

Since a logical log is decoupled from the storage engine internals, it can more easily be kept backward compatible, allowing the leader and the follower to run different versions of the database software, or even different storage engines.

TRIGGER-BASED REPLICATION

The replication approaches described so far are implemented by the database system, without involving any application code.

• In some circumstances, more flexibility is needed.
  • For example, if you want to only replicate a subset of the data, or want to replicate from one kind of database to another,then you may need to move replication up to the application layer.
• An alternative is to use features that are available in many relational databases: triggers and stored procedures.
• A trigger lets you register custom application code that is automatically executed when a data change (write transaction) occurs in a database system.
  • The trigger has the opportunity to log this change into a separate table, from which it can be read by an external process. That external process can then apply any necessary application logic and replicate the data change to another system.
  • Databus for Oracle and Bucardo for Postgres work like this, for example.
• Disadvantages
  • Trigger-based replication typically has greater overheads than other replication methods.
  • more prone to bugs and limitations than the database’s built-in replication.

PROBLEM WITH REPLICATION LAG

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.

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 asynchronous replication.

• Unfortunately, if an application reads from an asynchronous follower, it may see outdated information if the follower has fallen behind.
• This leads to apparent inconsistencies in the database: if you run the same query on the leader and a follower at the same time, you may get different results, because not all writes have been reflected in the follower.
• 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
• In general, there is no limit to how far a replica can fall behind. In normal operation, the replication lag may be only a fraction of a second, and not noticeable in practice or if there is a problem in the network, the lag can easily increase to several seconds or even minutes.
• In this section we will highlight three examples of problems that are likely to occur when there is replication lag and outline some approaches to solving them.

READING YOUR OWN WRITES

With asynchronous replication, there is a problem, if the user views the data shortly after making a write, the new data may not yet have reached the replica.(Updating the profile picture and viewing it) To the user, it looks as though the data they submitted was lost, so they will be understandably unhappy.

• In this situation, we need read-after-write consistency, also known as read-your-writes consistency.

This is a guarantee that if the user reloads the page, they will always see any updates they submitted themselves. It makes no promises about other users: other users’ updates may not be visible until some later time.

How can we implement read-after-write consistency in a system with leader-based replication? There are various possible techniques. To mention a few

• When reading something that the user may have modified, read it from the leader; otherwise, read it from a follower.

  • For example, user profile information on a social network is normally only editable by the owner of the profile, not by anybody else.
  • Thus, a simple rule is: always read the user’s own profile from the leader, and any other users’ profiles from a follower.
  • If most things in the application are potentially editable by the user, that approach won’t be effective, as most things would have to be read from the leader (negating the benefit of read scaling).

• In that case, other criteria may be used to decide whether to read from the leader.

  • For example, you could track the time of the last update and, for one minute after the last update, make all reads from the leader. You could also monitor the replication lag on followers and prevent queries on any follower that is more than one minute behind the leader.

• Another complication arises when the same user is accessing your service from multiple devices, for example a desktop web browser and a mobile app.

  • In this case you may want to provide cross-device read-after-write consistency: if the user enters some information on one device and then views it on another device, they should see the information they just entered.

• In this case, there are some additional issues to consider:

• Approaches that require remembering the timestamp of the user’s last update become more difficult, because the code running on one device doesn’t know what updates have happened on the other device. This metadata will need to be centralized.

• If your replicas are distributed across different datacenters, there is no guarantee that connections from different devices will be routed to the same datacenter. (For example, if the user’s desktop computer uses the home broadband connection and their mobile device uses the cellular data network, the devices’ network routes may be completely different).

  • If your approach requires reading from the leader, you may first need to route requests from all of a user’s devices to the same datacenter.

MONOTONIC READS

Our second example of an anomaly that can occur when reading from asynchronous followers is that it’s possible for a user to see things moving backward in time.

• This can happen if a user makes several reads from different replicas.
  • For example, a user making the same query twice, first to a follower with little lag, then to a follower with greater lag.
  • (This scenario is quite likely if the user refreshes a web page, and each request is routed to a random server.)
  • Let say, the first query returns a comment that was recently added by user , but the second query doesn’t return anything because the lagging follower has not yet picked up that write.
  • In effect, the second query is observing the system at an earlier point in time than the first query.
• Monotonic reads is a guarantee that this kind of anomaly does not happen.

  It’s a lesser guarantee than strong consistency, but a stronger guarantee than eventual consistency. When you read data, you may see an old value; monotonic reads only means that if one user makes several reads in sequence, they will not see time go backward— i.e., they will not read older data after having previously read newer data.

• One way of achieving monotonic reads is to make sure that each user always makes their reads from the same replica (different users can read from different replicas).
  • For example, the replica can be chosen based on a hash of the user ID, rather than randomly. However, if that replica fails, the user’s queries will need to be rerouted to another replica.

CONSISTENT PREFIX READS

Our third example of replication lag anomalies concerns violation of casuality. Imagine the following short dialog between Mr. Poons and Mrs. Cake:

Mr. Poons

How far into the future can you see, Mrs. Cake?

Mrs. Cake

About ten seconds usually, Mr. Poons.

There is a causal dependency between those two sentences: Mrs. Cake heard Mr. Poons’s question and answered it. Now, imagine a third person is listening to this conversation through followers. The things said by Mrs. Cake go through a follower with little lag, but the things said by Mr. Poons have a longer replication lag (see Figure 5-5). This observer would hear the following:

Mrs. Cake

About ten seconds usually, Mr. Poons.

Mr. Poons

How far into the future can you see, Mrs. Cake?

• Preventing this kind of anomaly requires another type of guarantee: consistent prefix reads.
• This guarantee says that if a sequence of writes happens in a certain order, then anyone reading those writes will see them appear in the same order.
  • This is a particular problem in partitioned (sharded) databases. If the database always applies writes in the same order, reads always see a consistent prefix, so this anomaly cannot happen. However, in many distributed databases, different partitions operate independently, so there is no global ordering of writes: when a user reads from the database, they may see some parts of the database in an older state and some in a newer state.
• One solution is to make sure that any writes that are causally related to each other are written to the same partition but in some applications that cannot be done efficiently.
• There are also algorithms that explicitly keep track of causal dependencies, a topic that we will return to in “The “happens-before” relationship and concurrency”