Distributed Transactions: 2PC, Saga & Transactional Outbox

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

// NAIVE ANTI-PATTERN: The Dual-Write Hazard
async function handleCreateOrder(orderData: Order): Promise<void> {
  // 1. Commit to local PostgreSQL database
  await postgres.query('INSERT INTO orders (id, total, status) VALUES ($1, $2, $3)', [
    orderData.id,
    orderData.total,
    'PENDING',
  ]);

  // 2. Publish event to Apache Kafka for downstream inventory and payment services
  // CATASTROPHE SCENARIO:
  // If the Kafka broker times out, the network drops, or the Node.js process runs out of
  // memory right here, the database transaction has ALREADY COMMITTED!
  // The event is NEVER published!
  // Downstream services never process the order; inventory is never reserved;
  // the order remains stuck in PENDING forever!
  await kafkaProducer.send({
    topic: 'order-events',
    messages: [{ key: orderData.id, value: JSON.stringify(orderData) }],
  });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

// PRODUCTION ORCHESTRATED SAGA STATE MACHINE WITH COMPENSATIONS
export interface SagaStep<TContext> {
  name: string;
  execute: (ctx: TContext) => Promise<void>;
  compensate: (ctx: TContext) => Promise<void>;
}

export class SagaOrchestrator<TContext> {
  private steps: SagaStep<TContext>[] = [];

  public addStep(step: SagaStep<TContext>): this {
    this.steps.push(step);
    return this;
  }

  public async execute(context: TContext): Promise<{ success: boolean; failedStep?: string; error?: Error }> {
    const executedSteps: SagaStep<TContext>[] = [];

    for (const step of this.steps) {
      try {
        console.log(`[Saga] Executing forward step: ${step.name}`);
        await step.execute(context);
        executedSteps.push(step);
      } catch (err) {
        console.error(`[Saga] Step ${step.name} failed. Initiating backward compensating rollback...`, err);
        await this.rollback(executedSteps, context);
        return { success: false, failedStep: step.name, error: err as Error };
      }
    }

    console.log('[Saga] All steps completed successfully.');
    return { success: true };
  }

  private async rollback(executedSteps: SagaStep<TContext>[], context: TContext): Promise<void> {
    // Compensations MUST execute in reverse order of forward steps
    const reverseSteps = [...executedSteps].reverse();
    for (const step of reverseSteps) {
      try {
        console.log(`[Saga] Executing compensating rollback: ${step.name}`);
        await step.compensate(context);
      } catch (compensationError) {
        // CRITICAL INVARIANT: Compensations MUST NEVER permanently fail.
        // If a compensation fails, alert on-call engineer and retry with exponential backoff!
        console.error(`[Saga CRITICAL] Compensation for ${step.name} failed! Queued for dead-letter retry.`, compensationError);
      }
    }
  }
}

// 2. TRANSACTIONAL OUTBOX PATTERN WITH POSTGRESQL ACID COMMIT
export async function executeTransactionalOutboxOrder(client: any, orderId: string, total: number): Promise<void> {
  // Execute BOTH domain insert AND event outbox insert in the SAME local database transaction!
  await client.query('BEGIN');
  try {
    // 1. Insert domain record
    await client.query(
      'INSERT INTO orders (id, total, status) VALUES ($1, $2, $3)',
      [orderId, total, 'PENDING']
    );

    // 2. Insert event record into outbox table
    await client.query(
      `INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload, created_at)
       VALUES ($1, $2, $3, $4, NOW())`,
      ['Order', orderId, 'OrderCreated', JSON.stringify({ orderId, total })]
    );

    // Commit guarantees both records hit the WAL simultaneously and atomically!
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  }
}