Two-phase commit

Deep dive

How it works

Parts:

• coordinator — drives both phases and owns the decision record
• participants — resource managers (shards, databases, queues) that vote and apply
• coordinator log — durable COMMIT/ABORT/END records; the commit point
• participant log — durable PREPARED record with redo data
• recovery procedure — in-doubt participants ask the coordinator for the outcome

Operations: prepare, decide, commit, abort, recovery (steps below).

1. Client finishes the transaction's reads/writes on participants (under locks or write intents).
2. Prepare: coordinator sends PREPARE to all participants.
3. Each participant validates (constraints, conflicts), forces a PREPARED record with its redo data to its log, replies YES; or rolls back and replies NO.
4. Decide: all YES → coordinator forces COMMIT(tx, participants) to its log. This write is the commit point; it can now reply to the client.
5. Commit: coordinator sends COMMIT; participants apply, force a commit record, release locks, ACK. When all ACK, the coordinator writes END (lazily) and forgets the transaction.
6. Any NO or timeout before step 4 → ABORT; under presumed abort nothing is forced.
7. Recovery: a participant with PREPARED but no outcome asks the coordinator; the coordinator answers COMMIT if it has the record, otherwise ABORT.

states: participant: Active ─PREPARE/YES─► Prepared ─COMMIT─► Committed
                       └──────NO/ABORT───► Aborted (Prepared leaves only on the decision)

coord        P1            P2
  ├─PREPARE──►│─fsync YES──►│
  ├─PREPARE───┼────────────►│─fsync YES
  │◄──YES─────┤◄────YES─────┤
  fsync COMMIT  ← commit point, reply to client
  ├─COMMIT───►│ apply, unlock ...

Code

use std::collections::{HashMap, HashSet};

pub type TxId = u64;
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Decision { Commit, Abort }

#[derive(Clone, Debug)]
pub enum Rec { Prepared(TxId), Outcome(TxId, Decision), CommitDecision(TxId, Vec<u32>), End(TxId) }

/// Stand-in for a write-ahead log; `force` = append + fsync before returning.
#[derive(Default)]
pub struct Log { pub recs: Vec<Rec> }
impl Log { pub fn force(&mut self, r: Rec) { self.recs.push(r); } pub fn lazy(&mut self, r: Rec) { self.recs.push(r); } }

pub struct Participant { pub id: u32, pub log: Log, pub locked: HashSet<TxId>, pub can_commit: bool }

impl Participant {
    pub fn prepare(&mut self, tx: TxId) -> bool {
        if !self.can_commit { self.locked.remove(&tx); return false; } // presumed abort: no log
        self.log.force(Rec::Prepared(tx)); // redo data must be durable with this record
        true                                // from here on we may NOT abort unilaterally
    }
    pub fn finish(&mut self, tx: TxId, d: Decision) {
        match d {
            Decision::Commit => self.log.force(Rec::Outcome(tx, d)), // forced, then ACK
            Decision::Abort => self.log.lazy(Rec::Outcome(tx, d)),   // no force, no ACK needed
        }
        self.locked.remove(&tx);
    }
    /// After a crash: prepared transactions with no outcome are in doubt and keep their locks.
    pub fn in_doubt(&self) -> Vec<TxId> {
        let done: HashSet<TxId> = self.log.recs.iter()
            .filter_map(|r| if let Rec::Outcome(t, _) = r { Some(*t) } else { None }).collect();
        self.log.recs.iter().filter_map(|r| match r {
            Rec::Prepared(t) if !done.contains(t) => Some(*t), _ => None }).collect()
    }
}

#[derive(Default)]
pub struct Coordinator { pub log: Log, acks: HashMap<TxId, HashSet<u32>> }

impl Coordinator {
    pub fn commit(&mut self, tx: TxId, parts: &mut [Participant]) -> Decision {
        // Phase 1 (sent in parallel in a real system; any timeout counts as NO).
        let all_yes = parts.iter_mut().all(|p| p.prepare(tx));
        if !all_yes {
            for p in parts.iter_mut() { p.finish(tx, Decision::Abort); }
            return Decision::Abort;               // nothing forced: presumed abort
        }
        let ids: Vec<u32> = parts.iter().map(|p| p.id).collect();
        self.log.force(Rec::CommitDecision(tx, ids)); // COMMIT POINT; may reply to client now
        for p in parts.iter_mut() {                   // Phase 2, retried until every ACK arrives
            p.finish(tx, Decision::Commit);
            self.acks.entry(tx).or_default().insert(p.id);
        }
        if self.acks[&tx].len() == parts.len() {
            self.log.lazy(Rec::End(tx));
            self.acks.remove(&tx);
        }
        Decision::Commit
    }
    /// Answer an in-doubt participant. No record => it never reached the commit point.
    pub fn query(&self, tx: TxId) -> Decision {
        let committed = self.log.recs.iter().any(|r| matches!(r, Rec::CommitDecision(t, _) if *t == tx));
        if committed { Decision::Commit } else { Decision::Abort }
    }
}

With presumed abort the coordinator may forget a transaction only after END; a COMMIT record must survive until every participant has acknowledged.

Complexity and constants

• Messages: PREPARE + vote + COMMIT (+ ACK) per participant = 3-4n.
• Forced writes on the critical path: participant PREPARED (parallel across participants) then coordinator COMMIT: 2 sequential fsyncs, order of 0.05-10 ms each depending on the device.
• Lock hold time: from first write through phase 2 delivery; contention on hot rows scales with it.
• With replicated participants (each a Raft/Paxos group), each "fsync" becomes a consensus round: a classic distributed-SQL commit costs 2 consensus rounds, which CockroachDB's parallel commits cut to 1.

Efficiency

• Cost: 3-4n messages for n participants (PREPARE, vote, COMMIT, ACK). Two sequential forced writes on the critical path, participant PREPARED (parallel across participants) then coordinator COMMIT, each order of 0.05-10 ms depending on the device. Client latency is about 1 RTT plus both fsyncs; locks are held at least 2 RTT. With Raft/Paxos-replicated participants each "fsync" becomes a consensus round. Bound: log-force latency and lock hold time.
• Levers:
  • single-shard fast path — one phase, one log write — well-partitioned schemas
  • read-only participants vote READ-ONLY and drop out of phase 2 — fewer messages and log writes (Mohan et al., R*) — read-mostly transactions
  • presumed abort or presumed commit — no forced write or ACK for the presumed outcome (R*) — abort-heavy or commit-heavy workloads
  • parallel commits — one consensus round fewer (CockroachDB docs) — replicated participants
  • coordinator on the first-written participant — one hop and one log less — shard-local transactions
• Naive → standard → state of the art:
  • 2PC with every record forced and every participant ACKing · 3-4n messages, 2 forced writes plus ACK round
  • presumed abort, read-only vote, coordinator on a participant · fewer forced writes and messages
  • replicated decision record plus parallel commits (CockroachDB) · 2 consensus rounds cut to 1 on the commit path

Implementation notes

• Single-shard fast path: one participant → commit locally (1 phase); the common case in well-partitioned schemas.
• Coordinator on a participant (often the first-written shard): saves a hop and one log.
• Read-only participants vote READ-ONLY at prepare and drop out of phase 2.
• Replicate the decision: store the transaction record in a replicated log (Spanner, CockroachDB, TiDB/Percolator's primary key) so any node can resolve in-doubt transactions; this removes the blocking problem in practice.
• Participants must persist enough redo/undo before voting YES; "prepared" must survive a crash.
• PostgreSQL PREPARE TRANSACTION is disabled by default (max_prepared_transactions = 0); forgotten prepared transactions hold locks and block vacuum.

Pitfalls

• Participant aborting after voting YES (e.g. on lock timeout): breaks atomicity. Once prepared, only the decision can end it.
• Coordinator replying to the client before the COMMIT record is durable.
• Garbage-collecting the decision before all ACKs: a recovering participant then gets "abort" by presumption for a committed transaction.
• Orphaned prepared transactions (XA) after a coordinator is decommissioned: monitor and resolve them manually.
• Timeouts too short in phase 1 → spurious aborts under load; too long → locks held across failures.

Variants and alternatives

In production

Distributed SQL: Google Spanner (2PC across Paxos groups), CockroachDB (parallel commits), TiDB/TiKV (Percolator model), YugabyteDB. Messaging: Kafka transactions (transaction coordinator + commit markers). Application servers and classic databases: XA in MySQL/PostgreSQL/Java EE transaction managers.

Implementations

Sources

• Jim Gray, Notes on Data Base Operating Systems, 1978.
• C. Mohan, Bruce Lindsay, Ron Obermarck, Transaction Management in the R* Distributed Database Management System, ACM TODS 1986. paper
• Jim Gray, Leslie Lamport, Consensus on Transaction Commit, ACM TODS 2006. paper
• Daniel Peng, Frank Dabek, Large-scale Incremental Processing Using Distributed Transactions and Notifications, OSDI 2010. paper
• Philip A. Bernstein, Vassos Hadzilacos, Nathan Goodman, Concurrency Control and Recovery in Database Systems, 1987.
• PostgreSQL documentation, PREPARE TRANSACTION. docs
• Apache Kafka, KIP-98: Exactly Once Delivery and Transactional Messaging. spec