Das Saga Pattern für verteilte Transaktionen

// ❌ Two-phase commit — reasons it fails in microservices
 
const twoPhaseCommitProblems = {
  availability: 'If ANY participant is down, the entire transaction blocks',
  latency: 'Coordinator must wait for ALL participants to respond',
  coupling: 'All services must implement the 2PC protocol',
  scalability: 'Locks are held across services during the prepare phase',
  singlePointOfFailure: 'The coordinator failing leaves transactions in limbo',
};
 
// ✅ Saga pattern — eventual consistency without distributed locks
 
const sagaBenefits = {
  availability: 'Each service operates independently',
  latency: 'No waiting for distributed locks',
  coupling: 'Services communicate via events, not protocols',
  scalability: 'No cross-service locks',
  resilience: 'Compensations handle partial failures gracefully',
};

// Core saga types
interface SagaStep<TContext> {
  name: string;
  execute: (context: TContext) => Promise<TContext>;
  compensate: (context: TContext) => Promise<TContext>;
}
 
interface SagaResult<TContext> {
  success: boolean;
  context: TContext;
  completedSteps: string[];
  failedStep?: string;
  error?: string;
}
 
// Saga executor — orchestration style
async function executeSaga<TContext>(
  steps: SagaStep<TContext>[],
  initialContext: TContext
): Promise<SagaResult<TContext>> {
  let context = { ...initialContext };
  const completedSteps: string[] = [];
 
  for (const step of steps) {
    try {
      console.log(`Executing step: ${step.name}`);
      context = await step.execute(context);
      completedSteps.push(step.name);
    } catch (error) {
      console.error(`Step '${step.name}' failed:`, error);
 
      // Run compensations in reverse order
      for (const completedStep of [...completedSteps].reverse()) {
        const stepDef = steps.find((s) => s.name === completedStep);
        if (stepDef) {
          try {
            console.log(`Compensating step: ${completedStep}`);
            context = await stepDef.compensate(context);
          } catch (compError) {
            console.error(
              `Compensation for '${completedStep}' failed:`,
              compError
            );
            // Log for manual intervention — compensation failures
            // are the hardest problem in sagas
          }
        }
      }
 
      return {
        success: false,
        context,
        completedSteps,
        failedStep: step.name,
        error: String(error),
      };
    }
  }
 
  return { success: true, context, completedSteps };
}

interface OrderSagaContext {
  orderId: string;
  customerId: string;
  items: { productId: string; quantity: number; price: number }[];
  total: number;
  reservationId?: string;
  paymentId?: string;
  shipmentId?: string;
}
 
const orderSagaSteps: SagaStep<OrderSagaContext>[] = [
  {
    name: 'reserve-inventory',
    execute: async (ctx) => {
      const reservation = await inventoryService.reserve(
        ctx.items.map((i) => ({
          productId: i.productId,
          quantity: i.quantity,
        }))
      );
      return { ...ctx, reservationId: reservation.id };
    },
    compensate: async (ctx) => {
      if (ctx.reservationId) {
        await inventoryService.releaseReservation(ctx.reservationId);
      }
      return { ...ctx, reservationId: undefined };
    },
  },
  {
    name: 'charge-payment',
    execute: async (ctx) => {
      const payment = await paymentService.charge({
        customerId: ctx.customerId,
        amount: ctx.total,
        orderId: ctx.orderId,
      });
      return { ...ctx, paymentId: payment.id };
    },
    compensate: async (ctx) => {
      if (ctx.paymentId) {
        await paymentService.refund(ctx.paymentId, ctx.total);
      }
      return { ...ctx, paymentId: undefined };
    },
  },
  {
    name: 'schedule-shipping',
    execute: async (ctx) => {
      const shipment = await shippingService.schedule({
        orderId: ctx.orderId,
        items: ctx.items,
      });
      return { ...ctx, shipmentId: shipment.id };
    },
    compensate: async (ctx) => {
      if (ctx.shipmentId) {
        await shippingService.cancel(ctx.shipmentId);
      }
      return { ...ctx, shipmentId: undefined };
    },
  },
];
 
// Execute the saga
async function placeOrder(order: OrderSagaContext): Promise<void> {
  const result = await executeSaga(orderSagaSteps, order);
 
  if (result.success) {
    await orderDb.updateStatus(order.orderId, 'confirmed');
    console.log(`Order ${order.orderId} placed successfully`);
  } else {
    await orderDb.updateStatus(order.orderId, 'failed');
    console.error(
      `Order ${order.orderId} failed at step '${result.failedStep}'`
    );
  }
}

// ORCHESTRATION — central coordinator controls the flow
// Pros: Clear flow, easy to understand, centralized error handling
// Cons: Single point of failure, coordinator can become a bottleneck
 
class OrderSagaOrchestrator {
  async execute(order: OrderSagaContext): Promise<SagaResult<OrderSagaContext>> {
    // Coordinator explicitly calls each service in order
    // and handles compensations on failure
    return executeSaga(orderSagaSteps, order);
  }
}
 
// CHOREOGRAPHY — services react to events autonomously
// Pros: No central coordinator, services are fully decoupled
// Cons: Hard to track flow, distributed error handling, debugging is harder
 
// Order service publishes: order.created
// → Inventory service listens, reserves stock, publishes: inventory.reserved
// → Payment service listens, charges payment, publishes: payment.charged
// → Shipping service listens, schedules shipment, publishes: shipment.scheduled
// → Order service listens, marks order as confirmed

// ❌ Choreography when flow is complex — becomes hard to follow
// Event chain: A → B → C → D → E → F
// If D fails, who triggers compensation for C, B, and A?
// Each service must know about compensations for upstream services
// Debugging: events are scattered across service logs
 
// ✅ Orchestration for complex flows — clear control flow
// Orchestrator: A, then B, then C, then D, then E, then F
// If D fails: compensate C, compensate B, compensate A
// All compensation logic is in one place
// Debugging: one service shows the entire saga lifecycle

interface CompensationRecord {
  sagaId: string;
  stepName: string;
  status: 'pending' | 'completed' | 'failed';
  attempts: number;
  lastError?: string;
  context: Record<string, unknown>;
}
 
async function compensateWithRetries(
  step: SagaStep<OrderSagaContext>,
  context: OrderSagaContext,
  sagaId: string,
  maxRetries: number = 3
): Promise<void> {
  let lastError: Error | null = null;
 
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      await step.compensate(context);
 
      await db.query(
        `UPDATE compensation_log
         SET status = 'completed', attempts = $2
         WHERE saga_id = $1 AND step_name = $3`,
        [sagaId, attempt, step.name]
      );
      return;
    } catch (error) {
      lastError = error as Error;
      console.error(
        `Compensation retry ${attempt}/${maxRetries} for ${step.name}:`,
        error
      );
 
      // Exponential backoff between retries
      if (attempt < maxRetries) {
        await sleep(Math.pow(2, attempt) * 1000);
      }
    }
  }
 
  // All retries exhausted — log for manual intervention
  await db.query(
    `UPDATE compensation_log
     SET status = 'failed', attempts = $2, last_error = $3
     WHERE saga_id = $1 AND step_name = $4`,
    [sagaId, maxRetries, lastError?.message, step.name]
  );
 
  // Alert operations team
  await alerting.critical(
    `Saga compensation failed after ${maxRetries} retries`,
    {
      sagaId,
      step: step.name,
      error: lastError?.message,
    }
  );
}

// Store saga state in the database
interface SagaState {
  id: string;
  type: string;
  status: 'running' | 'completed' | 'compensating' | 'failed';
  currentStep: number;
  context: Record<string, unknown>;
  completedSteps: string[];
  createdAt: Date;
  updatedAt: Date;
}
 
async function persistSagaState(state: SagaState): Promise<void> {
  await db.query(
    `INSERT INTO sagas (id, type, status, current_step, context, completed_steps, updated_at)
     VALUES ($1, $2, $3, $4, $5, $6, NOW())
     ON CONFLICT (id) DO UPDATE SET
       status = $3,
       current_step = $4,
       context = $5,
       completed_steps = $6,
       updated_at = NOW()`,
    [
      state.id,
      state.type,
      state.status,
      state.currentStep,
      JSON.stringify(state.context),
      JSON.stringify(state.completedSteps),
    ]
  );
}
 
// Recovery: find and resume incomplete sagas after service restart
async function recoverIncompleteSagas(): Promise<void> {
  const incomplete = await db.query(
    `SELECT * FROM sagas
     WHERE status IN ('running', 'compensating')
       AND updated_at < NOW() - INTERVAL '5 minutes'`
  );
 
  for (const saga of incomplete.rows) {
    console.log(`Recovering saga ${saga.id} (status: ${saga.status})`);
 
    if (saga.status === 'compensating') {
      await runCompensations(saga);
    } else {
      await resumeSagaFromStep(saga, saga.current_step);
    }
  }
}