Chapter 13: Distributed Transactions — Atomicity Across Machines

Curriculum: Distributed Systems from First Principles Chapter: 13 of 24 Part: 5 — Consensus Prerequisites: Chapter 2 (ACID, transactions, WAL), Chapter 6 (consistency models), Chapter 7 (sharding), Chapter 11 (consensus, why 2PC is not consensus), Chapter 12 (Raft, multi-Raft in CockroachDB) Estimated reading time: 85–95 minutes

Opening

You have a banking application. User Alice has an account on Shard 1. User Bob has an account on Shard 7. Alice sends Bob $100. This operation must be atomic: the debit from Alice's account and the credit to Bob's account must either both happen or neither happens. On a single database, this is a transaction — four lines of SQL, one COMMIT, done. The database's WAL (Chapter 2) guarantees atomicity even across a crash.

Across shards, there is no shared WAL. Shard 1 and Shard 7 are independent machines with independent storage engines, independent logs, and no way to atomically commit to both simultaneously. If you debit Alice and then crash before crediting Bob, the money disappears. If you credit Bob and then crash before debiting Alice, the bank creates money from nothing. Both outcomes are catastrophic. This is the distributed transaction problem: achieving ACID atomicity across machines that share nothing.

There is no perfect solution. Every approach to distributed transactions is a trade-off — between blocking (the system halts during coordinator failure), performance (transactions take more round trips), complexity (the protocol is hard to implement correctly), and correctness (some approaches allow certain anomalies). This chapter builds each approach from first principles, explains precisely which trade-off it makes, and shows where each one is used in production.

Part 1: What Single-Machine ACID Actually Does — The Foundation

Before distributing transactions, understand exactly what a single-machine transaction gives you, because the distributed problem is about recreating each of these guarantees across machine boundaries.

Atomicity: All operations in a transaction succeed or all fail. The WAL (Chapter 2) provides atomicity: before committing, the database writes the full transaction record to the WAL. If it crashes after the WAL write but before applying to heap pages, recovery replays the WAL and completes the write. If it crashes before the WAL write, recovery sees no record and the transaction never happened. The WAL is the atomicity mechanism.

Consistency: The database moves from one valid state to another. Foreign keys, CHECK constraints, and application invariants are all checked at commit time. If any constraint is violated, the transaction aborts.

Isolation: Concurrent transactions don't interfere with each other. The database uses locks (or MVCC — multi-version concurrency control) to give each transaction the illusion that it runs alone. The isolation level determines exactly which anomalies are permitted.

Durability: Once committed, the write survives crashes. The WAL fsync before commit acknowledgement guarantees this — the WAL record is on durable storage before the client is told "success."

For distributed transactions, atomicity is the hardest property to preserve across machines. Durability per-shard is easy (each shard has its own WAL). Isolation across shards requires distributed lock management or distributed MVCC. Consistency requires all shards to agree on the same commit/abort decision. The commit decision — the thing that is atomic on a single machine — becomes the central problem across machines.

Part 2: Two-Phase Commit — The Foundational Protocol

Two-phase commit (2PC) is the canonical protocol for distributed atomicity. It introduces a coordinator — a node whose job is to orchestrate the commit decision across all participants (the shards involved in the transaction).

The protocol has two phases, which map almost exactly to what a single database does internally during a commit.

Phase 1 — Prepare

The coordinator sends a Prepare message to every participant. Each participant must determine: can I commit my part of this transaction? To answer yes, it must:

• Execute all the transaction's operations on its local data
• Acquire all necessary locks
• Write the prepared state to its local WAL (durably — this write must survive a crash)
• Reply Yes to the coordinator

If any participant cannot commit (constraint violation, deadlock detected, lock timeout), it replies No and releases its locks.

Phase 2 — Commit or Abort

If the coordinator receives Yes from all participants, it writes Commit to its own log and sends Commit to all participants. Each participant commits its local transaction, releases locks, and acknowledges. If any participant replied No, the coordinator sends Abort to all, and each participant rolls back.

The correctness of 2PC rests on one invariant: once a participant writes Prepared to its WAL and votes Yes, it has irrevocably promised to commit if told to and cannot unilaterally abort. This is why the prepared state is written to the WAL durably before voting — the participant must honour this promise even across a crash and recovery.

2PC appears in: Postgres's PREPARE TRANSACTION and COMMIT PREPARED commands (which allow an external coordinator to drive 2PC across Postgres instances), MySQL's XA transactions, Java's JTA (Java Transaction API), and as the inner commit protocol in Spanner, CockroachDB, and TiDB.

Part 3: The Blocking Problem — 2PC's Fatal Flaw

2PC is correct under normal operation. Under coordinator failure, it produces the most operationally painful situation in distributed systems: blocking.

Trace this failure scenario:

Now Shard 1 and Shard 7 are in the PREPARED state. They have voted Yes. They are holding locks. They cannot unilaterally commit (they don't know if all participants voted Yes) and they cannot unilaterally abort (they already promised to commit). They must wait for the coordinator to recover and tell them what to do.

How long do they wait? Until the coordinator recovers. This might be seconds (quick restart), minutes (disk replacement), or hours (hardware failure requiring manual intervention). During this wait, Alice's account is locked and Bob's account is locked. No other transaction can modify either account. The bank is frozen for two accounts until the coordinator comes back.

Participant in PREPARED state:
  - Holds all locks for this transaction
  - Cannot commit without coordinator's Commit message
  - Cannot abort without coordinator's Abort message
  - Must wait, potentially indefinitely
  
This is 2PC blocking: participants are blocked in an uncertain state
waiting for a coordinator that may never return.

This is why Chapter 11 noted that 2PC is not consensus: it violates the termination property. A correct consensus algorithm guarantees that non-faulty nodes eventually decide. In 2PC, the participants are non-faulty — Shard 1 and Shard 7 are running fine — yet they can never decide without the coordinator. Faulty coordinator = blocked participants forever.

The operational mitigation: make the coordinator itself highly available. Run the coordinator as a Raft group (3 replicas). If the coordinator leader crashes, a new Raft leader takes over and has the coordinator's log — it sees the COMMIT record at t=4 and sends the Commit messages. This is Paxos Commit or Raft-backed 2PC — the approach used by CockroachDB, Spanner, and TiDB. It doesn't eliminate 2PC's two-round-trip overhead, but it makes coordinator failure a brief hiccup (Raft election: ~150–300 ms) rather than a permanent block.

Part 4: Three-Phase Commit — Attempting to Eliminate Blocking

Three-phase commit (3PC) was proposed as a non-blocking alternative. It adds a third phase between Prepare and Commit: a PreCommit phase.

The key property 3PC adds: if a participant receives PreCommit, it knows all participants voted Yes. If the coordinator then crashes, a participant that received PreCommit can safely commit on its own — it knows the decision was commit. A participant that hasn't received PreCommit can safely abort — if the coordinator crashed before PreCommit, some participant must have voted No (or the coordinator would have sent PreCommit before crashing).

3PC is non-blocking under crash failures. It sounds like a clear win. Why isn't it used in production?

Network partitions break 3PC's safety. 3PC's non-blocking guarantee assumes crash failures only — nodes either follow the protocol or stop. Under a network partition:

In a world with network partitions (the real world), 3PC can produce split decisions — some participants commit while others abort — which is exactly the data corruption we were trying to prevent. 3PC is safe only in a system with no network partitions, which is not a realistic assumption.

The practical conclusion: 3PC is not used in production. The right solution to 2PC blocking is making the coordinator highly available via Raft (crash-fault-tolerant) rather than trying to eliminate the coordinator.

Part 5: MVCC and Timestamps — The Foundation for Modern Distributed Transactions

Modern high-performance distributed databases (Spanner, CockroachDB, TiDB) don't use classic 2PC with locks held during network round trips. They use Multi-Version Concurrency Control (MVCC) — storing multiple timestamped versions of each row — combined with distributed timestamps to implement optimistic concurrency.

MVCC on a single machine (Postgres uses this — Chapter 2 touched on xmin/xmax): instead of overwriting a row, the database writes a new version with the current transaction ID. Older versions coexist for concurrent readers. A reader at snapshot timestamp T sees the latest version of each row committed before T.

The key insight: with MVCC, readers never block writers and writers never block readers. Only write-write conflicts require locks. This eliminates most lock contention, improving throughput dramatically for read-heavy workloads.

For distributed transactions, the challenge is assigning timestamps that are globally consistent across shards — a timestamp on Shard 1 must be comparable to a timestamp on Shard 7 in a way that preserves the happens-before relationship. If Shard 1 commits at timestamp 100 and Shard 7 then commits a conflicting write at timestamp 99 (because Shard 7's clock is slightly behind), MVCC's ordering is violated.

Two approaches to globally consistent timestamps:

Centralised timestamp oracle: A single service that hands out monotonically increasing timestamps. Every transaction begins by fetching a timestamp from the oracle, and commits by fetching another. Google's Spanner uses TrueTime (atomic clocks + GPS), which bounds clock uncertainty to ~7ms, allowing each node to generate timestamps locally within a bounded window. CockroachDB uses a hybrid logical clock (HLC): each node's clock is adjusted whenever it receives a message with a higher timestamp, maintaining causality without atomic hardware.

Centralized authority (Percolator model): Google's Bigtable-based distributed transaction system, which influenced TiKV/TiDB and CockroachDB's transaction model. Worth understanding in detail.

Part 6: Percolator — Optimistic Distributed Transactions Without a Coordinator

Google's Percolator (2010) is a distributed transaction system built on top of Bigtable (a distributed key-value store). It's the conceptual ancestor of TiDB's transaction model and influenced CockroachDB. Percolator's insight: store lock information in the same rows as data, and use a two-phase commit without a persistent coordinator.

The key data model

Every row stores three columns alongside its value:

• data: the actual value, with a timestamp key: (key, timestamp) → value
• lock: the transaction's lock record if this row is locked: (key, any_ts) → lock_info
• write: a pointer from commit timestamp to the data version: (key, commit_ts) → data_ts

The transaction lifecycle

Begin: The transaction fetches a start timestamp t_start from the timestamp oracle. All reads use this timestamp — the transaction sees a consistent snapshot of all data committed before t_start.

Read: To read key K, check the write column at timestamps ≤ t_start to find the latest committed version. Return that version's data. Also check the lock column — if K has a lock from another transaction, the current transaction must wait or choose to abort the conflicting transaction (if it's stale).

Write (buffered): All writes are buffered in the client. Nothing is written to the store yet. This is optimistic — the transaction assumes it won't conflict and defers validation to commit time.

Commit — Phase 1 (Prewrite): The transaction picks one row as the primary lock (arbitrary choice). For every row it wants to write:

• Check for conflicts: is there a lock from another transaction? Is there a newer committed version than t_start (write-write conflict)?
• If no conflicts: write the new data at t_start and write a lock record pointing to the primary lock

Prewrite for "debit Alice $100, credit Bob $100":

Primary: Alice's row
Write: alice.data[t_start=100] = balance - 100
Write: alice.lock[100] = {primary: "alice", txn_id: T1}

Write: bob.data[t_start=100] = balance + 100
Write: bob.lock[100] = {primary: "alice", txn_id: T1}

Commit — Phase 2 (Commit point): Fetch commit timestamp t_commit from the oracle. Write to the primary lock's write column atomically (this single write is the commit point — either it happens or it doesn't):

Commit point (single atomic write):
alice.write[t_commit=105] = {data_ts: 100}  ← this single write = transaction committed
Delete alice.lock[100]                        ← release primary lock

Then, asynchronously, clean up secondary locks:

bob.write[t_commit=105] = {data_ts: 100}    ← secondary cleanup (async)
Delete bob.lock[100]                          ← release secondary lock

The magic: the transaction is considered committed the moment the primary write column is written. If the process crashes after the primary write but before cleaning up secondary locks, any reader that encounters the stale secondary lock can check the primary lock: if the primary's write column shows a commit, the reader knows the transaction committed and can clean up the secondary lock itself (and read the new value). If the primary's write column shows no commit, the transaction aborted and any lock can be discarded.

Why this is better than 2PC with a coordinator

In classic 2PC, the coordinator is a single point of failure. In Percolator, the coordinator is the client process, and the "commit decision" is a single atomic write to one row (the primary lock). If the client crashes:

• If the primary commit write happened: any node can reconstruct the committed state by reading the primary and cleaning up secondaries. No blocking.
• If the primary commit write did not happen: any node can abort the transaction by cleaning up the pre-write locks. No blocking.

The persistent coordinator state that causes 2PC blocking simply doesn't exist. The commit decision is stored in the data itself, not in an external coordinator. Any reader can determine the transaction's fate by inspecting the primary lock.

The cost: two round trips to the timestamp oracle (begin + commit), plus the prewrite round trip to all participants, plus the commit write. Higher latency than 2PC for small transactions. The benefit: no blocking on coordinator failure, and optimistic execution means no lock contention during the transaction body.

Part 7: Sagas — Transactions Without Locking

Distributed transactions via 2PC or Percolator work for short-lived operations (milliseconds to seconds). They fail for long-running business processes — workflows that take minutes, hours, or days, like:

• An e-commerce order: reserve inventory → charge payment → ship → update loyalty points
• A loan application: credit check → income verification → appraisal → approval → funding
• A hotel booking: check availability → reserve room → charge card → send confirmation

These processes span multiple services, multiple databases, and potentially external third-party APIs. You cannot hold database locks across a 10-minute credit check. The transaction would hold locks on inventory records for 10 minutes, blocking everyone else from purchasing those items. This is why 2PC is unsuitable for long-running workflows.

The Saga pattern (Hector Garcia-Molina, 1987) is the alternative: break the long transaction into a sequence of local transactions, each of which commits immediately. If a step fails, execute compensating transactions — specific rollback logic for each completed step.

Each step is a normal local transaction. Each compensating step is also a normal local transaction. No distributed locks held across steps. No coordinator blocking.

The crucial difference from ACID atomicity

A saga is not atomic in the ACID sense. Between Step 1 committing and the compensating transaction running, other processes can observe the intermediate state — the inventory is reserved, the payment is charged, but no shipment is scheduled. During this window, the saga is in a partially-applied state visible to the outside world.

For some business processes, this is acceptable. A payment hold that's refunded within seconds is operationally fine. For others, it's not — a banking transfer that temporarily shows the money deducted from Alice before appearing in Bob's account would be visible to both Alice and Bob's concurrent reads, which is a problem for financial reporting.

The formal name for what sagas provide: ACD (Atomicity - eventual, Consistency - of each step, Durability - per step). They lack Isolation. Sagas are eventually atomic — the final state is as if the whole saga succeeded or none of it did, but the path to that state has observable intermediate states.

Choreography vs Orchestration

Two implementation styles for sagas:

Orchestration: A central coordinator process (the "orchestrator") knows the full saga workflow and tells each service what to do. The orchestrator persists its state (which steps are complete, which are compensating) to a database. If it crashes, it recovers from its persisted state and resumes.

The orchestrator is a single place to look for saga state. Easy to monitor, debug, and audit. The risk: the orchestrator is a centralised point of coupling — each new step type requires updating the orchestrator. Also, the orchestrator must be highly available.

Choreography: Each service listens to events and takes action, publishing its own events in response. No central coordinator. Services are loosely coupled through an event bus (Kafka, Chapter 15).

No central coordinator. Each service is autonomous. Easy to add new steps (just add a new listener). The risk: the overall workflow logic is distributed across services, making it hard to trace the state of a saga. A saga stuck in the middle (e.g., InventoryReserved published but PaymentCharged never comes) requires inspecting multiple services' logs to diagnose.

In practice: orchestration is better for complex workflows with many failure modes and regulatory audit requirements (financial transactions, healthcare). Choreography is better for simple, well-understood workflows in microservice architectures where loose coupling matters more than central oversight.

Compensating transactions must be idempotent

A compensating transaction might be executed more than once — if the orchestrator crashes between publishing the compensation command and receiving acknowledgement, it will retry on recovery. The compensation must be idempotent: running it twice must produce the same result as running it once.

"Refund payment for order #123" is idempotent if the payment service checks whether order #123 has already been refunded before processing. "Subtract $99.99 from account" is not idempotent — running it twice subtracts $199.98. Every compensating action in a saga must be designed with idempotency in mind.

Part 8: Distributed Deadlocks

On a single machine, deadlock detection is straightforward: the database maintains a wait-for graph of which transaction is waiting for which lock, and periodically checks for cycles. A cycle means deadlock. One transaction is chosen as victim and aborted.

Across shards, each shard has a local wait-for graph, but the cycle might span multiple shards:

Each shard sees no cycle locally. Global wait-for graph: A → B → A ← cycle! Deadlock.

Three approaches to distributed deadlock:

Timeout-based deadlock prevention: Every transaction has a maximum duration. If it doesn't commit within that time, it's aborted and retried. Simple, no distributed graph needed. The cost: legitimate long-running transactions are falsely killed. Timeout must be tuned longer than the worst-case legitimate transaction, but short enough to recover from actual deadlocks promptly. Used by most practical systems because it's simple.

Wait-Die and Wound-Wait (priority-based prevention): Assign each transaction a priority based on its timestamp. Older transactions have higher priority (lower timestamp = higher priority).

• Wait-Die: A low-priority transaction waiting for a lock held by a high-priority transaction aborts itself (dies) rather than waiting. A high-priority transaction can wait for a lower-priority one.
• Wound-Wait: A high-priority transaction needing a lock held by a low-priority transaction forces the low-priority one to abort (wounds it). A low-priority transaction can wait for a high-priority one.

Both prevent deadlocks by ensuring the wait-for graph has no cycles — priority ordering means the graph is always a DAG. The cost: unnecessary aborts when there is no actual deadlock.

Distributed deadlock detection: A background process periodically collects wait-for graphs from all shards and assembles the global graph, checking for cycles. When a cycle is found, one transaction is aborted. Correct but complex: the graph collection itself requires coordination, and by the time the global graph is assembled, the deadlock may have already resolved (one transaction timed out) or new ones may have formed.

CockroachDB uses a combination of timeout-based detection and a distributed deadlock detection algorithm. Spanner uses wound-wait for lock management — it aborts the lower-priority (younger) transaction when there's a conflict, preventing deadlocks from forming.

Part 9: Serializable Snapshot Isolation in Distributed Context

Chapter 2 mentioned isolation levels — the spectrum from Read Uncommitted to Serializable. Most production databases run at Read Committed (Postgres default) or Repeatable Read. Full Serializability (every transaction appears to execute sequentially, with no concurrency anomalies) is expensive to achieve at scale.

Snapshot Isolation (SI) — used by Postgres with MVCC — is almost serializable but allows one anomaly: write skew. Write skew occurs when two transactions each read overlapping data and each write non-overlapping data, but their combined effect violates a constraint:

Serializable Snapshot Isolation (SSI) — implemented in Postgres 9.1+ and CockroachDB — detects write skew by tracking read-write dependencies between transactions and aborting transactions that would produce a serialization cycle.

In a distributed context, SSI requires tracking these dependencies across shards. CockroachDB implements distributed SSI by maintaining a timestamp cache per range (shard) — a record of the highest timestamp at which each key was read. When a write arrives, it checks the timestamp cache: if any read at a timestamp newer than the write's start timestamp touched the same key, a potential serialization conflict exists. The write is either rejected (forcing a retry) or proceeds if the dependency doesn't form a cycle.

This is expensive — every read and write must update and check the timestamp cache — but it provides full Serializability without holding locks for the duration of a transaction. This is how CockroachDB achieves serializable isolation at distributed scale.

Part 10: Google Spanner — Transactions with TrueTime

Spanner is Google's globally-distributed relational database, serving Google Ads, Google Play, and other critical infrastructure. It provides external consistency — a stronger property than linearizability — across a planet-scale distributed database. Understanding how it achieves this with 2PC is the clearest example of production distributed transaction design.

Spanner uses 2PC with Raft-backed participant groups (each shard is a Raft group, like CockroachDB). The 2PC coordinator is also a Raft group. Coordinator failure causes a Raft election (~150ms) rather than indefinite blocking. This is the state of the art for 2PC in practice.

The novel mechanism: TrueTime. Spanner's datacenters use GPS receivers and atomic clocks to bound the uncertainty in wall clock time. TrueTime API returns an interval [earliest, latest] such that the true current time is guaranteed to be within [earliest, latest]. The uncertainty is typically 7ms or less.

Why does this matter for transactions? With bounded clock uncertainty, Spanner can assign globally ordered commit timestamps without a centralised timestamp oracle. When committing, Spanner waits until the latest time has definitely passed (a "commit wait" of ~7ms) before revealing the committed data to readers. This ensures that any transaction that started after this one will get a start timestamp strictly greater than the commit timestamp — guaranteeing external consistency.

The commit wait (~7ms for GPS-synchronized clocks) is the price of globally consistent timestamps without a centralised oracle. For Spanner's use case — global transactions across continents — 7ms is negligible compared to the 50–200ms inter-datacenter network latency.

CockroachDB achieves similar guarantees without atomic hardware using hybrid logical clocks (HLC) and a more conservative uncertainty window — reads may need to "wait out" clock uncertainty when they encounter uncertain timestamps. The result is similar external consistency guarantees, implemented in software.

Part 11: Choosing the Right Approach — Decision Framework

The choice between distributed transaction approaches maps to your transaction characteristics:

Use 2PC with Raft-backed coordinator when: transactions are short-lived (< 1 second), involve 2–10 shards, and you need strong ACID guarantees. This is what CockroachDB, TiDB, and Spanner use for standard SQL transactions. The key: make the coordinator fault-tolerant via Raft so blocking is a brief election delay, not an indefinite halt.

Use Percolator/MVCC-based optimistic transactions when: you have high read-to-write ratio (optimistic concurrency is efficient when conflicts are rare), transactions are short-lived, and you're building on top of a KV store that supports conditional writes (compare-and-swap). TiKV's transaction model is directly based on Percolator.

Use Sagas when: the operation spans external systems (payment gateways, shipping APIs), takes more than a few seconds, or must maintain loose coupling between services. Sagas are the right pattern for microservice workflows where each service owns its own database. The concession: visible intermediate states.

Use local transactions + careful schema design (avoid distributed transactions entirely) when: possible. Co-locating related data on the same shard (Chapter 7's co-location) turns distributed transactions into local transactions. The most powerful optimisation for distributed transaction performance is designing your data model so that the atomic unit of work fits on one shard.

Vocabulary Map

Practice Problems

Problem 1 — Failure Analysis

Trace the exact blocking scenario in 2PC:

A coordinator manages a transfer: debit Shard 1 (Alice -$100), credit Shard 7 (Bob +$100). Both shards vote Yes and write PREPARED to their WALs. The coordinator writes COMMIT to its own log and then immediately crashes (hardware failure — it won't restart for 4 hours).

What is the state of Shard 1? What is the state of Shard 7? What locks are held? What is the effect on other users trying to read Alice's or Bob's balance? Can the application serve Alice's balance query? At what point can the shards safely abort the prepared transaction on their own?

Now re-run the scenario with a Raft-backed coordinator (3 replicas, current leader crashes): what changes? What is the time to recovery? What does the new Raft leader do when it sees the COMMIT record in the coordinator's log? What if the coordinator crashes before writing the COMMIT record — what do the participants do when the new coordinator leader contacts them?

Problem 2 — Design from Scratch

Design the transaction protocol for an e-commerce checkout. Operations: (a) decrement inventory for each cart item (items may be on different inventory shards), (b) create the order record, (c) charge the customer's payment method (external Stripe API call), (d) send confirmation email.

The business requirements: inventory must never go negative, the customer must never be charged if the order fails, the confirmation email should be sent at most once. Operations (a) and (b) are in your database. Operations (c) and (d) are external APIs.

Identify which operations can use 2PC, which require Sagas, and why. Design the full saga: list each step, its compensating transaction, and why the compensation is or isn't idempotent. What is the customer-visible intermediate state if the system crashes between each step? Which crash points are acceptable? Which require special handling?

Problem 3 — Synthesis

Connect distributed transactions to Chapter 6's consistency models and Chapter 12's Raft.

CockroachDB uses Raft for per-range replication and a Percolator-style transaction protocol for cross-range atomicity. A transaction T1 writes to range R1 (Raft group 1) and range R7 (Raft group 7). T1's coordinator is in Raft group 1.

Trace the full commit sequence: how many total network round trips are required from client to committed? (Count: start timestamp fetch, prewrite to R1 and R7 in parallel, commit point write to R1, secondary cleanup on R7.) At each step, which Raft group's leader is being written to? If R7's Raft leader is in a different datacenter (100ms round trip) and R1's Raft leader is local (0.5ms), what is the minimum end-to-end latency for this transaction?

Now: a read-only transaction T2 starts after T1 commits. T2 reads a key in R1 and a key in R7. Using CockroachDB's MVCC model, what timestamp does T2 read at? Is it guaranteed to see T1's writes? Trace through the write column to show why the read is consistent, even though T2's read of R7 might arrive before T1's secondary cleanup has run on R7.

Problem 4 — Go Deeper

The Saga pattern says intermediate states are visible. This raises a question: what consistency guarantees do sagas actually provide?

Define formally: after a successful saga completion (all steps committed), is the final state serializable with respect to concurrent transactions? (Hint: each step is a local transaction, so it's locally serializable. Does local serializability of steps compose into global serializability?) What anomalies can a concurrent transaction observe during a saga's execution that it would never observe during a successful 2PC transaction?

Now consider a specific anomaly: a saga does Step 1 (reserve inventory), then Step 2 (charge payment). A concurrent read transaction R1 observes the state between Step 1 and Step 2: inventory is reduced but no payment has been taken. This "phantom reservation" state is inconsistent with any complete saga execution (either inventory and payment are both changed, or neither). Under what consistency model (Chapter 6 vocabulary) is this observation possible? Under what model would it be prohibited? What application-level design prevents users from seeing phantom reservations even though the database allows it?

Problem 5 — Connect to Previous Chapters

The chapter described two solutions to coordinator failure in 2PC: Raft-backed coordinator (CockroachDB, Spanner) and Percolator (no persistent coordinator). Compare them across the dimensions built across previous chapters.

Using Chapter 4's failure taxonomy: which failure modes does each approach handle? Which does it not? Using Chapter 6's consistency vocabulary: what consistency level does each provide in the normal case? In the degraded case (coordinator/primary crashed, Raft election in progress)? Using Chapter 11's safety/liveness framing: during the Raft election window (~150ms) for the Raft-backed coordinator, is the distributed transaction system providing safety or liveness? What happens to in-flight transactions during this window?

For Percolator: after the client crashes mid-transaction (after prewrite, before commit), the system is in a state with orphaned locks. How long before another transaction can force-abort these locks? What determines the timeout? How does this connect to Chapter 4's timeout-based failure detection?

What's Coming Next

You now understand how distributed systems achieve atomicity across machines — from 2PC's simplicity and blocking problem, through Percolator's lock-in-data elegance, to Saga's acceptance that some workflows cannot be atomic in the ACID sense. The next chapter shifts from coordination to communication: message queues. So far, every interaction in this curriculum has been synchronous — a client sends a request and waits for a response. Message queues break this synchrony: a producer writes a message and returns immediately; a consumer reads and processes it later. This decoupling enables patterns that synchronous RPC cannot: buffering traffic spikes, guaranteed delivery across service restarts, fan-out to multiple consumers, and replay of past messages. Chapter 14 builds message queues from first principles. Chapter 15 applies those principles to Kafka — the dominant distributed log used for real-time data pipelines at scale.

Chapter 13 complete. Problem 2 (e-commerce checkout transaction design) is the most interview-relevant — it's asked verbatim at FAANG system design rounds. The key insight that every candidate misses: you cannot use 2PC for the Stripe API call (external, not transactional), so you must use a Saga. Identifying which boundary forces the switch from 2PC to Saga is the skill being tested. Problem 5 (comparing approaches across prior chapter dimensions) is the synthesis exercise that locks in the cross-chapter connections. Work through it on paper before moving to Chapter 14.