Deterministic Distributed Databases and the Calvin Protocol: Eliminating Two-Phase Commit at Global Scale

Building globally distributed, multi-region transactional databases (NewSQL) has long been considered the pinnacle of distributed systems engineering. Achieving Strict Serializability (ACID) across cross-continental data centers requires solving two brutal adversaries: the speed of light and network partitions.

For decades, traditional distributed databases (and systems following Google Spanner's lineage) relied on the classical pairing of Two-Phase Locking (2PL) for concurrency control and Two-Phase Commit (2PC) for distributed atomicity. However, under high write contention and cross-region WAN latency (50ms to 200ms round trips), 2PC causes severe throughput collapse, holding database locks across continents, creating deadlock abort storms, and stalling during coordinator failures.

The paradigm shift that completely eradicates the need for distributed Two-Phase Commit is Deterministic Transaction Execution, pioneered by Alexander Thomson and Daniel Abadi in the Calvin Protocol (and commercially realized in engines like FaunaDB).

By shifting consensus to an independent Sequencer Layer *before* transactions are executed, Calvin guarantees that every replica across the world executes the exact same sequence of deterministic state transitions. This unlocks active-active multi-region writes without cross-datacenter locking during execution.

In this architectural deep dive, we examine why Two-Phase Commit fails at scale, dissect the three layers of the Calvin architecture, compare Calvin with Google Spanner and CockroachDB, and build a working deterministic database engine in Python.

The Achilles' Heel of Distributed ACID: Two-Phase Commit (2PC)

In a conventional non-deterministic distributed database (such as CockroachDB, TiDB, or Spanner without atomic clocks), a distributed transaction spanning multiple shards or geographic regions executes as follows:

sequenceDiagram
    autonumber
    participant Client
    participant Coordinator (Region US-East)
    participant Shard 1 (Region US-West)
    participant Shard 2 (Region EU-West)

    Client->>Coordinator: Begin Transaction & Execute Reads/Writes
    Coordinator->>Shard 1: Acquire 2PL Row Locks (Lock X)
    Coordinator->>Shard 2: Acquire 2PL Row Locks (Lock Y)
    Note over Coordinator,Shard 2: Phase 1: Prepare (WAN Round-Trip ~100ms)
    Coordinator->>Shard 1: PREPARE?
    Coordinator->>Shard 2: PREPARE?
    Shard 1-->>Coordinator: VOTE_YES
    Shard 2-->>Coordinator: VOTE_YES
    Note over Coordinator,Shard 2: Phase 2: Commit (WAN Round-Trip ~100ms)
    Coordinator->>Shard 1: COMMIT & Release Locks
    Coordinator->>Shard 2: COMMIT & Release Locks
    Coordinator-->>Client: Transaction Committed (Total Duration: 200ms+)

Why 2PC Destroys Scalability

1. Extended Lock Duration: In 2PL + 2PC, physical row locks are held across multiple cross-datacenter network hops. If two transactions contend for the same record across continents, the second transaction is blocked for 150–300 milliseconds. At 300ms per lock, maximum theoretical throughput on that record drops to just ~3 transactions per second!
2. Blocking Coordinator Failures: If the transaction coordinator crashes after shards vote VOTE_YES in Phase 1, participating shards cannot decide on their own whether to commit or abort. They must keep all row locks frozen indefinitely until the coordinator recovers.
3. Distributed Deadlocks & Abort Cascades: When multiple concurrent cross-region transactions acquire locks in different orders, distributed deadlocks occur, requiring expensive deadlock detection algorithms and triggering high abort rates.

The Deterministic Paradigm: Order First, Execute Later

The core thesis of the Calvin Protocol is simple yet revolutionary:

The Deterministic Thesis: If two or more independent replicas start from the exact same initial state and execute the exact same sequence of deterministic transactions in the exact same order, they are mathematically guaranteed to arrive at the exact same final state—with ZERO coordination or locking between them during execution.

In traditional databases, non-determinism (due to OS thread scheduling, arbitrary lock acquisition timing, network packet arrivals, and dynamic aborts) forces nodes to coordinate at commit time via 2PC. Calvin eliminates non-determinism by decoupling Consensus (Ordering) from Execution.

flowchart TD
    subgraph Traditional 2PC Model
        T_Exec[Execute Transactions Interactively] --> T_Lock[Acquire Distributed Locks across WAN]
        T_Lock --> T_2PC[Two-Phase Commit Coordination]
        T_2PC --> T_Done[Commit or Abort]
    end

    subgraph Calvin Deterministic Model
        C_Seq[1. Global Paxos Sequencer: Order Transactions in Batches] --> C_Sched[2. Deterministic Scheduler: Analyze Read/Write Sets]
        C_Sched --> C_Exec[3. Parallel Lock-Free Local Execution]
        C_Exec --> C_Done[Deterministic Commit (NO 2PC!)]
    end

    style T_2PC fill:#e11d48,stroke:#fda4af,color:#fff
    style T_Lock fill:#e11d48,stroke:#fda4af,color:#fff
    style C_Seq fill:#0284c7,stroke:#38bdf8,color:#fff
    style C_Sched fill:#7c3aed,stroke:#c084fc,color:#fff
    style C_Exec fill:#059669,stroke:#6ee7b7,color:#fff

The Three Layers of Calvin Architecture

A Calvin-based distributed database is structured into three clean, decoupled architectural layers:

flowchart LR
    subgraph Layer 1: Sequencer Layer
        S1[Node Sequencer A] -->|Batch Epochs| Paxos[Global Paxos / Raft Consensus]
        S2[Node Sequencer B] -->|Batch Epochs| Paxos
        Paxos --> GlobalLog[Global Ordered Transaction Sequence Log]
    end

    subgraph Layer 2: Scheduler Layer
        GlobalLog --> DLM1[Deterministic Lock Manager Shard 1]
        GlobalLog --> DLM2[Deterministic Lock Manager Shard 2]
    end

    subgraph Layer 3: Execution Layer
        DLM1 --> Exec1[Worker Threads: Storage Shard 1]
        DLM2 --> Exec2[Worker Threads: Storage Shard 2]
    end

    style Paxos fill:#0284c7,stroke:#38bdf8,color:#fff
    style GlobalLog fill:#7c3aed,stroke:#c084fc,color:#fff
    style Exec1 fill:#059669,stroke:#6ee7b7,color:#fff
    style Exec2 fill:#059669,stroke:#6ee7b7,color:#fff

1. The Sequencer Layer (Agreement on Sequence)

• Incoming client transactions are collected by local sequencers and grouped into Epoch Batches (typically 5ms to 10ms windows).
• A global consensus protocol (such as Multi-Paxos or Raft) replicates the epoch batches across all storage replicas and establishes a single, unambiguous global transaction order.
• Crucial Difference: Consensus happens in batch on raw transactions before any locks are acquired, amortizing WAN consensus latency across thousands of transactions simultaneously.

2. The Scheduler Layer (Deterministic Lock Management)

• Every storage node receives the globally ordered transaction log.
• Before executing a transaction, the Deterministic Lock Manager (DLM) inspects the transaction's declared Read Set and Write Set.
• Locks are granted strictly in the order transactions appear in the global log. Because the lock order is identical on every node, deadlocks are mathematically impossible. No deadlock detection or abort mechanisms are needed!

3. The Execution Layer (Zero-Lock Distributed Storage)

• Worker threads execute transactions as soon as their local locks are acquired.
• If a transaction spans multiple shards, the node holding the required read record simply sends a single one-way message containing the data to the node executing the write.
• Once a transaction finishes execution, it commits locally immediately. There is no second phase, no voting, and no coordinator rollback!

Handling Dynamic Transactions (OGRE & Pre-processing)

One common critique of deterministic databases is: "What if the read/write set cannot be known in advance (e.g., UPDATE users SET balance = balance - 10 WHERE account_id = (SELECT id FROM accounts WHERE name = 'Alice'))?"

Calvin solves dynamic transactions through Two-Phase Pre-processing (Optimistic Gather / OGRE):

1. Phase 1 (Reconnaissance Read): An initial, non-blocking snapshot read is executed against local storage to determine the specific record keys (e.g., resolving Alice to account_id = 49201).
2. Phase 2 (Deterministic Submission): The transaction is submitted to the sequencer with its concrete, resolved read/write key sets. If the underlying data changed between reconnaissance and execution, the transaction deterministically aborts and retries.

Furthermore, non-deterministic functions like NOW() or RAND() are evaluated once at the sequencer and embedded into the transaction payload as static deterministic values.

Calvin vs. Spanner vs. CockroachDB: Architectural Matrix

Hands-On Python Implementation: A Deterministic Calvin-Style Engine

Let's build a complete, runnable simulation of a Deterministic Distributed Database Engine in Python, demonstrating sequencer batching, deterministic dependency scheduling, and multi-replica execution with zero 2PC:

from dataclasses import dataclass, field
from typing import Callable, Dict, List, Set, Any
import time

@dataclass
class Transaction:
    txn_id: int
    read_set: Set[str]
    write_set: Set[str]
    logic: Callable[[Dict[str, Any]], Dict[str, Any]]
    result: Dict[str, Any] = field(default_factory=dict)

class DeterministicScheduler:
    """Schedules and executes transactions deterministically on local storage."""
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.storage: Dict[str, Any] = {}
        self.executed_txns: List[int] = []

    def execute_batch(self, ordered_txns: List[Transaction]):
        """Executes a batch of transactions in strict global sequence order."""
        for txn in ordered_txns:
            # 1. Read phase: Gather all required keys deterministically
            local_reads = {k: self.storage.get(k, 0) for k in txn.read_set}

            # 2. Execute deterministic application logic
            writes = txn.logic(local_reads)

            # 3. Write phase: Apply mutations directly without 2PC
            for k, v in writes.items():
                self.storage[k] = v

            txn.result = writes
            self.executed_txns.append(txn.txn_id)

class CalvinCluster:
    """Simulates a globally distributed Calvin database cluster."""
    def __init__(self, region_names: List[str]):
        self.global_txn_log: List[Transaction] = []
        self.replicas: Dict[str, DeterministicScheduler] = {
            r: DeterministicScheduler(node_id=r) for r in region_names
        }
        self.next_txn_id = 1

    def submit_transaction(self, read_set: Set[str], write_set: Set[str],
                           logic: Callable[[Dict[str, Any]], Dict[str, Any]]) -> Transaction:
        txn = Transaction(
            txn_id=self.next_txn_id,
            read_set=read_set,
            write_set=write_set,
            logic=logic
        )
        self.next_txn_id += 1
        # In real Calvin, Paxos orders transactions into the global log here
        self.global_txn_log.append(txn)
        return txn

    def run_epoch(self):
        """Dispatches the globally sequenced log to all independent regional replicas."""
        print(f"\n--- DISPATCHING EPOCH OF {len(self.global_txn_log)} ORDERED TXNS TO REPLICAS ---")
        for region_name, replica in self.replicas.items():
            replica.execute_batch(self.global_txn_log)
            print(f"[{region_name} Replica] State: {replica.storage}")

        # Clear processed epoch
        self.global_txn_log.clear()

# --- Simulation Demo ---
if __name__ == "__main__":
    # Initialize 3 global regions: US-East, EU-Central, AP-Southeast
    cluster = CalvinCluster(["US-East", "EU-Central", "AP-Southeast"])

    # Initialize initial bank balances on all replicas
    for r in cluster.replicas.values():
        r.storage["account_alice"] = 1000
        r.storage["account_bob"] = 500
        r.storage["account_charlie"] = 300

    print("Initial Balances: Alice=$1000, Bob=$500, Charlie=$300")

    # 1. Transaction 1 (From US): Alice transfers $200 to Bob
    def transfer_alice_to_bob(reads):
        alice_bal = reads["account_alice"] - 200
        bob_bal = reads["account_bob"] + 200
        return {"account_alice": alice_bal, "account_bob": bob_bal}

    cluster.submit_transaction(
        read_set={"account_alice", "account_bob"},
        write_set={"account_alice", "account_bob"},
        logic=transfer_alice_to_bob
    )

    # 2. Transaction 2 (From EU): Bob transfers $100 to Charlie (Depends on Txn 1)
    def transfer_bob_to_charlie(reads):
        bob_bal = reads["account_bob"] - 100
        charlie_bal = reads["account_charlie"] + 100
        return {"account_bob": bob_bal, "account_charlie": charlie_bal}

    cluster.submit_transaction(
        read_set={"account_bob", "account_charlie"},
        write_set={"account_bob", "account_charlie"},
        logic=transfer_bob_to_charlie
    )

    # 3. Transaction 3 (From Asia): Charlie transfers $50 to Alice
    def transfer_charlie_to_alice(reads):
        charlie_bal = reads["account_charlie"] - 50
        alice_bal = reads["account_alice"] + 50
        return {"account_charlie": charlie_bal, "account_alice": alice_bal}

    cluster.submit_transaction(
        read_set={"account_charlie", "account_alice"},
        write_set={"account_charlie", "account_alice"},
        logic=transfer_charlie_to_alice
    )

    # Run deterministic epoch across all independent replicas
    cluster.run_epoch()

    # Verify that all 3 replicas arrived at the exact same state without 2PC!
    states = [replica.storage for replica in cluster.replicas.values()]
    all_identical = all(s == states[0] for s in states)
    print(f"\nConsensus Verification: All Global Replicas Identical without 2PC? {all_identical}")
    print(f"Final Account Balances: {states[0]}")

Execution Output of Deterministic Cluster

Initial Balances: Alice=$1000, Bob=$500, Charlie=$300

--- DISPATCHING EPOCH OF 3 ORDERED TXNS TO REPLICAS ---
[US-East Replica] State: {'account_alice': 850, 'account_bob': 600, 'account_charlie': 350}
[EU-Central Replica] State: {'account_alice': 850, 'account_bob': 600, 'account_charlie': 350}
[AP-Southeast Replica] State: {'account_alice': 850, 'account_bob': 600, 'account_charlie': 350}

Consensus Verification: All Global Replicas Identical without 2PC? True
Final Account Balances: {'account_alice': 850, 'account_bob': 600, 'account_charlie': 350}

Key Takeaways for Distributed Systems Architects

• Decouple Consensus from Execution: Traditional databases coordinate via 2PC after execution. Calvin achieves consensus on transaction sequence before execution, eliminating runtime lock contention.
• No Two-Phase Commit Overhead: By guaranteeing deterministic state transitions, replicas commit transactions locally without distributed commit protocols or network round-trips during execution.
• Deadlocks are Impossible: Locks are allocated in strict sequence order according to the global log, completely eradicating distributed deadlock abort storms.
• Active-Active Global Scale: Calvin allows true active-active multi-region deployments with high throughput even under severe write contention.