Transactions

An interactive walk through Chapter 8 of Designing Data-Intensive Applications (2nd ed.). It covers every guarantee, anomaly and algorithm in the chapter. Most sections have a demo: press Next ▶ to step through what each client and the database do, and use the toggles to compare isolation levels and strategies.

1The big picture

Real data systems fail in many ways at once:

A transaction groups several reads and writes into one logical unit. Either the whole thing commits or it aborts (rolls back), and after an abort the application can safely retry. The application no longer has to deal with partial failure.

A short history. Almost every relational database (MySQL, PostgreSQL, Oracle, SQL Server) implements transactions in a way that is remarkably similar to IBM’s System R (1975), the first SQL database. In the late 2000s many NoSQL databases dropped transactions, or redefined the word to mean something weaker, for the sake of replication and sharding. That led to a popular belief that transactions don’t scale. “NewSQL” systems (CockroachDB, TiDB, Spanner, FoundationDB, YugabyteDB) have since shown this is wrong: they combine sharding with consensus protocols to provide strong ACID guarantees at scale.

Some applications can do without transactions, or weaken them for performance or availability. They do prevent a lot of grief, though: the technical cause of the UK Post Office Horizon scandal was probably an accounting system without ACID transactions. To decide whether you need them, you have to know exactly what they guarantee and what they cost.

Chapter map

Click any node to jump there.

2The meaning of ACID

ACID (Atomicity, Consistency, Isolation, Durability) was coined in 1983 by Härder and Reuter to give fault-tolerance mechanisms precise names. In practice one database’s ACID is not another’s; isolation in particular is ambiguous. “ACID compliant” has become largely a marketing term.

Atomicity: really “abortability”

In multithreaded programming, “atomic” means no other thread can see a half-finished operation. ACID atomicity is not about concurrency (that’s the I). It describes what happens when a client makes several writes and a fault occurs partway through: a crash, a dropped connection, a full disk, a violated constraint. The transaction is aborted and the database discards every write it made.

Without that guarantee, the application can’t tell which changes took effect. Retrying risks applying some changes twice, and not retrying leaves the data inconsistent. With atomicity, an aborted transaction changed nothing, so a retry is safe.

Consistency: your invariants, mostly your job

“Consistency” has at least five unrelated meanings:

An invariant is something that must always be true, e.g. “credits and debits balance across all accounts”. If every transaction starts from a valid state and preserves validity, the invariants always hold. They may be temporarily broken during a transaction, as long as they hold again at commit.

The database can enforce invariants you declare as constraints: foreign keys, uniqueness, check constraints, and sometimes triggers or materialized views. Anything more complex is the application’s responsibility. If you write bad data that breaks an undeclared invariant, the database can’t stop you. So C is a property of how the application uses the database, not of the database alone. (A, I and D are database properties.)

Isolation: concurrent transactions don’t step on each other

Many clients access the database at once. If they touch the same records, you can get race conditions. In the classic example below, two clients increment a counter without a database-level increment.

The textbook definition of isolation is serializability: each transaction can pretend it is the only one running, and the committed result equals running them serially, one after another. Serializability costs performance, so many databases offer only weaker levels. Oracle’s level called “serializable” is actually snapshot isolation, which is weaker.

Durability: don’t forget committed data

Once a transaction has committed, its data must survive hardware faults and crashes.

• Single node: written to non-volatile storage. Plain file writes sit in the OS page cache, so databases call fsync to force data to disk. A write-ahead log allows recovery from a crash mid-write, and checksums (MySQL, MongoDB, PostgreSQL) detect torn or corrupt log entries.
• Replicated: copied to some number of nodes before reporting success.

The database must finish these writes before acknowledging the commit. Try it:

Replication vs disk: nothing is perfect

There are no absolute guarantees, only risk-reduction techniques: disk writes, replication and backups. Use them together.

3Single-object and multi-object operations

Atomicity and isolation matter most when a transaction modifies several objects that must stay in sync:

• Atomicity: if an error occurs halfway through, earlier writes are discarded (all or nothing).
• Isolation: if a transaction makes several writes, another transaction sees all of them or none of them, never a subset.

Example: an email app stores the unread count in a separate field (denormalized) because SELECT COUNT(*) … WHERE unread_flag is too slow. Now inserting an email and incrementing the counter must happen together.

How does the database know which operations belong together? In relational databases, by the client’s TCP connection: everything between BEGIN TRANSACTION and COMMIT on that connection is one transaction, and if the connection drops, the transaction is aborted. Many non-relational databases have no such grouping. A multi-put that updates several keys at once may succeed for some keys and fail for others, so it isn’t necessarily a transaction.

Single-object writes

Even writing one 20 kB JSON document raises questions. If the network drops after 10 kB, is a broken fragment stored? If power fails mid-overwrite, do old and new values get spliced? Can a concurrent reader see half a document? Storage engines almost universally make single-object writes on one node atomic (via a log for crash recovery) and isolated (via a per-object lock).

Some databases offer richer single-object operations:

• Atomic increment: no read-modify-write cycle needed. Strictly, this should be called an isolated or serializable increment, since it’s about concurrency.
• Conditional write / compare-and-set: write only if the value hasn’t changed since you read it (more below).

These prevent lost updates on one object, but they are not transactions in the usual sense. Aerospike’s “strong consistency” mode and the “lightweight transactions” of Cassandra and ScyllaDB give linearizable reads and conditional writes on a single object, with no multi-object guarantees.

The need for multi-object transactions

You can build all of these without transactions, but error handling gets much harder without atomicity, and missing isolation causes the concurrency problems covered below.

Handling errors and aborts

ACID databases would rather abandon a transaction entirely than leave it half-finished. Leaderless-replication datastores work on a best-effort basis instead: they do as much as they can, and on error they don’t undo what’s done, so recovery is the application’s job. Meanwhile popular ORMs (Rails ActiveRecord, Django) don’t retry aborted transactions; the exception bubbles up and user input is lost. That wastes the main benefit of aborts.

Retrying isn’t a silver bullet, though:

4Weak isolation levels: read committed

Race conditions only arise when one transaction reads data another is modifying, or both modify the same data. They are timing-dependent, so they are rarely triggered in testing and hard to reason about in large codebases. Serializable isolation hides all of them but costs performance, so databases commonly default to weaker levels that prevent some anomalies and not others.

The ladder, from weakest to strongest:

Read committed

The most basic real isolation level makes two guarantees:

1. No dirty reads: you only see committed data.
2. No dirty writes: you only overwrite committed data.

No dirty reads

A dirty read means seeing another transaction’s writes before it commits or aborts. Under read committed, a transaction’s writes become visible only when it commits, and then all at once. Toggle the level and the ending below:

Why it matters: (1) a reader may see some but not all of a multi-row update, which is confusing and can lead to wrong decisions; (2) if the writer aborts, the reader has seen data that never existed, and undoing that leads to cascading aborts.

No dirty writes

A dirty write means overwriting a value another transaction has written but not yet committed. Read committed prevents it, usually by making the second writer wait until the first commits or aborts.

Implementing read committed

Read committed is the default in Oracle, PostgreSQL, SQL Server and many others.

• Dirty writes → row-level locks. To modify a row you must acquire its lock and hold it until commit or abort. Only one transaction can hold it, and others wait. Databases do this automatically at read committed and above.
• Dirty reads → remember the old value. You could make readers briefly take the same lock, and IBM Db2 and SQL Server with read_committed_snapshot=off do. But one long write transaction then stalls every reader, hurting response time and letting a slowdown in one part of the app spread to others. Most databases instead keep both the old committed value and the new uncommitted value, and give readers the old one until commit. (This is MVCC in miniature; see below.)

Read uncommitted is weaker still: it prevents dirty writes but allows dirty reads, returning the latest value even if uncommitted. It can be faster, since only one version is kept, and it makes lost updates less likely but doesn’t prevent them.

5Snapshot isolation and repeatable read

Read committed allows aborts, hides incomplete results and keeps concurrent writes from mixing. Plenty of bugs remain. Here is one where nothing uncommitted is ever read:

This is read skew, a kind of nonrepeatable read. (This “skew” is a timing anomaly, unrelated to the hot-spot skew of sharding.) For Aaliyah it’s temporary, since a reload a few seconds later looks fine. It is not tolerable for:

• Backups: copying a large database takes hours while writes continue. Parts of the backup are older than others, and after a restore the inconsistency (vanished money) becomes permanent.
• Analytical queries and integrity checks: scans over large parts of the database return nonsense if they see different parts at different times.

Snapshot isolation is the usual fix. Each transaction reads from a consistent snapshot: all data committed when it started, frozen, even if other transactions change it afterwards. This is a big help for long read-only queries. PostgreSQL, MySQL/InnoDB, Oracle and SQL Server all support variants of it. Oracle, TiDB and Aurora DSQL even make it their highest level, and cloud warehouses like BigQuery use it for point-in-time analytics.

Multiversion concurrency control (MVCC)

Like read committed, snapshot isolation uses write locks to prevent dirty writes, so writers can block writers. Reads take no locks. The core principle:

Generalizing the two-version trick from read committed: the database keeps several committed versions of each row, because different in-progress transactions may need to see the database at different points in time. Hence multiversion. In PostgreSQL-style MVCC:

• Each transaction gets a unique, always-increasing transaction ID (txid). In PostgreSQL these are 32-bit and wrap after about 4 billion; the vacuum process keeps that safe.
• Each row version has inserted_by (the txid that created it) and deleted_by (empty until deleted).
• A delete doesn’t remove the row; it sets deleted_by. An update = delete + insert, so updating account 2 from $500 to $400 leaves two versions.
• All versions live in the same heap, linked newest↔oldest. A garbage collector removes versions no transaction can see any more.

Visibility rules for a consistent snapshot

When a transaction reads, txids decide which versions it sees:

1. At transaction start, the database lists all transactions in progress. Their writes are ignored, even if they commit later.
2. Writes by transactions with a later txid (started after us) are ignored, whether committed or not.
3. Writes by aborted transactions are ignored, whenever they aborted. So aborted rows don’t need immediate removal; GC cleans them up later.
4. All other writes are visible.

These apply to insertions and deletions. Put differently, a row version is visible if:

• the inserting transaction had already committed when the reader started, and
• the row isn’t marked deleted, or the deleting transaction had not committed when the reader started.

Never updating in place means a long-running transaction can keep reading values that everyone else has long since overwritten, at only a small overhead. In the lab above, “Load Figure 8-7” shows exactly this: txn 12 still sees $500 + $500 after txn 13 committed its transfer.

Indexes and snapshot isolation

Commonly, each index entry points at one version of a matching row (oldest or newest), and each version links to the next. A query walks the chain to find a version that is both visible and matching. When GC removes a dead version, its index entries go too. Implementation details matter a great deal: PostgreSQL skips index updates when versions of a row fit on the same page, and some databases store only the deltas between versions.

A different approach is used by CouchDB, Datomic and LMDB: immutable, copy-on-write B-trees. Pages are never overwritten. A write copies the modified leaf and every parent up to the root, and unaffected pages are shared. Each root is a consistent snapshot, so there is no txid filtering. It still needs background compaction and GC.

Snapshot isolation, repeatable read, and naming confusion

The SQL standard is based on System R’s 1975 isolation levels and predates snapshot isolation, so it has no such level. It defines repeatable read, which looks superficially similar. The standard’s definitions are also ambiguous and imprecise. The result:

Research has defined repeatable read formally, but most implementations don’t satisfy that definition. Nobody really knows what “repeatable read” means. Test your actual database (the Hermitage project does this).

6Preventing lost updates

So far we have looked at what a reader sees. With two concurrent writers, the best-known problem is the lost update: two transactions do a read-modify-write cycle on the same object, and the second write clobbers the first because it doesn’t include the first one’s change. It shows up in:

• incrementing a counter or updating an account balance
• making a local change to a complex value, like appending to a list inside a JSON document
• two users editing a wiki page, each saving the whole page

Compare the five main solutions:

Atomic write operations

UPDATE counters SET value = value + 1 WHERE key = 'foo';

This is concurrency-safe in most relational databases. MongoDB has atomic operations on parts of a JSON document, and Redis on data structures like priority queues. They are usually the best choice when the change can be expressed this way; arbitrary text edits can’t (those need CRDTs or operational transformation, ch. 6). Atomic operations are implemented with an exclusive lock on the object from read to write, or by running them all on one thread. A common pitfall: ORMs make it easy to write an unsafe read-modify-write by accident.

Explicit locking

When the logic is too complex for an atomic op, such as checking that a game move is legal, the application can lock the rows it’s about to modify:

BEGIN TRANSACTION;
SELECT * FROM figures
  WHERE name = 'robot' AND game_id = 222
  FOR UPDATE;            -- lock every row this returns
-- check whether the move is valid, then:
UPDATE figures SET position = 'c4' WHERE id = 1234;
COMMIT;

It works, but it’s easy to forget a lock somewhere. Locking several objects also risks deadlock; the database detects it and aborts one transaction, which the app must retry.

Automatically detecting lost updates

Instead of forcing read-modify-write cycles to run one after another, let them run in parallel. If the transaction manager detects a lost update, it aborts the offender, which retries. This fits naturally with snapshot isolation: PostgreSQL repeatable read, Oracle serializable and SQL Server snapshot isolation do it automatically. MySQL/InnoDB repeatable read does not, so by some definitions it doesn’t provide snapshot isolation at all. The big advantage is that it needs no special code, so there’s no lock to forget. You still need to retry.

Conditional writes (compare-and-set)

Databases without transactions often offer a conditional write that applies only if the value hasn’t changed since you read it, the database version of a CPU’s CAS instruction:

-- may or may not be safe, depending on the implementation
UPDATE wiki_pages SET content = 'new content'
  WHERE id = 1234 AND content = 'old content';

If it updates 0 rows, retry. Comparing a version number column that you increment on every write is common, and is called optimistic locking. Caveat: under MVCC, the other transaction’s new content might be invisible in your snapshot. Many MVCC implementations make an exception and let UPDATE/DELETE WHERE clauses see the latest committed writes. Check that yours does.

Conflict resolution and replication

Locks and CAS assume a single up-to-date copy of the data. Multi-leader and leaderless replication allow concurrent writes on different nodes, replicated asynchronously, so that assumption fails (more under linearizability, ch. 10). Instead they keep conflicting versions (siblings) and merge them afterwards with application code or special data structures.

• Merging prevents lost updates if the operations are commutative (counter increments, set additions). That is the idea behind CRDTs. Conditional writes can’t be made commutative.
• Last write wins (LWW), the default in many replicated databases, is prone to lost updates.

7Write skew and phantoms

Dirty writes and lost updates involve two transactions writing the same object. Subtler races remain. A hospital must always have at least one doctor on call per shift. A doctor may go off call only if a colleague remains. Aaliyah and Bryce, the only two on call, both feel sick and click at the same moment:

Characterizing write skew

Write skew is neither a dirty write nor a lost update: the two transactions update different objects. It is still clearly a race, since serially the second doctor would have been refused.

Write skew is a generalization of the lost update. The options for fixing it are more limited:

• Atomic single-object operations don’t help, because several objects are involved.
• Automatic lost-update detection doesn’t help either. Write skew is not detected by PostgreSQL repeatable read, MySQL/InnoDB repeatable read, Oracle serializable or SQL Server snapshot isolation. Only true serializable isolation catches it.
• Constraints rarely help, because most databases don’t support constraints spanning multiple objects. Triggers or materialized views can sometimes emulate them.
• Second best: explicitly lock the rows the decision depends on (SELECT … FOR UPDATE). Try the second option in the demo above.

More examples of write skew

Phantoms causing write skew

All these examples follow one pattern:

1. A SELECT checks a requirement by searching for rows matching a condition.
2. Based on the result, the application decides whether to go ahead.
3. If it goes ahead, it writes (INSERT/UPDATE/DELETE) and commits, and that write changes the result of step 1’s search. Re-running the query now would give a different answer.

(The steps can come in a different order, e.g. write first, then query, then decide whether to commit.)

In the doctors case, the modified row was among those returned in step 1, so locking them works. The other examples check for the absence of rows and then add one. A SELECT FOR UPDATE that returns nothing locks nothing.

A write in one transaction that changes the result of another transaction’s search query is called a phantom. Snapshot isolation prevents phantoms in read-only queries, but in read-write transactions they cause especially tricky write skew. ORM-generated SQL is prone to it too.

Materializing conflicts

If the problem is that there’s no object to lock, create one. Pre-create a table of (room, 15-minute time slot) rows for the next six months. A booking transaction first locks the slot rows for its room and time (SELECT FOR UPDATE), then checks and inserts as before. The table stores no booking data; it is purely a collection of locks. This turns a phantom into a lock conflict on concrete rows (see the third option in the demo above).

It is hard to get right and error-prone, and it leaks concurrency control into the data model. Treat it as a last resort; serializable isolation is usually better.

8Serializability

So far: isolation levels are hard to understand and inconsistently implemented. It’s hard to tell from application code whether it’s safe at a given level, especially in a large app where you don’t know what else runs concurrently. And there are no good tools: static analysis exists in research but isn’t practical yet, and concurrency bugs are nondeterministic. Researchers have given the same answer since the 1970s: use serializable isolation.

Why isn’t everyone using it? The answer lies in the cost of the three implementation techniques:

Actual serial execution

Remove the concurrency entirely: one transaction at a time, on a single thread. Conflicts can’t happen, so the result is serializable by definition. Database designers only decided this was feasible in the 2000s, for two reasons:

• RAM got cheap, so the whole active dataset often fits in memory, and transactions don’t wait on disk.
• OLTP transactions are short and make few reads and writes. Long analytical queries are read-only and can run on a snapshot, outside the serial loop.

VoltDB/H-Store, Redis and Datomic work this way. With no locking overhead, a single-threaded system can sometimes beat a concurrent one, but its throughput is capped at one CPU core, and transactions must be structured differently.

Encapsulating transactions in stored procedures

The early hope was that one transaction could span an entire user flow (search flights, choose, book seats, enter details, pay). Humans are slow, though, and a database can’t efficiently support huge numbers of mostly idle transactions. So OLTP apps keep transactions within one HTTP request.

Even so, transactions usually run interactively: one statement, wait for the result, the next statement, with network round trips in between. A single-threaded database would sit idle during those waits. So serial-execution systems don’t allow interactive multi-statement transactions. You must either use single statements or submit the whole transaction up front as a stored procedure.

Procedures also help where logic can’t live elsewhere, for example validation behind a GraphQL proxy that exposes the database directly (otherwise you’d need a separate validation service). VoltDB also replicates with stored procedures: it executes the same procedure on every replica instead of copying writes. That requires procedures to be deterministic, so the current time etc. comes from special deterministic APIs. This is state machine replication, covered further in ch. 10.

Sharding

To go beyond one core, shard the data so each transaction touches one shard. Each shard then gets its own thread, and throughput scales linearly with cores. A cross-shard transaction must run in lockstep across all its shards, which is vastly slower: VoltDB reports about 1,000 cross-shard writes per second, which doesn’t improve with more machines. (Research such as Lotus explores scaling multi-shard transactions.) Key-value data shards easily; data with many secondary indexes needs a lot of cross-shard coordination.

Two-phase locking (2PL)

For about 30 years, 2PL (strictly, strong strict two-phase locking, SS2PL) was the only widely used serializability algorithm. It strengthens the write locks used against dirty writes: many transactions may read an object concurrently, but writing needs exclusive access.

• A has read an object and B wants to write it: B waits until A commits or aborts, so B can’t change data behind A’s back.
• A has written an object and B wants to read it: B waits. Reading the old version, as in read committed, isn’t allowed.

So writers block readers and readers block writers, the opposite of snapshot isolation’s mantra. In exchange you get protection from all races, including lost updates and write skew. It is used by serializable in MySQL/InnoDB and SQL Server, and by repeatable read in Db2.

Implementation: shared/exclusive locks and deadlock

Every object has a lock that is either shared (S) or exclusive (X), i.e. a multi-reader single-writer lock:

With so many locks, deadlocks are common: A waits for B and B waits for A. The database detects the cycle and aborts one transaction, and the application retries it.

Performance of two-phase locking

2PL’s throughput and response times are significantly worse than weak isolation’s, which is why it hasn’t been the default since the 1970s. Part of the cost is acquiring and releasing locks, but mostly it is reduced concurrency: anything that might race has to wait.

• A transaction reading a whole table (backup, analytics, integrity check) needs a shared lock on all of it. It first waits for every in-progress writer, then blocks all writers for as long as it reads. The database is effectively unavailable for writes.
• Latencies are unstable, with very slow high percentiles under contention. One slow or lock-heavy transaction can grind everything to a halt. Timeouts and slow-query monitoring help contain it.
• Deadlocks are much more frequent than with lock-based read committed, and each one throws away the aborted transaction’s work.

Predicate locks

A serializable database must also prevent phantoms. If a transaction searched for bookings of room 123 from noon to 1 p.m., no other transaction may insert or update a booking matching that search. Bookings for other rooms or times are fine. Conceptually this needs a predicate lock, which belongs to all objects matching a condition:

SELECT * FROM bookings
  WHERE room_id = 123 AND
    end_time   > '2026-01-01 12:00' AND
    start_time < '2026-01-01 13:00';

• To read matching objects, A acquires a shared predicate lock on the condition. If B holds an exclusive lock on any matching object, A waits.
• To insert, update or delete, A first checks whether the old or new value matches any existing predicate lock. If B holds a matching one, A waits.

The key point is that a predicate lock covers objects that don’t exist yet. 2PL with predicate locks prevents all write skew and race conditions.

Index-range locks

Checking every write against every active predicate is slow. Most 2PL databases use index-range locking (also called next-key locking), a simplified approximation. It is safe to lock a superset of the predicate: lock room 123 at any time, or all rooms from noon to 1 p.m. Any write matching the original predicate matches the approximation too.

• With an index on room_id, put a shared lock on index entry 123.
• With a time index, put a shared lock on the range noon–1 p.m. in that index.

Any conflicting insert, update or delete has to touch the same part of the index, finds the lock, and waits. This is less precise than predicate locking, since it may block harmless writes, but much cheaper. With no suitable index the database falls back to a shared lock on the whole table: safe, but bad for performance.

Serializable snapshot isolation (SSI)

So far serializability has meant poor performance (2PL) or poor scalability (serial execution), and weak isolation has meant races. SSI, first described in 2008, gives full serializability for a small cost over snapshot isolation. It’s used by PostgreSQL’s serializable level, SQL Server’s In-Memory OLTP (Hekaton), HyPer, CockroachDB, FoundationDB and BadgerDB.

Pessimistic vs optimistic concurrency control

Optimistic control is an old idea. It does badly under high contention, because many aborts happen, and near maximum throughput the retries add load and make things worse. With spare capacity and moderate contention it usually beats pessimistic control. Commutative atomic operations reduce contention: concurrent increments of a counter that isn’t read in the same transaction don’t conflict.

Decisions based on an outdated premise

Write skew always looks like this: read → examine → act on what was seen. By commit time the read may be stale. The transaction acted on a premise (“two doctors are on call”) that is no longer true. The database doesn’t know how the app uses a query result, so to be safe it assumes that any change to the result invalidates the writes. It must detect when a transaction may have acted on an outdated premise and abort it. There are two cases:

1. Stale MVCC reads: an uncommitted write happened before the read and was hidden by snapshot visibility.
2. Writes that affect prior reads: the write happens after the read.

SSI detection, step by step

• Case 1: the database tracks when a transaction ignores another’s writes because of MVCC rules. At commit, if any ignored write has since committed, it aborts. It waits until commit because the reader might be read-only, which is harmless, or the writer might still abort. Avoiding needless aborts keeps long read-only snapshots cheap.
• Case 2: like index-range locks, except they don’t block. Reads are recorded on index entries (e.g. shift_id = 1234, or at table level without an index). A write looks up recent readers and notifies them, which works like a tripwire. The first to commit wins, and a later committer whose premise was overwritten by a committed write aborts. The read records can be dropped once the reader and all transactions concurrent with it have finished.

Performance of SSI

• Tracking granularity is a trade-off: precise tracking means fewer unnecessary aborts but more bookkeeping, and coarse tracking is faster but aborts more. PostgreSQL uses theory showing that some reads of overwritten data are still serializable, which cuts aborts.
• vs 2PL: no transaction blocks on another’s locks. Readers and writers don’t block each other, latency is more predictable, and read-only queries take no locks, which suits read-heavy workloads.
• vs serial execution: not limited to one core. FoundationDB distributes conflict detection across machines, and transactions can span shards and stay serializable.
• vs plain SI: there is some checking overhead. How much matters is debated: some say serializability checking isn’t worth it, others say it’s now so cheap that weaker SI is unnecessary.
• Abort rate dominates: long read-write transactions are likely to conflict and abort, so they should be short. Long read-only ones are fine. SSI is less sensitive to slow transactions than 2PL or serial execution.

Comparing the three

9Distributed transactions

A transaction that touches multiple nodes, such as several shards or a global secondary index whose entries live on other nodes, is a distributed transaction. Concurrency control carries over (serial execution with sharding, 2PL across nodes, distributed SSI checkers). The hard new problem is atomicity.

How a single node commits

On one node, atomic commit is the storage engine’s job. It makes the writes durable (in the WAL), then appends a commit record. After a crash, recovery keeps the transaction if the commit record made it to disk and rolls it back otherwise. The deciding moment is when the disk finishes writing the commit record, so a single device on a single node makes the commit atomic.

With several nodes, simply sending COMMIT to each one independently doesn’t work. One may see a constraint violation, a request may be lost and time out, or a node may crash before its commit record is durable:

Once committed on a node, a transaction can’t be retracted. Under read committed or stronger, other transactions may already have read the data and acted on it. Making all nodes commit or all abort is the atomic commitment problem.

Two-phase commit (2PC)

2PC is the classic algorithm for atomic commit across nodes. It is used inside some databases and exposed to applications as XA transactions (e.g. via the Java Transaction API) or WS-AtomicTransaction for SOAP. It adds a coordinator (transaction manager), often a library inside the application process (e.g. a Java EE container), sometimes a separate service: Narayana, JOTM, BTM, MSDTC. The database nodes are participants.

• Phase 1: the coordinator sends prepare to every participant: “can you commit?”
• Phase 2: if all say yes, it sends commit; if any says no, it sends abort to all.

It resembles a Western wedding ceremony: the officiant asks each partner individually, both say “I do”, and only then is the couple pronounced married and the news broadcast. If either says no, the ceremony is aborted.

A system of promises

Prepare and commit messages can be lost just as easily as a one-phase commit, so why does 2PC work? Look at the details:

1. The app requests a globally unique transaction ID from the coordinator.
2. It starts a single-node transaction on each participant, tagged with that ID, and does its reads and writes. Until now anyone may abort (crash, timeout).
3. Ready to commit, the coordinator sends prepare (tagged with the ID) to all participants. If any request fails or times out, it sends abort to all.
4. A participant that receives prepare makes sure it can commit under all circumstances: all data on disk (a crash, power failure or full disk is no excuse later), constraints and conflicts checked. Voting yes is a promise: it gives up the right to abort, without actually committing.
5. With all responses in, the coordinator decides (commit only if all voted yes) and writes the decision to its log on disk. This is the commit point.
6. It sends commit/abort to all participants, and retries forever on failure. If a participant crashed meanwhile, it commits on recovery, because it voted yes.

In the wedding analogy: before “I do” you may still say “No way!”. After it you can’t take it back. If you faint and miss the pronouncement, you are still married. When you come round, you can ask the officiant about your global transaction ID, or wait for their next retry.

Coordinator failure

Participant or network failures are handled: failed prepares lead to an abort, and failed commits/aborts are retried. If the coordinator crashes:

• Before sending prepare, a participant can safely abort.
• After a participant has voted yes, it cannot abort unilaterally. It is in doubt (uncertain) and must wait. A timeout doesn’t help: aborting could contradict a node that already committed, and committing could contradict a node that voted no.

Participants could in principle ask each other, but that isn’t part of 2PC. The only way to finish is for the coordinator to recover, which is why it logs its decision before sending it. On recovery it reads the log, and any transaction without a commit record is aborted. So the commit point of 2PC comes down to a regular single-node atomic commit on the coordinator. (Choose “Coordinator crash after deciding” in the 2PC demo above to see this.)

Three-phase commit

2PC is a blocking atomic commit protocol: it can get stuck waiting for the coordinator. Three-phase commit (3PC) aims to be nonblocking, but it assumes bounded network delay and bounded response times. Real systems have unbounded delays and process pauses (ch. 9), so 3PC can’t guarantee atomicity in practice. The practical fix is to replace the single coordinator with a fault-tolerant consensus protocol (ch. 10).

Distributed transactions across different systems

Distributed transactions have a mixed reputation: they provide a safety guarantee that’s hard to get otherwise, but they are blamed for operational problems, killing performance and overpromising. Many cloud services don’t offer them. Much of 2PC’s cost comes from the extra fsyncs for crash recovery and the extra network round trips. Two very different kinds often get lumped together:

Exactly-once message processing (via distributed transactions)

Heterogeneous transactions allow a message from a queue to be acknowledged if and only if the database transaction that processed it committed. The ack and the DB writes are committed atomically. If either fails, both abort and the broker safely redelivers later. The message is effectively processed exactly once, even after retries. This is exactly-once semantics.

It only works if every affected system speaks the same atomic commit protocol. If processing also sends an email and the email server doesn’t support 2PC, retries can send the email twice. Only if all side effects roll back on abort can processing be retried as if nothing happened.

XA transactions

X/Open XA (eXtended Architecture, 1991) is the standard for 2PC across heterogeneous systems. It is supported by PostgreSQL, MySQL, Db2, SQL Server, Oracle, and by brokers such as ActiveMQ, HornetQ, MSMQ and IBM MQ. XA is not a network protocol; it is a C API for talking to a transaction coordinator, with bindings like JTA (backed by JDBC drivers for databases and JMS drivers for brokers).

If the application process crashes or its machine dies, the coordinator dies with it, and every prepared participant is stuck in doubt. The server must restart and the coordinator library must read its log before it can tell the participants the outcome through the drivers’ XA callbacks.

Holding locks while in doubt

Why care so much about in-doubt transactions? Because of locks. Modified rows carry exclusive locks against dirty writes, and with 2PL, read rows carry shared locks. None can be released until commit or abort. If the coordinator takes 20 minutes to restart, the locks are held for 20 minutes. If its log is lost, they are held forever, or until an administrator steps in. Other transactions touching those rows block, and large parts of the application can become unavailable. (In the 2PC demo, the shaded lifelines show the locks being held.)

Orphaned in-doubt transactions, whose outcome the coordinator can’t determine (lost or corrupted log, software bugs), do happen. They hold their locks forever, and rebooting doesn’t help: correct 2PC must preserve in-doubt locks across restarts. An administrator has to look at every participant, see whether any has committed or aborted, and apply the same outcome to the rest, usually under pressure during an outage.

The escape hatch is heuristic decisions: a participant unilaterally commits or aborts an in-doubt transaction. Here “heuristic” is a euphemism for probably breaking atomicity, so they are for emergencies only.

Problems with XA transactions

• A single-node coordinator is a single point of failure, and running it inside the application server makes that server’s local log as critical as the databases.
• Even a replicated coordinator can’t talk to participants directly; everything goes through application code and drivers. So the application becomes the single point of failure. Fixing that would need restartable, replicated application code, something like durable execution, but no tools do this in practice.
• It’s a lowest common denominator: no deadlock detection across systems (there’s no standard way to exchange lock info), and it doesn’t work with SSI (there’s no protocol for cross-system conflict detection).

These problems are inherent to heterogeneous transactions. Keeping heterogeneous systems consistent is still important, so other approaches are needed (below, and in ch. 12).

Database-internal distributed transactions

Internal distributed transactions are a defining feature of NewSQL databases (CockroachDB, TiDB, Spanner, FoundationDB, YugabyteDB); Kafka has them too. Many use 2PC for multi-shard writes without XA’s problems, because they don’t interoperate with anything else and can use better, faster protocols. They fix XA’s biggest problems by:

• replicating the coordinator, with automatic failover
• letting the coordinator and shards talk directly, without application code in between
• replicating the participating shards, so a fault in one is less likely to force an abort
• coupling atomic commit with distributed concurrency control that detects deadlocks and gives consistent cross-shard reads

Consensus algorithms (ch. 10) replicate the coordinator and shards and fail over automatically while keeping strong guarantees. Snapshot isolation and SSI are both possible across shards.

Exactly-once message processing revisited

You don’t need a distributed transaction for exactly-once semantics. Transactions inside the database are enough:

1. Every message has a unique ID, and the database has a table of processed IDs. Start a transaction and check the ID. If it is present, the message was already processed: ack and drop it.
2. Otherwise insert the ID, process the message (more writes), and commit, all in one transaction.
3. After the commit, ack the message to the broker.
4. After the ack, delete the ID (separate transaction), since the broker won’t redeliver it.

Recording the ID makes processing idempotent, so retries are safe. Kafka Streams uses a similar approach (ch. 12). Internal distributed transactions still help this pattern scale, e.g. storing message IDs on one shard and the data they update on others, with atomic commit across those shards.

10Summary

Transactions are an abstraction layer that lets an application pretend certain concurrency problems and certain hardware and software faults don’t exist. A large class of errors becomes a simple abort, and the application just retries. Apps with trivial access patterns (single-record reads and writes) may not need them. For anything more complex they greatly reduce the number of error cases you have to think about. Without them, crashes, network interruptions, power outages, full disks and unexpected concurrency leave data inconsistent in many ways; denormalized data drifts out of sync, for example. Transactions are valuable regardless of data model.

Anomaly matrix

The book’s Table 8-1. Dirty writes are prevented by almost every implementation, so they aren’t listed.

Serializable implementations: serial execution (fast, short stored procedures, fits one core or shards cleanly); two-phase locking (the classic, poor performance); SSI (optimistic, non-blocking, aborts at commit if not serializable). Distributed atomicity: 2PC works well inside one database system, and is fragile across heterogeneous systems (XA): sensitive to coordinator and application faults, and poor with concurrency control. Idempotence provides exactly-once semantics without cross-system atomic commit.

Self-check quiz

Glossary