فلسفهٔ سیستم‌های جریانیA Philosophy of Streaming Systems

This chapter opens with a simple observation that stitches the whole book together: for a problem like “store data and find it later” there is no single right solution; every tool is built for a particular consumption pattern. So a complex application inevitably combines several pieces of software: an OLTP database for recording, a search index for words, a warehouse for analytics, a cache for display, an ML model for recommendations, and a notification service for news. The hard question isn't choosing the tool; it's mapping tools to conditions — and then keeping all these views together.

The proposed fix: declare some systems the record and keep the rest derived from them. E.g. the record database's changes are captured with CDC and applied in the same order onto the search index. If CDC is the index's only update path, you can be sure the index is fully derived from the record and (barring software bugs) consistent with it. Writing to the database is the only way to inject fresh input into the system. This is the state-machine replication idea (Chapter 10), applied this time at the scale of a whole organization's architecture.

If the application writes directly to both the database and the search index, it falls into the Fig. 12-4 trap: two clients concurrently send two conflicting writes and the two systems apply them in two different orders. Here neither the database nor the index is “in charge of order”; each decides contradictorily and they stay inconsistent forever. The fix is funnelling all input through one system that decides every write's order — an application of the same state-machine-replication approach. Updating the derived system from the event log can usually be made deterministic and idempotent, simplifying error recovery.

The classic approach to keeping systems in sync is the distributed transaction (2PC, Chapter 8). At an abstract level, both chase one goal with two means: transactions guarantee atomic application via atomic commit; the log-based system reaches correctness via deterministic retry and idempotence. The biggest difference is that a transactional system usually guarantees “after writing, you immediately read the fresh value” (read-your-writes), while an asynchronous derived system offers no such guarantee.

Distributed transactions have succeeded in environments willing to pay their performance and operational cost; but XA has poor fault tolerance and performance that sharply limits its use. Building a better protocol is possible, but getting it adopted into existing tools is unlikely to happen soon. So in the absence of a good universal protocol, log-based derived data is the most promising way to integrate different systems. Of course, saying “eventual consistency is inevitable, build and burn” without a “how to burn” guide isn't useful either — and the rest of this chapter is exactly that guide: how to build stronger guarantees on asynchronous derived systems and reach the middle ground between distributed transactions and the asynchronous log.

While the system is small, building a fully ordered event log is feasible (single-leader databases build exactly that log). But with growing scale, limits surface:

• Building a fully ordered log usually means all events pass through a single leader that decides the order. If the event rate exceeds one machine's power, you must shard the log — and then the order of two shards' events is ambiguous.
• If servers sit in several geographic regions (to survive losing a whole datacenter), each datacenter usually has its own leader since synchronous cross-continent coordination is slow — so the order of two datacenters' events is undefined.
• In a microservice architecture each service has its own durable state with no state shared between services — two different services' events have no order.
• Apps that immediately update client-side state on user input (even offline) make client and server see events in different orders.

Formally, deciding the total order of events is total order broadcast, which (Chapter 10) equals consensus. And consensus algorithms are designed for where one node's power suffices for the whole event flow — they have no mechanism for several nodes to share the “ordering” work.

Now the subtler question: if there is no causal link between events, lacking total order isn't a big problem — concurrent events can be ordered arbitrarily. Multiple updates of one object can be totally ordered by routing all of that ID's updates to one shard. But causal dependencies are sometimes hidden. The book's example: in a social network, a user unfriends someone (unfriend) and then messages remaining friends about that ex-friend — intending the ex-friend not to see the harsh message. But if friendship state lives in one place and messages in another, the “unfriend first, then message” order may be lost, and the notification service may process the message before the unfriend and wrongly notify the ex-friend too. These notifications are really a join between messages and the friend list — the same join-timing problem.

The book's suggested starting points for capturing causality without the total-order-broadcast bottleneck:

• Logical timestamps give total order without coordination (Chapter 10), but the receiver must still tolerate out-of-order events and extra metadata is carried.
• You can log an event recording the user-seen state and give it a unique ID; later events cite that ID to record the causal dependency.
• Conflict-resolution algorithms (Chapter 6) help for unexpectedly ordered events — good for keeping state, but useless when the action has an external effect (like sending a notification).

Integration's goal is data sitting in the right shape in all the right places: taking input, transforming, joining, filtering, aggregating, training models, evaluating, and writing to the right output. The tools are Chapters 11–12's batch and stream processors; and their outputs are those derived datasets: search indexes, materialized views, recommendations, aggregate metrics. We know their fundamental difference: stream input is unbounded, batch input bounded and known.

Batch processing has a strong functional flavor: pure deterministic functions whose output depends only on input, with no side effects but the explicit output; immutable input and append-only output. Stream processing is the same, only extending the operators to have managed, fault-tolerant state too. This “deterministic function with clear input/output” principle isn't just good for fault tolerance; it simplifies reasoning about an organization's dataflows: derived index, statistical model or cache — all are pipelines deriving something from something else.

In principle the derived system could be kept synchronous — as a relational database updates its secondary index in the same write transaction. But asynchrony is exactly what makes the log-based system strong: errors in one corner stay contained; while a distributed transaction aborts if one participant dies, spreading the error to the rest. Likewise the most reliable, scalable way to maintain a shard-crossing secondary index is asynchronous maintenance.

Streams reflect input changes in derived views with low latency; batch lets you reprocess the accumulated historical volume and build a fresh view of the existing dataset. And reprocessing is a good mechanism for application evolution: without it, schema evolution stays limited to simple changes (a fresh optional field, a fresh record); with it, a dataset can be rebuilt into a wholly different model that serves the fresh need better.

Derived views allow the same: to rebuild a dataset, migration needn't be a sudden switch. Keep old and new schemas as two independent views derived from the same underlying data side by side; then move a few users to the fresh view to gauge its performance and find its bugs while most users still go to the old view. Gradually raise the fresh view's share and drop the old view at the end. Its beauty is that every step easily reverses; you always have a healthy system to return to. And lowering the risk of irreversible damage increases the courage to move forward and the speed of improvement.

The original proposal for integrating batch and stream was the lambda architecture, which had its own problems and is retired. Newer systems let batch computation (history reprocessing) and stream computation (fresh-event processing) be implemented in the same one system — an approach sometimes called kappa. Unifying in one system needs three things:

• History replay from the same processing engine that eats the fresh-event flow (a log-based broker that knows replay; a processor that also reads from DFS/object stores).
• Exactly-once semantics for the stream processor — output identical to as if no error ever occurred, even when some did; like batch, by discarding broken tasks' partial output.
• Windowing on event time, not processing time — since when reprocessing history, processing time is meaningless (like the Beam API running on Flink or Dataflow).

At an abstract level, a database, a batch/stream processor and an operating system do one job: hold data and let you process and query it. A database keeps data in the records of a data model, a filesystem in files — but at their core both are “information management systems”; and the batch processor (Chapter 11) is like the distributed version of Unix. Real practical differences exist, of course (a directory with 10 million tiny files is a disaster for many filesystems; a table with 10 million tiny records is ordinary for a database). But the similarities and differences of operating systems and databases are worth exploring.

Unix and the relational database took two different philosophies to information management: Unix gave the programmer a low-level hardware abstraction; the relational database a high-level one hiding disk layout, concurrency, crash-recovery complexity. Unix built pipes and files (byte sequences); the database SQL and transactions. Which is better? Depends what you want. Unix is “simpler” in being a thin cover over hardware; the relational database is “simpler” in that a short declarative query can use powerful machinery (optimizer, indexes, concurrency control, replication) without the query author knowing implementation details. This decades-long tension is still unresolved (the NoSQL movement can be seen as trying to bring Unix-style lightweight low-level approaches into distributed OLTP storage). This section wants to reconcile the two and have the best of both.

Think of database features: secondary indexes, materialized views (precomputed caches of query results), replication logs, full-text indexes. And in Chapters 11–12 we talked about building exactly these with batch and stream. There seems a deep parallel between “internal database feature” and “external derived system”.

Watch what happens when you run CREATE INDEX in a relational database: the database must scan a consistent snapshot of the table, take the indexed field's values, sort them and write the index; then process the backlog of writes that arrived since the snapshot (assuming the table wasn't locked and writes continued); and from then on keep the index fresh with every write transaction. This process is surprisingly similar to starting a fresh follower (Chapter 6) and very similar to bootstrapping CDC in a streaming system. Every time you run CREATE INDEX, the database effectively reprocesses the existing dataset and derives the index as a fresh view of existing data.

From this angle, a whole organization's dataflow looks like one very large database: every batch, stream or ETL process moving data from one place and shape to another acts like the database subsystem keeping an index or materialized view fresh. Batch and stream processors become elaborate implementations of triggers and stored procedures and the materialized-view maintenance algorithm; and the derived systems they maintain become different “index types”. The difference is that instead of all living inside one integrated product, they're fragmented: different software, on different machines, owned by different teams.

If you accept that no single data model or storage format fits every access pattern, there are two ways to combine different tools into one coherent system:

Federation and unbundling are two sides of one coin: building a reliable, scalable system from diverse parts. Mapping one data model to another (for federated read-only queries) takes thought but is ultimately a manageability problem. The hard engineering problem is keeping writes in sync across several storage systems — and that is our focus here. Its traditional approach is distributed transactions over heterogeneous systems, which is problematic; transactions inside one storage system or stream are feasible, but when data crosses the boundary of two different technologies, the asynchronous event log with idempotent writes is far stronger and more practical. E.g. distributed transactions inside some stream processors work well for exactly-once semantics; but when a transaction must involve systems written by different groups (writing from a stream processor to a distributed key-value store or a search index), the lack of a standard transaction protocol makes integration hard — and the ordered log with an idempotent consumer is a far simpler abstraction, implementable across heterogeneous systems.

The big advantage of log-based integration is loose coupling between parts, on two levels. At the system level: asynchronous flows make the whole system resilient to one part's outage or slowness — if a consumer slows or dies, the log buffers the messages; producer and other consumers continue unaffected; the broken consumer catches up when fixed, no data is lost, and the error stays contained. At the human level: unbundling lets different teams develop and maintain parts and services independently of each other.

• System level: asynchronous flows harden the whole system against one part's outage or slowness. If a consumer slows or dies, the log buffers messages; the producer and other consumers continue unaffected. The broken consumer catches up when fixed, no data is lost and the error stays contained. Synchronous distributed-transaction interaction, by contrast, amplifies a local error into a large failure.
• Human level: unbundling lets different teams develop and maintain parts and services independently. Specialization means each team focuses on “one job well” with a clear interface to other teams' systems. The event log is an interface simultaneously powerful enough (event durability and order → fairly strong consistency properties) and general enough for almost any data type.

The idea of “updating derived data when underlying data changes” isn't new. Spreadsheets have strong dataflow power: you put a formula in a cell (e.g. a column sum) and whenever the formula's input changes the result auto-recomputes — exactly what we want at data-system scale: a database record changes, its index auto-updates, dependent views and aggregations auto-refresh; without worrying about refresh technicalities and trusting it works. The difference from spreadsheets is that today's data system must be fault-tolerant, scalable and durable; and must integrate different technologies written by different groups over time. Expecting all software built with one language and framework is unrealistic.

When one dataset derives from another, a transform function sits in the middle. Some examples:

• A secondary index is built with a simple function: for each row/document of the base table, take the indexed column's values and sort by them (if the index is an SSTable or B-tree, which are key-sorted).
• A full-text index is built with language-processing functions (language detection, word segmentation, stemming, spell correction, synonym finding) then an efficient search structure (inverted index).
• In an ML system the model can be seen as derived from training data (via feature extraction and statistical analysis); and the model's output on fresh input is derived from that input and the learned parameters (i.e. indirectly, from training data).
• A cache is often an aggregation of data in the shape the UI will display; so filling the cache needs knowing which fields the UI references, and a UI change may redefine cache filling and require rebuilding it.

The secondary-index derivation function is so common it's embedded in many databases as a core feature, invoked with one CREATE INDEX. But advanced full-text needs domain-specific tuning; and in ML, feature engineering is notorious for being application-specific, needing precise knowledge of user interaction and deployment. When the derived-dataset-building function isn't a standard template, custom code is needed — and that is exactly where many databases struggle: they have triggers, stored procedures and UDFs, but these were somewhat “bolted on” in database design.

In theory the database could be the deployment environment for arbitrary application code, like the operating system. But in practice it turned out badly: it pairs poorly with modern development needs (dependency and package management, version control, rolling upgrades, evolvability, monitoring, calling network services, integrating with external systems). Cluster deployment and management tools (Kubernetes, Docker, Mesos, YARN and the rest) are built exactly for this — and since they focus on “one job well”, they do it far better than a database for which running UDFs is just one of dozens of features.

Most web apps today deploy as stateless services: any request can go to any server, and after responding the server forgets everything. Adding/removing servers is easy — but state must go somewhere (usually a database). The trend has been keeping stateless logic separate from state management: no application logic in the database, no durable state in the application. (To borrow the functional programmers' joke: “we believe in the separation of Church and state” — a nod to Alonzo Church and the state-free lambda calculus.)

In this common model the database is like a shared mutable variable, read and updated synchronously over the network; and the database takes on durability plus some concurrency control and fault tolerance. But in most languages you can't subscribe to a variable's change — you can only poll it periodically (the observer pattern must be hand-built; it isn't a built-in language feature). Databases inherited the same passive approach: to learn database content changed, often the only way is polling (periodically repeating a query). Subscribing to changes is only now emerging as a feature.

Today's dominant style is breaking capability into services talking over synchronous network requests (REST); its advantage over monoliths is organizational scalability with loose coupling (different teams, independent deploys). Composing stream operators in a dataflow system has many of the same properties, but its communication mechanism differs: one-way asynchronous message flow instead of synchronous request interaction. Beyond advantages like better fault tolerance, a dataflow system can also be faster than REST/RPC. The book's example: a customer buys an item priced in one currency but pays in another; conversion needs the day's exchange rate. Two ways:

• Microservices: the purchase-processing code queries the exchange-rate service or database for the current rate.
• Dataflow: the purchase-processing code already subscribes to the exchange-rate update flow and records the current rate in a local database every time it changes; when processing the purchase it reads that same local database directly (perhaps on the same machine, even the same process).

The second way replaces a synchronous network request to another service with a local query. The first way could cache the rate locally too, but keeping the cache fresh would need periodic polling or subscribing to the change flow — which is exactly the second way's work. So the dataflow way is not only faster but also more resilient to the other service's failure: the fastest, most reliable network request is no request at all! Instead of RPC you now have a streaming join between purchase events and rate-update events. That join is time-dependent: if you later reprocess purchase events, rates changed; to reproduce the original output you need that purchase moment's historical rate — whether by querying the service or subscribing to the flow, this time dependency must be managed. Subscribing to the change flow (instead of querying current state at need-time) brings us closer to a spreadsheet-like computation model: data changes, every dependent derivative updates fast.

The previous section's dataflow systems give you a process for building derived datasets and keeping them fresh — call it the write path: whenever information is written, it may pass several batch and stream stages until all derived datasets update; Fig. 13-1 shows search-index updating as an example. But why build a derived dataset at all? Because you'll probably want to query it later — that is the read path: when serving a user request, you read from the derivative, perhaps process further and build the answer. Together, the whole data journey from collection moment to consumption moment (probably by another human). The write path is the journey's precomputed part — eager, as soon as data arrives, unasked; the read path is the part happening only when someone asks (lazy). If you know functional languages, the write path resembles eager evaluation and the read path lazy evaluation.

The derived dataset is where the two paths meet. So it's a tradeoff between “how much work at write time” and “how much at read time”. The full-text index is a good example: writing must update the index entries of every document term; reading must look up each query word and apply Boolean logic (AND for “all words”, OR for synonyms). Without an index, queries would scan all documents (like grep) — cheap writes, very expensive reads. At the spectrum's other end, you could precompute every possible query's answer — reading just “find and return”, but horribly expensive writing; and since the possible-query set is infinite (exponential in term count), not feasible at all. The middle option: precompute only the most-used queries' answers (a cache of common queries — also nameable a materialized view, since fresh documents must update it) and serve rare queries from the index. So the role of caches, indexes and materialized views is simple: moving the boundary between read path and write path — precomputing means doing more work at write time to do less at read time. (This boundary-shifting was the subject of the Chapter-1 social-timeline case study — with different boundaries for celebrities and ordinary users. After 500 pages we're back where we started!)

The write/read-path boundary idea shows up in another context too. Browsers used to be stateless clients (offline you could only scroll the previous page); but single-page web apps and mobile apps now keep lots of state on-device. With durable local state, a class of apps can work offline and sync in the background when the network returns — and since the UI needn't wait for synchronous requests, the experience is far better. On-device state can be seen as a cache of server state: the page's pixels are a materialized view of the client app's model objects; and the model objects are a local replica of the far datacenter's state.

But the ordinary web page doesn't learn the server's next change until reload — a stale cache you don't poll. Newer protocols (server-sent events, WebSocket) give a channel for the server to push state changes down the same path, cutting client-side staleness. In the paths model, this means pulling the write path all the way to the user: the client still needs the read path for initial state, but then relies on the server's change flow. Devices are sometimes offline and miss notifications — but the log-based broker's consumer-offset technique works here too: each device is a tiny subscriber to a tiny flow that loses nothing after reconnecting. State changes can travel an end-to-end write path: from one device's interaction, through logs, derivatives and processors, to another device's UI — with sub-second latency. Why aren't all apps like this? Because the “stateless client + request/response” assumption is rooted in databases, libraries and protocols; many datastores have “one request, one response” operations and far fewer know “one request, a flow of responses over time” (subscribing to change). Pulling the write path to the user means fundamental rethinking toward publish/subscribe — costly, but a more responsive UI and better offline.

One step further: so far writes went through the log and reads were transient network requests going straight to the holding node. But a read request can also be shown as an event flow, routing both read and write events through one stream operator; the operator answers the read event by publishing the result to the output flow. When reads and writes are both events reaching one operator, you're doing a stream-table join between the read-query flow and the database: each read event must go to the shard holding the relevant data — just like copartitioning inputs on the same key in batch and stream joins. This “serving requests = join” correspondence is fundamental: a one-off read passes through the join operator and the operator immediately forgets it; a subscribe request is a durable join with the other side's past and future events. Logging read events also helps causality and provenance tracking (reconstructing what the user saw before deciding — e.g. the send history and balance a customer saw affects their purchase); but it costs extra storage and I/O, and optimizing it is an open research problem.

For single-shard queries, sending through the stream and gathering the streaming answer may be overkill; but this idea opens the door to distributed execution of complex multishard queries with the same routing, sharding and join machinery of the stream processor. Example: distributed RPC in Storm for “unioning the follower sets of everyone who posted a URL” (combining many shards' results), or fraud-risk scoring of a purchase by successive joins with reputation databases each sharded separately. If you need this kind of multishard join, it's probably simpler to use a database offering the feature than to implement it with a stream processor — but “queries as flows” is an option for big apps hitting ready-made solutions' ceilings.

With a read-only stateless service, if something breaks it's no problem: you fix the bug, restart, and everything returns. A stateful system like a database isn't like that: it's supposed to remember (more or less) everything forever, so a failure's effect potentially lasts forever too — and needs more careful thought. We want to build apps that are reliable and correct (clear, understood semantics, despite various errors). Four decades of atomicity, isolation and durability were the tools for this; but these foundations are shakier than they look (witness the confusion of weak isolation levels). In some domains transactions are abandoned altogether, replaced with a model giving better performance and scalability but messier semantics. Everyone says “consistency” and few define it precisely. Kyle Kingsbury's Jepsen tests showed the naked gap between some products' “claimed guarantee” and “actual behavior” under network trouble and crashes — and even if the infrastructure were flawless, application code must still use features correctly, which is error-prone with hard-to-understand configuration (weak isolation, quorums and the rest).

If your app tolerates occasionally breaking or losing data, life is simple (close your eyes and hope). But if you need stronger guarantees, serializability and atomic commit are the established paths — at a cost: they usually work only in one datacenter (ruling out geographic architectures) and limit scale and fault tolerance. The traditional transactional approach isn't leaving, but neither is it the last word on making apps correct and resilient. This section explores other ways of thinking about correctness in dataflow architectures.

That an app uses a data system with a strong feature (like a serializable transaction) doesn't mean it's safe from lost or corrupted data: a bug writing or deleting wrong data isn't saved by a serializable transaction — itself an argument for immutable, append-only data, since recovering from mistakes is simpler if broken code can't destroy healthy data. But immutability alone doesn't cure everything; the subtler example is “exactly-once execution” of an operation.

The idea: if something breaks while processing a message, either give up (lost message = lost data) or retry. But if you retry, the first attempt may actually have succeeded with only its acknowledgment lost — so the message gets processed twice. Processing twice is itself a data failure: charging the customer twice or double-counting a counter. Exactly-once means arranging for the final effect to equal as if no error ever happened, even if errors caused retries. One of the most effective ways is making operations idempotent: same effect whether run once or several times. But idempotizing an inherently non-idempotent operation takes work and care: it may need extra metadata (like the set of operation IDs that updated a value), and on failover you must fence (Chapter 9).

The same “duplicate suppression” pattern appears elsewhere. TCP orders packets with sequence numbers, resends lost ones and drops duplicates — but only inside one TCP connection. Suppose the TCP connection is a client-to-database connection running Example 13-1's transaction (in many databases a transaction is tied to the client connection, since several queries on the same connection mean the same transaction). If the client, after sending COMMIT and before hearing the reply, hits a network break and timeout, it doesn't know whether the transaction committed or aborted (the same situation as Fig. 9-1).

The client can reconnect and retry the transaction, but now it's outside TCP's duplicate-suppression territory. And since this transaction isn't idempotent, $22 may move instead of $11. So even this code — the textbook atomicity example — is incorrect — and real banks don't work this way. The 2PC protocol breaks the one-to-one “TCP connection = transaction” mapping (the coordinator must reconnect after failure and tell the database to commit or abort the in-doubt transaction) — but even that isn't enough for “executing just once”. Even suppressing duplicates between client and database server leaves the network between the user's device and the app server: the user on weak internet sends the POST but loses signal before the reply; sees an error and manually retries (the browser even asks “are you sure you want to resubmit?” — and the user says yes, wanting the operation done). To the web server the retry is a separate request; to the database, a separate transaction. Ordinary dedup mechanisms don't help.

This code relies on the request_id column's uniqueness constraint: if a transaction tries inserting a duplicate ID, the INSERT breaks and the transaction aborts, never applying twice. Relational databases usually uphold uniqueness correctly even under weak isolation (while check-then-insert at app level breaks under non-serializable isolation — Chapter 8's write skew). The requests table is also effectively an event log, useful for event sourcing or CDC: the balance updates needn't be in the same transaction as the event insert, since they're additive and can be derived from the event in a downstream consumer — provided the event is processed exactly once, which the same ID guarantees.

We saw end-to-end duplicate suppression with an ID carried from the client to the recording database. What about other constraints? Focus on the uniqueness constraint (the one Example 13-2 relied on): usernames or emails must identify the user uniquely; a file store can't hold two same-named files; two people can't book one flight or theater seat. Other constraints are much alike: account balances must not go negative; don't sell beyond warehouse stock; meeting rooms must have no overlapping bookings. The uniqueness-enforcement technique often works for this kind of constraint too.

A uniqueness constraint in a distributed environment needs consensus (Chapter 10): if several concurrent requests share a value, the system must decide which conflicting operation is accepted and reject the rest as constraint violations. The most common way to this consensus is making one node the leader and entrusting it with every decision — fine while you tolerate funnelling every request through one node (even if the client is across the world) and that node breaking. Consensus algorithms like Raft solve safely electing a fresh leader after the leader's failure and preventing split-brain. Uniqueness checks scale by sharding on the value that must stay unique: for request-ID uniqueness, route all requests with the same request ID to the same shard; for username uniqueness, shard by username hash. But asynchronous multi-leader replication is out, since different leaders can concurrently accept conflicting writes and values stop being unique. If you want every constraint-violating write rejected immediately, synchronous coordination is unavoidable.

Uniqueness in log-based messaging: the shared log guarantees every consumer sees messages in the same order (the total-order-broadcast guarantee that equals consensus). In the unbundled-database approach with log-based messaging, the uniqueness constraint can be enforced with a very similar approach: the stream processor consumes one log shard's messages sequentially on one thread; so if the log is sharded by the value that must stay unique, the processor can unambiguously and deterministically decide which of several conflicting operations in the log came first. E.g. several users want one username: (1) each request appends as a message to the shard set by the username's hash; (2) the stream processor reads requests sequentially and tracks with a local database which names are taken — for each free name it records taken and emits a success message to the output flow, and for taken names a reject message; (3) the requesting client watches the output flow and waits for its own success/reject. This algorithm is the same “reaching consensus with a shared log” structure as Chapter 10; and it scales easily to high throughput with more shards since each shard processes independently. This approach isn't only for uniqueness — it works for many other constraints; its fundamental principle is routing every potentially conflicting write to the same shard for sequential processing. What counts as “conflict” depends on the application, but the processor can validate requests with arbitrary logic.

When several shards are involved, atomically enforcing operations with constraints gets more attractive. In Example 13-2 there are potentially three shards: one for the request ID, one for the recipient account, one for the payer account — with no reason for all three to share one shard since they're independent. In the traditional database approach, running this transaction wants atomic commit across all three shards, effectively funnelling it with every other transaction of all three shards into one total order; and with cross-shard coordination now present, shards no longer process independently and throughput drops. But equivalent correctness is reachable without cross-shard transactions, using sharded logs and stream processors. Fig. 13-2 shows an example: a payment transaction that must check the source account holds enough money and, if so, atomically move the amount to the destination account minus a fee. The steps:

1. The transfer request from source to destination account takes a unique request ID from the user's client and appends in the log shard keyed by the source account ID.
2. The stream processor reads the request log and keeps a database holding the source account's state and the request IDs it already processed (this database's content is fully derived from the log). On reaching an unseen-ID request, it checks in the local database whether the source account holds enough money. If so, it updates the local database to reserve the amount on the source account and sends events to several other logs: the outgoing-payment event to the source account's log shard (its own input log), the incoming-payment event to the destination account's log shard, and the incoming-payment event to the fee account's log shard — with the original request ID present in those published events.
3. Finally the outgoing-payment event returns to the source-account processor (which may have taken unrelated events in between). From the request ID the processor knows this is a payment it already reserved, executes the payment and updates the source account's local state again; and ignores duplicates by request ID.
4. The destination- and fee-account log shards are consumed by independent stream-processing tasks; on taking the incoming-payment event they update local state and dedup events by request ID.

In the figure all three accounts sit in three separate shards, but they could share one shard too — it makes no difference. The only conditions: each account's events process exactly in log order with at-least-once semantics, and processors are deterministic. E.g. see what happens if the source-account processor crashes mid-payment: the output messages may or may not have published before the crash. After recovery the same request reprocesses (thanks to at-least-once) and takes the same “payment allowed or not” decision (since deterministic) — so it sends the same output messages with the same request ID to the outgoing/incoming/fee shards; and if messages duplicate, downstream consumers ignore them by request ID. Atomicity in this system doesn't come from transactions but from the fact that writing the initial request event to the source account's log is one atomic action. Once that one event is in the log, all downstream events eventually get written — perhaps after processors recover from crashes, and perhaps duplicated, but eventually. Exactly-once semantics simplifies implementation, since the processor's local state stays consistent with the message set it processed; so if it crashes and reprocesses some messages, its local state returns to just before those messages too.

If Fig. 13-2's user wants to know whether their transfer confirmed, they can subscribe to the source account's log shard and wait for the outgoing-payment event. And to explicitly tell the user when funds were insufficient, the processor can send a “payment declined” event to the same shard. By breaking the multishard transaction into stages with different shards and using an end-to-end request ID, we reach the same correctness property (each request applied exactly once on both payer and payee accounts), even under failure — with no atomic-commit protocol.

The comfortable property of many transactional systems is that once a transaction commits, its writes are immediately visible to other transactions — formalized as strict serializability (Chapter 10). When you break an operation across several stream-processor stages, that doesn't hold. Log consumers are inherently asynchronous, so the producer doesn't wait for consumers to process its message. But the client may wait for its message to appear in the output flow — like the user waiting for the outgoing- or declined-payment event in Fig. 13-2, depending on whether the source account held enough money. In this example, the correctness of the source-balance check doesn't depend on whether the user waits for the result; waiting only synchronously informs the user whether the payment succeeded — that notification is separate from processing the request's effect.

More generally, the word “consistency” mixes two needs worth examining separately:

As a slogan: timeliness violations are permitted under eventual consistency; but an integrity violation means permanent inconsistency. In most apps integrity matters far more than timeliness: timeliness violations are annoying and confusing, but integrity violations can be catastrophic. E.g. in a credit-card statement it's unsurprising if a last-24-hours transaction hasn't arrived yet — these systems normally lag; we know banks reconcile and settle transactions asynchronously and timeliness matters little here. But it would be very bad if the statement balance didn't equal the transactions' sum plus the previous balance (summing errors), or a transaction charged you without paying the merchant (lost money) — those would violate the system's integrity.

ACID transactions usually give both timeliness (like linearizability) and integrity (like atomic commit) guarantees; so viewed from the ACID-transaction angle, the timeliness/integrity distinction matters little. But the interesting property of this chapter's event-driven dataflow systems is that they separate timeliness from integrity: in asynchronous event-flow processing there is no timeliness guarantee, unless you explicitly build a consumer that waits for the message before returning (e.g. the user requests a payment then reads their account state before the processor executed the request — not seeing the just-requested payment). But integrity is actually the streaming systems' focus: exactly-once or effectively-once semantics is an integrity-preserving mechanism — if an event is lost or applied twice, system integrity can break; so fault-tolerant message delivery and duplicate suppression (idempotent operations) matter for preserving integrity against failure. And as the previous section showed, reliable stream processors can preserve integrity without distributed transactions and atomic-commit protocols — i.e. reach comparable correctness with far better performance and operational robustness. That integrity was taken with a combination of mechanisms:

• Showing a write operation's content as a single message that can be written atomically — an approach pairing very well with event sourcing;
• Deriving the rest of the state updates from that one message with a deterministic derivation function, like a stored procedure;
• Passing the client-generated request ID through all these processing layers, enabling end-to-end duplicate suppression and idempotence;
• Making messages immutable and allowing occasional reprocessing of derived data, simplifying bug recovery.

Enforcing a uniqueness constraint needs consensus, which usually means funnelling one shard's every event through one node — and if you want the traditional form of the uniqueness constraint, that limit is unavoidable and stream processing doesn't bypass it. But many real apps have a business need that permits violating what you think of as a “hard constraint”:

• If a customer orders beyond warehouse stock, you can oversell, apologize for the delay and discount — the same thing you'd do if a forklift drove into the warehouse goods and stock fell short of imagination. So the apology workflow must be part of the business process for such incidents anyway, and a hard constraint on stock count may be unnecessary.
• Many airlines overbook assuming some passengers miss the flight, and many hotels overbook rooms assuming some guests cancel. Here the “one person per seat” constraint is deliberately violated for business reasons, with a compensation process (refunds, upgrades, a free room at the neighboring hotel) laid on for over-demand. Even without overbooking, apology-and-compensation processes were needed for events like weather- or strike-cancelled flights — recovering from such issues is just an ordinary part of business.
• If someone withdraws beyond their account balance, the bank can charge an overdraft fee and ask the debt repaid; and by capping total daily withdrawals the bank's risk stays bounded.
• In systems integrating data across organizations, inconsistency inevitably arises and a correction mechanism is needed; inter-bank payment settlement (Chapter 11) is an example.

In many business contexts it's thus acceptable to temporarily violate a constraint and fix it later with an apology. This kind of mistake-fixing is called a compensating transaction. The apology's cost (money or reputation) varies but is often small: a sent email can't be unsent, but a corrective follow-up email can be sent; if you double-charged a card, one charge can be refunded at only the processing fee plus perhaps a customer complaint. (Cash out of an ATM can't directly be taken back — though in principle a collector can be sent for a negative balance.) Whether the apology cost is acceptable is a business decision. If acceptable, the traditional “check every constraint before even writing data” model is needlessly restrictive; it may be perfectly reasonable to write optimistically and check the constraint later. Validation can still sit before actions whose recovery is expensive — but that doesn't mean validating before even writing data. These apps want integrity (you don't want a reservation lost or money vanished with mismatched debit-credit) but no timeliness in constraint enforcement: oversold stock can be fixed later — like Chapter 6's “handling conflicting writes” conflict-resolution approaches.

Now two interesting observations sit side by side: (1) dataflow systems can hold integrity guarantees on derived data without atomic commit, linearizability or synchronous cross-shard coordination; (2) although hard uniqueness constraints want timeliness and coordination, many apps are comfortable with a loose constraint that may temporarily break and later heal, as long as integrity always holds. Together they mean dataflow systems can serve data management for many apps with no coordination required, yet still give strong integrity guarantees. Such coordination-avoiding systems are highly attractive: better performance and fault tolerance than systems needing synchronous coordination. E.g. such a system can run distributed across several datacenters in a multi-leader configuration, replicating asynchronously between regions; each datacenter can work independently of the rest since no synchronous cross-region coordination is needed. Such a system has weak timeliness guarantees (no linearizability without coordination) but can still have strong integrity guarantees. In this space, serializable transactions remain useful as part of derived-state maintenance, but can run in a small scope where they perform well; heterogeneous XA-style distributed transactions aren't needed. Synchronous coordination can still be brought where needed (e.g. enforcing a hard constraint before an irrecoverable action), but not everything must pay coordination's price if only a small app part needs it. Another view of coordination and constraints is that they reduce apology counts for inconsistency but potentially cut system performance and availability, thus potentially raising apology counts for outages. Apology counts can't reach zero, but you can find the best tradeoff for your needs — a sweet spot with neither much inconsistency nor many availability problems.

All the correctness, integrity and fault-tolerance discussion assumed some things can break and some can't — called the system model (Chapter 9). E.g. we assume processes crash, machines lose power, networks delay or drop messages arbitrarily; and assume disk-written data after fsync isn't lost, memory data doesn't corrupt, and the CPU's multiply always answers right. These assumptions are often true, and if we constantly worried about computer mistakes nothing would get done. But the traditional system model has a binary view of faults (some happen, some never); in reality it's mostly probability: some likelier, some less likely; and the question is whether assumption violations happen enough to meet them in practice. We saw data can corrupt in memory (Chapter 1), on disk (Chapter 6) and on the network (Chapter 9) — should we attend to it more? If you work at large enough scale, even the very unlikely happen.

Past hardware, there's always software-bug risk that low-level network, memory and filesystem checksums don't catch. Even heavily used databases have bugs — old MySQL versions didn't uphold uniqueness correctly and PostgreSQL's serializable level showed write-skew anomalies, though both are strong battle-tested databases; in less mature software it's probably far worse. Despite all design, test and review effort, bugs creep in; rare, eventually found and fixed, but there's a window where they can corrupt data. And about application code you must assume far more bugs, since most apps see less review and testing than database code; many don't even correctly use the database's integrity-preserving features (foreign keys, uniqueness constraints). “Consistency” in the ACID sense rests on the idea that the database starts from a consistent state and transactions move it from one consistent state to another — so you expect the database always consistent; but that only makes sense assuming bug-free transactions. If the app misuses the database (e.g. unsafely using weak isolation), database integrity can't be guaranteed.

Designing for auditability: if a transaction mutates several objects in the database, understanding the underlying reason later is hard; even capturing the transaction log, inserts, updates and deletes across tables needn't paint a clear picture of “why these mutations happened” — the application-logic call deciding those mutations is transient and irreproducible. Event-driven systems, by contrast, give better auditability: in event sourcing, user input to the system shows as one immutable event, and every state update derives from that event. Derivation can be made deterministic and repeatable, so running the same event log with the same derivation-code version yields the same state updates. Being explicit about dataflow makes data provenance far clearer and integrity checking feasible: for an event log, hashing can check event storage isn't corrupted; and for each derived state, the batch and stream processors that derived it from the event log can be rerun to see if the same result comes — or even run a redundant derivative in parallel. Deterministic, well-defined dataflow also simplifies debugging and tracing system execution to learn why it did something — valuable when something unexpected happens: the ability to reproduce exactly the conditions that led to the unexpected event — a kind of time-travel debugging.

The end-to-end argument again: if we can't fully trust every system part failure-free — that all hardware is flawless and all software bug-free — then we must at least periodically check our data's integrity; unchecked, we won't learn of corruption until too late, when it has caused downstream damage and tracing is far harder and pricier. Checking data-system integrity is better done end to end: the more systems included in the integrity check, the less chance for failure to pass unseen at some pipeline stage. If we can check that the whole derived-data pipeline is correct end to end, then every disk, network, service and algorithm along the way is implicitly included. Continuous end-to-end integrity checking raises trust in system correctness, which in turn lets you move faster (like automated tests, auditing raises the chance of early bug discovery and cuts the risk that a change or a fresh storage technology causes damage); if you don't fear change, you can evolve the app far better for changing needs.

Auditable data-system tools: currently few data systems make auditability a first-class concern. Some apps implement their own audit machinery (e.g. logging every change in a separate audit table) but the audit log's and the database state's integrity guarantees remain hard. The transaction log can be made tamper-proof by periodic signing with a hardware security module, but that doesn't guarantee the right transactions entered the log at all. Blockchains like Bitcoin and Ethereum are shared append-only logs with cryptographic consistency checks; the transactions they store are events and smart contracts are basically stream processors; and their consensus protocol ensures all nodes agree on the same event sequence. Their difference from Chapter-10 consensus protocols is that blockchains are Byzantine fault-tolerant — they work even with some participants holding corrupt data, since replicas constantly check each other's integrity. For most apps blockchains carry too much overhead to be useful; but some of their cryptographic tools apply in lighter contexts too: e.g. Merkle trees are hash trees that can efficiently prove a record's presence in a dataset (and several other things). Certificate Transparency uses cryptographically verified append-only logs and Merkle trees for checking TLS/SSL certificate validity; and by having one leader per log it avoids needing a consensus protocol. Integrity-checking and auditing algorithms, like certificate transparency and distributed ledgers, may see wider data-system use in future; some work is needed to scale them to non-cryptographically-audited systems' scale while keeping the performance penalty low — but they're interesting regardless.

1. “We have serializable transactions so our app is correct” — transactions don't cover app bugs and database misuse.
2. “The stream processor's exactly-once means users can't double-submit” — its scope is message processing only; a user retry is a separate request needing an end-to-end ID.
3. “TCP drops duplicates so retry is safe” — only inside one connection; a fresh connection is a fresh scope.
4. “Eventual consistency means giving up on correctness” — no: integrity (permanent) must be kept; only timeliness (temporary) is relaxed.
5. “We enforce uniqueness in the app with check-then-insert” — breaks under non-serializable isolation; needs a database constraint or shard + sequential check.
6. “Async multi-leader + hard uniqueness” — impossible together; different leaders concurrently accept conflicts.
7. “We integrate with dual writes” — with no ordering authority you get permanent divergence.
8. “CDC or event sourcing, no difference” — true for the “decide the total order” principle, but each has its own operational tradeoffs; say the full sentence.
9. “We solve everything with one integrated database” — great if one technology covers all needs; but the unbundle debate is breadth (workload range), not depth — don't mistake the metric.
10. “For more timeliness, add coordination” without accounting outage apologies — the sweet spot is two-sided.
11. “Logical timestamps fully solve causal ordering” — no: the receiver must still tolerate unordered delivery, and external effects (notifications) stay unfixed.
12. “We have backups so data is safe” — until you've tested a restore, you don't know the backup is healthy; blind trust banned.

1. Say record vs derived with one example (database + search index + CDC) in 30 seconds.
2. Have the four total-order ceilings (single leader, regions, microservices, offline clients) memorized?
3. Write Examples 13-1/13-2 and UNIQUE(request_id)'s role on paper.
4. Explain the end-to-end argument on three levels (TCP, transaction, ID) with each scope.
5. Review Fig. 13-2 in four steps (atomic append → reserve + 3 events → output return → destination dedup).
6. Recite the timeliness/integrity definitions + “which violation is temporary/permanent” + dataflow's four mechanisms.
7. Have two cheap-apology examples (warehouse, flights) and one forced-coordination example (ATM cash) ready?
8. One-line the federated/unbundled difference + “the hard problem = writes”.
9. Explain the write/read path + the spectrum's two ends (no index, precompute-everything).
10. Prepare one closing line: “coordination only as much as needed; integrity always, timeliness separately.”