Distributed Transactions: 2PC, Saga, and When to Avoid Both

TL;DR: When a single business action touches multiple databases or services, you need atomicity across nodes. Two-phase commit (2PC) delivers it but blocks during coordinator failures. Percolator and Spanner layer 2PC on consensus-replicated state to eliminate the blocking problem, at the cost of cross-region latency (~100 ms for Spanner read-write transactions^[1]). Sagas break the big transaction into local ACID steps with compensating actions and are the default for distributed business workflows. The outbox pattern solves the "atomically write DB + publish event" problem without XA. Most distributed transaction problems dissolve under idempotency + eventual consistency + user-visible reconciliation. If you take one thing from this chapter: reach for the outbox and sagas first, 2PC last.

Learning Objectives#

After this module, you will be able to:

• Walk through 2PC and identify its blocking failure mode
• Describe Percolator and how it gets snapshot isolation across shards
• Explain how Spanner and CockroachDB make 2PC non-blocking via consensus
• Design a saga with compensating actions for each step
• Implement the transactional outbox pattern for atomic state + event publish
• Recognize when you actually need a distributed transaction vs eventual consistency

Intuition#

You are closing on a house. The buyer, seller, bank, and title company each have documents to sign. Nobody wants to sign first because if the other parties back out, they are stuck. So everyone signs their documents and hands them to an escrow agent. The agent holds all signatures. Only when every party has signed does the agent release everything simultaneously. If any party refuses, the agent returns all documents and the deal is off.

That is two-phase commit. The escrow agent is the coordinator. "Signing and handing over" is the prepare phase. "Releasing simultaneously" is the commit. The problem: if the escrow agent has a heart attack after collecting all signatures but before releasing them, everyone is stuck holding nothing, unable to proceed or back out.

Now consider a different scenario. You and three friends split a restaurant check. Each person pays their own card independently. If one card declines, that person pays cash or Venmos later. There is no single coordinator holding everything. Each payment is a local transaction. If something fails, you compensate (refund, retry, settle up later). That is a saga: a sequence of independent local transactions with compensating actions on failure.

The rest of this chapter makes both patterns precise, shows you the modern variants, and argues that most of the time you want the restaurant approach, not the escrow approach.

Theory#

The problem#

A distributed transaction requires that a single logical operation touching N independent resource managers commits everywhere or nowhere, despite concurrent access, partial failures, and an asynchronous network^[2].

The canonical example: transfer $100 from account A on shard 1 to account B on shard 2. The intermediate state where money has left A but has not arrived at B must never be durable and never observed. Preserving atomicity across nodes is hard because a participant that has durably prepared its change has no safe local action if the coordinator disappears^[2:1][3].

Two-phase commit#

The protocol has two phases^[2:2]:

1. Prepare. The coordinator sends PREPARE to all participants. Each participant durably writes its changes to a log, acquires locks, and votes YES (commit-ready) or NO.
2. Commit/Abort. If all participants vote YES, the coordinator logs COMMIT and broadcasts it. Otherwise it broadcasts ABORT. Participants act on the decision and release locks.

The XA specification from The Open Group defines the standard interface (xa_start, xa_prepare, xa_commit, xa_rollback) between a transaction manager and resource managers^[4]. All major RDBMSs and message brokers implement it.

The blocking problem. After voting YES, a participant has surrendered its right to abort unilaterally. If the coordinator crashes between collecting all YES votes and writing COMMIT, every participant is stuck: locks held, no authority to proceed or roll back^[2:3][3:1]. Three-phase commit (Skeen) adds an extra round to avoid this, but it assumes bounded message delay and is unsafe under partitions. Nobody runs 3PC in production.

The coordinator crashes after collecting PREPARE-yes votes, leaving participants prepared and holding locks indefinitely.

The modern fix is Gray and Lamport's Paxos Commit (tech report 2004; ACM TODS 2006): replicate the coordinator's state via consensus so any majority can recover the outcome^[3:2]. This is exactly what Spanner and CockroachDB do.

Percolator: snapshot isolation across Bigtable#

Google's Percolator (OSDI 2010) layers ACID snapshot-isolation transactions on top of Bigtable using a per-cell lock and a centralized timestamp oracle^[5].

Each logical column is stored as three Bigtable columns: data (timestamped value), lock (uncommitted lock), and write (committed write pointer). A transaction gets a start_ts from the oracle, prewrites all cells (writing data and acquiring locks), designates one lock as the primary, then gets a commit_ts and atomically replaces the primary lock with a write pointer. That single atomic write on the primary row is the commit point^[5:1][6].

Secondary locks are resolved lazily: any later reader that encounters a stale lock checks the primary. If the primary was committed, roll forward. If not, roll back. This means no central transaction manager is needed for recovery.

The atomic transition from "primary lock" to "primary write" is the single commit point; secondaries are resolved lazily by later readers.

The timestamp oracle serves ~2 million timestamps/sec from a single machine via batching^[5:2]. Percolator cut average document age in Google's search index by 50% versus the MapReduce system it replaced, with median doc-to-index latency improving ~100x^[5:3][7].

Trade-off: Percolator deliberately accepts tens-of-seconds tail latency for lock cleanup. Acceptable for web indexing, unacceptable for OLTP. It provides snapshot isolation, not serializability, so write skew is possible^[5:4][6:1].

Spanner and CockroachDB: 2PC over consensus groups#

Google Spanner (OSDI 2012) runs 2PC across Paxos-replicated tablet groups^[1:1]. Each tablet is a Paxos group spanning multiple data centers. A read-write transaction acquires locks at each involved tablet leader, picks one group as the 2PC coordinator, and runs the protocol where every PREPARE and DECISION are Paxos-replicated log entries. The coordinator cannot block because it is itself a consensus group^[1:2][3:3].

Spanner uses the TrueTime API (GPS receivers + atomic clocks per data center) to assign globally meaningful commit timestamps. Cross-USA read-write transactions take ~100 ms. Read-only transactions skip locks and 2PC entirely, reading from a local replica, and are ~10x faster^[1:3][8].

CockroachDB Parallel Commits (19.2, November 2019) collapses two sequential Raft rounds into one^[9]. The key insight: a transaction is implicitly committed when its transaction record is in STAGING state and all its in-flight intent writes have Raft-committed. The coordinator pipelines the STAGING record write in parallel with intent writes, so the client sees acknowledgment after one round of consensus instead of two. TPC-C benchmarks show transaction latency scales at 1x inter-node RTT with Parallel Commits versus 2x RTT without^[9:1]. The protocol was formally verified in TLA+^[9:2].

TiDB inherits the Percolator protocol directly, running it over TiKV (a Raft-replicated key-value store). TiDB Cloud targets < 105 ms P95 transaction latency on Sysbench workloads^[10][6:2].

Sagas: compensating actions instead of distributed locks#

A saga (Garcia-Molina and Salem, SIGMOD 1987) models a long-lived transaction as a sequence of local ACID transactions T1..Tn, each paired with a compensating transaction Ci that semantically undoes Ti^[11][12]. If step k fails, the orchestrator runs C(k-1)..C1 in reverse.

Two coordination styles exist^[13]:

• Orchestrated saga. A central orchestrator (state machine) sends commands to each service in order, waits for replies, and drives compensation on failure. Implementations: Temporal (9.1 trillion lifetime actions, 150,000+ actions/sec peak^[14]), AWS Step Functions (Express Workflows advertise up to 100,000 state transitions/sec^[15]; current service quotas list the Express StateTransition bucket as Unlimited^[16]), Netflix Conductor (reported >2.6M process flows in its first year^[17]).
• Choreographed saga. Each service subscribes to events from others, runs its local transaction on the trigger, and emits its own event. No central flow owner. Works for 2-3 step flows; becomes unobservable beyond that.

Sagas lack the "I" in ACID. This is the single most important thing to understand. Between T1 and T2, another concurrent saga can observe the intermediate state. Every saga bug you will hit in production is an isolation bug^[13:1]. Countermeasures are domain-specific: semantic locks (reservation rows), commutative updates (deltas instead of read-then-write), pessimistic views (filter "pending" states from user queries).

Step T3 fails; the orchestrator runs C2 then C1 in reverse to restore semantic consistency.

The outbox pattern#

The problem: your service writes to its database and then publishes an event to Kafka. If the process crashes between the DB commit and the Kafka publish, the event is lost. If you publish first and the DB write fails, you have a phantom event.

The solution: write the event to an outbox table in the same local ACID transaction as the business write. A separate relay process delivers outbox rows to the broker^[18][19].

Two relay styles:

• Polling publisher: periodically SELECT unsent rows and publish. Simple but adds latency.
• CDC (change data capture): tail the database's write-ahead log with Debezium or Maxwell. INSERTs into the outbox table become broker messages automatically. Lower latency, preserves per-aggregate ordering^[19:1].

One local ACID transaction writes both the business row and the outbox row; Debezium tails the WAL and publishes to Kafka, giving at-least-once event delivery without XA.

Because CDC delivery is at-least-once, consumers must be idempotent. Track processed event IDs in a local table (INSERT ... ON CONFLICT DO NOTHING). This is covered in depth in Idempotency and Exactly-Once.

The outbox pattern is not glamorous. It is the honest answer to "atomically write DB + publish event." If you take one thing from this chapter, it should be this pattern.

When you do not need distributed transactions#

Pat Helland's "Life Beyond Distributed Transactions" (CIDR 2007) argues that systems designed for nearly unbounded scale should reject distributed transactions entirely^[20][21]. ACID lives inside one entity (one user, one order, one account). All cross-entity interaction is messaging-based, idempotent, and eventually consistent.

Practical restatements:

• Money transfer is a ledger append pattern. Each account is its own entity. The transfer is two append-only ledger entries linked by a transfer ID, plus a reconciliation job that verifies conservation of money^[20:1].
• Inventory reservation is a saga: reserve, charge, fulfill. If charging fails, release the reservation.
• "Book flight + hotel + car" is a saga with explicit cancellation flows. No 2PC across three travel providers exists or could exist.
• Analytics pipelines use idempotent writes keyed by event ID plus eventual consistency.

The design cost is real: idempotency keys, compensations, reconciliation, and user-visible recovery flows are all extra engineering. But the alternative (distributed locks across services) is worse at scale because the failure of any participant stalls commit, and larger systems are more likely to be blocked at any moment^[21:1].

Real-World Example#

Temporal: durable execution at trillion-action scale#

Temporal is a durable-execution orchestrator that originated as Uber's Cadence (2016) and was forked by its original founders in 2019^[22]. As of February 2026, Temporal Cloud has processed 9.1 trillion lifetime action executions at a peak throughput exceeding 150,000 actions/sec^[14:1]. Production users include OpenAI, ADP, Yum! Brands, and Block.

How it works. A Workflow is durable-executed code (Go, Java, TypeScript, Python). Every invocation of a non-deterministic operation (activity call, timer, signal) is logged to a durable event history. If a worker crashes, a replacement replays the event history to reconstruct local state and resumes from the last recorded decision^[22:1]. Activities are the steps that call external systems (DB writes, RPCs, payments); they must be idempotent because Temporal retries them at-least-once on timeout.

Saga orchestration in Temporal is natural: the workflow code is a for-loop over steps with a try/catch that runs compensations in reverse on failure. Unlike JSON DSL orchestrators (Step Functions, Conductor), branches, loops, and sleeps are expressed in the host language. The trade-off: Step Functions gives you a visual state machine and 1-year execution durability with a 5-minute Express mode for high-throughput; Temporal gives you code-level expressiveness and unbounded execution duration but requires you to run (or pay for) the Temporal server infrastructure.

The non-determinism trap. Temporal requires workflows to be deterministic modulo the SDK's workflow.Now() and activity calls. If workflow code reads time.Now() or a random number directly, replay diverges and the workflow fails with a NonDeterministicError^[22:2]. This is the most common production pitfall for teams adopting Temporal.

Trade-offs#

The rows below split into two groups: coordination mechanisms (Percolator/Spanner-style 2PC over consensus, orchestrated sagas, choreographed sagas) are substitutable choices for a single business action. Outbox pattern and Avoid the problem are complementary: outbox runs alongside any saga that writes DB + publishes events, and "avoid the problem" is the meta-question you should answer before picking anything else.

Common Pitfalls#

Exercise#

Design the checkout flow for an e-commerce site: reserve inventory (inventory service), authorize payment (payment service), create order (orders service), emit shipping event. Pick orchestrated vs choreographed saga, define compensations for each step, and describe what happens when the payment authorization times out after inventory is reserved.

Key Takeaways#

• 2PC works and is correct, but it is blocking and slow. Reach for it only when no saga is acceptable and both participants are in one data center.
• Percolator-style 2PC layered on a consensus-replicated store is how global databases (Spanner, CockroachDB, TiDB) achieve cross-shard atomicity without blocking.
• CockroachDB's Parallel Commits halve commit latency (1x RTT vs 2x RTT) by pipelining the transaction record write with intent writes^[9:3].
• Sagas are the default for distributed business workflows. Orchestrators like Temporal handle retries, timeouts, and visibility.
• Sagas lack isolation. Every saga bug you will hit in production is an isolation bug. Design countermeasures per domain.
• The outbox pattern is not glamorous but it is the honest answer to "atomically write DB + publish event." Use CDC (Debezium) for low-latency relay.
• Most distributed transaction problems dissolve under idempotency + eventual consistency + user-visible reconciliation. Start by asking "do I actually need atomicity across these services?" The answer is usually no.

Further Reading#

• Consensus on Transaction Commit (Gray and Lamport, 2004): reframes 2PC as a consensus problem and introduces Paxos Commit as a non-blocking coordinator design; the theoretical foundation for Spanner's approach.
• Percolator paper (Peng and Dabek, OSDI 2010): canonical source for snapshot-isolation 2PC over Bigtable; read alongside the TiKV deep-dive for a modern implementation.
• Spanner: Google's Globally-Distributed Database (OSDI 2012): 2PC over Paxos groups with TrueTime; the paper that proved externally consistent distributed transactions were possible at global scale.
• Parallel Commits (CockroachDB, 2019): 1x RTT commit via STAGING transaction record with TLA+ verification; essential reading for understanding modern distributed database commit protocols.
• Life Beyond Distributed Transactions (Helland, CIDR 2007): the philosophical argument against distributed transactions at scale; entity boundaries, messaging, and idempotency as the alternative.
• Pattern: Saga (Chris Richardson): orchestration vs choreography, isolation countermeasures, and practical implementation guidance for microservices.
• Reliable Microservices Data Exchange With the Outbox Pattern (Debezium): reference implementation of the outbox pattern with Debezium CDC and Kafka; includes the Outbox Event Router SMT configuration.
• DDIA Chapters 7 and 9 (Kleppmann): the most readable textbook treatment of distributed transactions, consensus, and exactly-once semantics; pairs directly with this chapter.

Flashcards#

References#