Distributed SQL, Multi-Raft Consensus & Two-Phase Commit

1. Problem and Motivation

For more than two decades, system architects were forced to choose between two mutually exclusive database paradigms:

+---------------------------------------------------------------------------------------------------+
|                                  THE CLASSICAL DATABASE PARADOX                                   |
+---------------------------------------------------+-----------------------------------------------+
| Traditional Relational DBMS (PostgreSQL / MySQL)  | Distributed NoSQL (Cassandra / DynamoDB)      |
+---------------------------------------------------+-----------------------------------------------+
| [x] Strong ACID Transactions (Multi-Row / Table)  | [ ] No Multi-Row ACID Transactions            |
| [x] Rich SQL Querying, Secondary Indexes & Joins  | [ ] Limited Key-Value / Partition Key Queries |
| [ ] Vertical Scale Ceiling (Single Write Master)  | [x] Linear Horizontal Scale (100s of Nodes)   |
| [ ] Manual Sharding Nightmare (Vitess, Citus)     | [x] Automated Sharding & High Availability    |
+---------------------------------------------------+-----------------------------------------------+

When relational databases grow beyond a single machine’s storage or write throughput, engineering teams resort to application-level sharding. This introduces immense complexity: cross-shard transactions require fragile application-level Two-Phase Commit, schema migrations must be applied across hundreds of independent shards, and secondary indexes become non-atomic and stale.

Distributed SQL (NewSQL)—pioneered by Google Spanner (2012) and open-source systems like TiDB/TiKV and CockroachDB—bridges this divide. It delivers standard SQL schemas, foreign keys, secondary indexes, and true Serializable ACID transactions combined with the horizontal scaling, automatic sharding, and multi-region fault tolerance of NoSQL engines.

2. Mental Model & Architectural Overview

A modern Distributed SQL database is architected in decoupled, modular tiers:

SYSTEM ARCHITECTURE BLUEPRINTDISTRIBUTED SYSTEMS

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 3

1. Ordered Global Keyspace & Independent Raft Consensus Groups

Establishes the ingress and structural boundary across Ordered Global Keyspace & Independent Raft Consensus Groups: Continuous range: [-inf, +inf); Sliced into 64MB - 96MB continuous Ranges; Table rows & indexes mapped to keys Enables ultra-fast range scans in SQL!.

Architectural Invariant: Range 2 [m..z): Leader Node B, Peers A, D Single physical server hosts 1,000s of Rafts! Failure of Range 1 never affects Range 2.

ARCHITECTURAL FLOW · STEP 2 OF 3

2. Phase 1: Prewrite (Percolator Model) & Phase 2: Atomic Primary Commit

Coordinates core data-plane execution and state transitions through Phase 1: Prewrite (Percolator Model) & Phase 2: Atomic Primary Commit: SQL Coordinator coordinates transaction; Selects Key 1 on Range 1 as PRIMARY; Writes Primary Lock via Raft Group 1.

Architectural Invariant: Raft Group 1 commits log record THE ATOMIC COMMIT MOMENT: Once Primary is committed, the entire transaction is legally committed permanently!.

ARCHITECTURAL FLOW · STEP 3 OF 3

3. TrueTime Uncertainty Window (2ε) & Commit Wait Rule

Enforces durability, state convergence, and output verification across TrueTime Uncertainty Window (2ε) & Commit Wait Rule: GPS + Atomic Clocks across datacenters; TT.now() returns: [earliest, latest]; Clock uncertainty: ε ≈ 1ms - 7ms.

Architectural Invariant: Real time t is guaranteed inside interval Solves distributed clock drift mathematically.

The Two-Tier Architecture:

1. Stateless Compute Tier (SQL Layer - e.g., TiDB / CockroachDB SQL):
   • Accepts MySQL or PostgreSQL wire-protocol connections.
   • Parses, optimizes, and compiles SQL queries into distributed physical execution plans (Distributed Hash Joins, Index Scans).
   • Coordinates distributed Two-Phase Commit (2PC) transactions across storage nodes.
   • Holds zero persistent disk state: can scale horizontally behind a standard Layer 4 load balancer.
2. Distributed Storage Tier (Multi-Raft Engine - e.g., TiKV / CockroachDB Pebble):
   • Organizes the entire cluster’s data as a single global, monotonically sorted byte-array keyspace: $[-\infty ,+\infty )$.
   • Automatically slices the keyspace into thousands of continuous 64MB–96MB partitions called Ranges (or Regions).
   • Each Range is an independent Raft consensus group replicated across 3 or 5 storage nodes.
   • High-throughput local storage is backed by an embedded LSM-Tree or B-Tree engine (RocksDB or Pebble).

3. Mathematical Foundations & Formal Mechanisms

3.1 Global Keyspace Mapping & Range Splitting

In a relational database, table rows and secondary indexes are encoded into sorted byte keys:

The keyspace is partitioned into $M$ disjoint, contiguous ranges:

When a Range exceeds a threshold $S_{\text{max}}$ (e.g., $96\text{ MB}$):

1. The Raft leader identifies the median key $K_{\text{split}}$ where bytes are roughly balanced ($48\text{ MB}$ each).
2. The leader proposes a Raft log entry: $\text{SplitOp}(K_{\text{split}})$.
3. Upon quorum commitment, the original range is atomically converted into two sub-ranges:

3.2 Spanner TrueTime & The External Consistency Theorem

Google Spanner achieves External Consistency (Strict Serializability / Linearizability) globally using the TrueTime API:

Where:

• $ϵ$ is the dynamically tracked worst-case clock uncertainty bound (typically $1\text{ ms}\le ϵ\le 7\text{ ms}$).
• The true absolute physical UTC time $t_{\text{absolute}}$ is guaranteed to fall within the interval: $t_{\text{earliest}}\le t_{\text{absolute}}\le t_{\text{latest}}$.

The External Consistency Invariant:

If transaction $T_2$ is initiated after transaction $T_1$ commits in real world time:

Where $s_{Ti}$ is the assigned commit timestamp of transaction $T_i$.

The Commit Wait Rule:

1. The transaction coordinator picks a proposed commit timestamp $s\ge t_{\text{latest}}$.
2. Commit Wait: The coordinator guarantees $s$ has passed in real time by sleeping until:

3. The sleep duration is at least $2ϵ$.
4. Once the sleep finishes, locks are released and success is returned to the client.

Because $T_1$ waits until $s_{T1}$ is strictly in the past, any subsequent transaction $T_2$ is mathematically guaranteed to pick $s_{T2}>s_{T1}$, eliminating the need for cross-datacenter communication during reads!

4. The Google Percolator Distributed 2PC Model

Most open-source Distributed SQL engines (TiDB/TiKV) implement the Google Percolator distributed transaction model layered over Multi-Raft.

Column Families per Key:

For every key $K$, the storage engine maintains three internal columns:

• data:K_startTS: Stores raw row values written at transaction start timestamp.
• lock:K: Stores active lock records indicating an in-flight transaction.
• write:K_commitTS: Stores the committed pointer linking commit timestamp to startTS.

====================================================================================================
THE PERCOLATOR TWO-PHASE COMMIT PROTOCOL
====================================================================================================

Phase 1: Prewrite (Acquire Locks)
1. Coordinator requests Start Timestamp (startTS) from central Timestamp Oracle (TSO).
2. Selects one key as the PRIMARY KEY (Key 1); remaining keys become SECONDARIES (Key 2..N).
3. For Key 1 (Primary):
   - Check write conflict: ensure no write record exists with commitTS >= startTS.
   - Check lock conflict: ensure no lock record exists at any timestamp.
   - Write lock: lock:Key1 = { primary: Key1, startTS: startTS } via Raft Group 1.
4. For Secondary Keys (Key 2..N):
   - Check conflicts.
   - Write lock: lock:Key2 = { primary: Key1, startTS: startTS } (Pointers back to Primary!)

----------------------------------------------------------------------------------------------------

Phase 2: Commit (The Atomic Moment)
1. Coordinator requests Commit Timestamp (commitTS) from TSO.
2. Commit PRIMARY KEY:
   - Atomically remove lock:Key1 and write write:Key1_commitTS = startTS via Raft Group 1.
   - THIS IS THE SINGLE ATOMIC POINT OF COMMIT!
   - If this succeeds, the transaction is 100% committed permanently.
3. Commit SECONDARY KEYS:
   - Asynchronously remove lock:Key2 and write write:Key2_commitTS = startTS.
   - Even if the coordinator crashes right now, concurrent readers encountering Key 2's lock
     will inspect Key 1, discover it is committed, and roll Key 2 forward!
====================================================================================================

5. Concrete Production Implementation

Below is a complete, production-grade TypeScript implementation simulating:

1. Multi-Raft Range partitioning with automatic range splits.
2. The Percolator Distributed Two-Phase Commit transaction coordinator.
3. Lock conflict detection, atomic primary commit, and secondary lock roll-forward.

/**
 * Distributed SQL & Multi-Raft Architecture Engine
 * Demonstrates:
 * - Multi-Raft Range Partitioning & Splitting
 * - Google Percolator-style Distributed 2PC
 * - Primary Lock Commit & Asynchronous Roll-Forward
 */

export interface StorageRecord {
  value: string;
  startTS: number;
}

export interface LockRecord {
  primaryKey: string;
  startTS: number;
  ttl: number;
}

export interface WriteRecord {
  startTS: number;
  commitTS: number;
}

/**
 * An individual Range acting as an independent Raft consensus group
 */
export class RaftRange {
  public readonly id: string;
  public startKey: string;
  public endKey: string;
  public dataCF: Map<string, StorageRecord> = new Map(); // data:key_startTS
  public lockCF: Map<string, LockRecord> = new Map();     // lock:key
  public writeCF: Map<string, WriteRecord> = new Map();   // write:key_commitTS
  public byteSize = 0;

  constructor(id: string, startKey: string, endKey: string) {
    this.id = id;
    this.startKey = startKey;
    this.endKey = endKey;
  }

  public contains(key: string): boolean {
    return key >= this.startKey && (this.endKey === "" || key < this.endKey);
  }

  public prewrite(key: string, value: string, primaryKey: string, startTS: number): boolean {
    // 1. Conflict check: Has any transaction committed after our startTS?
    for (const [wKey, wRec] of this.writeCF.entries()) {
      if (wKey.startsWith(`${key}_`) && wRec.commitTS >= startTS) {
        return false; // Write-write conflict!
      }
    }

    // 2. Lock check: Is the key locked by another transaction?
    if (this.lockCF.has(key)) {
      return false; // Lock contention!
    }

    // 3. Write data & lock
    this.dataCF.set(`${key}_${startTS}`, { value, startTS });
    this.lockCF.set(key, { primaryKey, startTS, ttl: Date.now() + 5000 });
    this.byteSize += value.length + key.length;

    return true;
  }

  public commitKey(key: string, startTS: number, commitTS: number): boolean {
    const lock = this.lockCF.get(key);
    if (!lock || lock.startTS !== startTS) {
      return false;
    }

    // Remove lock and add committed write entry atomically via Raft log
    this.lockCF.delete(key);
    this.writeCF.set(`${key}_${commitTS}`, { startTS, commitTS });
    return true;
  }

  public readSnapshot(key: string, readTS: number): string | null {
    // Find latest committed write with commitTS <= readTS
    let latestCommit: WriteRecord | null = null;

    for (const [wKey, wRec] of this.writeCF.entries()) {
      if (wKey.startsWith(`${key}_`) && wRec.commitTS <= readTS) {
        if (!latestCommit || wRec.commitTS > latestCommit.commitTS) {
          latestCommit = wRec;
        }
      }
    }

    if (!latestCommit) return null;
    const data = this.dataCF.get(`${key}_${latestCommit.startTS}`);
    return data ? data.value : null;
  }
}

/**
 * Distributed SQL Multi-Raft Storage Cluster
 */
export class DistributedStorageCluster {
  private ranges: RaftRange[] = [];
  private tsoClock = 100;

  constructor() {
    // Initial cluster starts with a single unbounded Range
    this.ranges.push(new RaftRange("range-0", "", ""));
  }

  public getTimestamp(): number {
    return ++this.tsoClock;
  }

  public getRangeForKey(key: string): RaftRange {
    const range = this.ranges.find((r) => r.contains(key));
    if (!range) throw new Error(`Routing error: No range covers key '${key}'`);
    return range;
  }

  /**
   * Dynamically split range when it exceeds size threshold
   */
  public splitRange(rangeId: string, splitKey: string): void {
    const index = this.ranges.findIndex((r) => r.id === rangeId);
    if (index === -1) return;

    const oldRange = this.ranges[index];
    const newRangeId = `range-${Date.now()}`;
    const newRange = new RaftRange(newRangeId, splitKey, oldRange.endKey);
    oldRange.endKey = splitKey;

    // Migrate keys
    for (const [k, v] of oldRange.dataCF.entries()) {
      const bareKey = k.slice(0, k.lastIndexOf("_"));
      if (newRange.contains(bareKey)) {
        newRange.dataCF.set(k, v);
        oldRange.dataCF.delete(k);
      }
    }
    for (const [k, v] of oldRange.writeCF.entries()) {
      const bareKey = k.slice(0, k.lastIndexOf("_"));
      if (newRange.contains(bareKey)) {
        newRange.writeCF.set(k, v);
        oldRange.writeCF.delete(k);
      }
    }
    for (const [k, v] of oldRange.lockCF.entries()) {
      if (newRange.contains(k)) {
        newRange.lockCF.set(k, v);
        oldRange.lockCF.delete(k);
      }
    }

    this.ranges.splice(index + 1, 0, newRange);
  }

  public getRanges(): RaftRange[] {
    return this.ranges;
  }
}

/**
 * Distributed 2PC Transaction Coordinator (Percolator Protocol)
 */
export class DistributedTxnCoordinator {
  private cluster: DistributedStorageCluster;

  constructor(cluster: DistributedStorageCluster) {
    this.cluster = cluster;
  }

  public async executeTransaction(mutations: Map<string, string>): Promise<{ committed: boolean; commitTS?: number; error?: string }> {
    const entries = Array.from(mutations.entries());
    if (entries.length === 0) return { committed: true };

    const startTS = this.cluster.getTimestamp();
    const primaryKey = entries[0][0];

    // Phase 1: Prewrite All Keys (Primary first, then secondaries)
    for (const [key, val] of entries) {
      const range = this.cluster.getRangeForKey(key);
      const ok = range.prewrite(key, val, primaryKey, startTS);
      if (!ok) {
        // Rollback any acquired locks on failure
        this.rollback(entries, startTS);
        return { committed: false, error: `Conflict during prewrite on key '${key}'` };
      }
    }

    // Phase 2: Commit Primary Key
    const commitTS = this.cluster.getTimestamp();
    const primaryRange = this.cluster.getRangeForKey(primaryKey);
    const primaryOk = primaryRange.commitKey(primaryKey, startTS, commitTS);

    if (!primaryOk) {
      this.rollback(entries, startTS);
      return { committed: false, error: "Failed to commit primary key" };
    }

    // ATOMIC POINT OF COMMIT ACHIEVED!
    // Asynchronously roll forward secondary keys
    for (let i = 1; i < entries.length; i++) {
      const [secKey] = entries[i];
      const secRange = this.cluster.getRangeForKey(secKey);
      secRange.commitKey(secKey, startTS, commitTS);
    }

    return { committed: true, commitTS };
  }

  private rollback(entries: Array<[string, string]>, startTS: number): void {
    for (const [key] of entries) {
      const range = this.cluster.getRangeForKey(key);
      const lock = range.lockCF.get(key);
      if (lock && lock.startTS === startTS) {
        range.lockCF.delete(key);
      }
    }
  }
}

async function runDemo(): Promise<void> {
  console.log("=== Distributed SQL: Multi-Raft & Percolator 2PC Distributed Transactions ===");
  const cluster = new DistributedStorageCluster();
  const coordinator = new DistributedTxnCoordinator(cluster);
  console.log("Initialized DistributedStorageCluster with Multi-Raft key ranges");
  const mutations = new Map<string, string>([
    ["acc_1001", "balance: 900"],
    ["acc_2002", "balance: 1100"]
  ]);
  const res = await coordinator.executeTransaction(mutations);
  console.log(`Percolator 2PC Transaction execution: ${res.committed ? 'COMMITTED' : 'ABORTED'} at commitTS=${res.commitTS}`);
  const range1 = cluster.getRangeForKey("acc_1001");
  const readVal = range1.readSnapshot("acc_1001", (res.commitTS ?? 200) + 1);
  console.log(`Snapshot read at commitTS+1: acc_1001 = '${readVal}'`);
}

runDemo();

Simulation Execution Output & State Transitions

=== Distributed SQL: Multi-Raft & Percolator 2PC Distributed Transactions ===
Initialized DistributedStorageCluster with Multi-Raft key ranges
Percolator 2PC Transaction execution: COMMITTED at commitTS=102
Snapshot read at commitTS+1: acc_1001 = 'balance: 900'

6. Real-World Failure Modes, Edge Cases & Operational War Stories

War Story 1: The Monotonic Key Range Split Storm

Incident:
A high-throughput e-commerce application migrated from MySQL to a 30-node Distributed SQL cluster.
During a flash-sale marketing campaign, write latency on the `orders` table suddenly degraded from
4 ms to 820 ms, and nodes began reporting Raft election timeouts.

Investigation:
The table schema used a standard auto-incrementing integer:
CREATE TABLE orders (
    order_id BIGINT AUTO_INCREMENT PRIMARY KEY,
    user_id BIGINT,
    amount DECIMAL
);

Root Cause:
Because `order_id` values increased monotonically, 100% of incoming `INSERT` transactions mapped to the
single tail Range covering keys `[CurrentMax, +inf)`. A single Raft leader node processed every insert!
The range continuously exceeded 96 MB, triggering endless Range Splits and Raft leader elections.
While 29 nodes sat completely idle (0% CPU), 1 node melted at 100% CPU.

Remediation:
Switched the primary key to a hash-distributed prefix:
`order_id BIGINT PRIMARY KEY /*T![auto_random(5)]*/`
The top 5 bits of the ID were randomized, scattering concurrent inserts evenly across 32 independent
Multi-Raft ranges and scaling cluster throughput to 180,000 inserts/sec.

War Story 2: Cross-Region 2PC Latency Explosion

Incident:
A multinational enterprise deployed a Distributed SQL database spanned across three cloud regions:
`us-east` (Northern Virginia), `eu-central` (Frankfurt), and `ap-southeast` (Singapore).
A financial checkout query that touched an inventory record in Singapore and a user balance in Frankfurt
took 680 ms to commit.

Root Cause:
Distributed Two-Phase Commit requires cross-network round-trips:
1. Prewrite Primary in Frankfurt (local: 2ms).
2. Prewrite Secondary in Singapore (cross-globe RTT: 160ms).
3. Raft log consensus replication within Singapore group (20ms).
4. Commit Primary in Frankfurt (160ms RTT).
Total serial network travel: $160\text{ms} + 160\text{ms} + \dots \approx 640\text{ms}$.

Remediation:
Implemented **Data Locality Partitioning**:
Configured table row placement policies (`PLACEMENT POLICY = 'region_eu'`) so related customer balances
and regional inventory records reside on the same local Multi-Raft ranges, keeping 2PC local to single regions.

War Story 3: Orphaned Secondary Locks Stall Analytical Readers

Incident:
During nightly ETL batch ingestion, an analytical dashboard query:
`SELECT SUM(balance) FROM accounts;`
hung indefinitely and eventually timed out with `TransactionLockTimeoutException`.

Root Cause:
A background batch worker executing an update crashed midway through Phase 2:
It committed the Primary Key on Range 1, but crashed before rolling forward the Secondary Lock on Range 42.
The analytical reader encountered the active `lock:account_42` record.
Because the lock resolution timeout had not yet expired, the reader waited for the dead coordinator.

Remediation:
Implemented **Percolator Lazy Lock Resolution**:
When a reader encounters a lock older than TTL (5 seconds), it does not wait: it reads the `primaryKey` pointer
stored inside the lock record. It inspects the Primary Key in Range 1:
- If Primary is COMMITTED: The reader rolls the secondary lock forward itself and reads the new value!
- If Primary is ROLLBACK or MISSING: The reader rolls the orphan lock back.

7. Benchmarks, Capacity Planning & Sizing Math

7.1 Distributed SQL Transaction Overhead Comparison

+---------------------------------------------------------------------------------------------------+
|                            TRANSACTION COMMIT LATENCY BENCHMARK                                   |
+--------------------------+---------------------+-------------------+------------------------------+
| Architecture             | 1-Row Read Latency  | 1-Row Write Lat   | Cross-Node 2PC Write Lat     |
+--------------------------+---------------------+-------------------+------------------------------+
| Local Single-Node MySQL  | 0.3 ms              | 0.8 ms            | N/A (Single node)            |
| Multi-Raft Single Range  | 0.8 ms (Leader read)| 3.5 ms (Raft ack) | N/A (Single Raft Group)      |
| Distributed 2PC (Same DC)| 1.2 ms              | 4.2 ms            | 8.5 ms (Prewrite + Commit)   |
| Distributed 2PC (Cross-DC| 2.5 ms              | 18.0 ms           | 65.0 - 180.0 ms (WAN RTT)    |
| Spanner TrueTime Wait    | 1.0 ms (Snapshot)   | 14.0 ms (Local)   | 28.0 ms (Includes 2ε sleep)  |
+--------------------------+---------------------+-------------------+------------------------------+

7.2 Multi-Raft Range Sizing Math

In a 100 TB production storage cluster with target range size $S_{\text{range}}=96\text{ MB}$:

With $3\times$ replication across 100 physical storage servers:

Raft Heartbeat Scaling Challenge: If 31,250 Raft groups send individual heartbeats every 200 ms: $$\text{Heartbeats/sec}=31,250\times 5=156,250\text{ msgs/sec per server!}$$

Architectural Solution: Multi-Raft engines implement Heartbeat Batching: a single physical TCP message between Node A and Node B batches the heartbeats for all 10,000 co-located Raft groups, slashing network packet overhead by 99.9%.

8. Comparative Trade-off Matrix

+---------------------------------------------------------------------------------------------------+
|                                 NEWSQL VS TRADITIONAL ARCHITECTURES                               |
+----------------------+--------------------+--------------------+----------------------------------+
| Dimension            | Single-Node RDBMS  | Sharded MySQL (Vitess)| Distributed SQL (TiDB/Spanner) |
+----------------------+--------------------+--------------------+----------------------------------+
| Write Scalability    | Vertical only      | High (Per-shard)   | Elastic (Multi-Raft auto-split)  |
| Cross-Shard ACID     | Local only         | Complex 2PC        | Native transparent distributed   |
|                      |                    | application burden | 2PC (Percolator / Spanner)       |
| Schema Migrations    | Fast (single host) | Painful (N shards) | Online schema changes (Google F1)|
| Operational Overhead | Low                | High (Sharding key | Zero manual re-sharding          |
|                      |                    | planning, routers) | (Handled by storage engine)      |
+----------------------+--------------------+--------------------+----------------------------------+

9. Verification, Testing & Failure Injection

9.1 Verification Checklist & Jepsen Tests

• Multi-Raft Partitioning Test: Sever network connections to a random node during peak multi-row transactions. Verify that unaffected ranges continue serving reads and writes with zero interruption.
• Range Split Concurrency: Inject high-rate concurrent writes directly across a range split boundary key. Confirm zero lost updates or torn writes.
• Orphaned Lock Resolution: Terminate the SQL coordinator process (kill -9) precisely between Prewrite and Commit phases. Verify that subsequent client queries lazily resolve or roll back orphan locks within 5 seconds.

10. Key Takeaways & Best Practices Checklist

+---------------------------------------------------------------------------------------------------+
|                                 DISTRIBUTED SQL CHECKLIST                                         |
+---------------------------------------------------------------------------------------------------+
| [x] Decouple compute from storage: Stateless SQL query nodes scale independently from Multi-Raft  |
|     storage ranges.                                                                               |
| [x] Never use monotonically increasing primary keys (auto-increment): Prepend hash bits or UUIDs |
|     to distribute write traffic evenly across all Multi-Raft ranges.                              |
| [x] Leverage Percolator's single atomic point of commit: Committing the Primary Key guarantees    |
|     transaction success, allowing secondary keys to roll forward asynchronously.                  |
| [x] In multi-region deployments, configure table placement policies to keep related transaction |
|     keys within the same regional Multi-Raft groups, avoiding cross-continent WAN 2PC latency.     |
| [x] Batch Multi-Raft heartbeats across co-located ranges to prevent network saturation.           |
+---------------------------------------------------------------------------------------------------+

11. Guided Design Exercises

Exercise 1: Multi-Raft Range Count, Heartbeat Coalescing & Split Threshold Math

A financial ledger deployed on TiKV/CockroachDB stores $240\text{ TB}$ of logical user data with a replication factor $\text{RF}=3$, target range size $S_{\text{range}}=96\text{ MB}$, across $N=120$ storage nodes. Each Raft group leader sends heartbeats every $T_{\text{hb}}=200\text{ ms}$.

1. Calculate the total number of Multi-Raft Ranges in the cluster, the total number of Range Replicas, and the average number of Range Replicas and Range Leaders hosted per node.
2. Calculate the raw unbatched Raft heartbeat RPC rate (messages/second) emitted per node if each Range leader sends independent heartbeats to its 2 followers.
3. Show how Multi-Raft Store-to-Store Heartbeat Coalescing reduces the per-node RPC rate when any two nodes share hundreds of Raft ranges, and compute the exact coalesced heartbeat packet rate per node in a 120-node cluster.

Exercise 2: Percolator 2PC Crash Recovery & Lazy Lock Resolution Trace

Transaction $T_1$ (startTS = 210) transfers $500 from "acc_1001" (hosted on Range A) to "acc_2002" (hosted on Range B). The coordinator designates "acc_1001" as the Primary Key and "acc_2002" as a Secondary Key.

1. Trace the exact contents of dataCF, lockCF, and writeCF on Range A and Range B immediately after Phase 1 (Prewrite) succeeds on both ranges.
2. Suppose the coordinator obtains commitTS = 215, commits the Primary Key "acc_1001" on Range A, and immediately crashes (SIGKILL) before sending the Phase 2 commit RPC to Range B. Trace what happens when a concurrent reader $T_2$ (readTS = 220) queries "acc_2002" on Range B.
3. Contrast that outcome with the case where the coordinator crashed before Range A’s Raft group committed "acc_1001"’s writeCF record, and explain why a concurrent reader rolling back "acc_1001"’s expired lock must also write a Rollback tombstone record at write:acc_1001_210.

Exercise 3: Hot-Range Write Contention & Spanner TrueTime Commit-Wait Sizing

A global payment table uses payment_id BIGINT AUTO_INCREMENT PRIMARY KEY and experiences $60,000\text{ inserts/sec}$ during a peak event, saturating a single TiKV/CockroachDB node at $100\%$ CPU while 119 other nodes sit at $<8\%$ CPU.

1. Redesign the primary key and secondary index schema (using AUTO_RANDOM(5) / hash-sharded buckets) so inserts distribute uniformly across at least $32$ Multi-Raft ranges without breaking user-level point queries.
2. If the cluster uses a TrueTime-style commit-wait protocol where clock uncertainty $ϵ$ degrades from $1.5\text{ ms}$ to $9\text{ ms}$ during an atomic clock reference failover, compute the minimum commit-wait sleep duration $2ϵ$ and quantify its impact on per-key lock-hold duration and maximum single-row update throughput ${\lambda }_{\text{row,max}}=\frac{1}{T_{\text{raft}}+2ϵ}$.

12. Authoritative References & Version Context

• Rebecca Taft et al. (2020), “CockroachDB: The Resilient Geo-Distributed SQL Database”, ACM SIGMOD ‘20 — doi.org/10.1145/3318464.3386134
• Dongxu Huang et al. (2020), “TiDB: A Raft-based HTAP Database”, VLDB ‘20 — doi.org/10.14778/3415478.3415535
• Peng and Dabek (2010), “Large-scale Incremental Processing Using Distributed Transactions and Notifications”, USENIX OSDI ‘10 (Google Percolator) — usenix.org
• Version Context: Distributed SQL invariants reflect CockroachDB 23.x / 24.x LTS (Multi-Raft Ranges) and TiKV 7.x (Multi-Raft Regions with Percolator 2PC).