Database Sharding, Partitioning & Distributed Transactions: 2PC vs. Sagas

1. The Core Engineering Challenge: The Limits of Vertical Scaling

When an application’s data volume or transaction throughput surges, engineering teams initially scale vertically (buying larger cloud instances with 128 vCPUs, 1 TB RAM, and enterprise NVMe RAID arrays).

However, vertical scaling inevitably hits hard physical and economic boundaries:

1. The Economic Cliff: Moving from a 64 GB cloud instance to a 1 TB high-memory instance increases costs exponentially, not linearly.
2. Hardware Ceilings: No single server exists with 500 terabytes of RAM or 1,000 physical CPU cores.
3. Database Connection & Mutex Bottlenecks: Relational database storage engines (such as MySQL InnoDB or PostgreSQL) suffer severe lock contention on buffer pool mutexes, transaction logs, and internal catalog locks when thousands of concurrent threads access a single operating system kernel.
4. Catastrophic Blast Radius: If your monolithic multi-terabyte database server suffers a kernel panic or drive array corruption, 100% of your global business operations halt instantly.

Horizontal Scaling: Sharding

To scale beyond the limits of a single machine, we deploy Horizontal Database Sharding (Partitioning):

The dataset is partitioned into $S$ mutually exclusive subsets, each hosted on an independent physical database server with its own CPU, RAM, and storage.

SYSTEM ARCHITECTURE BLUEPRINTDATABASES

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 3

1. Application Client Tier & Shard Router Tier (e.g. Vitess / Citus)

Establishes the ingress and structural boundary across Application Client Tier & Shard Router Tier (e.g. Vitess / Citus): Issues standard SQL queries; SELECT * FROM orders WHERE user_id=42; Parses SQL AST & extracts shard key.

Architectural Invariant: Zero cross-node network overhead.

ARCHITECTURAL FLOW · STEP 2 OF 3

2. Shard 1 Database (Node A) & Shard 2 Database (Node B)

Coordinates core data-plane execution and state transitions through Shard 1 Database (Node A) & Shard 2 Database (Node B): Partition: hash(user_id) % 3 == 0; Users: [3, 6, 9, 42...]; Local B-Tree Index: orders(user_id, id).

Architectural Invariant: Horizontal Scale: 3x storage & IOPS capacity.

ARCHITECTURAL FLOW · STEP 3 OF 3

3. Resharding: Consistent Hash Virtual Nodes & Two-Phase Commit (2PC) - Strict ACID

Enforces durability, state convergence, and output verification across Resharding: Consistent Hash Virtual Nodes & Two-Phase Commit (2PC) - Strict ACID: Atomic balance transfer between Shard 1 & 2; Phase 1: PREPARE; Coordinator -> Shard 1, 2: "Can you commit?" Shards lock rows, flush WAL, vote "YES".

Architectural Invariant: Emits completion event onto message broker. Failure Rollback Path: If Step 2 fails (e.g. account closed): Trigger Compensating Action on Shard 1: Refund Account (Reverse of Step 1) Non-blocking, highly resilient for cloud.

However, sharding imposes profound architectural trade-offs:

• Single-node relational ACID guarantees, foreign keys, and unique secondary indexes disappear across shard boundaries.
• Cross-shard queries require distributed network coordination.
• Transactions touching multiple shards require distributed consensus (Two-Phase Commit or Sagas).

Curriculum Bridge (Synthesizing Partitioning & Consensus into Application Schema & Query Routing): Building directly upon storage-layer partitioning (CORE-M19-001 & CORE-M19-002) and distributed transaction protocols (CORE-M22-001: 2PC, Percolator, Sagas & TCC), this lesson focuses on Application Data Modeling Across Paradigms (Relational 3NF, Document, Wide-Column, Graph & the ESR Index Rule), Co-Located Sharding & Broadcast Reference Tables, Scatter-Gather Query Routing (Vitess / Citus), and Zero-Downtime Monolith-to-Sharded Cutovers.

2. Mental Model: The Shard Query Routing Middleware

In a sharded architecture, application services should not maintain hardcoded database connection pools to hundreds of distinct shards. Instead, systems introduce an intermediate Shard Query Routing Tier (such as Vitess for MySQL, Citus for PostgreSQL, or custom application proxies):

+-------------------------------------------------------------------------------+
|                       SHARD ROUTING ENGINE ANATOMY                            |
+-------------------------------------------------------------------------------+
| Component           | Responsibility                                          |
+---------------------+---------------------------------------------------------+
| 1. SQL Parser       | Inspects AST to identify target tables & WHERE filters  |
| 2. Shard Resolver   | Extracts Shard Key value and computes target Shard ID   |
| 3. Query Planner    | Decides: Single-Shard Route vs. Multi-Shard Scatter     |
| 4. Connection Pool  | Multiplexes persistent connections to physical shards   |
| 5. Result Merger    | Performs streaming in-memory merge, sort, and LIMIT     |
+-------------------------------------------------------------------------------+

When an application executes:

SELECT * FROM orders WHERE customer_id = 42;

1. The Router identifies customer_id as the designated Shard Key.
2. It evaluates the partitioning hash function: $$\text{ShardID}=\text{Hash}(42)(\mathrm{mod}S)=\text{Shard }1$$
3. It dispatches the SQL query strictly to Shard 1. Shards 2 through $S$ remain completely untouched, achieving true $\mathcal{O}(1)$ horizontal scalability.

Foundational Bridge: Relational Normalization (3NF) vs. Access-Pattern Denormalization & The ESR Index Rule

Before partitioning a database across multiple nodes, architects must choose the right logical data model and single-node index layout:

1. Third Normal Form (3NF) vs. Access-Pattern Denormalization:
   • Relational 3NF: Every non-key attribute depends solely on the primary key (“the key, the whole key, and nothing but the key”). This eliminates update anomalies and storage duplication, making it ideal for OLTP systems where ad-hoc query predicates evolve frequently.
   • Access-Pattern-Driven Denormalization (Wide-Column / Document): At distributed scale, cross-shard JOIN operations incur $O(N)$ network fan-out. Wide-column stores (Cassandra, DynamoDB) and document stores model data around pre-joined query access patterns, duplicating attributes across tables so every high-QPS read is served by a single partition seek.
   • Graph Models: When queries traverse arbitrary-depth many-to-many relationships (e.g., fraud rings or social graphs), index-free adjacency in graph engines avoids recursive relational join explosions.
2. Composite B+Tree Indexing & The ESR (Equality, Sort, Range) Rule:
   • The column order of a multi-column composite index INDEX (col_A, col_B, col_C) dictates whether the storage engine executes a single contiguous leaf scan or falls back to an expensive in-memory filesort.
   • Equality (E) First: Place exact-match predicate columns (WHERE tenant_id = ? AND status = 'ACTIVE') first to narrow the B+Tree search to a single contiguous prefix range.
   • Sort (S) Second: Place ORDER BY created_at DESC columns immediately after Equality columns so matching tuples are already encountered in sorted order with zero CPU sort overhead.
   • Range (R) Last: Place inequality predicates (AND amount > 500) last, because a range condition breaks lexicographical contiguity for any subsequent index columns.

3. Mathematical Model: Shard Key Selection & Hotspot Variance

The single most critical architectural decision in distributed database engineering is the choice of Shard Key. A poorly chosen shard key produces severe hotspots that destroy cluster stability.

The Three Sharding Partitioning Strategies

+-------------------------------------------------------------------------------+
|                      SHARDING STRATEGY COMPARISON                             |
+-------------------------------------------------------------------------------+
| Strategy            | Mechanism               | Strengths     | Vulnerabilities|
+---------------------+-------------------------+---------------+---------------+
| 1. Hash-Based       | Shard = hash(key) % S   | Uniform write | Destroys range|
|    Sharding         |                         | distribution  | scan queries  |
|                     |                         |               |               |
| 2. Range-Based      | Shard 1: [A - M]        | Blazing fast  | Severe write  |
|    Sharding         | Shard 2: [N - Z]        | range queries | hotspots on   |
|                     |                         |               | sequential keys|
|                     |                         |               |               |
| 3. Directory /      | Lookup service maps     | Highly dynamic| Lookup service|
|    Lookup-Based     | EntityID -> ShardID     | custom routing| is an extra   |
|                     |                         | per tenant    | latency hop   |
+-------------------------------------------------------------------------------+

The Sequential Timestamp Antipattern

A classic rookie mistake in distributed database design is sharding time-series or order tables by created_at:

• Shard 1: January data
• Shard 2: February data
• Shard 3: March data

Every single insert operation right now has timestamp NOW(), routing $100\%$ of global write traffic to the single shard hosting today’s date! All historical shards sit completely idle. Sharding by timestamp converts a multi-node cluster back into an overloaded single server.

The Celebrity / Hot Tenant Monopoly (Zipfian Skew)

In real-world multi-tenant B2B applications, account size follows a Power-Law (Zipfian) distribution:

• $99\%$ of tenants have $<10,000$ records.
• $1$ enterprise tenant (e.g. Amazon or Uber) has $500,000,000$ records.

If you shard strictly by tenant_id:

• The enterprise tenant is pinned to a single physical shard.
• That shard’s NVMe drive reaches $100\%$ capacity and its CPU throttles under read contention.
• Mitigation: Deploy Compound Sharding: $$\text{Shard Key}=(\text{tenant\_id},\text{hash}(\text{user\_id})(\mathrm{mod}K))$$ Large enterprise tenants are internally sub-sharded across $K$ database instances, while small tenants remain consolidated on single shards.

4. Scatter-Gather Query Execution & Tail Latency Amplification

What happens when an application executes a query that does not include the shard key?

SELECT * FROM orders WHERE status = 'PENDING' ORDER BY total_amount DESC LIMIT 10;

Because the router cannot know which shard contains pending orders, it must execute a Scatter-Gather operation:

1. Scatter: Broadcasts the SQL query concurrently to all $S$ physical database shards.
2. Local Execution: Every shard executes the query locally against its local B-Tree index, returning its local Top 10 pending orders.
3. Gather & Merge: The router collects $S\times 10$ candidate rows over the network, executes an in-memory priority queue sort across all candidates, and returns the global Top 10 to the client.

Tail Latency Amplification Theorem

Let $p$ be the probability that a single database shard responds within an acceptable latency threshold (e.g. $p=0.99$ for $50\text{ ms}$).

When a query must scatter across $S$ independent shards, the query completes only when the slowest single shard responds:

The probability that at least one shard suffers a 99th-percentile tail latency stall is:

Scatter-Gather Tail Latency Amplification Table:
---------------------------------------------------------------------------------
Cluster Shard Count (S)   | Single-Shard SLA (p99) | Probability of Query Stall
+-------------------------+------------------------+----------------------------+
| S = 1 Shard             | 99% (p = 0.99)         | 1.0%
| S = 10 Shards           | 99%                    | 1 - (0.99)^10 = 9.6%
| S = 50 Shards           | 99%                    | 1 - (0.99)^50 = 39.5%
| S = 100 Shards          | 99%                    | 1 - (0.99)^100 = 63.4%!
| S = 500 Shards          | 99%                    | 1 - (0.99)^500 = 99.3%!!
+-------------------------+------------------------+----------------------------+

This mathematical reality explains why production distributed database architectures mandate that $>95\%$ of production OLTP queries must resolve through a single shard key.

5. Distributed Transactions: Two-Phase Commit (2PC)

When a business operation must update data residing on two different shards atomically (e.g., transferring $100 from User A on Shard 1 to User B on Shard 2), single-node database ACID transactions cannot help us.

The classical protocol to guarantee atomic multi-node commitment is Two-Phase Commit (2PC) (Gray, 1978).

Two-Phase Commit (2PC) Protocol Execution:
---------------------------------------------------------------------------------
Coordinator                     Shard 1 (Debit User A)        Shard 2 (Credit User B)
    |                                    |                              |
    |========= PHASE 1: PREPARE ========|                              |
    |--- PREPARE (Can you commit?) ----->|                              |
    |--- PREPARE (Can you commit?) ------------------------------------>|
    |                                    |                              |
    |                                [Acquires Row Lock]            [Acquires Row Lock]
    |                                [Flushes WAL to disk]          [Flushes WAL to disk]
    |<-- VOTE YES -----------------------|                              |
    |<-- VOTE YES ------------------------------------------------------|
    |                                    |                              |
    | [All participants voted YES!]      |                              |
    | [Writes COMMIT decision to WAL]    |                              |
    |                                    |                              |
    |========= PHASE 2: COMMIT =========|                              |
    |--- COMMIT ------------------------>|                              |
    |--- COMMIT ------------------------------------------------------->|
    |                                    |                              |
    |                                [Commits data]                 [Commits data]
    |                                [Releases Row Lock]            [Releases Row Lock]
    |<-- ACK ----------------------------|                              |
    |<-- ACK -----------------------------------------------------------|
    |                                    |                              |
    |-> Transaction complete! Returns success to application.

The Blocking Coordinator Disaster

While 2PC guarantees safety (atomicity), it has a fatal liveness flaw: it is a blocking protocol.

Consider this catastrophic failure scenario:

1. Shard 1 and Shard 2 vote YES in Phase 1 (Prepare).
2. Both shards lock the affected rows and wait for the coordinator’s Phase 2 decision.
3. The Coordinator server suffers a power outage and crashes before sending Phase 2!

What can Shard 1 and Shard 2 do?

• They cannot abort autonomously: The coordinator might have sent COMMIT to Shard 2 before dying. If Shard 1 aborts while Shard 2 commits, money is created or destroyed!
• They cannot commit autonomously: Shard 2 might have voted NO.
• Result: Total Resource Lockup: Shard 1 and Shard 2 must hold local row locks indefinitely, blocking all subsequent transactions on those user accounts until the coordinator server recovers from disk.

Because of this vulnerability, modern cloud-native systems minimize or completely avoid synchronous 2PC across microservices, opting instead for Asynchronous Sagas.

6. Asynchronous Sagas: Orchestration vs. Choreography

A Saga (Garcia-Molina & Salem, 1987) is an architectural pattern that coordinates distributed workflows by breaking them into a sequence of independent local ACID transactions:

Each local transaction $T_i$ updates data in a single shard/service and commits immediately, releasing locks in milliseconds.

Compensating Transactions

If step $T_j$ fails (e.g. insufficient funds, card declined, out of stock), the system achieves eventual consistency by executing a sequence of Compensating Transactions in reverse order:

Where $C_i$ semantically undos the effect of transaction $T_i$:

Saga Execution Pipeline (E-Commerce Order):
---------------------------------------------------------------------------------
Forward Happy Path:
[T1: Authorize Payment] --> [T2: Reserve Inventory] --> [T3: Create Shipping Label]
      (Commits!)                   (Commits!)                    (Commits!)

Failure Rollback Path (Inventory out of stock at T2):
[T1: Authorize Payment] --> [T2: Reserve Inventory: FAILS!]
      |
      v (Trigger Compensation)
[C1: Refund Payment]
      (Semantically reverses T1! Balance restored!)

Orchestration vs. Choreography

+-------------------------------------------------------------------------------+
|                       SAGA IMPLEMENTATION PARADIGMS                           |
+-------------------------------------------------------------------------------+
| Dimension            | Choreography (Event-Driven) | Orchestration (State Mach)|
+----------------------+-----------------------------+--------------------------+
| Coordination         | Decentralized (Events/PubSub| Centralized Orchestrator |
| Flow Definition      | Implicit across handlers    | Explicit workflow code   |
| Debuggability        | Difficult (Spaghetti events)| Simple (Audit log state) |
| Cyclic Dependencies  | High risk                   | Zero risk                |
| Ideal For            | Simple 2-3 step flows       | Complex multi-step flows |
+-------------------------------------------------------------------------------+

The Lack of Isolation Hazard in Sagas

Unlike 2PC, Sagas lack Isolation (the ‘I’ in ACID). Because local transaction $T_1$ commits immediately, other concurrent users can read uncommitted business state before the Saga finishes.

Semantic Lock Mitigation

To prevent dirty writes during Sagas:

• When $T_1$ reserves money, it sets the account state to PENDING_TRANSFER.
• Subsequent transactions check for the PENDING_TRANSFER flag and queue or reject updates until the Saga completes, re-introducing isolation at the application layer.

7. Real-World Architectures: Vitess, Citus, and CockroachDB

+-------------------------------------------------------------------------------+
|                       PRODUCTION SHARDING PLATFORMS                           |
+-------------------------------------------------------------------------------+
| Technology     | Base Engine   | Sharding Strategy    | Distributed Tx Model  |
+----------------+---------------+----------------------+-----------------------+
| 1. Vitess      | MySQL         | Horizontal Sharding  | 2PC across shards;    |
|    (YouTube)   | (VTGate/VTTablet)| via VSchema (Keyspace) Saga workflows     |
|                |               |                      |                       |
| 2. Citus       | PostgreSQL    | Distributed Tables   | Distributed 2PC with  |
|    (Microsoft) | Extension     | across worker nodes  | coordinator master    |
|                |               |                      |                       |
| 3. CockroachDB | Native Cloud- | Multi-Raft Consensus | Multi-Version (MVCC)  |
|                | Distributed SQL| across 64MB Ranges   | with Spanner-like Tx  |
+-------------------------------------------------------------------------------+

Vitess Case Study: Scaling YouTube MySQL

Vitess decouples MySQL into a distributed database without modifying the underlying MySQL engine:

• VTGate: Stateless proxy servers that receive application SQL, parse queries, look up shard routing metadata from etcd, and scatter-gather or point-route queries.
• VTTablet: Sidecar daemon running alongside every physical MySQL instance, managing connection pooling, enforcing query memory limits, and terminating runaway queries.
• VSchema: Declarative JSON schema defining shard keys (e.g., using hash(user_id) or lookup vindexes).

8. Non-Blocking Shard Migrations & Online Resharding

As data volumes double, a cluster sharded across 4 nodes must eventually reshard to 8 nodes. In production, you cannot take the database offline for 12 hours to repartition data.

The 4-Phase Online Resharding Pipeline

Online Non-Blocking Resharding Pipeline:
---------------------------------------------------------------------------------
Phase 1: Background Historical Backfill
Copy historical data from Old Shard to New Shards using point-in-time snapshot.

Phase 2: Change Data Capture (CDC) Replication
Tail the MySQL Binlog / PostgreSQL WAL to replicate all real-time mutations
from Old Shard to New Shards with low replication lag.

Phase 3: Dual-Write & Shadow Verification
Router sends live writes to Old Shard AND New Shards concurrently.
Validate checksum consistency across random sampled keys.

Phase 4: Instantaneous Atomic Cutover
Update Router metadata (VSchema / etcd) in a single atomic transaction.
All subsequent reads and writes immediately route to New Shards!
Old Shard is decommissioned after 48 hours.

9. Operational Failure Modes & Production Troubleshooting

+-------------------------------------------------------------------------------+
|                    SHARDING OPERATIONAL TROUBLESHOOTING                       |
+-------------------------------------------------------------------------------+
| Failure Mode            | Root Cause                | Mitigation Strategy     |
+-------------------------+---------------------------+-------------------------+
| 1. Shard Memory / Disk  | Monopolized by a single   | Split hot tenant into   |
|    Exhaustion           | giant enterprise customer | composite sub-shards    |
|                         |                           |                         |
| 2. Cross-Shard Deadlock | Two transactions locking  | Enforce global deterministic|
|    Freeze               | rows on Shard 1 and 2 in  | shard lock ordering     |
|                         | opposite sequences        | (Lock Shard A before B) |
|                         |                           |                         |
| 3. Shard Key Regret     | Shard key chosen early    | Build online dual-write |
|    (Schema Lock-In)     | becomes sub-optimal as    | shadow migration pipeline|
|                         | access patterns evolve    | to re-shard table       |
|                         |                           |                         |
| 4. Connection Pool      | Scatter-gather queries    | Dedicated proxy pools   |
|    Starvation           | opening 100 connections   | with aggressive timeout |
|                         | per incoming request      | deadlines (200ms)       |
+-------------------------------------------------------------------------------+

Deep Dive: Enforcing Global Deterministic Lock Ordering

In cross-shard transactions, if Transaction 1 locks Shard 1 then Shard 2, while Transaction 2 locks Shard 2 then Shard 1:

• A classic Distributed Deadlock occurs.
• Neither database engine can detect the deadlock autonomously because the cycle spans multiple physical MySQL/PostgreSQL kernels.

Remediation: Always enforce Deterministic Global Shard Ordering: $$\text{If a transaction touches }\{{\text{Shard}}_A,{\text{Shard}}_B\},\text{ it must ALWAYS acquire locks in ascending order: }{\text{Shard}}_{\min (A,B)}\to {\text{Shard}}_{\max (A,B)}$$ Eliminating circular lock dependency graphs completely prevents distributed deadlocks.

10. Concrete Implementation: Shard Router & Saga Orchestrator

The following complete, runnable TypeScript implementation demonstrates:

1. Hash-based shard routing for point queries.
2. Parallel scatter-gather query execution with in-memory result merging.
3. An Asynchronous Saga Orchestrator with automatic compensating transaction rollback upon failure.

// sharding-saga-engine.ts
export interface OrderRecord {
  id: string;
  tenantId: string;
  amount: number;
  status: 'PENDING' | 'COMPLETED' | 'CANCELLED';
}

/**
 * Represents a single independent physical database shard
 */
export class DatabaseShard {
  public readonly id: number;
  public records = new Map<string, OrderRecord>();

  constructor(id: number) {
    this.id = id;
  }

  public insert(order: OrderRecord): void {
    this.records.set(order.id, { ...order });
  }

  public updateStatus(id: string, status: 'PENDING' | 'COMPLETED' | 'CANCELLED'): boolean {
    const record = this.records.get(id);
    if (!record) return false;
    record.status = status;
    return true;
  }

  public getById(id: string): OrderRecord | null {
    return this.records.get(id) || null;
  }

  public queryByStatus(status: string): OrderRecord[] {
    const matched: OrderRecord[] = [];
    for (const record of this.records.values()) {
      if (record.status === status) matched.push({ ...record });
    }
    return matched;
  }
}

/**
 * Distributed Shard Query Router
 */
export class ShardRouter {
  private shards: DatabaseShard[];

  constructor(numShards = 3) {
    this.shards = Array.from({ length: numShards }, (_, i) => new DatabaseShard(i));
  }

  public hashShardKey(tenantId: string): number {
    let hash = 0;
    for (let i = 0; i < tenantId.length; i++) {
      hash = (hash << 5) - hash + tenantId.charCodeAt(i);
      hash |= 0;
    }
    return Math.abs(hash) % this.shards.length;
  }

  /**
   * Point Query: Directly routed to a single shard using Shard Key
   */
  public executePointInsert(order: OrderRecord): number {
    const shardIdx = this.hashShardKey(order.tenantId);
    this.shards[shardIdx].insert(order);
    return shardIdx;
  }

  /**
   * Scatter-Gather Query: Broadcast to all shards and merge in-memory
   */
  public executeScatterGather(status: 'PENDING' | 'COMPLETED' | 'CANCELLED', limit = 10): OrderRecord[] {
    const allResults: OrderRecord[] = [];

    // Parallel scatter broadcast across all physical shards
    for (const shard of this.shards) {
      const partial = shard.queryByStatus(status);
      allResults.push(...partial);
    }

    // In-memory gather, sort by amount DESC, and apply limit
    return allResults.sort((a, b) => b.amount - a.amount).slice(0, limit);
  }

  public getShard(index: number): DatabaseShard {
    return this.shards[index];
  }
}

/**
 * Asynchronous Saga Step Definition
 */
export interface SagaStep {
  name: string;
  forwardAction: () => Promise<boolean>;
  compensatingAction: () => Promise<void>;
}

/**
 * Asynchronous Saga Orchestrator with compensating rollback
 */
export class SagaOrchestrator {
  private executedSteps: SagaStep[] = [];

  public async executeSaga(steps: SagaStep[]): Promise<boolean> {
    this.executedSteps = [];

    for (const step of steps) {
      try {
        const success = await step.forwardAction();
        if (!success) {
          await this.rollback();
          return false;
        }
        this.executedSteps.push(step);
      } catch (err) {
        await this.rollback();
        return false;
      }
    }

    return true;
  }

  private async rollback(): Promise<void> {
    // Execute compensating actions in reverse chronological order
    while (this.executedSteps.length > 0) {
      const step = this.executedSteps.pop();
      if (step) {
        await step.compensatingAction();
      }
    }
  }
}

async function runDemo(): Promise<void> {
  console.log("=== Database Sharding & Distributed Saga Coordinator ===");
  const router = new ShardRouter(3);
  console.log("Initialized ShardRouter with 3 database shards");
  const s1 = router.executePointInsert({ id: "ord-101", tenantId: "tenant-A", amount: 5000, status: "PENDING" });
  const s2 = router.executePointInsert({ id: "ord-102", tenantId: "tenant-B", amount: 15000, status: "PENDING" });
  console.log(`Order ord-101 routed to shard ${s1}, ord-102 routed to shard ${s2}`);
  const orders = router.executeScatterGather("PENDING", 5);
  console.log(`Scatter-gather queried ${orders.length} orders across all 3 shards (top order: ${orders[0].id}, amount=${orders[0].amount})`);

  // Demonstrate Saga Orchestrator with compensating rollback on cross-shard failure
  const saga = new SagaOrchestrator();
  const sagaSuccess = await saga.executeSaga([
    {
      name: "Step1_CreatePendingOrder",
      forwardAction: async () => {
        router.executePointInsert({ id: "ord-103", tenantId: "tenant-A", amount: 9500, status: "PENDING" });
        return true;
      },
      compensatingAction: async () => {
        const shardIdx = router.hashShardKey("tenant-A");
        router.getShard(shardIdx).updateStatus("ord-103", "CANCELLED");
      },
    },
    {
      name: "Step2_ReserveInventoryShard",
      forwardAction: async () => false, // Simulate downstream inventory out-of-stock failure
      compensatingAction: async () => {},
    },
  ]);
  const compensated = router.getShard(router.hashShardKey("tenant-A")).getById("ord-103");
  console.log(`Saga completed: success=${sagaSuccess}, ord-103 status after compensation=${compensated?.status}`);
}

runDemo();

Simulation Execution Output & State Transitions

=== Database Sharding & Distributed Saga Coordinator ===
Initialized ShardRouter with 3 database shards
Order ord-101 routed to shard 2, ord-102 routed to shard 1
Scatter-gather queried 2 orders across all 3 shards (top order: ord-102, amount=15000)
Saga completed: success=false, ord-103 status after compensation=CANCELLED

11. Exercises with Observable Deliverables

Exercise 1: Scatter-Gather Tail Latency Simulation

Deliverable:

1. Write a script simulating a cluster of $S$ shards where each shard responds with latency drawn from a normal distribution with $p99=50\text{ ms}$ and an occasional $200\text{ ms}$ tail stall ($1\%$ probability).
2. Measure overall scatter-gather query latency for $S=1,10,50,100$.
3. Plot or print the empirical probability of query latency $>150\text{ ms}$. Confirm that the empirical curve matches $P=1-(0.99{)}^S$.

Exercise 2: Verify Saga Compensating Rollback

Deliverable: Using ShardRouter and SagaOrchestrator:

1. Initialize a 3-shard router.
2. Formulate a 2-step Saga:
   • Step 1: Insert an order for Tenant A into Shard 1 with status PENDING.
   • Step 2: Simulate an inventory reservation on Shard 2 that fails deliberately (throws error).
   • Compensating Action for Step 1: Set order status on Shard 1 to CANCELLED.
3. Execute the Saga and assert that:
   • executeSaga returns false.
   • The record on Shard 1 is in CANCELLED state, leaving zero orphaned active reservations.

12. Summary & The Database Architect’s Rules of Thumb

+-------------------------------------------------------------------------------+
|                    THE DATABASE SHARDING DECISION GUIDE                       |
+-------------------------------------------------------------------------------+
| Decision Dimension        | Recommended Production Architectural Pattern      |
+---------------------------+---------------------------------------------------+
| Shard Key Selection       | High-cardinality entity ID (tenant_id / user_id)  |
| Sharding Mechanism        | Consistent hashing with virtual nodes             |
| OLTP Query Discipline     | > 95% of queries must resolve through Shard Key   |
| Cross-Shard Atomicity     | Asynchronous Sagas with compensating rollbacks    |
| Synchronous 2PC Usage     | Strictly isolated to internal engine range merges |
| Shard Routing Tier        | Decoupled proxy cluster (Vitess, Citus)           |
| Online Resharding         | 4-phase CDC shadow replication with atomic cutover|
+-------------------------------------------------------------------------------+

1. Avoid Sharding Until Necessary: Exhaust vertical scaling, read replicas, connection pooling, and table indexing before sharding. Sharding permanently fractures relational semantics.
2. Never Shard by Low-Cardinality Keys: Sharding by status (status='ACTIVE') or country code routes massive traffic imbalances to small subsets of machines.
3. Scatter-Gather Queries Do Not Scale: Every scatter-gather query multiplies database connection overhead across all shards. If analytics require global scans, replicate data asynchronously to a columnar OLAP warehouse (ClickHouse, BigQuery).
4. Prefer Sagas Over Two-Phase Commit: 2PC holds synchronous database row locks across network round-trips, making clusters vulnerable to coordinator crash freezes. Use Sagas with compensating actions for long-lived distributed workflows.

13. Design Alternatives & Sharding Trade-Off Matrix

14. Authoritative References & Version Context

• Jim Gray (1978), “Notes on Data Base Operating Systems”, IBM Research Report RJ2188 (Foundational Two-Phase Commit derivation) — semanticscholar.org
• Hector Garcia-Molina and Kenneth Salem (1987), “Sagas”, ACM SIGMOD ‘87 — doi.org/10.1145/38713.38742
• Vitess Architecture Guide (Vitess 19.x LTS), Horizontal Sharding and VReplication Workflows — vitess.io/docs/
• Version Context: Sharding patterns reflect PostgreSQL 16+ declarative partitioning, Citus 12.x, Vitess 19.x LTS, and CockroachDB 23.x distributed transaction engines.