Let’s think about correctness in the context of the ideas around unbundling databases. We saw that end-to-end duplicate suppression can be achieved with a request ID that is passed all the way from the client to the database that records the write. What about other kinds of constraints?
In particular, let’s focus on uniqueness constraints, such as the one we relied on in Example 13-2. In “Constraints and uniqueness guarantees” we saw several other examples of application features that need to enforce uniqueness: a username or email address must uniquely identify a user, a file storage service cannot have more than one file with the same name, and two people cannot book the same seat on a flight or in a theater.
Other kinds of constraints are very similar—for example, ensuring that an account balance never goes negative, that you don’t sell more items than you have in stock in the warehouse, or that a meeting room does not have overlapping bookings. Techniques that enforce uniqueness can often be used for these kinds of constraints as well.
Uniqueness constraints require consensus
In the related discussion we saw that in a distributed setting, enforcing a uniqueness constraint requires consensus. If several concurrent requests have the same value, the system somehow needs to decide which one of the conflicting operations is accepted and reject the others as violations of the constraint.
The most common way of achieving this consensus is to make a single node the leader and put it in charge of making all the decisions. That works fine as long as you don’t mind funneling all requests through a single node (even if the client is on the other side of the world), and as long as that node doesn’t fail. Consensus algorithms like Raft tackle the problem of safely electing a new leader if the current leader has failed (or is believed to have failed because of a network problem) and preventing split brain.
Uniqueness checking can be scaled out by sharding based on the value that needs to be unique. For example, if you need to ensure uniqueness by request ID, as in Example 13-2, you can ensure that all requests with the same request ID are routed to the same shard. If you need usernames to be unique, you can shard by hash of username.
However, asynchronous multi-leader replication is ruled out, because different leaders could concurrently accept conflicting writes, and thus the values are no longer unique. If you want to be able to immediately reject any writes that would violate the constraint, synchronous coordination is unavoidable.
Uniqueness in log-based messaging
A shared log ensures that all consumers see messages in the same order (the total order broadcast guarantee that, as we established in “The Many Faces of Consensus”, is equivalent to consensus). In the unbundled database approach with log-based messaging, we can use a very similar approach to enforce uniqueness constraints.
A stream processor consumes all the messages in a log shard sequentially on a single thread. Thus, if the log is sharded based on the value that needs to be unique, a stream processor can unambiguously and deterministically decide which one of several conflicting operations came first in the log. For example, in the case of several users trying to claim the same username:
-
Every request for a username is encoded as a message and appended to a shard determined by the hash of the username.
-
A stream processor sequentially reads the requests in the log, using a local database to keep track of which usernames are taken. For every request for a username that is available, it records the name as taken and emits a success message to an output stream. For every request for a username that is already taken, it emits a rejection message to an output stream.
-
The client that requested the username watches the output stream and waits for a success or rejection message corresponding to its request.
This algorithm is the same as the construction for achieving consensus using a shared log, which we saw in the related discussion. It scales easily to a large request throughput by increasing the number of shards, as each shard can be processed independently.
The approach works not only for uniqueness constraints but also for many other kinds of constraints. Its fundamental principle is that any writes that may conflict are routed to the same shard and processed sequentially. The definition of a conflict may depend on the application, but the stream processor can use arbitrary logic to validate a request.
Multishard request processing
Ensuring that an operation is executed atomically, while satisfying constraints, becomes more interesting when several shards are involved. In Example 13-2, there are potentially three shards: one containing the request ID, one containing the payee account, and one containing the payer account. There is no reason those three things should be in the same shard, since they are all independent from one another.
In the traditional approach to databases, executing this transaction would require an atomic commit across all three shards, which essentially forces it into a total order with respect to all other transactions on any of those shards. Since there is now cross-shard coordination, different shards can no longer be processed independently, so throughput is likely to suffer.
However, equivalent correctness can be achieved without cross-shard transactions by using sharded logs and stream processors. Figure 13-2 shows an example of a payment transaction that needs to check whether the source account contains sufficient funds and, if so, atomically transfers an amount to a destination account while deducting fees.
Diagram illustrating the process of transferring money from a source account to a destination and fees account using event logs and stream processors.
Figure 13-2. Checking whether a source account has enough money and atomically transferring money to a destination account and a fees account, using event logs and stream processors
This process works as follows:
-
The request to transfer money from the source account to the destination account is given a unique request ID by the user’s client and appended to a log shard based on the source account ID.
-
A stream processor reads the log of requests and maintains a database containing the state of the source account and the IDs of requests it has already processed. The contents of this database are entirely derived from the log. When the stream processor encounters a request with an ID that it has not seen before, it checks in its local database whether the source account has enough money to perform the transfer.
-
If so, it updates its local database to reserve the payment amount on the source account and emits events to several other logs: an outgoing payment event to the log shard for the source account (its own input log), an incoming payment event to the log shard for the destination account, and an incoming payment event to the log shard for the fees account. The original request ID is included in those emitted events.
-
Eventually, the outgoing payment event is delivered back to the source account processor (which may have received unrelated events in the meantime). The stream processor recognizes based on the request ID that this is a payment it previously reserved and executes the payment, again updating its local state for the source account. It ignores duplicates based on request ID.
-
The log shards for the destination and fees accounts are consumed by independent stream processing tasks. When they receive an incoming payment event, they update their local state to reflect the payment, and they deduplicate events based on request ID.
Figure 13-2 shows the three accounts as being in three separate shards, but they could just as well be in the same shard—it doesn’t matter. The only requirements are that the events for any given account are processed strictly in log order with at-least-once semantics and that the stream processors are deterministic.
For example, consider what happens if the source account processor crashes while processing a payment request. The output messages may or may not have been emitted before the crash occurred. After recovering from the crash, the processor will process the same request again (because of at-least-once semantics), and it will make the same decision on whether to allow the payment (since it’s deterministic). It will therefore emit the same output messages with the same request ID to the outgoing, incoming, and fees account shards. If the messages are duplicates, the downstream consumers will ignore them based on the request ID.
Atomicity in this system comes not from transactions, but from the fact that writing the initial request event to the source account log is an atomic action. Once that one event is in the log, all the downstream events will eventually be written as well—possibly after stream processors have recovered from crashes, and possibly with duplicates, but they will appear eventually.
With exactly-once semantics, this example becomes easier to implement, since the semantics ensure that the stream processor’s local state is consistent with the set of messages it has processed. Thus, if it crashes and reprocesses some messages, its local state is also reset to what it was before those messages were processed.
If the user in Figure 13-2 wants to find out whether their transfer was approved, they can subscribe to the source account log shard and wait for the outgoing payment event. To explicitly notify the user if the balance is insufficient, the stream processor can emit a “declined payment” event to that log shard.
By breaking the multishard transaction into several differently sharded stages and using the end-to-end request ID, we achieve the same correctness property (every request is applied exactly once to both the payer and payee accounts), even in the presence of faults, without using an atomic commit protocol.