Distributed Transactions

A distributed transaction involves a coordinator (the node that initiates the transaction) and one or more participants (nodes that must execute part of the transaction). The goal is atomic commit: every participant commits, or every participant aborts — there is no partial outcome.

Formally, an atomic commit protocol must satisfy:

1. Agreement: All nodes that reach a decision reach the same decision.
2. Validity: If any node votes to abort, the final decision is Abort. If all vote to commit and there are no failures, the final decision is Commit.
3. Termination: Every non-failed node eventually reaches a decision.
4. Integrity/Durability: Once committed, the effects are durable.

The hard part is termination under failure. If a participant crashes mid-protocol, the coordinator must not be stuck forever. If the coordinator crashes, the participants must not be stuck forever. The tension between agreement (which requires coordination) and termination (which requires progress even when nodes fail) is the heart of the problem.

Why you cannot just use a single lock: A naive approach — have the coordinator grab locks on all participants, apply the writes, then release — fails because locks are not durable. If the coordinator crashes after writing to some participants but before writing to others, those partial writes are either rolled back (inconsistent with durability) or left committed (inconsistent with atomicity). You need a two-round protocol.

# The problem in pseudocode — this naive approach BREAKS atomicity
def transfer(amount, src_node, dst_node):
    src_node.debit(amount)   # What if dst_node crashes here?
    dst_node.credit(amount)  # Or the network fails here?
    # No recovery possible: src debited but dst not credited

Two-phase commit is the classic solution, invented by Jim Gray in the 1970s and still used inside PostgreSQL's PREPARE TRANSACTION, MySQL XA, and virtually every distributed RDBMS.

Phase 1 — Prepare (Voting): The coordinator sends a PREPARE message to all participants. Each participant:

• Acquires all locks needed for the transaction.
• Writes a prepare record to its durable write-ahead log (WAL).
• Sends VOTE-COMMIT if it is ready to commit, or VOTE-ABORT if it cannot.

Writing the prepare record is critical: it means the participant has promised it can commit. Even if it crashes and restarts, it must honor this promise.

Phase 2 — Commit/Abort: If the coordinator receives VOTE-COMMIT from all participants, it writes a commit record to its own WAL, then sends COMMIT to all participants. Otherwise, it sends ABORT.

Each participant, upon receiving COMMIT, applies the changes and writes a commit record. Upon ABORT, it rolls back and releases locks.

Correctness intuition: A participant can only commit if it has received a COMMIT message from the coordinator. The coordinator only sends COMMIT if it received VOTE-COMMIT from everyone. So if any participant voted to abort, no one ever commits.

The blocking problem: 2PC is a blocking protocol. If the coordinator crashes after sending PREPARE but before sending the phase-2 decision, participants are stuck: they have voted COMMIT (and cannot unilaterally abort, because another participant may have already committed on receiving a COMMIT they didn't see), but they cannot proceed. They hold locks indefinitely. This is the fundamental flaw of 2PC.

# Two-Phase Commit — Coordinator side
class Coordinator:
    def run_2pc(self, participants, txn):
        # Phase 1: Prepare
        votes = []
        for p in participants:
            vote = p.prepare(txn)  # blocks until response or timeout
            votes.append(vote)

        # Decision
        if all(v == 'COMMIT' for v in votes):
            self.log.write(CommitRecord(txn.id))  # durable!
            decision = 'COMMIT'
        else:
            self.log.write(AbortRecord(txn.id))
            decision = 'ABORT'

        # Phase 2: Broadcast decision
        for p in participants:
            p.notify(txn.id, decision)  # must retry until ack

# Participant side
class Participant:
    def prepare(self, txn):
        try:
            self.acquire_locks(txn)
            self.apply_to_buffer(txn)
            self.log.write(PrepareRecord(txn.id))  # durable promise!
            return 'COMMIT'
        except Exception:
            return 'ABORT'

    def notify(self, txn_id, decision):
        if decision == 'COMMIT':
            self.log.write(CommitRecord(txn_id))
            self.flush_buffer(txn_id)
        else:
            self.rollback(txn_id)
        self.release_locks(txn_id)

Understanding 2PC deeply requires tracing what happens in every failure scenario. There are four critical moments where failures matter:

1. Participant crashes before sending VOTE-COMMIT: The coordinator either times out or receives VOTE-ABORT. It aborts the transaction. Safe: no one committed.

2. Participant crashes after writing PREPARE but before Phase 2: When it recovers, it finds its prepare record. It asks the coordinator: "What was the decision for txn X?" The coordinator replies and the participant follows. Safe, with recovery protocol.

3. Coordinator crashes after writing COMMIT but before all participants are notified: When the coordinator recovers, it reads its commit record and resends COMMIT to participants it hasn't heard from. Phase 2 messages must be idempotent and retried until acknowledged. Safe, with recovery.

4. Coordinator crashes after Phase 1 but before writing its Phase-2 decision (THE BLOCKING CASE): This is the problematic case. Participants have voted COMMIT. They hold locks. They cannot abort (another participant might already have received COMMIT from the old coordinator before it crashed, which is impossible in 2PC but participants don't know that). They cannot commit (the coordinator hasn't decided). They are blocked until the coordinator recovers.

A timeout-based workaround is to elect a new coordinator, but a new coordinator cannot determine the original coordinator's decision without reading its log — and if that log was not flushed before the crash, the decision is lost. This is not just a performance problem; it is a correctness problem if we weaken durability.

Practical mitigation: Modern systems use PREPARE TRANSACTION (XA) and monitor for long-running prepared transactions. PostgreSQL, for instance, will show them in pg_prepared_xacts. Operators must manually resolve them if the coordinator dies permanently.

-- PostgreSQL XA / prepared transactions (real SQL)
-- Coordinator side:
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
PREPARE TRANSACTION 'txn_abc123'; -- writes prepare record to WAL

-- ... coordinator sends COMMIT to all participants ...
-- If coordinator crashes here, txn sits in pg_prepared_xacts
-- After recovery or manual intervention:
COMMIT PREPARED 'txn_abc123';
-- or:
ROLLBACK PREPARED 'txn_abc123';

-- Check for stuck prepared transactions:
SELECT * FROM pg_prepared_xacts;

Three-phase commit (3PC), proposed by Dale Skeen in 1981, attempts to fix the blocking problem by adding a third phase between prepare and commit. The key insight: 2PC blocks because the coordinator can crash in a state where participants do not know whether to commit or abort. 3PC prevents this by ensuring that before the coordinator sends COMMIT, all participants know that all other participants voted to commit.

Phase 1 — Can-Commit (Voting): Same as 2PC phase 1. Coordinator sends CAN-COMMIT?. Participants vote yes/no.

Phase 2 — Pre-Commit: If all voted yes, coordinator sends PRE-COMMIT to all. Participants acknowledge. This phase propagates the information: "everyone voted yes". A participant that receives PRE-COMMIT knows it is safe to commit if the coordinator fails — it just needs to wait for the final confirmation, and if the coordinator doesn't arrive, it can ask any other participant who is also in the pre-commit state.

Phase 3 — Do-Commit: Coordinator sends DO-COMMIT. Participants commit.

Why it seems to fix blocking: If the coordinator crashes between Phase 2 and Phase 3, any participant can take over as coordinator. Since all participants that received PRE-COMMIT know everyone voted yes, they can safely send DO-COMMIT to the rest and commit.

The fatal flaw — network partitions: 3PC assumes the "fail-stop" model: nodes either run correctly or crash permanently, and failures are detectable via timeouts. In a real network, a partition can make a live node look crashed. If the coordinator is partitioned from some participants after Phase 2, the non-partitioned group elects a new coordinator and commits. The partitioned group, having timed out, aborts. When the partition heals, you have committed on one side and aborted on the other — split-brain. This is a safety violation.

3PC is therefore not used in practice for systems that must tolerate network partitions (which is essentially all WAN systems). It solves the blocking problem only in a synchronous network with fail-stop crashes, which is not the real world.

# 3PC State machine for a participant
from enum import Enum, auto

class State(Enum):
    INIT = auto()
    WAITING = auto()    # voted yes, waiting for PRE-COMMIT
    PREPARED = auto()   # received PRE-COMMIT, can coordinate if needed
    COMMITTED = auto()
    ABORTED = auto()

class Participant3PC:
    def __init__(self):
        self.state = State.INIT

    def on_can_commit(self, txn):
        if self.can_commit(txn):
            self.state = State.WAITING
            return 'YES'
        else:
            self.state = State.ABORTED
            return 'NO'

    def on_pre_commit(self, txn_id):
        # Key: now we KNOW all others voted yes
        self.log.write(PreCommitRecord(txn_id))
        self.state = State.PREPARED
        return 'ACK'

    def on_do_commit(self, txn_id):
        self.commit(txn_id)
        self.state = State.COMMITTED

    def on_timeout_in_prepared_state(self, txn_id, peers):
        # Safe to commit: everyone was in PREPARED, meaning all voted yes
        # Contact peers to confirm, then commit
        self.coordinate_commit_among_peers(txn_id, peers)

Two theoretical results explain why there is no perfect distributed transaction protocol.

FLP Impossibility (Fischer, Lynch, Paterson, 1985): In an asynchronous distributed system where even one process can crash, there is no deterministic consensus algorithm that always terminates. More precisely: no protocol can simultaneously guarantee safety (never decide incorrectly) and liveness (always eventually decide) in the presence of even a single crash failure in an asynchronous model.

This does not mean consensus is impossible in practice — it means you must weaken one assumption. Real systems either:

• Accept that termination might not hold (2PC in the coordinator-crash case), or
• Use randomized algorithms (Raft/Paxos with randomized timeouts), or
• Assume partial synchrony (messages arrive within a known bound eventually).

CAP Theorem (Brewer, 2000; formalized by Gilbert and Lynch, 2002): A distributed system can guarantee at most two of:

• Consistency (linearizability — every read sees the most recent write)
• Availability (every request receives a non-error response)
• Partition tolerance (the system operates even when the network partitions)

Since network partitions are not optional in any real WAN system, you must choose between C and A when a partition occurs. Two-phase commit chooses C: it blocks (sacrifices A) rather than risk inconsistency. Eventual consistency systems (like Dynamo-style databases) choose A: they continue serving reads and writes during a partition, but may return stale data.

The practical implication for transactions: You cannot have atomic distributed transactions AND high availability under partition. 2PC gives you C at the cost of A. Sagas (discussed below) give you A at the cost of strong C — they provide eventual consistency with compensating transactions instead of true atomicity.

# CAP in pseudocode — the forced choice during a partition

# CP system (choose Consistency over Availability)
def read_cp(key):
    if not can_reach_quorum():
        raise ServiceUnavailable  # refuse to answer rather than serve stale data
    return quorum_read(key)

# AP system (choose Availability over Consistency)
def read_ap(key):
    if not can_reach_quorum():
        return local_stale_read(key)  # serve possibly stale data
    return quorum_read(key)

# You CANNOT implement both behaviors simultaneously:
# if partition happens and you must return a value,
# that value might be stale -> not consistent
# if you refuse to return a value, you're not available

Sagas, introduced by Hector Garcia-Molina and Kenneth Salem in their 1987 paper, are a pattern for long-lived transactions that explicitly trade atomicity for availability. A saga is a sequence of local transactions T1, T2, ..., Tn, each with a corresponding compensating transaction C1, C2, ..., Cn that semantically undoes its effect.

If Ti fails, the saga executes Ci-1, Ci-2, ..., C1 to undo the completed steps. There is no blocking coordinator, no prepare phase, and no global lock. Each step commits locally and immediately.

Two coordination styles:

1. Choreography: Each service, upon completing its step, publishes a domain event. The next service subscribes to that event and starts its step. There is no central coordinator; the flow emerges from event subscriptions. Simple for small sagas, but hard to trace and debug for large ones.
2. Orchestration: A central saga orchestrator (a state machine) sends commands to each service and waits for responses before proceeding. Easier to trace and monitor. The orchestrator is itself a service that must be fault-tolerant.

The semantic guarantee: Sagas provide atomicity in the sense that the saga either completes all steps or rolls back all completed steps via compensations. But compensations are not true rollbacks — they are new forward-moving transactions that undo the semantic effect. Between steps, intermediate states are visible to other transactions. This is called ACI (without D) or eventual atomicity.

Compensating transactions must be idempotent and always succeed. This is a hard constraint: if a compensation itself fails, you are in trouble. Real systems design compensations to be retryable (idempotent) and non-failable (e.g., instead of deleting a record, mark it as cancelled).

Classic example — order saga:

• T1: Reserve inventory → C1: Release inventory
• T2: Charge payment → C2: Issue refund
• T3: Ship order → C3: Arrange return

If T3 fails, the saga runs C2 (refund) then C1 (release inventory). Each compensation must be designed to be retryable and idempotent.

# Saga Orchestrator (simplified)
from dataclasses import dataclass, field
from typing import List, Callable

@dataclass
class SagaStep:
    name: str
    action: Callable      # forward transaction
    compensate: Callable  # compensating transaction

class SagaOrchestrator:
    def __init__(self, steps: List[SagaStep]):
        self.steps = steps

    def execute(self, context: dict):
        completed = []
        for step in self.steps:
            try:
                print(f"Executing: {step.name}")
                step.action(context)
                completed.append(step)
            except Exception as e:
                print(f"Step {step.name} failed: {e}. Rolling back.")
                # Execute compensations in reverse order
                for done in reversed(completed):
                    try:
                        done.compensate(context)  # must be idempotent!
                    except Exception as ce:
                        # Log and alert — compensation failed
                        # This is a 'saga stuck' scenario requiring manual intervention
                        raise RuntimeError(f"Compensation {done.name} failed: {ce}")
                return False  # saga aborted
        return True  # saga committed

# Usage:
order_saga = SagaOrchestrator([
    SagaStep(
        name="reserve_inventory",
        action=lambda ctx: inventory_svc.reserve(ctx['item_id'], ctx['qty']),
        compensate=lambda ctx: inventory_svc.release(ctx['item_id'], ctx['qty'])
    ),
    SagaStep(
        name="charge_payment",
        action=lambda ctx: payment_svc.charge(ctx['user_id'], ctx['amount']),
        compensate=lambda ctx: payment_svc.refund(ctx['user_id'], ctx['amount'])
    ),
    SagaStep(
        name="create_shipment",
        action=lambda ctx: shipping_svc.create(ctx['order_id']),
        compensate=lambda ctx: shipping_svc.cancel(ctx['order_id'])
    ),
])

Both 2PC and sagas require operations to be retried after failures. Phase 2 of 2PC retries COMMIT until acknowledged. Saga compensations are retried until they succeed. This means every operation in a distributed transaction must be idempotent: applying it multiple times has the same effect as applying it once.

The canonical technique is idempotency keys: assign each operation a globally unique ID. Before executing, the receiver checks if this ID has been processed before (via a deduplication table). If yes, return the cached result. If no, execute and store the result.

Exactly-once delivery is impossible in a distributed system (you can have at-most-once or at-least-once delivery). But exactly-once processing is achievable with idempotent operations: deliver at-least-once, deduplicate at the receiver.

For database operations, idempotency is achieved by:

• Using INSERT ... ON CONFLICT DO NOTHING with a natural key.
• Checking and setting a status column atomically: UPDATE orders SET status='shipped' WHERE id=X AND status='processing'.
• Storing the idempotency key in the same transaction as the business operation (transactional outbox pattern).

The transactional outbox pattern is critical for sagas: when a service commits a local transaction, it also writes the outgoing event/command to an outbox table in the same local database transaction. A separate relay process reads the outbox and publishes to the message broker. This ensures the event is published if and only if the local transaction commits.

-- Transactional Outbox Pattern (PostgreSQL)
-- Step 1: In the same transaction, write the business change AND the outbox event
BEGIN;
  -- Business operation
  UPDATE accounts SET balance = balance - 100 WHERE id = 1;
  INSERT INTO orders (id, user_id, status) VALUES ('ord_123', 1, 'pending');
  
  -- Outbox: event to be published to message broker
  INSERT INTO outbox_events (id, aggregate_type, aggregate_id, event_type, payload)
  VALUES (
    gen_random_uuid(),
    'order',
    'ord_123',
    'OrderCreated',
    '{"order_id": "ord_123", "user_id": 1, "amount": 100}'
  );
COMMIT;
-- The relay process polls outbox_events and publishes to Kafka/RabbitMQ
-- If relay crashes, it re-polls and re-publishes -> at-least-once
-- Consumers must be idempotent (use event.id as idempotency key)

Modern distributed databases have developed sophisticated protocols that go beyond vanilla 2PC by incorporating Paxos/Raft for the coordinator role, eliminating the single-point-of-failure blocking problem.

Google Spanner (2012): Uses 2PC where the coordinator is a Paxos group, not a single node. Since a Paxos group can make progress as long as a majority is alive, the coordinator cannot be permanently blocked by a single crash. Spanner also uses TrueTime — a globally synchronized clock with bounded uncertainty from GPS and atomic clocks — to assign globally consistent timestamps without inter-node coordination for reads. This enables external consistency (linearizability across the globe) with high availability.

Google Percolator (2010): Built on Bigtable, Percolator implements distributed transactions using a lock column in Bigtable to encode 2PC state. The prepare phase writes a lock entry; the commit phase atomically replaces the lock with a write entry. Crucially, Percolator is lazy: it does not clean up locks eagerly. Instead, a transaction that encounters a lock checks if the lock-holder is still alive. If the lock-holder has crashed, the new transaction can steal the commit decision or roll back the old transaction. This eliminates coordinator blocking by making recovery opportunistic.

CockroachDB and TiDB: Both use Raft groups as participants and implement their own variant of Percolator-style optimistic concurrency control with distributed timestamps. They replace the 2PC coordinator with a consensus group so the coordinator itself is fault-tolerant.

The lesson: 2PC's blocking problem is solved not by moving to 3PC (which fails under partitions) but by making the coordinator itself highly available via Paxos/Raft.

# Spanner-style external consistency read (conceptual)
# TrueTime API: TT.now() returns an interval [earliest, latest]
# with guaranteed: true_time in [earliest, latest]

def commit_transaction(txn):
    # Step 1: Get commit timestamp candidate
    tt = TrueTime.now()
    s = tt.latest  # pick latest bound
    
    # Step 2: Run 2PC (with Paxos coordinator)
    coordinator.two_phase_commit(txn, timestamp=s)
    
    # Step 3: "Commit wait" — wait until true time has passed s
    # This ensures any future transaction starts after s,
    # guaranteeing external consistency
    while TrueTime.now().earliest < s:
        sleep(1ms)
    
    return s  # commit timestamp

# A read at timestamp t sees all commits with timestamp < t
# This gives linearizable reads across globally distributed nodes
# without any cross-node coordination for read-only transactions

Atomicity is only one property of the ACID model. Isolation — the I in ACID — determines what intermediate states of a transaction are visible to concurrent transactions. In a distributed setting, weak isolation interacts badly with atomic commit protocols.

The SQL standard defines four isolation levels (weakest to strongest):

1. Read Uncommitted: Can read uncommitted (dirty) data. Prohibited in any reasonable system.
2. Read Committed: Only reads committed data, but non-repeatable reads are possible (re-reading the same row can give different results within a transaction).
3. Repeatable Read: Reads within a transaction are stable, but phantom reads are possible (new rows can appear on re-queries).
4. Serializable: Transactions appear to execute one at a time in some serial order. No anomalies.

Serializable Snapshot Isolation (SSI): Used by PostgreSQL, CockroachDB, and Spanner. Transactions read from a consistent snapshot (MVCC) and the system detects serialization conflicts by tracking read-write dependencies. It provides serializability with much less contention than two-phase locking.

In distributed transactions: Weak isolation levels can cause anomalies that span nodes — a transaction reading node A and writing to node B might see an inconsistent cross-shard state even if each shard uses snapshot isolation. This is the write skew anomaly. True distributed serializability requires that the snapshot used by a transaction is consistent across all nodes, which requires global snapshot management (as in Spanner's TrueTime or CockroachDB's hybrid logical clocks).

-- Write skew anomaly (possible under snapshot isolation without SSI)
-- Two doctors are on call. At least one must always be on call.
-- T1: Doctor A checks: are there 2 doctors on call? Yes. Marks A as off-call.
-- T2: Doctor B checks: are there 2 doctors on call? Yes (snapshot before T1 commits). Marks B as off-call.
-- Result: 0 doctors on call. Neither transaction saw the other's write.

-- With SSI, the system detects the read-write dependency cycle and aborts one:
-- T1 reads the on-call count, T2 reads the on-call count (both see 2)
-- T1 writes (modifies what T2 read), T2 writes (modifies what T1 read)
-- SSI detects this rw-anti-dependency cycle and aborts T2

-- PostgreSQL:
BEGIN ISOLATION LEVEL SERIALIZABLE;
  SELECT COUNT(*) FROM on_call WHERE active = true;  -- reads 2
  UPDATE on_call SET active = false WHERE doctor = 'alice';
COMMIT;
-- If concurrent T2 does the same, one will get:
-- ERROR: could not serialize access due to read/write dependencies

The choice between 2PC and sagas is not always obvious. Here is a principled framework.

Use 2PC when:

• All participants are in the same database cluster or within the same organization's trusted network.
• You need true atomicity (no intermediate states visible).
• Latency is low (LAN, not WAN).
• You can afford to block during coordinator failure (acceptable availability trade-off).
• The participants are a small, fixed set (2PC scales poorly beyond 10-20 participants due to lock contention).
• Examples: bank transfers within one database cluster, shopping cart checkout within a single RDBMS, PostgreSQL PREPARE TRANSACTION for XA.

Use Sagas when:

• The transaction spans multiple independent services (microservices, bounded contexts).
• Services are maintained by different teams and cannot be coupled to a shared transaction coordinator.
• Long-running processes (minutes to hours) where holding locks is infeasible.
• Availability is more important than strict atomicity.
• You can design meaningful compensating transactions.
• Examples: order fulfillment, travel booking (flights + hotels + cars), multi-step onboarding workflows.

Key warning about sagas: Sagas do not isolate intermediate states. Between T2 and T3, another transaction can read the partially-committed state. If you need to prevent this, you need additional patterns: semantic locks (mark records as "pending" and refuse concurrent modifications), commutative updates (design updates to be order-independent), or process managers with explicit state machines.

Never use neither: The worst pattern is to issue cross-service writes without any coordination, hoping they all succeed. This is the "distributed monolith" anti-pattern and leads to data inconsistencies that are extremely hard to detect and repair.

# Decision matrix in code
def choose_protocol(scenario):
    if scenario.services_share_a_database:
        return "2PC (or single-DB transaction)"

    if scenario.max_latency_ms < 50 and not scenario.crosses_wan:
        if scenario.participants <= 10:
            return "2PC with Paxos coordinator (e.g., CockroachDB)"

    if scenario.step_duration_seconds > 5:
        return "Saga (locks held for seconds are infeasible)"

    if scenario.can_design_compensating_transactions:
        if scenario.orchestration_preferred:
            return "Orchestrated Saga (explicit state machine)"
        else:
            return "Choreographed Saga (event-driven)"

    # Fallback: accept eventual consistency with monitoring
    return "Saga + monitoring + manual reconciliation for stuck sagas"