Saga Pattern: Managing Distributed Transactions Without Two-Phase Commit

snippet-2.sql

CREATE TABLE saga_log (
    id          BIGSERIAL PRIMARY KEY,
    saga_id     UUID        NOT NULL,
    state       TEXT        NOT NULL,
    payload     JSONB,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_saga_log_saga_id ON saga_log(saga_id);

CREATE TABLE sagas (
    saga_id     UUID        PRIMARY KEY,
    order_id    UUID        NOT NULL,
    state       TEXT        NOT NULL,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

snippet-3.go

type Orchestrator struct {
    db        *sql.DB
    publisher EventPublisher
}

func (o *Orchestrator) Handle(ctx context.Context, event SagaEvent) error {
    saga, err := o.loadSaga(ctx, event.SagaID)
    if err != nil {
        return fmt.Errorf("load saga: %w", err)
    }

    switch saga.State {
    case SagaStarted:
        return o.reserveInventory(ctx, saga)
    case InventoryReserved:
        return o.processPayment(ctx, saga)
    case PaymentProcessed:
        return o.confirmOrder(ctx, saga)
    case OrderConfirmed:
        return o.completeSaga(ctx, saga)

    // Compensation path
    case CompensatingPayment:
        return o.compensatePayment(ctx, saga)
    case CompensatingInventory:
        return o.compensateInventory(ctx, saga)
    }

    return nil
}

func (o *Orchestrator) reserveInventory(ctx context.Context, saga *OrderSaga) error {
    if err := o.publisher.Publish(ctx, "inventory.reserve", ReserveCommand{
        SagaID:  saga.SagaID,
        OrderID: saga.OrderID,
    }); err != nil {
        return err
    }
    return o.advanceSaga(ctx, saga, InventoryReserved)
}

snippet-4.go

func (o *Orchestrator) compensatePayment(ctx context.Context, saga *OrderSaga) error {
    if err := o.publisher.Publish(ctx, "payment.refund", RefundCommand{
        SagaID:  saga.SagaID,
        OrderID: saga.OrderID,
    }); err != nil {
        return err
    }
    // After publishing refund command, move to compensate inventory next
    return o.advanceSaga(ctx, saga, CompensatingInventory)
}

func (o *Orchestrator) compensateInventory(ctx context.Context, saga *OrderSaga) error {
    if err := o.publisher.Publish(ctx, "inventory.release", ReleaseCommand{
        SagaID:  saga.SagaID,
        OrderID: saga.OrderID,
    }); err != nil {
        return err
    }
    return o.advanceSaga(ctx, saga, SagaFailed)
}

snippet-5.go

func (svc *InventoryService) HandleReserve(ctx context.Context, cmd ReserveCommand) error {
    tx, err := svc.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // Idempotency guard — unique index on (saga_id, action)
    _, err = tx.ExecContext(ctx, `
        INSERT INTO processed_commands (saga_id, action, processed_at)
        VALUES ($1, 'reserve', now())
        ON CONFLICT (saga_id, action) DO NOTHING
    `, cmd.SagaID)
    if err != nil {
        return err
    }

    // Actual reservation logic
    _, err = tx.ExecContext(ctx, `
        UPDATE inventory SET reserved = reserved + $1
        WHERE product_id = $2 AND available >= $1
    `, cmd.Quantity, cmd.ProductID)
    if err != nil {
        return err
    }

    return tx.Commit()
}

snippet-6.yaml

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_PASSWORD: secret
    ports:
      - "5432:5432"

  kafka:
    image: confluentinc/cp-kafka:7.6.0
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  orchestrator:
    build: ./orchestrator
    environment:
      DATABASE_URL: postgres://postgres:secret@postgres:5432/sagas
      KAFKA_BROKERS: kafka:9092
    depends_on: [postgres, kafka]

  inventory-service:
    build: ./inventory
    environment:
      DATABASE_URL: postgres://postgres:secret@postgres:5432/inventory
      KAFKA_BROKERS: kafka:9092
    depends_on: [postgres, kafka]

  payment-service:
    build: ./payment
    environment:
      DATABASE_URL: postgres://postgres:secret@postgres:5432/payments
      KAFKA_BROKERS: kafka:9092
    depends_on: [postgres, kafka]

snippet-7.sql

-- Find sagas stuck in a non-terminal state for more than 10 minutes
SELECT saga_id, order_id, state, updated_at,
       now() - updated_at AS stuck_duration
FROM sagas
WHERE state NOT IN ('COMPLETED', 'FAILED')
  AND updated_at < now() - INTERVAL '10 minutes'
ORDER BY updated_at ASC;

snippet.go

type SagaState string

const (
    SagaStarted            SagaState = "STARTED"
    InventoryReserved      SagaState = "INVENTORY_RESERVED"
    PaymentProcessed       SagaState = "PAYMENT_PROCESSED"
    OrderConfirmed         SagaState = "ORDER_CONFIRMED"
    CompensatingInventory  SagaState = "COMPENSATING_INVENTORY"
    CompensatingPayment    SagaState = "COMPENSATING_PAYMENT"
    SagaFailed             SagaState = "FAILED"
    SagaCompleted          SagaState = "COMPLETED"
)

type OrderSaga struct {
    SagaID      string
    OrderID     string
    State       SagaState
    CreatedAt   time.Time
    UpdatedAt   time.Time
}