To give one example, it is common to need to integrate an OLTP database with a full-text search index in order to handle queries for arbitrary keywords. Although some databases (such as PostgreSQL) include a full-text indexing feature, which can be sufficient for simple applications, more sophisticated search facilities require specialist information-retrieval tools. Conversely, search indexes are generally not very suitable as a durable system of record. Therefore, many applications need to combine two tools to satisfy all of their requirements.
We touched on the issue of integrating data systems in “Keeping Systems in Sync”. As the number of representations of the data increases, the integration problem becomes harder. Besides the database and the search index, perhaps you need to keep copies of the data in analytical systems (data warehouses, or batch and stream processing systems); maintain caches or denormalized versions of objects that were derived from the original data; pass the data through machine learning, classification, ranking, or recommendation systems; or send notifications based on changes to the data.
Reasoning about dataflows
When copies of the same data need to be maintained in several storage systems in order to satisfy multiple access patterns, you need to be very clear about the inputs and outputs. Where is data written first, and which representations are derived from which sources? How do you get data into all the right places, in the right formats?
For example, you might arrange for data to first be written to a system-of-record database, after which the changes made to that database are captured (see “Change Data Capture”) and applied to the search index in the same order. If CDC is the only way of updating the index, you can be confident that the index is entirely derived from the system of record and therefore consistent with it (barring bugs in the software). Writing to the database is the only way of supplying new input into this system.
Allowing the application to directly write to both the search index and the database introduces the problem shown in Figure 12-4, in which two clients concurrently send conflicting writes, and the two storage systems process them in a different order. In this case, neither the database nor the search index is “in charge” of determining the order of writes, so they may make contradictory decisions and become permanently inconsistent with each other.
If it is possible for you to funnel all user input through a single system that decides on an ordering for all writes, it becomes much easier to derive other representations of the data by processing the writes in the same order. This is an application of the state machine replication approach that we saw in “Consensus in Practice”. Whether you use CDC or an event sourcing log is less important than the principle of deciding on a total order.
Updating a derived data system based on an event log can often be made deterministic and idempotent (see “Idempotence”), making it quite easy to recover from faults.
Derived data versus distributed transactions
The classic approach for keeping data systems consistent with each other involves distributed transactions, as discussed in “Two-Phase Commit”. How does the approach of using derived data systems fare in comparison to distributed transactions?
At an abstract level, they achieve a similar goal by different means. Distributed transactions use an atomic commit protocol to ensure that changes are applied atomically, while log-based systems achieve correctness through deterministic retry and idempotence.
The biggest difference is that transaction systems usually guarantee that after a value is written, you can immediately read the up-to-date value (see “Reading your own writes”). On the other hand, derived data systems are often updated asynchronously, so they do not by default guarantee that reads are up-to-date.
Distributed transactions have been used successfully in environments that are willing to absorb their performance and operational costs. However, XA has poor fault tolerance and performance characteristics (see “Distributed Transactions Across Different Systems”), which severely limit its usefulness. It might be possible to create a better protocol for distributed transactions, but getting such a protocol widely adopted and integrated with existing tools would be challenging, and it is unlikely to happen soon.
In the absence of widespread support for a good distributed transaction protocol, log-based derived data is the most promising approach for integrating different data systems. Guarantees such as reading your own writes are useful, though, and it is not productive to tell everyone “eventual consistency is inevitable—suck it up and learn to deal with it” (at least not without good guidance on how to deal with it).
Later in this topic we will discuss some approaches for implementing stronger guarantees on top of asynchronously derived systems and work toward a middle ground between distributed transactions and asynchronous log-based systems.
The limits of total ordering
With systems that are small enough, constructing a totally ordered event log is entirely feasible (as demonstrated by the popularity of databases with single-leader replication, which construct precisely such a log). However, as systems are scaled toward bigger and more complex workloads, limitations begin to emerge:
-
In most cases, constructing a totally ordered log requires all events to pass through a single leader node that decides on the ordering. If the throughput of events is greater than a single machine can handle, you need to shard the log across multiple machines. The order of events in two shards is then ambiguous.
-
If the servers are spread across multiple geographically distributed regions—for example, in order to tolerate an entire datacenter going offline—you typically have a separate leader in each datacenter, because network delays make synchronous cross-datacenter coordination inefficient. This implies an undefined ordering of events that originate in two different datacenters.
-
When applications are deployed as microservices, a common design choice is to deploy each service and its durable state as an independent unit, with no durable state shared between services. When two events originate in different services, those events have no defined order.
-
Some applications maintain client-side state that is updated immediately on user input (without waiting for confirmation from a server), and even continue to work offline. With such applications, clients and servers are very likely to see events in different orders.
In formal terms, deciding on a total order of events is known as total order broadcast, which, as we saw in “The Many Faces of Consensus”, is equivalent to consensus. Most consensus algorithms are designed for situations in which the throughput of a single node is sufficient to process the entire stream of events, and these algorithms do not provide a mechanism for multiple nodes to share the work of ordering the events.
Ordering events to capture causality
If no causal link exists between events, the lack of a total order is not a big problem, since concurrent events can be ordered arbitrarily. Some other cases are easy to handle—for example, multiple updates of the same object can be totally ordered by routing all updates for a particular object ID to the same log shard. However, causal dependencies sometimes arise in more subtle ways.
For example, consider a social networking service and two users who were in a relationship but have just broken up. One user removes the other as a friend, then sends a message to their remaining friends complaining about their ex-partner. The user’s intention is that their ex-partner should not see the rude message, since the message was sent after that individual’s friend status was revoked.
However, in a system that stores friendship status in one place and messages in another, that ordering dependency between the unfriend event and the message-send event may be lost. If the causal dependency is not captured, a service that sends notifications about new messages may process the message-send event before the unfriend event and thus incorrectly send a notification to the ex-partner.
In this example, the notifications are effectively a join between the messages and the friend list, making it related to the timing issues of joins that we discussed previously (see “Time dependence of joins”). Unfortunately, this problem doesn’t seem to have a simple answer. Starting points include the following:
-
Logical timestamps can provide total ordering without coordination (see “ID Generators and Logical Clocks”), so they may help when total order broadcast is not feasible. However, they still require recipients to handle events that are delivered out of order, and they require additional metadata to be passed around.
-
If you can log an event to record the state of the system that the user saw before making a decision and give that event a unique identifier, then any later events can reference that event identifier in order to record the causal dependency.
-
Conflict resolution algorithms (see “Automatic conflict resolution”) help with processing events that are delivered in an unexpected order. They are useful for maintaining state, but they do not help if actions have external side effects (such as sending a notification to a user).
Perhaps patterns for application development will emerge in the future that allow causal dependencies to be captured efficiently, and derived state to be maintained correctly, without forcing all events to go through the bottleneck of total order broadcast.