Transcript
Greef: The world is becoming more transactional, from servers to serverless, and per second billing. You used to buy a server every three years or rent dedicated by the month, then an instance by the hour, and now every second there's a serverless transaction. In the same way from coal to clean energy, our energy sources are changing to be more transactional. You have smart meters around the world that now transact energy every 30 minutes instead of once a month. It's an increase of 1440 times. Then there's the tsunami of instant payments. For example, India's UPI processed 10 billion real-time transactions in 2019. This year, in the month of August alone, UPI processed 10 billion transactions. India went from 10 billion a year to 10 billion a month, and will triple over the next 3 years. If India is leading, then other countries are not far behind. For example, Brazil, 3 billion a month, growing 200% a year. This is spreading, for example, to the U.S. with the introduction of FedNow. The world is becoming more transactional. The volume of online transactional processing or OLTP, across several sectors has grown by three orders of magnitude. Yet, the three most popular OLTP databases today, Postgres, MySQL, and SQLite are 20 to 30 years old, designed for a different world and a different scale, which I think is a testament to the people that design these databases. That they still power everything today, and at the same time, in the same 30 years, we've seen advances in hardware and research.
There are hints of the need to redesign OLTP for a new order of magnitude, starts with a creeping dependence on caching, escalates to deployments that have 64 shards of Postgres or MariaDB, and capitulates to the extreme of replacing the OLTP database entirely. For example, the switch that powers India's UPI runs on open source, but not open source OLTP. Instead, they append transactions to Kafka, then they've rewritten the transactions' processing logic in Java services above Redis. They're doing around 10,000 logical payments per second at peak, and 10 times that many physical database queries, so 100,000 transactions a second in terms of database transactions is the peak OLTP load. Yet even a tenth of this, 1000 TPS for systems a fraction of the size of UPI is still not easy to do. The other problem is that open source OLTP databases are not online, if you interpret online in the mission critical sense. They are a single node out of the box. You get synchronous replication, yes, but there's no consensus algorithm baked in, no Raft or Paxos. Failover is manual, and you risk data loss. I think this is why, on the other end of the spectrum, you see mission critical systems start to abandon open source entirely. They move to proprietary cloud databases to trust the cloud provider with the problems of scale, availability, and cost efficiency.
Redesign OLTP by Three Orders of Magnitude
How can we do better? Can we redesign OLTP by three orders of magnitude? We'll see not only that it's possible, but that it's exciting and needs to be done, especially if we change the way we think about scalability. I love this paper by Frank McSherry, because it shows, I think, perhaps we focus too much on the scalability of our systems and we've lost sight of the need to optimize our unit of scale. It's common in software or distributed systems to define scalability as doing more with more. You handle orders of magnitude more workload with orders of magnitude more machines. The trouble is that OpEx gets out of hand, firstly, operating expenditure, so you pay for more machines. Secondly, operating experience. One thing manages a single node server, it's another to manage 1000. I prefer the definition of scalability from industrial engineering, do more with less. You handle orders of magnitude more load with the same machine. You start to ask the question, how can we take our unit of scale, a single machine, and redesign the design so that we can go from 1000 real-world transactions a second to a million transactions a second? Thinking in terms of the exponent like this, tweaking that exponent. It's a fun challenge, because there are only so many microseconds in a second. To process a million transactions, your budget is one microsecond per transaction. It's not very long. To make it no less easy, we're also not going to take shortcuts, so no magic tricks, no eventual consistency, no weak durability here. We're going to have to work hard together if we want to handle orders of more scale with the same machine, and in fact, better guarantees, because the default isolation level, for example, in Postgres is actually read committed, which is not so safe. You might query an account balance, see that there's enough to transact, but then query again and see that another database transaction has withdrawn the balance you thought you had, so we need to offer the highest level of isolation and strict serializability. To do this, I think we need to think from first principles. To ask the question, how can we optimize our unit of scale? How can we take the four primary colors of computer science, network, storage, memory, compute, and blend them into a more optimal design?
TigerBeetle
This was the question we asked in July 2020 as we began to create TigerBeetle. At the time, we were working on an open source payment switch, we wanted a drop-in open source OLTP database to power the switch more cost efficiently. We wanted to tweak the exponents, and nail the scale with a distributed financial transactions database optimized for OLTP. Our goals were to improve performance by three orders using the same or less hardware, and for the design to be significantly safer and easier to operate. TigerBeetle is licensed under Apache 2.0, my favorite open source license. We're approaching our first production release. The dream is to run OLTP at any scale in any environment. Indeed, the beetle we named TigerBeetle after thrives in all kinds of environments. Also, one of the fastest creatures in the world, sprinting at 5.6 miles per hour, 125 body lengths per second. Scaled for size, the TigerBeetle is the fastest insect and land animal on earth. TigerBeetle is meant to be fast and small.
Everyday Business Transactions (OLTP, OLAP, OLGP)
What is OLTP? What do we mean by transactions, database transactions, or are database transactions named after another kind of transaction, the everyday transactions, everyday business in every sector, the who, what, when, where, why, and how much of business? I find it fascinating that the first popular OLTP benchmark was literally called DebitCredit. This wasn't just any benchmark either, so I love the Anon et al., anonymous, but in fact, there were 24 et als., 24 co-authors collaborating with Jim Gray, the transactions processing guru, who at tandem at the time also coined the term ACID, also gave us the 5-minute rule. Jim's DebitCredit benchmark inspired so many benchmark wars that soon the Transaction Processing Performance Council formed to standardize OLTP benchmarks. Then, again, the historical context of transactions was all financial, banking, banking, warehouse, brokerage. Once upon a time, OLTP actually included analytics in there somewhere, you didn't have the term OLAP. General purpose is also somewhere in between. Until 1993, Edgar Codd coined the term OLAP, as online analytical processing, and OLAP spun out of OLTP. Where OLTP is interested in the who, what, when, where, why, and how much, OLAP is more interested in the why and what if.
Where OLTP is write-heavy with a high volume of simple queries, OLAP is read-heavy with less volume and complex query. OLTP is actually not really a query execution engine. That's all the OLAP stuff. OLTP is just, how do we get stuff in the front door? OLAP is like, ok, now, how do we query?
After OLAP spun out of OLTP, object storage quickly followed suit, and we no longer store blobs in our OLTP database, we use the database as a queue. Today we use OLTP for business transactions, and anything general purpose, which means that the OLTP database is forced to generalize, and there's no separation of concerns. I think you see where I'm going with this. Just as DuckDB is better at OLAP than SQLite, this frees SQLite to focus on general-purpose workloads. I think we can unlock scale if we separate our general-purpose processing from OLTP. If we take all the metadata relating to business transactions, we move it out of the data plane into the control plane for OLGP to continue to carry the mixed read/write control plane workloads that have less volume than a data plane, which remains OLTP. If we apply separation of concerns to the OLGP control plane, and OLTP data plane like this, then I think we can start to reach three orders, in three design decisions across storage engine, consensus, and network.
Network Protocol (Client/Server)
Here with respect to the network, we find that scale increases aerodynamic drag. As you go through three orders of magnitude, the pressure on everything increases. Anything that's not isomorphic burns up. You suddenly see an impedance mismatch where there wasn't one before. That's because while the language of database transactions is SQL, the language of business transactions in the real world is double entry accounting. I would never want to replace SQL for analytics. It's great. Send one query in, get a lot of analysis out. There's also a lot going for double entry. If your workload is OLTP, and you want to track business transactions, the Jim Gray DebitCredit, then double entry is really the perfect schema, which means you see this tension in query amplification. Here it is, a switch wants to execute the DebitCredit logic for one financial transaction, but this needs on the order of 10 read/write database queries. You can use stored procedures to get this amplification down to 1. Then a million transactions a second is still a million network requests a second, more with more. It's not more with less. What if we could invert query amplification to our advantage? What if in one database transaction, we could process 10 business transactions, 1000 business transactions, even 10,000. This would give not 3 orders, but 4 orders more scale in one network request. One network request, and we're doing 10,000. One request, 10,000. That's a lot more with less. Then we can spend the extra order on safety. The advantage of double entry as simple flexible business schema is not only that it can handle any kind of OLTP business transaction, the who, what, when, where, why, and how much, but that it's standardized. You can pack a double entry transaction in a fixed size struct, put 10,000 of these in a single network request. You've got a UUID for distributed idempotency, debit and credit account ID to describe the who, the amount or the how much. This amount can be any unit of value, not only financial. Then the transaction timestamp, obviously, is the when. Finally, the transaction metadata which links back up to your OLGP control plane, and you've got your user data to record the why, what, and where. This is really what these systems need to do. This is what they all do.
If you have an OLGP database, again, in your control plane, then your data plane can operate on thousands of these 128-byte transactions, or 2 CPU cache lines. In a single 1-Meg query, you can process 8000 transactions and with low latency as well. Because if you have a single transaction that you need to do, you just send it right away, you don't batch. Then as your load increases, your requests fill up and you get more transactions per query. This gives not only the best throughput, but counterintuitively, also a sweet spot for latency because your system now has more capacity, more mechanical sympathy. For example, in your server with a single receive score, you can receive 8000 transactions from a client. If we then reuse this wire schema for the disk schema, we can write this 1-Meg request as-is to the write ahead log, plus replicate the same in parallel to a backup machine. This gives crash consistency, amortizes fsync very nicely, and gives you the best performance when you use direct I/O to do direct memory access to the disk. Direct I/O is actually faster than buffered I/O when you do it right. It's becoming more important also when you think about where network, storage, memory, bandwidth are going.
This graph by Roland Dreier, shows how network, storage, and memory bandwidth have increased, even just the last three years. What's interesting is not that everything got faster, everything did get faster: network, storage, memory, bandwidth, everything got faster, but look at that little flip. The relative ratio has completely flipped. Just three years ago, network was the bottleneck, then storage, then memory. You burned memory bandwidth to make the most of your network and storage bandwidth. You had Bloom filters. Today, everything has inverted, the bottleneck is now memory bandwidth. It's like drinking from a firehose, but through a straw. This is an incredible time in the history of database design. I haven't figured this out, it's like gravity just inverted. This is why we put so much effort into TigerBeetle into how we work with memory. Fixed size cache line aligned data structures. Little Endian is everywhere these days, so we no longer wanted to burn memory bandwidth on deserialization or memory copies. This in turn avoids thrashing the CPU cache, but most of all, we wanted to invest in static memory allocation, not the C technique, but the principle of zero memory allocations up to startup. At startup, we allocate everything we know that we need, and at runtime, as you run TigerBeetle, there's no more malloc or free: no malloc, no free. It's forgotten. It doesn't happen. The motivation for this is really the second order effects of the spring. Your memory layout is now handcrafted for efficiency, and locality of access. There's no waste, no hidden runtime allocation latency, no risk of global allocation panic, if you push TigerBeetle to the limit. No risk of fragmentation. Because memory bandwidth is so important for where we felt systems programming was going, if it was a choice between C, or Rust, and Zig between the last 30 years of systems programming, or investing in the next 30 years, then memory efficiency was really why we decided to invest in Zig. A highly readable, explicit language, great for writing a database, where memory bandwidth is more the bottleneck.
After appending to the log with a minimum of memory bandwidth, the next step is to execute the transactions, to apply them to our existing state using the architecture of a replicated state machine. We take the same inputs, we replay them in the same order through the state machine on every machine, to get the same state across the cluster. It's such a very simple technique. You've got a distributed log. All your machines processes in the same order, same business logic, same state. Finally, we send a small reply back to the client. This is an interesting trick in TigerBeetle because we assumed success, so that we only have to return error codes for the transactions that actually failed. You save 8 kilobytes on the wire. This is how TigerBeetle processes 8000 transactions in a single query, one database query with 4 syscalls, 4 memcopies, and 3 network requests. Do more, with less. These are really massive wins, it could be 30,000 syscalls or something like that. The biggest win with this network protocol is that it eliminates the need to take 16,000 row locks on accounts, for every transaction is two accounts. If we understand that the OLTP workload is business transactions, then we can see why a transactional workload can mean contention. Because if your debit or credit is against a cold account, then the contra account is almost always a hot account. I'm going to explain this again. Just bear in mind, in a general-purpose OLGP database like Postgres, SQLite, MySQL, this would force row locks. For example, you have a million customers, you can spread your updates across a million rows, you can horizontally shard, no problem. All your writes still need to be serialized through your bank account, and you only have one or you have four. The effect of row locks then is worse when you go horizontally distributed. Because when you shard across machines, now transactions have to talk across the network, but they're still a bottleneck on the bank account, which is on one shard. You get this effect also in OLGP databases as a single node, but this compounds, especially when you consider how compute advances relative to network latency.
There's a law, Moore's Law, that the number of people predicting the death of Moore's Law doubles every 2 years. I say this, because if you look back 10 years ago if you thought that Moore's Law was slowing, and I think I thought so too, you can understand why someone might have wanted to scale compute horizontally. However, 10 years later, Moore's Law is still on track. I think, this year or next, we're going to get to the M3 from Apple with 3 nanometer process, 100 billion transistors, we're on track. Yes, a single core has advanced since 2015, which is, after all the cloud databases, after all the horizontal scalability strategies, a single core is now an order of magnitude more powerful. All this time, the speed of light and fiber has remained constant. If you bet on that, if you bet on horizontal for compute, compute is an exponential resource. If your CPU has to wait on a network request to a shard to complete a transaction, this is going to appear slower every two years as Moore's Law progresses. You don't want to make the CPU wait on the network. If you do, if you make that decision, it's going to be twice as costly in two years. What TigerBeetle does then is to keep the hot, contentious transactional data local to every CPU core in the cluster. We replicated across all machines: you can, it's small. The principle again is separation of concerns because compute and storage scale differently, instead of coupling compute and storage, scaling only vertically or only horizontally, you can blend these techniques to scale diagonally. You go vertical with compute, and with the hot storage replicated across the cluster, and then you go horizontal to call the remote storage, object storage tiering, if necessary. We haven't needed to do this yet with TigerBeetle. I'm going to speak to the first two.
Storage Engine
Now that we've looked at the network, how to use it, and when not to, I want to look at TigerBeetle's storage engine. When transactions execute through the state machine, the results of this execution go into the storage engine. Transactions come off the wire, execute the logic, into the storage engine, into the in-memory components, and then they spill out to disk. This execution is interesting, because as we execute, there's no I/O going on. Again, Moore's Law. The CPU levitates, doesn't touch disk or network. Instead, before execution, we pre-cache all data dependencies into memory from disk, so they're really in-memory when execution happens. Then we execute the transaction serially, and it's all DebitCredit objects, they're all monomorphic, so the instruction cache is hot. There are zero row locks. The CPU becomes an Olympic sprinter that you let loose some 100 meters. From here, they insert spill-out to disk. For this, you have two main choices, B-Tree, LSM-Tree. B-Tree is optimized for read-heavy workloads at the expense of random writes. LSM-Tree is optimized for write-heavy workloads at the expense of multiple reads to find an object. You can always use caching to optimize those reads. The original LSM-Tree paper was published by O'Neil, 1996. Then a month later, Postgres was born. The three most popular OLTP or OLGP databases, they're actually optimized more for read/write OLGP, or read-intensive OLAP. They don't actually have the write optimized LSM-Tree engine, which is what you want for OLTP. This makes a difference, for example, Meta took MySQL, they swapped the engine for RocksDB and LSM-Tree, and they found that MyRocks was 10 times more write efficient, that's their phrase, from that blog post, 10 times more write efficient. Even RocksDB is now more than 10 years old. Most of the LSM research since then, it's recent, so today you can leapfrog. That's why if the best time to plant an LSM-Tree was 20 years ago, I think the second-best time is now.
I want to show you a few LSM techniques we've developed at TigerBeetle for predictable performance. When we talk with LSM researchers, for example, at FORTH and Crete, they tell us, they routinely see, in their words, 1 to 10 second write stalls in existing LSM-Tree. A client is trying to make a request, something goes wrong in the tree and it's a 1 to 10-second p100. The problem is that there's no end-to-end control loop between the client requests coming in and the background compaction work that needs to keep up. If the engine can't keep up, then it blocks client requests. There's been a lot of work in RocksDB to improve this. It's not a guarantee, it's not a solved problem. That's because to fix it, compaction really needs to be able to predict how much work the next client requests will generate, predict the future. Without limits, you can't predict the future. We wanted to eliminate these writes stalls, we wanted to have a hard guarantee. Because TigerBeetle does have limits on all work coming into the system, we could design compaction to be just in time, so compaction does just enough work to be able to absorb the next client request and no more. It's choreographed. You don't worry, you just need to get the big picture here of something being choreographed. I'll tell you what this is. We divide time into bars, bars into beats. Then each request and the corresponding compaction in terms of CPU, disk, and memory, it's spread out incrementally across these beats. We think of this like GC, and this is how you can move from stop-the-world, to paced, jitter-free compaction.
Another technique is how to deal with the problem of using one tree for all your data. If you store different types and sizes of data in the same LSM-Tree, then you get this tug of war, how you optimize the engine, as described in the RUM Conjecture, by Athanassoulis, Idreos, Callaghan, and others. The RUM Conjecture is that you can optimize for reads and writes, but you pay for it with more memory or space on disk. Pick any two of three formula. I think the context for the RUM Conjecture is you're storing everything in one tree, trying to optimize the tree for all workloads, one tree and now you try to optimize for all workloads. What if this is not always the case, could we then push past the RUM Conjecture? What if we broadened the model to assume many trees instead of one? Then we have each tree adapt to its individual workload to have workload awareness. We go from RUM, to RUMBA like this, can we then find a more efficient Pareto frontier? We found that if you store different key-value sizes in the same tree, then you need to have length prefixes. Writing these length prefixes to disk increases write and space amplification, you're burning more write bandwidth, using more space. For example, if you have many secondary indexes with values of 8 to 32 bytes, then a 32-bit length prefix consumes 10% to 30% of write bandwidth and space overhead. The 30% is a lot. However, the worst part of putting all your data in one tree is that different key values have different workloads. Half your data, for example, transactions might be immutable. It's double entry. If you store these in the same tree as your secondary indexes or your account balances, you're compacting and churning your immutable data again and again, for no reason. When you mix your data, it also becomes harder to find, because you're looking for a needle in a haystack of wood, brick, and straw. What if we just had a wood stack and a brick stack and a straw stack or haystack? The insight then is not to miss the forest for the trees, but to go from an LSM-Tree to an LSM-Forest. If you store disjoint data in disjoint trees, you can optimize writes, reads, and memory. You can order tune according to kind if you have a tree for every type. For example, TigerBeetle's forest is around 20 trees. This reduces read, write, and space amp, improves cache locality as well because now you have everything tuned according to your access patterns. This is the second technique in TigerBeetle's storage engine.
The third derives from the fact that more scale demands more durability. In a study with NetApp, Bairavasundaram, UW-Madison, found that commodity disks have a half a percent chance of corruption, or silent bit rot in a two-year window. At scale, this thing becomes common. Probability theory, you go from one disk to 100, you start to expect a 50% chance of running into compaction somewhere in your fleet. UW-Madison under Remzi, Andrea Arpaci-Dusseau, they've done wonderful storage fault research like this over the last 10 years. However, SQLite which was first released in 2000 before most of this work, does not add any redundancy to the database file for the purpose of detecting corruption or I/O errors. SQLite assumes that the data it reads is exactly the same data that it previously wrote. Postgres and MySQL have a similar crash consistency model. That's what you call it, crash consistency model. They aim to be consistent in the event of power loss, but they're not actually designed to recover from or even detect storage faults. You can understand this because, again, much of the research into storage faults was done only after these databases were designed. This can and does lead to data loss. 2018, this became known as fsyncgate. Users found that Postgres' handling of surprising fsyncgate in the Linux kernel, the way that Postgres handled that could actually accelerate what was an otherwise routine, recoverable, just a storage fault, but it could accelerate that into data loss. Just a temporary sector error, and that could lead to data loss. Postgres and MySQL were patched for this to panic, rather, and then recover from the log at startup. Then in 2020, UW-Madison looked into it and they found that the fix was not enough. This is not well known. At startup, when Postgres recovers the log, it actually reads from the kernel page cache in-memory, it doesn't read what was durably synced to disk, so it doesn't really know what's durable or not when it commits transactions. The risk of data loss is unfortunately still there. This is going to be fixed when Postgres finishes adding support for the direct I/O. There's a lot of movement there coming into Postgres even this year.
The third storage technique in TigerBeetle then is to move beyond a crash consistency model, and to actually design for an explicit storage fault model, to assume that storage faults do happen. You can expect latent sector errors, corruption, or even where the firmware or whatever file system will send your reads or writes to the wrong sector, or just not even do I/O at all. There are a myriad of detection and recovery techniques that TigerBeetle uses to solve this model. For example, to checkpoint the log ring buffer every time it wraps, we do special read/write quorums. We treat the disk like a distributed system and we check for quorum overlap. We use write and then we verify the writes. We read it back in, write verify techniques. We store checksums out of band, so the parent knows the checksum of the child in case the I/O to read the child is misdirected, it's a ZFS technique. Then we also have a second small write ahead log. We have two write ahead logs in TigerBeetle. The small one is for metadata to enable TigerBeetle to determine whether the log is corrupt in the middle of the log because of corruption, or if it's torn at the tail because of power loss. As another example, this is actual TigerBeetle code, it's a comment in the source. It shows how we enumerate all these kinds of faults under the model to see how they influence our recovery at startup. Because it's a matrix, the recovery actions can be generated dynamically, tested to be sure that every case is handled. As you start to accept that disk sectors do fail, you also start to see another problem with existing engines. That's because the data file they produce is not deterministic, what they produce on disk. If a single disk sector fails on one machine, the data files in other machines are all different, so you have to recover the whole data file from another machine. This can take hours or days, gigabytes, terabytes. To solve this, TigerBeetle's just-in-time compaction is also deterministic, so that the same log always produces the exact same data file across every machine in the cluster. The storage engine is deterministic, what it produces on disk, which is very unique. Now if you see that a 64-kilobyte block is corrupt, that's all you need to transfer over the network.
At this point, I'm sure you're wondering, why not RAID? Why not just put Postgres on RAID, or SQLite on RAID? Of course, the answer is that if you're really replicating your data across three machines for three times storage overhead, then extra local RAID redundancy triples this, now you're 9x storage overhead, but you also don't need to because you have global redundancy that you can tap into. It's more efficient to take your storage engine and integrate it with your consensus protocol, then you can recover from local faults using your global redundancy. You can also share one log between the storage engine and consensus. In a lot of systems, there's actually two logs, so you're halving your write bandwidth. If you can share the log, you double your write efficiency. These are all techniques in TigerBeetle storage engine to solve write stalls, push the limits of performance plus RAM, increase durability, optimize recovery.
Consensus Protocol
These are all integrated with the consensus protocol. We've already looked at how we optimize the network to process 8000 transactions in one query. We use the same technique to make consensus just as cheap. Instead of replicating transactions one by one through the consensus log, a single entry in the consensus log replicates 8000 transactions, one consensus commit, 8000. It's the same idea. It's very powerful. This is only, again, one round trip to backup, to replicate, optimal because you need data on more than one machine for durability anyway. Where consensus comes in is when you want to use this redundancy, not only for durability, but also for high availability, for automated failover when your primary machine fails. It's not enough to have synchronous replication, you also want automated failover. We do this using the pioneering replication and consensus protocol viewstamp replication by Brian Oki, Barbara Liskov, James Cowling. It was actually published a year before Paxos, and revised again two years before Raft. Raft is almost exactly the same protocol as VSR, names are changed, but it missed out on optimizations that Liskov and Cowling introduced in a 2012 VSR paper. If you haven't read it yet, I think this is the most intuitive of the consensus papers, also more optimal, which we're going to go into now. For example, in Raft, you don't know ahead of time who the next primary will be. If the primary current leader crashes, who's the next leader? You don't know. Whereas in VSR, you do have a pretty good idea, because the election of the new primary in VSR is actually round robin. Each machine has what is called a view number. They bump this number when a quorum of the cluster confirms that the old primary is down. Then a simple modulo tells you who the new primary is. That's consensus, like I've just explained it to you. It's so simple, but even more powerful because you've got more information encoded in the protocol.
There's also no protocol in Raft actually, to repair the log of what otherwise might be a perfect candidate, so Raft can sometimes get stuck. For example, you have three machines. Replica 0 on the left is primary, and then you've got two backups, so leader and two followers, but the classical terminology is always primary and backup, which is what we use. Replica 0 as the primary has three entries in the log, replica 1 has three, replica 2 only has two. It hasn't got the latest log entry. Then replica 2 crashes. Raft here would have to elect replica 1 as primary. This makes sense because replica 1 has a longer log than replica 2, it has more data. However, what if replica 1 has a sector fault in the first entry of its logs? In this scenario, your Raft cluster is actually stuck. It has to wait until replica 0 recovers, whereas VSR is able to elect either replica 1, or replica 2, and then transfer the missing log entries across. In other words, by limiting who can be primary, Raft doesn't fully utilize the global redundancy that you're paying for, as it could. At scale, this matters. Raft's formal proof also assumes stable storage, but that the stable storage disk is perfect, so there's no idea of storage faults. Whereas VSR can run only in memory, so it's a very nice protocol for prototyping, if you are afraid of working with the disk. VSR can also run with stable storage like Raft, which is what we do with TigerBeetle, except the storage engine that we have can detect disk faults and cooperate with VSR to recover. One of the ways we do this is by extending the consensus log with a cryptographic hash chain of the checksums of our ops that are going through consensus.
This means we can go beyond Raft to handle situations even where all machines have a corrupt log, but in different places. Here we've got ops 1, 2, and 3 corrupt across different machines. In Raft, the cluster is lost, there is no way to recover, in terms of the Raft protocol. What we do is we've got a checksum from C back to B, and then from B back to A's checksum. TigerBeetle can actually stitch this log back together and keep going. The hash chain also means that the primary doesn't need to have an op on its own disk before it commits. Usually, you append to your disk, replicate, then you commit, but you got to wait for your disk. What if the disk is slow? This technique, you don't have to wait for your disk, you can just go as fast as the fastest two of three disks in your cluster. Then backups can also receive ops out of order. They don't have to try and first catch up, they can take the op from the primary, say yes, I got this, and then they'll catch up in the background. Whereas in Raft, normally, they have to block the primary, and say, "Wait, I've got a gap, let me go and repair first and then I'll tell you it's ok," keeps the client waiting. This technique can also optimize backups that they give, very fast access to the primary.
So far, I've only shown you cases where Raft can get stuck, and that's only because it wasn't designed for storage faults. That's fine, it doesn't have a storage fault model. Again, new research since Raft was designed, protocol-aware recovery for consensus-based storage. If you're writing a consensus protocol, how do you do storage? This is really the paper you want to look at. Alagappan, Ganesan, UW-Madison won best paper, FAST 2018. Again, it's so recent, so we just didn't know this when Raft was designed. This shows how with Raft and Paxos, a single sector fault on one machine can actually just ruin your whole cluster and cause global cluster data loss. It can propagate through the protocol. For example, if the first open replica 1 is corrupt, we've seen this example, then its checksum isn't going to validate. What typically happens at startup, replica 1 will see, this checksum didn't validate, I must have been writing this and then the power went off, again, the crash consistency model. Then replica 1 will truncate its log all the way back to 0. You lose all those committed transactions. That's really because it's conflating the checksum error with a torn write off to crash. Actually, that was just corruption on disk, which is a half percent chance every two years. This will erase committed operations, and you get split brain and everything is messed up. In TigerBeetle, we implement protocol-aware recovery, and we ask the cluster to figure out, what is the correct action to take to recover from the storage fault?
I want to show you one more trick that we're working on still. Everything I've shown you here is in. We're fine tuning our compaction pacing, but everything is in. This is the insight with consensus. Normally, you want to replicate an op across your cluster, you wait for a quorum, then you execute, so you always replicate, get quorum, execute. The insight is that 99% of the time, your ops are always going to get the quorum. It's only if like, the primary is down, or some machines are down that you don't get the quorum. Ninety-nine percent of the time, why not just execute? Let's append to our disk, let's replicate, and let's execute through the state machine in parallel. We can remove that write barrier, then we just wait. We've executed, we cache that result, and when we do get the quorum, then we apply it. Basically, you're getting a head start on your state machine execution. Then you only throw the execution away if the op actually doesn't receive quorum. You get a round trip time of head start. This is really powerful, because this is basically saying consensus is now 100% free. Whether you're doing single node or cluster, doesn't matter, because that round trip, you're using it to execute CPU. Moore's Law, today, it's 15% win, 2 years' time it's a 30% win, 4 years' time, 8 years' time, then your CPU is even free.
It's been 3 years since the start of TigerBeetle, production release around the corner. With the design decisions I've showed you, TigerBeetle is able to do today 988,000 transactions per second. We're going for a million, we should get there. We've got some more techniques to come. That's on NVMe with primary indexes, everything you need to do Change Data Capture, a million a second. If we then index all columns in TigerBeetle, so that we can do very predictable, multi-attribute queries with zigzag merge join, which is fantastic. I actually think Postgres and MySQL don't have zigzag merge join so it's something that's newer again, but it's a great way to do multi-attribute queries, also in the works. If we add 20 secondary indexes, so we add a lot of indexes now, then TigerBeetle is still able to do 200,000 to 500,000 transactions per second, business transactions. That's again 20 secondary indexes for nice multi-attribute scans. The p99 is under 100 milliseconds.
Redesigning OLTP, Safely
More important than performance is safety. How can we hope to be safer than things that were 30 years tried and tested? You have to be a guru to get this right. I know a guru or two, we're certainly not. Instead, we adopted NASA's Power of 10 Rules for Safety-Critical Code. It's not a standard you see very often, but it means that there's 4000 assertions in TigerBeetle as it runs, it's checking itself. There are limits also on all the resources, memory, like I've showed you. Even concurrency even of while loops, we've put a safety counter on. Everything is worked out. We know how much resource any algorithm will use in terms of memory especially. You get a piece of software that is rock solid, it's got a well-defined shape, you can depend on the shape, it's not going to change. All these limits are known statically at compile time. Zig's got amazing compile time, obviously. Finally, we designed TigerBeetle not only as a distributed database, but as a deterministic, distributed database. This means then you can take this whole database cluster, you can run it in a simulated world that's deterministic. You can simulate different latencies of network, storage, different bandwidths. You can inject all kinds of faults, very high levels of storage faults. We do that. Then you can verify correctness and liveness of the consensus protocol, of the storage engine, up to the theoretical limit. You can say, I'm going to corrupt every log on every replica in different places, can you stitch that log together? Can you keep going, or do you shut down prematurely? Because this testing is also completely deterministic, you can replay bugs from a single seed, and then you get this incredible rapid debug velocity.
The most valuable thing with simulation testing, is that you can actually speed up time. Obviously, this is inspired by FoundationDB, who've really pioneered this, our heroes. Time itself is simulated, so you just speed it up in a while true loop, tick that second hand while true in the simulated world. For example, if you run the TigerBeetle simulator in your terminal just for 3.3 seconds, you've got the equivalent of 39 minutes of real test time on average. You run for two days, you get two years of test time. We run 10 of these simulators, 24/7, just burning CPU to put years on the clock. This is actually verifying the consensus, the actual code, the actual implementation is verified like this. It's all virtual. You're running a whole cluster of real TigerBeetle code in your terminal, all the I/O, the time is all simulated. This also means this is a whole simulated world. What if we can transplant this world? What if we wanted to, you could do something fun, like you could compile this to Wasm, and you could run this whole simulation in your browser also? Of course, this is what we went and did.
TigerBeetle Simulator
This is what I want to leave with you now, https://sim.tigerbeetle.com. I was walking on Golden Gate, and I thought, we've got this simulated world, what if we can take this cluster of tiger beetles running real TigerBeetle code, and just teleport them onto Golden Gate Bridge? Of course, this is what we went and did as a team. Now we're going to see no faults at all, no processes crash, network is perfect, storage is perfect. Golden Gate Bridge, traffic has stopped, the tiger beetles descend. You can see here, Captain America is the leader. The client requests at the top, they're sending the requests in. Cluster has started. The beetles don't even know that they're at QCon on Golden Gate Bridge. This is real TigerBeetle code, all VSR consensus. You can see the replication happening, and the ACKs and the replies going back to clients. All the faults on the left, everything is perfect.
This is now Red Desert, Andy Pavlo is in the bath doing his bath lecture that is famous. Here we've got network faults, so we're going to mess with the network, drop network packets on the ground, partition the beetles in that Mission Impossible 5 glass box. You can see the cluster is now starting to do leader elections. If you look at the primary you can see it's going round robin because it's VSR round robin. It's moving clockwise around the circle. You can see the one that holds the helmet when it's orange is the next primary. We added a few fun tools here. Let's find the primary, which will be this one. This is quite a hard game to play, because they automatically fail over so fast. There's our primary. Wait, we will succeed. There we go. You can crash it, and then watch it recover.
This is Radioactive. How does your database do if you just corrupt the disk, 8% of reads you corrupt them? Every machine is writing to disk and you're corrupting 9% of the time. Let's just do that because it's a simulation, see what happens. We found so many bugs in our code like this. Now we're actually exercising the whole storage fault model. We zap the tiger beetles with cosmic rays, and let's see what happens. The simulator is testing strict serializability just like Jepsen would, but it does it in real time, not after the fact. You can see we've got different latencies simulated, different packet loss, different read and write corruption, 8%, 9%, and everything is working, otherwise the simulator would crash. If you're a fan of DuckDB, there's a little Easter egg, if you click that duck. You can engage duck mode and see if OLTP can survive OLAP, which is not easy to do but TigerBeetle can do it. We put this in there just for our offerings there DuckDB. You can also do a bit of lightning, which is very violent. TigerBeetle survives and this is just thanks to FoundationDB, what you can do if you simulate everything deterministically, put all your faults in.
See more presentations with transcripts
