分散トランザクションにおけるサガパターン - Goによる例

type OrderSagaOrchestrator struct {
    orderService    OrderService
    paymentService  PaymentService
    inventoryService InventoryService
    shippingService ShippingService
}

func (o *OrderSagaOrchestrator) CreateOrder(order Order) error {
    sagaID := generateSagaID()
    
    // ステップ1: 注文を作成
    orderID, err := o.orderService.Create(order)
    if err != nil {
        return err
    }
    
    // ステップ2: 在庫を予約
    if err := o.inventoryService.Reserve(order.Items); err != nil {
        o.orderService.Cancel(orderID) // 補償
        return err
    }
    
    // ステップ3: 支払いを処理
    paymentID, err := o.paymentService.Charge(order.CustomerID, order.Total)
    if err != nil {
        o.inventoryService.Release(order.Items) // 補償
        o.orderService.Cancel(orderID)          // 補償
        return err
    }
    
    // ステップ4: 出荷を作成
    if err := o.shippingService.CreateShipment(orderID); err != nil {
        o.paymentService.Refund(paymentID)      // 補償
        o.inventoryService.Release(order.Items) // 補償
        o.orderService.Cancel(orderID)          // 補償
        return err
    }
    
    return nil
}

// 注文サービス
type OrderService struct {
    eventBus EventBus
    repo     OrderRepository
}

func (s *OrderService) CreateOrder(order Order) (string, error) {
    orderID, err := s.repo.Save(order)
    if err != nil {
        return "", err
    }
    
    s.eventBus.Publish("OrderCreated", OrderCreatedEvent{
        OrderID:    orderID,
        CustomerID: order.CustomerID,
        Items:      order.Items,
        Total:      order.Total,
    })
    
    return orderID, nil
}

// 注記: s.repo.Saveの後にs.eventBus.Publishが続くのは二重書き込みです。
// 本番環境では、トランザクショナルアウトボックスパターンに置き換えて、
// イベントが注文行とアトミックに書き込まれ、リレーによって公開されるようにします。

func (s *OrderService) HandlePaymentFailed(event PaymentFailedEvent) error {
    return s.repo.Cancel(event.OrderID) // 補償
}

// 支払いサービス
type PaymentService struct {
    eventBus EventBus
    client   PaymentClient
}

func (s *PaymentService) HandleOrderCreated(event OrderCreatedEvent) {
    paymentID, err := s.client.Charge(event.CustomerID, event.Total)
    if err != nil {
        s.eventBus.Publish("PaymentFailed", PaymentFailedEvent{
            OrderID: event.OrderID,
        })
        return
    }
    
    s.eventBus.Publish("PaymentSucceeded", PaymentSucceededEvent{
        OrderID:   event.OrderID,
        PaymentID: paymentID,
    })
}

func (s *PaymentService) HandleInventoryReservationFailed(event InventoryReservationFailedEvent) error {
    // 補償: 支払いを返金
    return s.client.Refund(event.PaymentID)
}

func (s *PaymentService) Refund(paymentID string) error {
    // すでに返金済みかチェック
    payment, err := s.getPayment(paymentID)
    if err != nil {
        return err
    }
    
    if payment.Status == "refunded" {
        return nil // すでに返金済み、冪等
    }
    
    // 返金を処理
    return s.processRefund(paymentID)
}

type SagaState struct {
    ID           string
    Status       SagaStatus
    Steps        []SagaStep
    CurrentStep  int
    CreatedAt    time.Time
    UpdatedAt    time.Time
}

type SagaStep struct {
    Service     string
    Operation   string
    Status      StepStatus
    Compensated bool
    Data        map[string]interface{}
}

type SagaOrchestrator struct {
    timeout time.Duration
}

func (o *SagaOrchestrator) ExecuteWithTimeout(step SagaStep) error {
    ctx, cancel := context.WithTimeout(context.Background(), o.timeout)
    defer cancel()
    
    done := make(chan error, 1)
    go func() {
        done <- step.Execute()
    }()
    
    select {
    case err := <-done:
        return err
    case <-ctx.Done():
        // タイムアウトが発生、補償
        if err := step.Compensate(); err != nil {
            return fmt.Errorf("compensation failed: %w", err)
        }
        return fmt.Errorf("step %s timed out after %v", step.Name(), o.timeout)
    }
}

func retryWithBackoff(operation func() error, maxRetries int) error {
    backoff := time.Second
    for i := 0; i < maxRetries; i++ {
        err := operation()
        if err == nil {
            return nil
        }
        
        if !isTransientError(err) {
            return err
        }
        
        time.Sleep(backoff)
        backoff *= 2
    }
    return fmt.Errorf("operation failed after %d retries", maxRetries)
}

type SagaEvent struct {
    SagaID    string
    EventType string
    Payload   []byte
    Timestamp time.Time
    Version   int64
}

type SagaEventStore struct {
    store EventRepository
}

func (s *SagaEventStore) AppendEvent(sagaID string, eventType string, payload interface{}) error {
    data, err := json.Marshal(payload)
    if err != nil {
        return fmt.Errorf("failed to marshal payload: %w", err)
    }
    
    version, err := s.store.GetNextVersion(sagaID)
    if err != nil {
        return fmt.Errorf("failed to get version: %w", err)
    }
    
    event := SagaEvent{
        SagaID:    sagaID,
        EventType: eventType,
        Payload:   data,
        Timestamp: time.Now(),
        Version:   version,
    }
    
    return s.store.Save(event)
}

func (s *SagaEventStore) ReplaySaga(sagaID string) (*Saga, error) {
    events, err := s.store.GetEvents(sagaID)
    if err != nil {
        return nil, fmt.Errorf("failed to get events: %w", err)
    }
    
    saga := NewSaga()
    for _, event := range events {
        if err := saga.Apply(event); err != nil {
            return nil, fmt.Errorf("failed to apply event: %w", err)
        }
    }
    
    return saga, nil
}

func (o *OrderSagaOrchestrator) CreateOrder(order Order) error {
    span := tracer.StartSpan("saga.create_order")
    defer span.Finish()
    
    span.SetTag("saga.id", sagaID)
    span.SetTag("order.id", order.ID)
    
    logger.WithFields(log.Fields{
        "saga_id": sagaID,
        "order_id": order.ID,
        "step": "create_order",
    }).Info("Saga started")
    
    // ... saga execution
    
    return nil
}