Saga Orchestration Pattern

Overview

The Saga pattern is a design pattern for managing distributed transactions across multiple services while maintaining data consistency. Unlike traditional ACID transactions that span multiple databases, sagas coordinate a sequence of local transactions, each with its own compensation logic for failure scenarios.

Why Sagas?

In a booking platform with multiple services (payments, capacity management, notifications), traditional distributed transactions don't scale well. Sagas provide:

• Eventual Consistency: System reaches consistent state over time
• Isolation: Each step is a local transaction
• Compensation: Automatic rollback via compensating transactions
• Resilience: Retry logic with exponential backoff
• Observability: Checkpointing and correlation tracking

---

How It Works

Saga Lifecycle

A saga progresses through distinct states tracked in the SagaExecution table:

not_started → running → completed ✓
                 ↓
              compensating → failed ✗
                 ↓
              timeout
              cancelled

State Descriptions:

State	Description	Next States
not_started	Saga created but not yet executing	running, cancelled
running	Actively executing forward steps	completed, compensating, timeout
compensating	Rolling back completed steps due to failure	failed
completed	All steps succeeded	Terminal state ✓
failed	Saga failed, compensation complete/failed	Terminal state ✗
timeout	Exceeded maximum execution time	compensating, failed
cancelled	Explicitly cancelled by user/admin	Terminal state ✗

Compensation States

When a saga fails, compensation tracks the rollback process:

none (success path)

pending → in_progress → completed ✓ (clean rollback)
                  ↓
               partial ⚠️ (needs review)
                  ↓
               failed ✗ (critical issue)

Compensation Status Meanings:

• none: No compensation needed (saga succeeded)
• pending: Compensation required but not started
• in_progress: Actively rolling back
• completed: All compensation steps succeeded
• partial: Some compensation succeeded, manual review needed
• failed: Compensation failed, urgent intervention required

---

Database Schema

SagaExecution Table

Tracks the overall saga execution:

model SagaExecution {
  id                 String  @id
  tenantId           String
  sagaType           String  // e.g., "booking_confirmation"
  aggregateId        String  // e.g., Booking ID
  correlationId      String  @unique  // UUID for tracing
  status             saga_state_enum
  compensationStatus compensation_status_enum?
  parameters         Json    // Input parameters
  context            Json?   // Intermediate results
  retryCount         Int     @default(0)
  maxRetries         Int     @default(3)
  startedAt          DateTime?
  completedAt        DateTime?
  nextRetryAt        DateTime?
  lastError          String?

  checkpoints        SagaCheckpoint[]
}

SagaCheckpoint Table

Tracks individual saga steps:

model SagaCheckpoint {
  id               String  @id
  tenantId         String
  sagaExecutionId  String
  stepName         String  // e.g., "hold_capacity", "process_payment"
  stepOrder        Int     // Execution order (determines compensation order)
  status           saga_state_enum
  output           Json?   // Step result data
  errorMessage     String?
  retryCount       Int     @default(0)
  compensationData Json?   // Data needed for rollback
  compensatedAt    DateTime?

  sagaExecution    SagaExecution @relation(...)
}

Key Points:

• stepOrder determines execution sequence and reverse compensation order
• compensationData stores information needed to undo the step
• Each checkpoint is idempotent (safe to retry)

---

Common Saga Patterns

1. Booking Confirmation Saga

Coordinates booking confirmation with payment capture and capacity locking.

Saga Type: booking_confirmation

Steps:

1. validate_pricing - Verify pricing hash
2. hold_capacity - Lock capacity in all slots
3. process_payment - Capture payment via Stripe/Paytrail
4. confirm_booking - Update booking status to CONFIRMED
5. send_confirmation - Queue confirmation email

Compensation Logic:

Step 5 fails (email) → Compensate steps 4, 3, 2 (reverse order)
  Step 4: Cancel booking confirmation
  Step 3: Refund payment
  Step 2: Release capacity

Example Flow:

// Saga initiated
const correlationId = uuid();
await sagaOrchestrator.start({
  sagaType: 'booking_confirmation',
  aggregateId: bookingId,
  correlationId,
  parameters: {
    bookingId,
    paymentIntentId,
    pricingHash,
  },
});

// Steps execute sequentially
// If step 3 (payment) fails, steps 2 and 1 compensate automatically

2. Booking Cancellation Saga

Handles booking cancellation with refund processing.

Saga Type: booking_cancellation

Steps:

1. validate_cancellation - Check cancellation policy
2. calculate_refund - Determine refund amount (fees)
3. process_refund - Issue refund via payment provider
4. release_capacity - Release reserved capacity
5. update_status - Mark booking as CANCELLED
6. send_notification - Queue cancellation email

Compensation Logic:

If refund succeeds but capacity release fails:

Step 4 fails (capacity) → Compensate step 3 (reverse refund)
Manual intervention required for data consistency

3. Payment Compensation Saga

Reverses payment when booking fails after payment capture.

Saga Type: payment_compensation

Steps:

1. verify_payment - Confirm payment exists
2. initiate_refund - Start refund process
3. process_refund - Execute refund via provider
4. update_ledger - Record refund transaction
5. notify_customer - Queue refund notification

---

Implementation

Service: SagaOrchestratorService

Location: apps/api/src/modules/saga/saga-orchestrator.service.ts

Core Methods:

@Injectable()
export class SagaOrchestratorService {
  // Start a new saga
  async startSaga(params: StartSagaParams): Promise<SagaExecution> {
    const correlationId = uuid();
    const saga = await this.db.sagaExecution.create({
      data: {
        sagaType: params.sagaType,
        aggregateId: params.aggregateId,
        correlationId,
        status: 'not_started',
        compensationStatus: 'none',
        parameters: params.parameters,
        tenantId: params.tenantId,
      },
    });

    // Start execution asynchronously
    await this.executeSaga(saga.id);
    return saga;
  }

  // Execute saga steps
  async executeSaga(sagaId: string): Promise<void> {
    const saga = await this.loadSaga(sagaId);
    const steps = this.getSagaSteps(saga.sagaType);

    try {
      await this.updateSagaStatus(sagaId, 'running');

      for (const step of steps) {
        await this.executeStep(saga, step);
      }

      await this.completeSaga(sagaId);
    } catch (error) {
      await this.handleSagaFailure(saga, error);
    }
  }

  // Execute individual step with checkpointing
  async executeStep(saga: SagaExecution, step: SagaStep): Promise<void> {
    const checkpoint = await this.createCheckpoint(saga, step);

    try {
      const result = await step.execute(saga.parameters, saga.context);

      await this.updateCheckpoint(checkpoint.id, {
        status: 'completed',
        output: result,
      });

      // Update saga context with step result
      await this.updateSagaContext(saga.id, result);
    } catch (error) {
      await this.updateCheckpoint(checkpoint.id, {
        status: 'failed',
        errorMessage: error.message,
      });
      throw error;
    }
  }

  // Handle saga failure with compensation
  async handleSagaFailure(saga: SagaExecution, error: Error): Promise<void> {
    await this.updateSagaStatus(saga.id, 'compensating');
    await this.updateCompensationStatus(saga.id, 'pending');

    try {
      await this.compensateSaga(saga);
      await this.updateCompensationStatus(saga.id, 'completed');
    } catch (compError) {
      await this.updateCompensationStatus(saga.id, 'failed');
      this.logger.error('Compensation failed', compError);
    } finally {
      await this.updateSagaStatus(saga.id, 'failed');
    }
  }

  // Compensate completed steps in reverse order
  async compensateSaga(saga: SagaExecution): Promise<void> {
    const completedCheckpoints = await this.getCompletedCheckpoints(saga.id);

    // Reverse order (LIFO)
    const reversedCheckpoints = completedCheckpoints.reverse();

    for (const checkpoint of reversedCheckpoints) {
      await this.compensateStep(checkpoint);
    }
  }

  // Compensate individual step
  async compensateStep(checkpoint: SagaCheckpoint): Promise<void> {
    const step = this.getStepByName(checkpoint.stepName);

    if (!step.compensate) {
      // Step doesn't require compensation
      return;
    }

    try {
      await step.compensate(checkpoint.compensationData);
      await this.updateCheckpoint(checkpoint.id, {
        compensatedAt: new Date(),
      });
    } catch (error) {
      this.logger.error(
        `Compensation failed for step ${checkpoint.stepName}`,
        error,
      );
      throw error;
    }
  }
}

Defining Saga Steps

// Example: Booking confirmation saga steps
const bookingConfirmationSteps: SagaStep[] = [
  {
    name: 'validate_pricing',
    order: 1,
    execute: async (params, context) => {
      const { bookingId, pricingHash } = params;
      const booking = await db.booking.findUnique({ where: { id: bookingId } });

      if (booking.pricingHash !== pricingHash) {
        throw new Error('Pricing hash mismatch');
      }

      return { validated: true };
    },
    compensate: null, // No compensation needed for validation
  },

  {
    name: 'hold_capacity',
    order: 2,
    execute: async (params, context) => {
      const { bookingId } = params;
      const booking = await db.booking.findUnique({
        where: { id: bookingId },
        include: { bookingSlots: true },
      });

      // Lock capacity for all slots atomically
      await capacityLockingService.lockSlots(booking.bookingSlots);

      return {
        slotsLocked: booking.bookingSlots.map((bs) => bs.slotId),
      };
    },
    compensate: async (compensationData) => {
      // Release locked capacity
      await capacityLockingService.releaseSlots(compensationData.slotsLocked);
    },
  },

  {
    name: 'process_payment',
    order: 3,
    execute: async (params, context) => {
      const { paymentIntentId } = params;

      // Capture payment
      const charge = await stripeAdapter.capturePayment(paymentIntentId);

      return {
        chargeId: charge.id,
        amountCharged: charge.amount,
        capturedAt: new Date(),
      };
    },
    compensate: async (compensationData) => {
      // Refund payment
      await stripeAdapter.refundPayment(compensationData.chargeId);
    },
  },

  {
    name: 'confirm_booking',
    order: 4,
    execute: async (params, context) => {
      const { bookingId } = params;

      await db.booking.update({
        where: { id: bookingId },
        data: {
          status: 'CONFIRMED',
          confirmedAt: new Date(),
        },
      });

      return { confirmed: true };
    },
    compensate: async (compensationData) => {
      // Revert to HOLD status
      await db.booking.update({
        where: { id: compensationData.bookingId },
        data: { status: 'HOLD', confirmedAt: null },
      });
    },
  },

  {
    name: 'send_confirmation',
    order: 5,
    execute: async (params, context) => {
      const { bookingId } = params;

      // Queue confirmation email via Outbox
      await outboxProducer.publishEvent({
        messageType: 'email.booking_confirmation',
        aggregateId: bookingId,
        eventData: { bookingId },
      });

      return { emailQueued: true };
    },
    compensate: null, // Email already sent, cannot undo
  },
];

---

Retry Logic

Exponential Backoff

Sagas automatically retry failed steps with exponential backoff:

function calculateNextRetry(retryCount: number): DateTime {
  const baseDelay = 1000; // 1 second
  const delay = baseDelay * Math.pow(2, retryCount);
  const jitter = Math.random() * 1000; // Add jitter to prevent thundering herd

  return new Date(Date.now() + delay + jitter);
}

// Retry attempts:
// Attempt 1: ~1 second
// Attempt 2: ~2 seconds
// Attempt 3: ~4 seconds
// Attempt 4: ~8 seconds (max retries reached)

Saga Retry Worker

Background worker processes failed sagas:

@Injectable()
export class SagaRetryWorkerService {
  @Cron('*/1 * * * *') // Every minute
  async processRetries() {
    const now = new Date();

    // Find sagas ready for retry
    const sagasToRetry = await this.db.sagaExecution.findMany({
      where: {
        status: { in: ['failed', 'timeout'] },
        retryCount: { lt: this.db.raw('max_retries') },
        nextRetryAt: { lte: now },
      },
    });

    for (const saga of sagasToRetry) {
      await this.retrySaga(saga);
    }
  }

  async retrySaga(saga: SagaExecution) {
    await this.sagaOrchestrator.executeSaga(saga.id);
  }
}

---

Correlation and Tracing

Correlation ID

Every saga has a unique correlation ID (UUID) that:

• Links related operations across services
• Enables distributed tracing
• Facilitates debugging with centralized logging
• Tracks saga through its lifecycle

// All logs include correlation ID
logger.info('Step executed', {
  correlationId: saga.correlationId,
  step: checkpoint.stepName,
  status: checkpoint.status,
});

// Query by correlation ID
const sagaEvents = await db.sagaCheckpoint.findMany({
  where: {
    sagaExecution: {
      correlationId: 'uuid-abc-123',
    },
  },
  orderBy: { stepOrder: 'asc' },
});

Audit Trail

Complete saga execution history is preserved:

SELECT
  se.correlation_id,
  se.status AS saga_status,
  se.compensation_status,
  sc.step_name,
  sc.step_order,
  sc.status AS step_status,
  sc.created_at,
  sc.error_message
FROM saga_execution se
JOIN saga_checkpoint sc ON sc.saga_execution_id = se.id
WHERE se.correlation_id = 'uuid-abc-123'
ORDER BY sc.step_order;

---

Best Practices

1. Keep Steps Idempotent

Each step must be safe to retry:

// Bad: Not idempotent
async execute(params) {
  await db.booking.update({
    where: { id: params.bookingId },
    data: { partySize: { increment: 1 } }  // ❌ Retry will double-count
  });
}

// Good: Idempotent
async execute(params) {
  await db.booking.update({
    where: { id: params.bookingId },
    data: { partySize: params.newPartySize }  // ✓ Same result on retry
  });
}

2. Store Compensation Data

Save information needed for rollback:

{
  name: 'reserve_inventory',
  execute: async (params) => {
    const reservation = await inventoryService.reserve(params.items);

    return {
      reservationId: reservation.id,  // ✓ Store for compensation
      itemsReserved: reservation.items
    };
  },
  compensate: async (data) => {
    await inventoryService.release(data.reservationId);
  }
}

3. Handle Partial Failures Gracefully

Some operations cannot be fully compensated:

{
  name: 'send_email',
  execute: async (params) => {
    await emailService.send(params.to, params.template);
    return { emailSent: true };
  },
  compensate: async (data) => {
    // Cannot un-send email
    // Send cancellation email instead
    await emailService.send(data.to, 'cancellation_template');
  }
}

4. Use Timeouts

Prevent sagas from running indefinitely:

const SAGA_TIMEOUT = 5 * 60 * 1000; // 5 minutes

async executeSaga(sagaId: string) {
  const timeoutPromise = new Promise((_, reject) =>
    setTimeout(() => reject(new Error('Saga timeout')), SAGA_TIMEOUT)
  );

  try {
    await Promise.race([
      this.executeSteps(sagaId),
      timeoutPromise
    ]);
  } catch (error) {
    if (error.message === 'Saga timeout') {
      await this.updateSagaStatus(sagaId, 'timeout');
    }
    throw error;
  }
}

5. Monitor Dead Letter Queue

Track sagas that failed compensation:

// Alert on compensation failures
const failedCompensations = await db.sagaExecution.count({
  where: {
    compensationStatus: { in: ['failed', 'partial'] },
  },
});

if (failedCompensations > 0) {
  await alertService.notify({
    severity: 'critical',
    message: `${failedCompensations} sagas require manual intervention`,
  });
}

---

Monitoring and Observability

Key Metrics

Track saga health:

// Success rate
const successRate = await db.sagaExecution.groupBy({
  by: ['status'],
  _count: true,
  where: {
    createdAt: { gte: last24Hours },
  },
});

// Average execution time
const avgDuration = await db.$queryRaw`
  SELECT
    saga_type,
    AVG(EXTRACT(EPOCH FROM (completed_at - started_at))) as avg_duration_seconds
  FROM saga_execution
  WHERE status = 'completed'
  GROUP BY saga_type
`;

// Compensation rate
const compensationRate =
  (await db.sagaExecution.count({
    where: {
      status: { in: ['compensating', 'failed'] },
      createdAt: { gte: last24Hours },
    },
  })) / totalSagas;

Dashboards

Monitor in real-time:

• Saga Throughput: Sagas completed per minute
• Failure Rate: % of sagas requiring compensation
• Average Duration: Time to complete by saga type
• Retry Count Distribution: How many sagas require retries
• Compensation Success Rate: % of successful rollbacks

---

Troubleshooting

Saga Stuck in Running State

Symptoms: Saga remains in running status for extended period

Causes:

• External service timeout
• Database connection lost
• Worker crash mid-execution

Resolution:

-- Manually mark saga for retry
UPDATE saga_execution
SET status = 'failed',
    next_retry_at = NOW() + INTERVAL '1 minute'
WHERE id = 'saga_id'
  AND status = 'running'
  AND started_at < NOW() - INTERVAL '10 minutes';

Compensation Failed

Symptoms: compensationStatus = 'failed' or 'partial'

Causes:

• External service unavailable during compensation
• Data inconsistency

Resolution:

1. Review saga checkpoints to identify failed compensation step
2. Manually execute compensation logic
3. Update saga status to failed with compensationStatus = 'completed'

// Manual compensation
const saga = await db.sagaExecution.findUnique({
  where: { id: 'saga_id' },
  include: { checkpoints: true },
});

for (const checkpoint of saga.checkpoints.reverse()) {
  if (!checkpoint.compensatedAt) {
    await manuallyCompensateStep(checkpoint);
  }
}

---

Related Documentation

• Database Schema - Saga table structure
• Architecture - Saga module design
• Outbox Pattern - Reliable event delivery

---

Summary

The Saga pattern enables:

• ✅ Distributed transaction coordination
• ✅ Automatic compensation on failure
• ✅ Retry logic with exponential backoff
• ✅ Complete audit trail
• ✅ Correlation tracking
• ✅ Graceful failure handling

Use sagas for complex, multi-step workflows where atomicity across services is required.