chapter-4

# 4 Replication

## 4.8 Putting it together: Replicating a key-value store

In the previous chapter, we examined how colocating data with compute using SQLite reduces latency compared to a more traditional client-server approach with PostgreSQL. However, if you need multiple copies of the data, replication is necessary.

In this section, we will implement a simple replicated in-memory key-value store using a primary-replica architecture, where writes only occur on the primary, which actively broadcasts updates to the replicas, allowing them to read values locally.

We start by implementing an in-memory key-value store using a hash map:

    pub struct KVStore {
     data: Mutex<HashMap<String, String>>,
    }

    impl struct KVStore {
     pub fn get(&self, key: &str) -> Option<String> {
       self.data.lock().unwrap().get(key).cloned()
     }

     pub fn put(&self, key: String, value: String) {
       self.data.lock().unwrap().insert(key, value);
     }
    }

The `KVStore` struct provides the backing store we use in both the primary and replica servers. In a real system, you could use a database that offers durability, such as SQLite or another embedded database.

We then specify the replication protocol we use between the primary and the replica:

    pub enum Message {
     Put { key: String, value: String },

     Join { replica_addr: String },

     Snapshot { entries: Vec<(String, String)> },
    }

The primary broadcasts the `Put` message to the replicas to actively push changes. A replica joins the primary replica set by sending a Join message with its network address. When a primary receives a `Join` message, it sends a snapshot of its data to the replica via a `Snapshot` message. The snapshot ensures that the replica has all the data the primary has and can, therefore, incrementally receive new updates.

We also implement a shell for the primary server that users can use to manipulate the state:

    loop {
      match rl.readline("primary> ") {
        Ok(line) => {
          let parts: Vec<&str> = line.trim().split_whitespace().collect();
          match parts.as_slice() {
            ["PUT", key, value] => {
              storage.put(key.to_string(), value.to_string());
              let message = Message::Put {
                key: key.to_string(),
                value: value.to_string(),
              };
              broadcast(&replicas, &message);
            }
            ["GET", key] => match storage.get(key) {
              Some(value) => println!("{} -> {}", key, value),
              None => println!("{} -> Not found", key),
            },
          }
        }
      }
    }

And also one for the replica:

    loop {
      match rl.readline("replica> ") {
        Ok(line) => {
          let parts: Vec<&str> = line.trim().split_whitespace().collect();
          match parts.as_slice() {
            ["GET", key] => match storage.get(key) {
              Some(value) => println!("{} -> {}", key, value),
              None => println!("{} -> Not found", key),
            },
            ["JOIN", host_port] => {
              let message = Message::Join {
                replica_addr: replica_addr.clone(),
              };
              join_primary(host_port, &storage, &message);
           }
        }
      }
    }

You can find the complete example of this in https://github.com/penberg/latency-book/tree/main/chapter-04/rust/replication-kv.

First, let's start up the primary server:

    $ cargo run --bin primary
    Primary server ready (port 8080)
    Commands:
     - PUT <key> <value>
     - GET <key>
     - EXIT

We now have a shell that allows us to interact with the primary key-value store. The server is also listening to port 8080 for incoming messages from replicas.

Now let's insert some data into the primary key-value store:

    primary> PUT a 1

    OK a = 1
    primary> PUT b 2
    OK b = 2

We now have (a, 1) and (b, 2) key-value pairs stored in the primary.

Next, let's start up a replica server:

    $ cargo run --bin replica
    Replica ready (port 8081)
    Commands:
    - GET <key>
    - JOIN <host:port>
    - EXIT

The replica server has a similar shell to the primary server, but with slightly different commands. When the replica starts up, it is not automatically connected to the primary. Therefore, reading from the replica yields no results:

    replica> GET a
    a -> Not found
    replica> GET b
    b -> Not found

We can register the replica to the replication set, with the `JOIN` command as follows:

    replica> JOIN localhost:8080
    Snapshot received

When the replica registers itself to the replication set of the primary, it receives a snapshot of the current primary key-value store. So now, if we read from the replica, we can see the same values:

    replica> GET a
    a -> 1
    replica> GET b
    b -> 2

As the replica is now part of the primary's replication set, writing to the primary:

    primary> PUT c 3
    OK c = 3

is pushed to the replica server:

    replica> GET c
    c -> 3

That's it, we now have an example of a primary/replica architecture in action. Of course, the only guarantee we provide is eventual consistency, without any durability guarantees, and we assume no errors. However, despite these shortcomings, the approach illustrates how the flow of real-world replication systems works, even though they utilize, for example, distributed consensus algorithms to provide fault tolerance and durability. Furthermore, if you are building an application, you will probably be using the replication feature of your database or key-value store that is built into it.

## 4.9 Summary

- Replicating data has multiple benefits reducing latency, improving reliability and availability, and helping prevent data loss due to hardware failures, power outages, or other unforeseen circumstances.
- Strong consistency (linearizability) is the gold standard of data access, whereas eventual consistency is the weakest consistency model. Between the two, there are other consistency levels such as causal and session consistency with different trade-offs.
- Replication can either be based on single-leader, multiple leaders, or be leaderless, depending on the requirements. Replication can also be synchronous or asynchronous, which impacts latency and consistency.