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.