Somtochi Onyekwere on Distributed Data Systems, Eventual Consistency and Conflict-free Replicated Data Types

Transcript

Introductions (00:20)

Srini Penchikala: Hi everyone. My name is Srini Penchikala. I serve as the lead editor for AI, ML and Data Engineering community at the InfoQ website, and I also host podcast episodes. In today’s episode, I will be speaking with Somtochi Onyekwere, software engineer at Fly.io organization. We will discuss the recent developments in distributed data systems, especially topics like eventual consistency and how to achieve fast, eventually consistent replication across distributed nodes. We’ll also talk about the conflict-free replicated data type data structures, also known as CRDTs and how they can help with conflict resolution when managing data in distributed data storage systems.

Hi Somtochi. Thank you for joining me today. Can you introduce yourself and tell our listeners about your career and what areas you have been focusing on recently?

Somtochi Onyekwere: Oh, I’m glad for the opportunity to be on the podcast. Thank you, Srini. So yes, like you said, my name is Somtochi. I’m a software engineer at Fly.io where I work primarily on Corrosion, which is our open source distributed system. And I’m also on the networking team because Corrosion powers a networking stack, so basically distributed systems and networking software. Fly is a cloud startup for deploy your applications.

Before Fly, I worked at a company called Weaveworks. One way or the other throughout my career I’ve worked on some form of open source software where it was primarily for building open source controllers for Kubernetes that helped users deploy the applications through GitOps, which is managing deployments through files stored on Git.

So one way or the other, I’ve been involved in developer tooling, deployment infrastructure, also part of the Kubernetes community just from Weaveworks, from even way before my career where I did Google Summer of Code. So I’ve mostly been working on Kubernetes, distributed system, infra, so that’s mostly about that.

Faster, Low Latency Writes [02:38]

Srini Penchikala: Thank you. Yes, that’s a lot of different areas of expertise. I’m definitely very curious about Corrosion framework, you know how it works and everything. So maybe we can start with more high level topics on that framework. So one of the goals of your project was to be able to write data on one server node and read it from other server nodes without much delay, without much latency. So can you talk about how you are able to achieve this? What kind of patterns and practices that you and your team invented to achieve these goals?

Somtochi Onyekwere: Yes, sure. So normally when a machine is created on Fly, we want to do that quickly. We provide a system where you can deploy applications anywhere. So when people create machines, create VMs, they expect to be able to reach it pretty quickly. So we favor speed over consistency, and this has also shown up in how we build software, right? Because if you’re trying to make everything consistent, making sure that everybody agrees before you move ahead, you’re going to have some trade-off in terms of latency.

Before Corrosion, we used to use Console, which is strongly consistent, uses Raft for consensus. So before that we used Console and we created Corrosion out of a problem and that was basically we just wanted data at different points very fast. So already the problem that we’re creating Corrosion for had to do with speed. We are pulling Console servers at first, pulling the API, trying to insert it into data and into JSON and have that everywhere and have the software read from that, but that didn’t really scale well trying to pull the API some of the time.

So we’ve seen that a lot of times, not just in Corrosion, but in other software that we use where we don’t necessarily want to make the best case decision but want to make a decision quickly, we want access to information to be able to make a decision quickly. So that has played out, especially in Corrosion where it’s a very relaxed system, is eventually consistent and is fast.

Eventual Consistency [04:50]

Srini Penchikala: That’s where the whole CAP Theorem comes from. The Consistency, Availability and Partition Tolerance. So can you talk about the application use cases and discuss how the eventual consistency did not affect the quality of the data?

Somtochi Onyekwere: The thing is of course there’s going to be systems that need to be strongly consistent. If you’re using a bank, you don’t want eventual consistency, it could come up in many funky ways, but for a vast majority of internet applications, eventual consistency works when you relax the consistency guarantees. You’re like, okay, this data might be a little out of style and maybe I might make a sort of optimization, but I can do that quickly.

Srini Penchikala: With the eventual consistency, how do you ensure the data quality of the application and that there are no issues?

Somtochi Onyekwere: How we ensure the quality of the applications and there are no issues is that we have a system that handles it. We have ways of dealing with stale data within the system, because most times it’s the two things with eventual consistency. It’s either all the data is arriving too late or there’s two different parts of the platform writing to the same place and it’s like, okay, what’s going to win? Which data wins? So with stale data it’s just finding a way to make good decisions even if the data might be out of style. For example, our networking software now, when a request comes in for a machine, so we have servers called edges, so all the requests come in through edges and then the edges route the request to the right worker. If a request comes into an edge and it matches like the IP to a particular machine and it picks a machine, but maybe that machine has been deleted, but it doesn’t know that yet. It’s like, okay, this machine is started. It’s going to send a request to a worker, which are the actual VMs that run the machine.

So it’s the one updating the machine state. It has the most recent state, definitely, because locally if you write to say Corrosion, it gets the read back. So Corrosion on the worker has the most recent state. So when the edge sends the request to the worker and it’s like, okay, send this request to a machine on there, it can send a response back to the edge saying, oh, this instance actually has been deleted or wasn’t found. So the proxy now reroutes it to another machine. So this is just a small example of how to deal with stale state. Even if the state is a little out of date, maybe you route it to a closer machine or closer server that has the closer state and have it respond backwards, okay, this is not right. Or if it is right, it would just do the writing of the request. In this case it will write it to the right VM.

The other one is conflict resolution, which is different things writing to the same data. So we have a maze of conflict resolution within the system, but even above that is that sometimes your data is even structured in a way that you don’t really need to worry about conflicts. We have different servers and each of those servers have a particular set of machines. It’s all in the same table, we have a table that contains the machine’s design SQL, but each server writes just for its own. So basically each row kind of has an owner because each server just writes for the machine that it’s in charge of.

So even though it’s a distributed system, there are multiple tier writes into the same table. There’s actually not a lot of conflicts because there are different things that are on different parts and different roles in database. So just little things like that where you just adjust your system a little bit to be able to work, relax the consistency constraint that you require from your databases or your storage systems and be able to gain some speed and still be able to work efficiently.

Conflict-free Replicated Data Types (CRDT) [08:40]

Srini Penchikala: You have used conflict-free replicated data types, CRDT’s they’re called to handle some of these requirements. Can you talk about these specialized data structures? How do they help with the conflict resolution? How can we use them in other applications?

Somtochi Onyekwere: Sure. CRDT is conflict-free replicated data types. I feel like they’ve been around for a while in the distributed system community, but it’s like they’re having a comeback now with just more research being used and more software just because of how they allow people build eventually consistent systems and resolve conflicts in them. So CRDT is mostly a concept around data types and possibly associated algorithms where you have independent replicas able to accept writes and update the data without communicating with each other. So each replica can accept writes and then those replicas at different time, doesn’t have to be a long time, can exchange the data they have, and then part of CRDTs is when they exchange the data, they converge to the same state. So there’s some background mechanism, some gradient, that probably runs when the data is exchanged that would ensure that regardless of the order in which the replicas have received the changes, they converge to the same states. So there are different ways and different data types that have been brought up.

I feel like a very popular example, which can make it easy for someone to grasp it is like let’s say a replicated counter. And so in CRDTs there are mostly two types of CRDTs where you have state-based CRDTs, so in state-based CRDTs the nodes exchange all of their states between each other and they find a way to merge to different states. And then you have an operation-based CRDT where they exchange the operations that they have received instead of the whole states. Each approach has their disadvantages and advantages. Of course, if you’re exchanging the whole states, you are going to have way more network communications, but it means that there are less guarantees. If you’re communicating operations immediately you have to more careful in terms of duplicates and stuff like that.

For something like a replicated counter, so I’ll give a state-based example and an operation-based example. So first a state-based example of a replicated counter. So you have a counter, two different nodes, they get updated, incremented, separately. Maybe one is incremented twice from zero to two, and the other one is incremented once, so one has the value of one, the other one has the value of two. When they exchange their state, your merging function can just be one that takes in max. So in this case, it’d be like a grow-only counter. So if your counter can only increment, a good example is maybe likes on an Instagram post because okay, you can unlike, but let’s say an Instagram post that you can’t unlike, so it’s a grow-only counter. So in that case, when you exchange the two states, you just know that the merge is just going to take the max of the two functions and you know that regardless of whatever order that you exchange state between different counters, you still get to the same state.

But for something like an operation-based CRDT, they exchange operations. So in this case, one would send the two increments it has received to another node while the other one just send the other one. Operation-based CRDTs don’t have as much network requirements, but it means that you need to be more careful about say duplicated states because let’s say there’s a network issue, I don’t know if the other node has received, then send the operation the increment counter again. It might increment it more than required, so there are many ways to work around this, maybe some versioning so that they can discard an already applied version kind of thing.

But basically if the two counters exchange the exact number of operations that they have gone through, they would come to the same state. So I think this is a very good example of, oh, this is a simple data structure that people interact with daily and this is how it can easily be kind of a CRDT.

CRDT Use Cases [12:44]

Srini Penchikala: So other than the Instagram counter, are there any other examples, other business use cases?

Somtochi Onyekwere: Yes, the counter is of course very simplified, but just easy to understand. So you have something like Grow-only Set. So set nice in that you can’t get duplicates, so once you add something to the set is being added, right? But you also have to now deal with concurrent additions and removals. So some sets have something like an add wins depending on what the set is being used for. It might be a case, so where the business logic now comes into play, some sets are like once there’s an add, there’s a concurrent add and remove, the add always wins, which is just static. There’s also remove means set too. There’s also an observed remove where each add and each remove, there’s some data to track what’s being added and what’s being removed. You don’t just have an arbitrary operation wins, but whichever operation has been done more times will.

There’s also CRDTs that use the concept of Last-Writer-Wins, which is they find some way to track timestamps on operations. So whichever one has the older timestamps will win. Of course there’s some trickiness to this because when you’re working with distributed systems, it can depend on system time. So you have to start thinking of something like Lamport’s clock to track the timestamps. Yes, there are also some data structures that kind of leave it up to the application to decide on how best to resolve the conflict too. So it keeps conflicting. It merges the changes, Git is a good example actually, because Git would merge whatever changes it can and then tell you, oh, I can’t merge this, there’s a conflict here that I can’t really understand. So there are some systems they will merge as much as they can without conflicts, but when there’s a conflict, they might leave it up to the application to decide.

Order of Data Changes [14:40]

Srini Penchikala: Onto a similar topic. In your presentation at QCon conference earlier this year, you mentioned the order of data changes should matter. So for the counters it makes sense, but how do you handle the data changes that come in at different times for other use cases where they’re a little bit more complex than just a counter example?

Somtochi Onyekwere: The main theme is whatever order the changes come in the two replicas should converge the same state. So there’s that consistency as much as is an eventual consistency. It’s a strong eventual consistency. Whatever algorithm people have, when you have a different order, they should eventually decide on the same state. The main thing is understanding the algorithm behind whatever CRDT you’re using is. Let’s say something like you’re breaking ties on the timestamp, it has to mean that each person is breaking ties on the time, someone is breaking ties or there are differing value, it becomes trickier or they’re not storing the right timestamp.

So let’s say it receives a change that has a greater timestamp, but it doesn’t store it. So it takes the value but not the timestamp. If another change comes in that maybe it has a smaller timestamp and it’s not supposed to be, because it still has an older timestamp it gets out of sync. So basically how the nodes accept and resolve conflict, the order of changes should not matter. So even if let’s say it’s breaking a tie on the node like an ID, the comparison, the way it compares and lands on which of these changes when should be the same at every replica, so that even though it might not be what you expect, you know that it’s going to be the same data everywhere as long as each of those replicas have committed to each other.

CRDT Limitations [16:26]

Srini Penchikala: Makes sense. So regarding the CRDTs, right, so just to get back to that, are there any limitations, like when they’re not good candidates to use in applications?

Somtochi Onyekwere: I think sure, there are definitely limitations to CRDTs. Most times commits associated metadata that helps in conflict resolution, especially the not so simple ones. For the observed removed set, it has a second set tracking what they call tombstones, which is the number of times an element has been separated or renewed. Depending on the CRDTs, if you’re working in very space constrained environments, I can’t have this additional thing storing extra data. It depends on the CRDTs you’re working with. So if it’s a CRDT where it has an excess metadata or there’s an extra garbage collection that is happening in background, it might not work for you if you care about the order of changes that you receive. Basically the type of data you’re managing is serialized, so if you write a two and then you write a five, you’d never want the five appearing before a two.

That could happen in CRDTs because they do not track the order of changes. It’s like if a five comes in first it would show you a five. So in those kind of cases where the order matters, you want strong consistency because a lot of times CRDTs are used in eventually consistent systems. That’s something to consider. Maybe your data is more complex and you can’t find a type of CRDT that will let you define a conflict resolution, right? Some CRDTs they come with already how they manage conflicts, but maybe you want to manage it differently.

Maybe instead of choosing the one with the bigger timestamp, you want the one with the bigger value in that kind of way. They’re just little things that you need to consider when you’re trying to pick up CRDTs. But the main thing you need to think about is like, well, how do they resolve conflicts? Am I okay with that? Am I okay with the concept of CRDTs and what it means for application? Meaning the writes might appear out of order, it might be a bit before all the changes quizzes because you have less of partitions. Do you understand how it works under that? So just little things like this, I think those current situations might be a limitation for applications here.

Data Replication Across Nodes [18:40]

Srini Penchikala: I also want to talk about the topic of data replication. The Corrosion framework replicates SQL data across the nodes. So can you talk about how eventual consistency makes sense to bring all the nodes up to date with the data changes, but how does the data replicate between the nodes?

Somtochi Onyekwere: Corrosion is the system that we built at Fly to replicate SQLite data to a global cluster. It’s eventually consistent using CRDTs. So as soon as you write, it responds almost immediately, but it would not be immediately available at all the global clusters. Part of what we try to do with Corrosion is ensure that it is quickly available. We try to keep the communication quick. Part of how we do this is that we use Gossip protocol. When you make the change, we broadcast that change to all the closest nodes. I think part of this is just that data is more likely to be read close to where it’s written. It’s just that locality associated with data. Doesn’t always happen in all writes, but first of all, we broadcast it to all local nodes and then we do a gossip-based dissemination to the rest of the cluster, which is sort this kind of epidemic protocol where one node sends to eight nodes and those eight nodes send to eight other nodes and the change kind of spreads exponentially across the cluster.

The broadcast is kind of final goal. So it’s possible maybe a node is down, didn’t receive the broadcast, missed it in this time and what happens then is that we also have a periodic sync protocol that run where a node contacts other nodes, they exchange states and the node contacting is able to find out which version is missing. We attach version numbers to each change that comes into Corrosion for state tracking. So when it contacts the other nodes and they exchange some data, it’s able to know why I’m missing this version. So these are called gaps in Corrosion.

It has these gaps and then it’s able to request from those gaps, from those nodes. So this sync protocol runs periodically. So even if a node misses a broadcast or doesn’t get broadcast too quickly enough, the sync protocol will also make sure that it’s communicating with disordered nodes and is able to exchange any states that is missing. Another thing we do within Corrosion is we also favor more recent states when we are processing data, let’s say with sync protocol where we request data from other nodes, we request the most recent data first just because it might have been over reading the more recent one. Just little things like that, little decisions that we’ve made within Corrosion to ensure that it’s very quick and it’s able to replicate data very quickly.

Srini Penchikala: Also, you mentioned about local nodes, they get the data faster and the global nodes may take some time. So can you share some statistics on this? What is the response time for local nodes? Is it like seconds or milliseconds?

Somtochi Onyekwere: Oh yes. Local nodes is definitely milliseconds. It pretty short. It’s almost the same thing as writing to an SQLite database, but you would write the changes to SQLite and return immediately. So it’s order of milliseconds. But for the global cluster we have a P99 of one to two seconds. So it’s pretty quick. One second is enough that before someone tries to read the data, most times it’s already there. Or like I said, retries, that’s also another thing to do with eventually consistent data. Sometimes you receive a request maybe for a personal machine or something and you know this data is supposed to be there, but it’s just not there right now, you can easily implement your retries. If the system replicates data on two seconds, you can have a two-second retry. If it’s not, then probably a node is lagging somewhere or it’s just salty.

Srini Penchikala: So eventual consistency doesn’t have to be that long. It can be just in a few seconds.

Somtochi Onyekwere: Yes, it doesn’t have to be very long.

Corrosion Framework Architecture [22:38]

Srini Penchikala: So this framework, can you talk about the technology stack behind it now? It definitely does a lot of great stuff. So what is it built on?

Somtochi Onyekwere: Yes, it’s written in Rust. Rust is just a good language for these kinds of systems where you are aiming for speed and safety at the same time. It has to do with data. We don’t want random crashes. We don’t want random garbage collection happening as maybe the system is processing, so Rust is good in that way. It’s type-safe, zero cost abstractions, little things like this. So a lot of things we use are within the Rust ecosystem.

We use the Tokio, which is very popular, async runtime that uses Rust projects. Tokyo comes with a host of utilities, like the channels for communications are mostly Tokyo channels. So basically we write a lot of async code within Corrosion and that’s used a lot. A lot of things, there are Rust libraries. We use Rust SQLites, which is the Rust SQLites for interacting with SQL. We use high-power, Corrosion exposes a HTTP API for writing and reading SQLite data. We also provide a Postgres endpoint. So yes, we use high-power and Axion for the HTTP API. I think that most of it is built or it uses QUIC protocol instead of TCP, which is this new protocol that has some advantages over TCP within Corrosion.

Srini Penchikala: I was wondering about the name of the project, Corrosion. So now I know where it came from, it’s built on Rust.

Somtochi Onyekwere: It’s built on Rust. So it is just like a Rust name. I think is it a nice name. When we’ve had issues where maybe there’s an issue with Corrosion and Corrosion is something rusted, it’s not a good thing in real life, that’s what it’s like. Of course it has issues, it’s called Corrosion, But yes, I think it’s a fun name. It’s come to stick definitely. But yes, just following the line of a lot of Rust based projects having Rust based names.

Srini Penchikala: Once you make the connection, it makes sense, right?

Somtochi Onyekwere: Yes.

Srini Penchikala: So also you talked about Local First software. Is that updating local nodes first and then go to global nodes? What is the concept?

Somtochi Onyekwere: Yes, I think that’s very close to that. It’s basically the same thing. I mean when I’m talking about nodes, it’s like database, but I’ll just give you examples of where CRDTs are used outside of storage systems. You still see them used within software and it’s very popular with the push to local first software where people want to be able to use a software and still collaborate. So instead of every change has to go through a central server, once there’s no Internet, you can’t access the software, you can’t use it.

There’s also, on the other hand, traditional software, like most of desktop applications where the application is completely available offline, but you can’t collaborate across devices. You can’t just move over your data because it’s kind of stuck on one machine. So the local-first software is kind of like an in-between, between the both of them, where you do have the benefits of being able to use software locally, which means access is quick but also some of the benefits too of online software or fully online software where you’re able to collaborate among multiple devices and you do it in this way because whenever the application is ready or has access to network in terms of mobile devices that disconnects from networks quickly or connect quickly, it can communicate with peers and send across data.

Srini Penchikala: Yes, that’s good. I think we can start wrapping up. Do you have any additional comments or any other thoughts before we close out today’s discussion?

Somtochi Onyekwere: Oh yes, I just realized I’ve talked a lot about CRDTs but not actually about the CRDT that Corrosion itself uses. So Corrosion is built on top of SQLite and we use an SQLite extension called CRSQL, which is just like a little spin of the name like CRDT and SQL joined together. So basically how CRSQL works is that this is an SQLite extension. So when Corrosion starts to be loaded into the database, is that basically when you load the extension and you indicate which tables you want to merge as CRDTs, it would basically create clock tables, these shadow tables, where it stores additional information for each of the rows, each column in each of the rows. So when you insert something like a data into a row in Corrosion, CRSQL creates metadata rows on these clock tables and it tracks the node ID, which is the ID of the node where the data was originally written, the number of times the row has been deleted, recreated or the number of times it’s been updated too.

So just information like that. And then these rows actually, once Corrosion communicates with a bunch of other nodes in the cluster, it would select rows from the clock tables and send out. So basically CRSQL has a lot of hooks. So before data is inserted, updated or deleted in the SQLite table, it has hooks so that its code runs basically and inserts data into this table. What the merge looks like is that it checks first which columns, so it does it on a per column, not a per row basis, so if you have a table that has one table and one row, it has three columns on the table, you create a row for each of those columns. So it tracks them at the columnar level.

So merge it, it basically checks which of them has more updates, whichever of them has been updated more frequently. It also tracks the leads. So if one has been updated frequently but never deleted, the one that has been deleted or inserted a couple of times would be in, but there’s still going to be cases where these things don’t work. So it still does a tie-break on the value, which can be useful for numerical data.

Then lastly, on the node ID, because each node has its own ID, so it does the last tie-break. It’s actually fundamentally very simple, it has a very simple way. But also in the way that we use Corrosion is, I mean there are times when there are conflicts, but it’s also very useful in the system where there are different parts of things, open data, so even though there’s possibly conflicts, we know that most rows do not actually have a lot of things. It’s really surprising what comes back from the database.

Srini Penchikala: Do you have any additional comments in general in distributed data systems and networking areas?

Somtochi Onyekwere: I mean, there’s a lot of stuff. I think where CRDTs become very interesting is when they’re being used in editing software like text, but things become more complicated there because there’s very granular stuff. You have to track the characters, where stuff is being inserted, where stuff is being deleted and a lot of times CRDTs provide a way to resolve conflicts, but the application still has to decide how does it send out the data? How does it synchronize dates? So those are just actively interesting areas to me in how CRDTs are being used everywhere.

Srini Penchikala: And definitely with the data being a little bit more textual and even other data types, we definitely have a lot of use cases. Somtochi, thank you very much for joining this podcast.

Somtochi Onyekwere: Thank you very much. Yes, I’ve had so much fun. I feel like this is my first podcast. I was a bit anxious, but it’s been very lovely talking to you. Thank you so much.

Srini Penchikala: Yes, sure. It’s great to discuss one of the important topics, right? Sometimes the CRDTs, the consistency doesn’t get as much attention as it should.

Somtochi Onyekwere: Yes,

Srini Penchikala: Definitely a good discussion. And for our listeners, thank you for listening to this podcast. If you would like to learn more about data engineering topics, please check out the AI, ML and data engineering community page on infoq.com website. I encourage you to listen to the recent podcast we are going to be publishing the AI ML trends report very soon, so please check it out. And also other trend reports for Architecture and Agile and other areas. The InfoQ editors here who are also practitioners like you assess what’s coming up in these areas and we publish those trends reports. Thanks, Somtochi. Thank you.

Somtochi Onyekwere: Thank you.

Mentioned: