Just because an application uses a data system that provides comparatively strong safety properties, such as serializable transactions, that does not mean the application is guaranteed to be free from data loss or corruption. For example, if an application has a bug that causes it to write incorrect data or delete data from a database, serializable transactions aren’t going to save you. This is an argument in favor of immutable and append-only data, because it is easier to recover from such mistakes if you remove the ability of faulty code to destroy good data.
Although immutability is useful, it is not a cure-all by itself. Let’s look at a more subtle example of how data corruption can occur.
Exactly-once execution of an operation
In “Distributed Transactions Across Different Systems” we introduced the idea of exactly-once (or effectively-once) semantics in the context of message processing. The idea is this: if something goes wrong while processing a message, you can either give up (by dropping the message and incurring data loss) or try again. If you try again, there is the risk that the processing actually succeeded the first time and you just didn’t get a confirmation, so the message ends up being processed twice.
Processing twice is a form of data corruption: it is undesirable to charge a customer twice for the same service (billing them too much) or to increment a counter twice (overstating a metric). In this context, exactly once means arranging the computation such that the final effect is the same as if no faults had occurred, even if the operation was retried because of a fault. We discussed a few approaches for achieving this goal.
One of the most effective approaches is to make the operation idempotent—that is, to ensure that it has the same effect, no matter whether it is executed once or multiple times. However, doing this for an operation that is not naturally idempotent requires effort and care. You may need to maintain additional metadata (such as the set of operation IDs that have updated a value) and ensure fencing when failing over from one node to another (see “Distributed Locks and Leases”).
Duplicate suppression
The same pattern of needing to suppress duplicates occurs in many other places besides stream processing. For example, TCP uses sequence numbers on packets to put them in the correct order at the recipient and to determine whether any packets were lost or duplicated on the network. Any lost packets are retransmitted and any duplicates are removed by the TCP stack before it hands the data to an application.
However, this duplicate suppression works only within the context of a single TCP connection. Imagine that the TCP connection is a client’s connection to a database, and it is currently executing the transaction in Example 13-1. In many databases, a transaction is tied to a client connection (if the client sends several queries, the database knows that they belong to the same transaction because they are sent on the same TCP connection). If the client suffers a network interruption and connection timeout after sending the COMMIT but before hearing back from the database server, it does not know whether the transaction has been committed or aborted (we saw this situation in Figure 9-1).
Example 13-1. A nonidempotent transfer of money from one account to another
BEGIN TRANSACTION;
UPDATE accounts SET balance = balance + 11.00 WHERE account_id = 1234;
UPDATE accounts SET balance = balance - 11.00 WHERE account_id = 4321;
COMMIT;
The client can reconnect to the database and retry the transaction, but now it is outside the scope of TCP duplicate suppression. Since the transaction in Example 13-1 is not idempotent, $22 could be transferred instead of the desired $11. Thus, even though code like this is a standard example for transaction atomicity, it is not correct, and real banks do not work like this.
2PC (see “Two-Phase Commit”) protocols break the one-to-one mapping between a TCP connection and a transaction, since they must allow a transaction coordinator to reconnect to a database after a network fault and tell it whether to commit or abort an in-doubt transaction. Is this sufficient to ensure that the transaction will be executed only once? Unfortunately not.
Even if we can suppress duplicate transactions between the database client and server, we still need to worry about the network between the end-user device and the application server. For example, if the end-user client is a web browser, it probably uses an HTTP POST request to submit an instruction to the server. Perhaps the user has a weak cellular data connection, and they succeed in sending the POST request, but they lose the signal before they are able to receive the response from the server.
In this case, the user will probably be shown an error message, and they may retry manually. Web browsers warn, “Are you sure you want to submit this form again?”—and the user says yes, because they want the operation to happen. (The Post/Redirect/Get pattern avoids this warning message in normal operation, but it doesn’t help if the POST request times out.) From the web server’s point of view, the retry is a separate request, and from the database’s point of view, it is a separate transaction. The usual deduplication mechanisms don’t help.
Uniquely identifying requests
To make a request idempotent through several hops of network communication, it is not sufficient to rely on a transaction mechanism provided by a database. You need to consider the end-to-end flow of the request.
For example, you could generate a unique identifier for each request (such as a UUID) and include it as a hidden form field in the client application, or calculate a hash of all the relevant form fields to derive the request ID. If the web browser submits a POST request twice, the two requests will have the same request ID. You can then pass that request ID all the way through to the database and check that you always execute only one request with a given ID, as shown in Example 13-2.
Example 13-2. Suppressing duplicate requests by using a unique ID
ALTER TABLE requests ADD UNIQUE (request_id);
BEGIN TRANSACTION;
INSERT INTO requests
(request_id, from_account, to_account, amount)
VALUES('0286FDB8-D7E1-423F-B40B-792B3608036C', 4321, 1234, 11.00);
UPDATE accounts SET balance = balance + 11.00 WHERE account_id = 1234;
UPDATE accounts SET balance = balance - 11.00 WHERE account_id = 4321;
COMMIT;
This code relies on a uniqueness constraint on the request_id column. If a transaction attempts to insert an ID that already exists, the INSERT fails and the transaction is aborted, preventing it from taking effect twice. Relational databases can generally maintain a uniqueness constraint correctly, even at weak isolation levels (whereas an application-level check-then-insert may fail under nonserializable isolation, as discussed in “Write Skew and Phantoms”).
Besides suppressing duplicate requests, the requests table in Example 13-2 acts as a kind of event log, which can be useful for event sourcing or CDC. The updates to the account balances don’t have to happen in the same transaction as the insertion of the event, since they are redundant and could be derived from the request event in a downstream consumer—as long as the event is processed exactly once, which can again be enforced using the request ID.
The end-to-end argument
This scenario of suppressing duplicate transactions is just one example of a more general principle called the end-to-end argument, which was articulated by Saltzer, Reed, and Clark in 1984:
The function in question can completely and correctly be implemented only with the knowledge and help of the application standing at the endpoints of the communication system. Therefore, providing that questioned function as a feature of the communication system itself is not possible. (Sometimes an incomplete version of the function provided by the communication system may be useful as a performance enhancement.)
In our example, the function in question was duplicate suppression. We saw that TCP suppresses duplicate packets at the TCP connection level, and some stream processors provide so-called exactly-once semantics at the message processing level, but that is not enough to prevent a user from submitting a duplicate request if the first one times out. By themselves, TCP, database transactions, and stream processors cannot entirely rule out these duplicates. Solving the problem requires an end-to-end solution: a transaction identifier that is passed all the way from the end-user client to the database.
The end-to-end argument also applies to checking the integrity of data. Checksums built into Ethernet, TCP, and TLS can detect corruption of packets in the network, but they cannot detect corruption due to bugs in the software at the sending and receiving ends of the network connection, or corruption on the disks where the data is stored. If you want to catch all possible sources of data corruption, you also need end-to-end checksums.
A similar argument applies with encryption. The password on your home WiFi network protects against people snooping your WiFi traffic, but not against attackers elsewhere on the internet; TLS/SSL between your client and the server protects against network attackers, but not against compromises of the server. Only end-to-end encryption and authentication can protect against all these things.
Although the low-level features (TCP duplicate suppression, Ethernet checksums, WiFi encryption) cannot provide the desired end-to-end features by themselves, they are still useful, since they reduce the probability of problems at higher levels. For example, HTTP requests would often get mangled if we didn’t have TCP putting the packets back in the right order. We just need to remember that the low-level reliability features are not by themselves sufficient to ensure end-to-end correctness.
Applying end-to-end thinking in data systems
This brings us back to the original thesis: just because an application uses a data system that provides comparatively strong safety properties, such as serializable transactions, that does not mean the application is guaranteed to be free from data loss or corruption. The application itself needs to take end-to-end measures, such as duplicate suppression, as well.
That is a shame, because fault-tolerance mechanisms are hard to get right. Low-level reliability mechanisms, such as those in TCP, work quite well, so the remaining higher-level faults occur fairly rarely. It would be really nice to wrap up the high-level fault-tolerance machinery in an abstraction so that application code needn’t worry about it—but it seems that we have not yet found the right one.
Transactions have long been seen as a useful abstraction. they take a wide range of possible issues (concurrent writes, constraint violations, crashes, network interruptions, disk failures) and collapse them down to two possible outcomes: commit or abort. That is a huge simplification of the programming model, but it is not enough.
Transactions are expensive, especially when they involve heterogeneous storage technologies (see “Distributed Transactions Across Different Systems”). When we refuse to use distributed transactions because they are too expensive, we end up having to reimplement fault-tolerance mechanisms in application code. As numerous examples throughout these notes have shown, reasoning about concurrency and partial failure is difficult and counterintuitive, and so most application-level mechanisms do not work correctly. The consequence is lost or corrupted data.
For these reasons, it is worth exploring fault-tolerance abstractions that make it easy to provide application-specific end-to-end correctness properties, but also maintain good performance and operational characteristics in a large-scale distributed environment.