Distributed Transactions
Two-phase and three-phase commit with their failure modes, Calvin, Spanner and TrueTime, partitioning, Percolator, invariant confluence and RAMP
A transaction that touches data on several nodes must still look atomic: every node applies its part or none does, even if a node crashes or a cable is cut halfway through. A local database gets atomicity from the write-ahead log and a single commit record; across machines there is no single disk write that commits everywhere, so a protocol is needed. This chapter covers atomic commitment in its classic form (two-phase commit), its attempted fix (three-phase commit), and four systems that each sidestep the classic cost differently: Calvin agrees on inputs instead of outcomes, Spanner replicates the coordinator and uses synchronised clocks, Percolator moves the coordinator into the client and leans on single-row atomicity, and the coordination-avoidance line of work asks whether coordination was needed at all.
The vocabulary (prepare, in-doubt, blocking, idempotent recovery) is how engineers talk about any step that updates two systems at once: charge the card, then reserve the seat.
In one minute
- Atomic commitment: commit only if every participant votes yes; the decision is the same everywhere and survives crashes.
- 2PC: cohorts prepare (do the work, log it durably, hold locks, vote); on unanimous yes the coordinator logs commit, the commit point, and tells everyone.
- 2PC blocks: a cohort that voted yes holds locks until it learns the decision; if the coordinator dies after collecting votes, cohorts are in doubt until it recovers.
- 3PC adds a pre-commit phase so cohorts can decide by timeout after a coordinator crash, but a network partition can commit one side and abort the other. It is not used in practice.
- Calvin fixes a global transaction order with a sequencer, then executes deterministically on every replica: agreement on inputs, so no 2PC on outcomes.
- Spanner runs 2PC across Paxos groups (the coordinator cannot permanently die) and uses TrueTime intervals plus commit-wait for externally consistent timestamps.
- Partitioning: consistent hashing rings with virtual nodes, or range partitioning with splits.
- Percolator layers snapshot isolation on Bigtable with client-driven 2PC, a timestamp oracle and a primary lock whose single-row commit is the commit point; stale locks are cleaned lazily by readers.
- Coordination avoidance: an I-confluent invariant survives merging independently evolved states, so transactions preserving it need no coordination. RAMP gives atomic visibility across partitions without blocking.
Atomic commitment: two-phase and three-phase commit
Making operations appear atomic
An atomic commitment protocol has to deliver: agreement (all participants decide the same way), validity (commit only if every participant voted yes), stability (a decision is never reversed) and termination (every non-failed participant eventually decides). Commit needs unanimity, which makes this a commitment and not a majority vote. The participants are cohorts; the process driving the protocol is the coordinator, which may be a cohort, a separate service or the client.
Two-phase commit: the protocol
Phase 1, prepare. The coordinator sends PREPARE to every cohort. Each cohort does whatever is needed to guarantee it could commit later no matter what: it writes the changes and a PREPARED record to its log, forces the log to disk, and keeps the transaction's locks. Then it votes YES. If it cannot (constraint violation, deadlock, disk full), it votes NO and may abort locally at once, since one NO fixes the outcome.
Phase 2, commit or abort. If every vote is YES, the coordinator forces a COMMIT record to its own log and only then sends COMMIT. The durable commit record is the commit point: from then on the transaction is committed regardless of further failures. Any NO or timeout leads to ABORT instead. Cohorts apply the decision, release locks and acknowledge; when all acks are in, the coordinator logs END and forgets the transaction.
The cost is two round trips and at least two forced log writes on the critical path, with locks held across both round trips. For cross-region participants that is tens of milliseconds during which no one else can touch those rows.
Cohort failures
- Before voting: the coordinator times out and aborts; on recovery the cohort finds no PREPARED record and undoes partial work.
- After voting NO: nothing to do; it already aborted.
- After voting YES: on recovery it finds PREPARED and is in doubt: it may neither commit (someone may have voted no) nor abort (the coordinator may have told others to commit), so it re-acquires the locks and asks the coordinator. This is why PREPARE logs both redo and undo, and why PostgreSQL's
PREPARE TRANSACTIONholds its locks across a crash untilCOMMIT PREPAREDorROLLBACK PREPARED. - After the decision was sent: the coordinator retries until acknowledged, so COMMIT must be idempotent.
Coordinator failures and blocking
- Before sending PREPARE: cohorts time out and abort.
- After sending PREPARE, before logging a decision: cohorts that voted YES may run a cooperative termination protocol, asking one another. If any peer has the decision, they adopt it; if any peer has not voted or voted NO, they may abort, since commit was impossible. But if every reachable cohort voted YES and none knows the decision, they are blocked: the coordinator may have logged COMMIT and told an unreachable cohort, or logged nothing. Waiting for the coordinator's log is the only safe option.
- After logging the decision, before sending it: on recovery the coordinator resends; cohorts were blocked for the outage.
Blocking is the central objection to 2PC. Blocked cohorts hold locks, so rows become unavailable to every other transaction and the outage spreads. Operators have resolved long in-doubt transactions by hand (heuristic completion), at the risk of inconsistency. Presumed abort lets the coordinator skip logging aborts and treats unknown transactions as aborted, saving log writes but not fixing blocking. The real fix is a highly available coordinator whose decision log is replicated with consensus, which is what Spanner, CockroachDB, YugabyteDB and TiDB do.
Where 2PC appears
X/Open XA standardises 2PC between a transaction manager and resource managers. PostgreSQL offers PREPARE TRANSACTION, MySQL XA PREPARE (and InnoDB runs an internal 2PC between binary log and storage engine), MongoDB uses 2PC across shards, and Kafka's transactional producer uses a 2PC-like protocol with commit markers.
Three-phase commit: the idea
2PC blocks because a cohort that voted YES cannot tell "the coordinator decided commit and told someone else" from "the coordinator decided nothing". 3PC (Skeen, 1981) adds a phase to remove the ambiguity:
- Propose (can-commit): collect votes as before; any NO or timeout aborts.
- Pre-commit: on unanimous YES, the coordinator tells every cohort "everyone voted yes, prepare to commit"; cohorts acknowledge. A cohort holding pre-commit knows all others voted yes.
- Commit: once all cohorts acknowledge pre-commit, the coordinator sends COMMIT.
Timeout rules make it non-blocking under crashes: a cohort that times out waiting for pre-commit aborts, because nobody can have committed without pre-commit having been sent to all. A cohort that times out waiting for COMMIT after receiving pre-commit commits, because everyone voted yes and the coordinator intended to commit. A replacement coordinator polls the cohorts: if any has pre-committed, drive to commit; otherwise abort.
Why partitions still break 3PC
The rules assume that if I have not heard, nobody has. Suppose pre-commit reached cohorts A and B, then the network split them from C and D before it arrived there. A and B time out and commit; C and D time out and abort. Both sides followed the protocol and the outcome diverged. 3PC is non-blocking only in a synchronous network with crash-only faults, and it costs an extra round trip on every transaction, so nobody ships it. The lesson survived: the robust way to remove blocking is to make the decision fault tolerant through consensus, not to add phases.
Distributed transactions with Calvin
Deterministic execution
Traditional systems let each node execute in whatever order local concurrency control yields, then agree on the outcome with 2PC. Calvin (Thomson et al., 2012) inverts this: nodes agree on the order of inputs first, then execute deterministically, so every replica given the same input sequence reaches the same state. Agreement on outcomes becomes unnecessary because the outcome is a function of the agreed input.
Sequencer, scheduler, storage
- Sequencer: clients submit requests to sequencers, which batch them into epochs (10 ms in the paper). Each batch is replicated to the partition's other replicas (Paxos, or asynchronously from a master replica), and all sequencers' batches for an epoch are interleaved deterministically into the global input log. Once a batch is replicated it is durable: it will execute even if every node then crashes.
- Scheduler: acquires locks in log order with a deterministic protocol, so every replica resolves the same conflicts the same way and deadlocks cannot occur. A transaction runs once it holds all its locks; worker threads execute the logic.
- Storage: any key-value or record store.
Why Calvin avoids 2PC
For a multi-partition transaction, each participant executes its part and sends the values it read to the other participants so each can evaluate the full logic. That is one exchange of reads, not a vote: nobody can abort for "my disk filled up" (durability already happened at the sequencer) or "I lost a lock" (locking is deterministic). The only aborts are logic-based and every replica reaches them identically. A failed node replays the input log from its last checkpoint.
The constraints: read and write sets must be known before sequencing, so transactions whose write set depends on a read run a cheap reconnaissance query first and are re-sequenced if it went stale. Interactive transactions do not fit because the log cannot wait for a client's next statement, and every transaction pays at least half an epoch of latency. Fauna is the best-known Calvin-based system.
Distributed transactions with Spanner
Architecture
Spanner (Corbett et al., 2012) splits data into tablets, each replicated across data centres by its own Paxos group with a long-lived leader holding a lock table for two-phase locking and a transaction manager. A transaction within one group commits through that group's Paxos log; one across groups runs 2PC across the leaders, one of which coordinates.
Every participant, including the coordinator, is a Paxos group: PREPARED and COMMIT records are replicated to a majority before being acted upon. If the coordinator leader dies, a new leader of that group continues from the replicated state. Blocking disappears because the coordinator is briefly leaderless, never gone.
| Operation | Locks | Where it runs | Timestamp |
|---|---|---|---|
| Read-write transaction | 2PL with wound-wait on leaders | Leaders of involved groups, 2PC if more than one | Chosen at commit |
| Read-only transaction | None | Any sufficiently up-to-date replica | Assigned at start from TrueTime |
| Snapshot read | None | Any sufficiently up-to-date replica | Client-chosen or bounded staleness |
Lock-free reads never block or are blocked by writers and can be served nearby, which is where Spanner's read scalability comes from. "Sufficiently up to date" is checked against the replica's safe time, the maximum timestamp at which it has applied all writes, bounded by Paxos progress and by prepared-but-uncommitted transactions.
TrueTime
Timestamps must respect real-time order across continents: if T1 commits before T2 starts, T1's timestamp must be smaller. Ordinary clocks cannot promise that. TrueTime, backed by GPS and atomic clocks in each data centre, returns from TT.now() an interval [earliest, latest] guaranteed to contain true time; its width, epsilon, is a few milliseconds (7 ms average in the paper).
Commit-wait and external consistency
The leader picks a commit timestamp s at least TT.now().latest and greater than anything the group has assigned, then commit-waits: it releases no locks and reports no success until TT.after(s) holds, so true time is certainly past s. By the time any client can learn of the commit, every TrueTime interval everywhere lies after s, and any later-starting transaction picks a larger timestamp. That is external consistency, Spanner's term for strict serializability. In 2PC, each participant leader proposes a prepare timestamp and logs PREPARED through Paxos; the coordinator chooses s as the maximum of those and its own TT.now().latest, logs COMMIT through Paxos, commit-waits, then sends the outcome. The wait mostly overlaps Paxos replication, adding roughly epsilon in the worst case.
The lesson: a timestamp with known uncertainty turns a real-time ordering requirement into a wait. CockroachDB, without special hardware, uses hybrid logical clocks with a configured maximum offset and restarts reads that fall inside the uncertainty window instead.
Database partitioning
Routing and schemes
Replication copies the dataset; partitioning (sharding) splits it so each node holds a subset, which is how a database grows beyond one machine. Each partition is then replicated. Every request carries a routing key, and all nodes must agree cheaply on the mapping from keys to partitions and partitions to nodes.
| Scheme | Mapping | Strengths | Weaknesses |
|---|---|---|---|
| Range partitioning (Bigtable, HBase, Spanner, CockroachDB, TiKV) | Contiguous key ranges, split as they grow | Local range scans; adaptive splits | Hot ranges from monotonic keys; needs a range directory |
| Hash partitioning (Cassandra, DynamoDB, Riak, Redis Cluster) | Hash of the key picks the partition | Even spread regardless of key skew | Range scans hit every partition; naive hash mod N remaps almost everything when N changes |
Consistent hashing and rings
Consistent hashing (Karger et al., 1997) fixes remapping. Hash values form a ring; each node owns points (tokens) on it; a key belongs to the first node clockwise from its hash. Adding a node takes part of one neighbour's range and removing one hands its range to a neighbour, so about 1/N of keys move. Dynamo, Cassandra and Riak add virtual nodes: each physical node owns many scattered tokens, so a join or leave spreads load across many nodes and heterogeneous hardware can take proportionally more tokens. Replicas are the next N distinct physical nodes clockwise (the preference list), skipping nodes in the same rack or data centre.
Rings still have hot keys (one celebrity row lives on one partition) and make range scans scatter-gather; range-partitioned systems instead move whole ranges under a load balancer and keep the range directory in a consensus-backed store.
Rebalancing
Move whole partitions, not rows; throttle streaming so foreground traffic is not starved; keep the old owner serving until the new one has caught up (Cassandra's joining node takes writes but serves no reads until bootstrap finishes); and keep the routing table consistent, by versioned gossip or a consensus store.
Distributed transactions with Percolator
Setting and ingredients
Percolator (Peng and Dabek, 2010) incrementally updated Google's search index on Bigtable, which offers single-row atomic operations but no multi-row transactions. Percolator adds snapshot isolation across arbitrary rows as a client library, keeping bookkeeping in extra columns and running 2PC from the client. TiKV implements the same protocol.
- Timestamp oracle (TSO): a single service handing out strictly increasing timestamps (fast because it allocates ranges and logs only the range). Each transaction takes a start timestamp and later a commit timestamp.
- Three columns per user column:
dataholds versions keyed by start timestamp;lockholds an uncommitted transaction's lock with a pointer to the primary;writeholds, keyed by commit timestamp, a pointer to the committed data version. A read at snapshottsfinds the latestwriteat or belowtsand follows it. - Primary lock: one written row is designated primary; every other row's lock points to it. The primary's state is the sole truth about whether the transaction committed.
Protocol
Prewrite. For each written row, in one single-row Bigtable transaction, the client checks for a write with commit timestamp above its start_ts (someone committed after our snapshot: write-write conflict, abort) and for any existing lock (concurrent writer: abort or wait). If clear, it writes data at start_ts and a lock pointing to the primary. The primary is prewritten first.
Commit. The client gets commit_ts, then in one single-row transaction on the primary verifies its lock is still present, writes the write record at commit_ts pointing to start_ts, and removes the lock. That write is the commit point. Secondaries are committed afterwards, possibly asynchronously, because anyone who finds a secondary's lock can learn the outcome from the primary.
Lazy cleanup
Clients crash and leave locks behind, and there is no coordinator to recover them. A reader that meets a lock older than a threshold follows it to the primary. If the primary's lock is gone and a write record exists, the transaction committed: the reader rolls forward the secondary. If the primary lock is still present, the reader rolls back by removing it, which is safe because the committer's own commit step checks for that lock and fails if it is gone. The primary lock is the one atomic decision everything hinges on; the rest is replayable. The price is latency (two oracle trips plus prewrite and commit per row), acceptable for a batch indexer and mitigated in TiDB by batching and a one-phase commit for single-region transactions.
Coordination avoidance
Invariant confluence
Everything above pays for coordination; Bailis and colleagues asked when it is actually necessary. An invariant I is invariant confluent (I-confluent) for a set of transactions if, whenever two replicas holding I-valid states diverge by running transactions independently, merging their states yields an I-valid state. If so, those transactions need no coordination; if not, some coordination is provably required.
| Invariant | I-confluent? | Why |
|---|---|---|
| Counter only increments | yes | merged increments only grow it |
| Balance never negative, with withdrawals | no | two replicas each approve a withdrawal that is fine alone |
| Insert a child row referencing an existing parent | yes | merging inserts removes no parent |
| Delete a parent row | no | a concurrent child insert elsewhere breaks it on merge |
| Uniqueness of a user-chosen id | no | two replicas accept the same id |
| Uniqueness of ids drawn from per-replica ranges | yes | the ranges cannot collide |
Applied to TPC-C, ten of twelve invariants were I-confluent, so most of the benchmark can run coordination-free. The practical move is to restate constraints so they become I-confluent (pre-allocated id ranges, escrow splits of bounded counters) and spend coordination only on the rest.
RAMP transactions
Read Atomic Multi-Partition (RAMP) transactions (Bailis et al., 2014) target one guarantee: read atomicity, a reader sees all or none of a transaction's writes. Secondary indexes, foreign-key pairs and denormalised copies need it, and 2PC with locks is a heavy way to get it.
RAMP never blocks: readers do not wait for writers or vice versa, and a partition failure affects only transactions touching it. Writers run two rounds: store each item's new version with metadata describing the transaction's write set, then make the versions visible. Readers fetch the latest visible version of each item, then check the metadata: if B's metadata says the same transaction wrote A but the fetched A is older, a second round fetches that version of A by timestamp (partitions keep recent versions). RAMP-Fast stores the full write set, RAMP-Small only a timestamp (always two read rounds), RAMP-Hybrid a Bloom filter. RAMP gives neither serializability nor protection against lost updates; it stops readers from seeing half a transaction.
Design trade-offs
| Approach | Agreement on | Blocking on coordinator failure | Latency per cross-shard transaction | Isolation | Constraints |
|---|---|---|---|---|---|
| Classic 2PC | Outcome | Yes, locks held until recovery | 2 round trips plus forced log writes | Participants' (usually 2PL) | Simple; blocking kills it at scale |
| 2PC over Paxos groups (Spanner, CockroachDB) | Outcome, replicated coordinator | No, new leader continues | 2PC plus Paxos per record plus commit-wait | Strict serializability | Consensus per shard; good clocks |
| Calvin | Input order | No, replicas replay the log | Epoch plus one exchange of reads | Serializable | Known read and write sets; no interactive transactions |
| Percolator | Outcome via one primary row | No coordinator; lazy cleanup | 2 oracle trips plus prewrite and commit per row | Snapshot isolation | Single oracle; batch-friendly latency |
| RAMP | Nothing across partitions | None | 1 to 2 read rounds, 2 write rounds | Read atomic only | No lost-update protection |
Serializable multi-shard interactive transactions: 2PC over consensus groups. Short stored-procedure-shaped transactions where throughput matters most: Calvin. Snapshot isolation on an existing key-value store: Percolator. First check whether the invariant is I-confluent, because then the cheapest correct protocol is none.
Interview angle
- Walk through 2PC and name the commit point. Prepare: cohorts log redo and undo plus PREPARED, hold locks, vote. Commit: on all yes the coordinator forces COMMIT to its log, then tells cohorts. The commit point is the coordinator's durable COMMIT record, not the last ack.
- What is the blocking problem and what are the ways out? A yes-voting cohort cannot decide alone if the coordinator vanishes, so it holds locks. Cooperative termination helps only if a peer knows the decision or has not voted; 3PC breaks under partitions; replicating the coordinator's log with consensus (Spanner, CockroachDB) is the real fix.
- How does Spanner order a commit in Tokyo before a later start in Belgium? TrueTime bounds uncertainty. The leader picks a timestamp at least
TT.now().latestand commit-waits until true time has passed it before replying; any later transaction anywhere sees an interval entirely beyond it. - Calvin versus Spanner for multi-partition transactions. Spanner agrees on outcomes with 2PC over Paxos groups and holds locks meanwhile. Calvin agrees on input order and executes deterministically, so there is no vote, at the cost of needing read and write sets up front and no interactive transactions.
- A Percolator client crashes after committing the primary but before the secondaries. The transaction is committed. The next reader that hits a stale secondary lock follows it to the primary, finds a write record, rolls the secondary forward and proceeds. Had the primary lock still been there, the reader would roll back by removing it, and a late committer would fail its primary check.
- Debit one shard, credit another: does it need coordination? Non-negative balance is not I-confluent for withdrawals, so yes for the debit side: 2PC, or a saga with compensation if temporary inconsistency is tolerable, or restructure into per-replica escrow allowances that are I-confluent.
Pitfalls and misconceptions
- 2PC and 2PL are the same thing. Two-phase commit is atomic commitment across nodes; two-phase locking is concurrency control. Spanner uses both; Percolator uses 2PC with optimistic conflict checks instead of 2PL.
- The transaction commits when the last participant acknowledges. It commits when the coordinator's decision record is durable; acks only let the coordinator forget.
- 3PC solves 2PC's problems. Only blocking under crash faults in a synchronous network, at the price of a round trip and inconsistency under partitions.
- Spanner is fast because of atomic clocks. TrueTime makes external consistency possible with a bounded wait; Paxos replication and 2PC still cost round trips.
- Calvin has no coordination. It coordinates once, at the sequencer, on input order; it avoids per-transaction agreement on outcomes.
- Snapshot isolation is serializability. Percolator's snapshot isolation permits write skew; Spanner's read-write transactions are serializable. Say which you mean.
Key terms
| Term | Meaning |
|---|---|
| Atomic commitment | Protocol ensuring all participants reach the same commit-or-abort decision, committing only on unanimous yes |
| Coordinator and cohort | The process driving the commit protocol, and the participants holding data and voting |
| Commit point | The instant the decision becomes irreversible: the coordinator's durable COMMIT in 2PC, the primary row write in Percolator |
| Three-phase commit | 2PC plus a pre-commit phase enabling timeout decisions; inconsistent under partitions |
| Deterministic database | Order fixed before execution so every replica computes the same result from the same log (Calvin) |
| Sequencer | Calvin's layer that batches and replicates transaction inputs into a global order |
| Paxos group | Replicas of one shard running Paxos with a leader; Spanner's unit of replication and 2PC participation |
| TrueTime | Clock API returning an interval containing true time, with bounded uncertainty epsilon |
| Commit-wait | Waiting until true time is certainly past the commit timestamp before releasing locks and replying |
| External consistency | Spanner's term for strict serializability |
| Consistent hashing | Keys and nodes on a ring so that adding or removing a node moves about 1/N of the keys |
| Timestamp oracle | Service handing out strictly increasing timestamps for transaction start and commit |
| Primary lock | Percolator's designated lock whose single-row removal is the commit point |
| Invariant confluence | An invariant that survives merging independently evolved states, implying no coordination is needed |
| Read atomicity | A reader sees all or none of a transaction's writes; RAMP's guarantee |
Further reading
- Database Internals, Chapter 13; DDIA Chapter 9 covers 2PC and XA from the application side.
- Jim Gray, "Notes on Data Base Operating Systems", 1978 (the original two-phase commit description).
- Dale Skeen, "Nonblocking Commit Protocols", SIGMOD 1981 (three-phase commit).
- Alexander Thomson et al., "Calvin: Fast Distributed Transactions for Partitioned Database Systems", SIGMOD 2012, http://cs.yale.edu/homes/thomson/publications/calvin-sigmod12.pdf
- James Corbett et al., "Spanner: Google's Globally-Distributed Database", OSDI 2012, https://www.usenix.org/conference/osdi12/technical-sessions/presentation/corbett
- Daniel Peng and Frank Dabek, "Large-scale Incremental Processing Using Distributed Transactions and Notifications", OSDI 2010 (Percolator).
- Peter Bailis et al., "Coordination Avoidance in Database Systems", VLDB 2015, http://www.vldb.org/pvldb/vol8/p185-bailis.pdf
- Peter Bailis et al., "Scalable Atomic Visibility with RAMP Transactions", SIGMOD 2014, http://www.bailis.org/papers/ramp-sigmod2014.pdf
- PostgreSQL documentation, PREPARE TRANSACTION, https://www.postgresql.org/docs/current/sql-prepare-transaction.html