Scaling the Big Peer Store
The Ditto Big Peer may appear like any another peer in a mesh, but it is different. It is a distributed database that is made up of a number of components, including:
- Subscription Servers that communicate with Small Peer meshes using the Ditto Replication Protocol
- API Servers that host the HTTP API
- A distributed transaction log
- A partitioned and replicated distributed database: the Big Peer Store
- A Change Data Capture (CDC) system that monitors changes and provides them to third-party systems, either through pre-defined Kafka topics or WebHooks.
This makes the Big Peer both a system of record with data at rest, as well as a a conduit of data from existing data systems, into the mesh, and back out to existing systems.
This article focuses on the Big Peer Store, specifically how the Store scales up and down, while continuing to provide transactional causal consistency.
The Big Peer Store
To understand how we can scale the Big Peer Store while maintaining functionality, it is first necessary to describe how it works. What follows describes the key primitives of its design and shows how we scale the system using transitions.
The Big Peer Store is inspired by the paper PaRiS: Causally Consistent Transactions with Non-blocking Reads and Partial Replication.
Partitioned
The Big Peer is a distributed database that stores Documents as CRDTs. It can store more documents than can fit on a cheap commodity disk, and so it partitions its state by storing subsets of documents on multiple Store Nodes.
To ensure a balanced partitioning mechanism, the Store uses Random Slicing, partitioning the available keyspace using the range of integers from 0 to 2^64 − 1. Each partition of the Store is responsible for multiple sub-ranges of this keyspace, called Intervals. These sub-ranges don't need to be contiguous, as random slicing uses the Cut-Shift algorithm, which might take sections of pre-existing partitions and use them to make new partitions. This ensures that the minimal amount of data is moved when creating partitions.
Using this strategy, each document identifier is hashed to produce an u64, which in turn
is mapped to a specific Interval, stored by a particular Store node.
Replicated
The other axis of distribution for the Big Peer Store is
replication. Each partition is fully replicated, resulting in storing
multiple copies of each document. For simplicity's sake, we can say that every
Big Peer Store cluster is made of P partitions replicated on R Store
nodes. For example, a cluster of 3 partitions with 2 replicas will
have 6 store nodes.
The Log, Transactions, and Transaction Timestamps
All data enters the Big Peer Store via the Transaction Log. Each
transaction may contain updates, inserts, and deletes for many
documents. The log is totally ordered, where each transaction gets
assigned a position in the log (the first transaction gets position 1, the second 2, and so on).
In the Big Peer Store these log positions become Transaction Timestamps, such that the logical
time of a transaction is represented by a single scalar value.
Each Big Peer Store Node consumes from the log, sequentially and independently. Although a transaction may affect documents that are owned by different partitions, each Store Node will only store modifications for the documents that it owns. Independently of this, each Store Node records the fact that it has seen a given transaction by storing the Transaction Timestamp in its metadata.
Periodically the Store Nodes gossip to each other the maximum Transaction Timestamp that they have committed to disk.
Universally Stable Timestamp (UST)
The Universally Stable Timestamp (UST) represents the timestamp of the latest transaction that is guaranteed to be stored (stable) at every node, meaning any read at that timestamp will be consistent. The UST is calculated by computed the minimum of the latest stored transaction, for every Store Node in the system.
NOTE: We actually use a subset of the Store Nodes to calculate a stable timestamp, but explaining how detracts from the purpose of the article.
Store Versions
Each Transaction Timestamp can be thought of as a version of the
database, meaning that a store node can contain multiple versions for a given document.
This means that if a read starts when the UST is 100, it must be allowed to complete consistently,
alongside another read at a UST of 200.
Garbage Collection Timestamp
Keeping all versions of all documents forever is prohibitively expensive. Instead, the Big Peer Store Nodes independently perform garbage collection of document versions that will never be read again.
The Garbage Collection Timestamp (GC Timestamp) is much like the UST. When a read transaction is served by a Store Node, it gets assigned a Read Timestamp equal to the UST at that node. Each Store Node gossips their lowest active Read Timestamp, and in turn computes the GC Timestamp by taking the minimum of all read transaction timestamps. Any version below the GC Timestamp can be discarded as that version will never be read again. For a given Store Node with no in-flight reads, its local GC Timestamp will be equal to its view of the UST.
One can imagine a sliding window of visible versions, with the lower edge being the GC Timestamp, and the upper edge the UST.
Cluster Configurations
As described above, the Big Peer Store is made up of Store Nodes, each being a replica of a partition. The partitions themselves are made up of intevals of the keyspace. All this information is stored in the Cluster Configuration, which aids in mapping Intervals to partitions, and partitions to store nodes.
Epochs and Configuration-aware Read Timestamps
Each Cluster Configuration has a unique monotonically increasing identifier,
starting from 0 (the empty cluster configuration). This identifier is also
called the Cluster Configuration Epoch, or Epoch for short.
Recall that we described Read Timestamps earlier as a single scalar. We now augment this concept by making read transactions Configuration-aware, such that a Read Timestamp is pair of Epoch and Transaction Timestamp (UST). Only nodes in the current Cluster Configuration co-ordinate read transactions.
This pair of scalars, rising monotonically, are used to manage cluster Transitions.
Transitions
Transitions are the process by which the Big Peer Store moves from the Current Cluster Configuration to the Next Cluster Configuration, allowing the Big Peer Store to scale up and down. The number of partitions and replicas in a Cluster Configuration describe the size and shape of the Big Peer Store.
For example, assume that the Big Peer Store is deployed with a single partition and two replicas (thus two Big Peer Store Nodes running).
To scale up the deployment, we add a new partition. This means that half
the documents that were owned by partition 1 (P1) will now be
owned by partition 2 (P2). The Transition must be performed in such
a way that writes and reads progress, and all queries remain causally
consistent.
Using this concrete example (1P*2R -> 2P*2R) we can name the Big
Peer Store Nodes P1R1 and P1R2 for the two original nodes and
P2R1 and P2R2 for the two new nodes. The Current Cluster Configuration
contains one partition made up of a single Interval covering
the whole keyspace, while the Next Cluster Configuration contains two partitions:
P1 is made up of an Interval of the top half of the keyspace, and P2 is the
Interval of the bottom half of the keyspace.
Backfill
The first stage of a Transition is Backfill. In our example we can
start P2R1 and P2R2 as new Big Peer Store Nodes, providing them
with the Current Cluster Configuration and the Next Cluster
Configuration. We also let P1R1 and P1R2 know about the Next Cluster Configuration,
so that they too know about the Transition.
As soon as P2R1 and P2R2 are brought online, they will start consuming from
the end of the transaction log. Using the Current Cluster Configuration, they can
send a Backfill Query to the current owners of any data they will own in the Next Cluster
Configuration. In this example, P2R1 and P2R2 send backfill queries to P1R1 and P1R2,
asking for the data from the bottom half of the keyspace.
Ownership and Gossip
During the Transition, Store Nodes gossip their view of the cluster to the union of all Store Nodes from both Cluster Configurations. Store Nodes also consume and store Transaction data that they might own in either the Current or Next Cluster Configurations.
In this example P1R1 and P1R2 continue to store transactions for
the whole keyspace during the transition.
New UST and Routing Configuration
The UST is calculated within the scope of a Cluster Configuration. The UST for the Next Cluster Configuration cannot be calculated until all Store Nodes in that Configuration are backfilled. Once Backfill is complete, the Store Nodes compare the UST in the Next Cluster Configuration with the UST in the Current Cluster Configuration. When the Next Cluster Configuration UST is equal or greater to the Current Cluster Configuration UST, it means that the Store Nodes in the Next Cluster Configuration have all the data they need, and can serve queries. A Store Node in the Current Cluster Configuration can now route queries to the Store Nodes in the Next Cluster Configuration. Each Store Node makes these calculations independently, based on received gossip messages.
In order to ensure that a Transition eventually completes, the UST in the Current Configuration stops rising once all Store Nodes in the Next Cluster Configuration have Backfilled. Since the new Store Nodes begin consuming from the Transaction Log before they begin to backfill, the UST in the Next Cluster Configuration should immediately match that of the Current Cluster Configuration. However, this might not be the case when we scale the Big Peer to a much larger number of nodes, given that the gossip time can slow the perceived rate of UST rise.
Installing new Cluster Configurations
As described above, read transactions occur at a given Read Timestamp, which is a pair of scalars, made up of the Cluster Configuration Epoch and the UST. The GC Timestamp is the locally calculated minimum of all Read Timestamps received by gossip. This means that each Store Node can independently calculate when all Store Nodes in the Current Cluster Configuration are routing queries in the Next Cluster Configuration, because all Read Timestamps are in the Next Cluster Configuration Epoch.
At this point, any of the Current Configuration Store Nodes can declare that the Transition is complete, and that the Next Cluster Configuration is installable as the Current Cluster Configuration.
When scaling down, nodes that are no longer in the Current Cluster Configuration can be shut down.
Garbage Collection
Garbage Collection also checks for ownership. Any data not owned by
the Store Node will be deleted. In the example above that means P1R1
and P1R2 can delete all the data from the bottom half of the
keyspace that is now owned, and safely stored, by P2R1 and P2R2.
Summary
This post described the process for transitioning a Big Peer Store cluster for scale up purposes. It describes the process in terms of two primitives: Transaction Timestamps and Cluster Configuration Epochs. This pair of scalars are composed to make the UST, Read Timestamps, and the GC Timestamp. Store Nodes gossip to each other about their Read Transactions and their latest consumed transaction, which allows each Store Node to independently calculate the progress of a Transition.
This post skips many implementation details of the engineering to ensure monotonicity, fault tolerance, and liveness of this process, for the sake of brevity.
Acknowledgements
With thanks to my colleague, Borja de Régil, for his review, edits and suggestions.