Saga Pattern with Temporal for Fault-Tolerant Distributed Transactions

sequenceDiagram
    participant Client
    participant OrderWorkflow
    participant InventoryService
    participant PaymentService
    participant NotificationService

    Client->>OrderWorkflow: StartOrderSaga(OrderDetails)
    OrderWorkflow->>InventoryService: ReserveInventoryActivity()
    InventoryService-->>OrderWorkflow: Reservation Success

    OrderWorkflow->>PaymentService: ProcessPaymentActivity()
    alt Payment Fails
        PaymentService-->>OrderWorkflow: Payment Failure
        OrderWorkflow->>InventoryService: ReleaseInventoryActivity() (Compensation)
        InventoryService-->>OrderWorkflow: Inventory Released
        OrderWorkflow-->>Client: Order Failed (Payment Error)
    else Payment Succeeds
        PaymentService-->>OrderWorkflow: Payment Success
        OrderWorkflow->>InventoryService: UpdateInventoryActivity()
        alt Inventory Update Fails
            InventoryService-->>OrderWorkflow: Inventory Failure
            OrderWorkflow->>PaymentService: RefundPaymentActivity() (Compensation)
            PaymentService-->>OrderWorkflow: Refund Success
            OrderWorkflow->>InventoryService: ReleaseInventoryActivity() (Compensation)
            InventoryService-->>OrderWorkflow: Inventory Released
            OrderWorkflow-->>Client: Order Failed (Inventory Error)
        else Inventory Update Succeeds
            InventoryService-->>OrderWorkflow: Inventory Success
            OrderWorkflow->>NotificationService: NotifyCustomerActivity()
            NotificationService-->>OrderWorkflow: Notification Sent
            OrderWorkflow-->>Client: Order Succeeded
        end
    end

// file: workflow.go
package app

import (
	"fmt"
	"time"

	"go.temporal.io/sdk/workflow"
)

// OrderSagaWorkflow orchestrates the e-commerce order saga.
func OrderSagaWorkflow(ctx workflow.Context, orderDetails OrderDetails) (string, error) {
	// Set up activity options with timeouts.
	// These are crucial for handling unavailable services.
	ao := workflow.ActivityOptions{
		StartToCloseTimeout: 10 * time.Second,
		RetryPolicy: &temporal.RetryPolicy{
			InitialInterval:    time.Second,
			BackoffCoefficient: 2.0,
			MaximumInterval:    100 * time.Second,
			MaximumAttempts:    5,
		},
	}
	ctx = workflow.WithActivityOptions(ctx, ao)

	logger := workflow.GetLogger(ctx)
	logger.Info("Order saga started.", "OrderID", orderDetails.OrderID)

	// Use a compensation stack. Functions are added to this stack and executed in LIFO order on failure.
	var compensations []func() error
	var executionErr error

	// Defer the execution of compensation functions.
	// This block will run if the workflow function returns, either successfully or with an error.
	defer func() {
		if executionErr != nil {
			logger.Error("Saga failed. Starting compensation.", "Error", executionErr)
			// Execute compensations in reverse order.
			for i := len(compensations) - 1; i >= 0; i-- {
				// Compensations themselves can fail. We execute them within a new disconnected context
				// to ensure they run even if the main workflow context is cancelled.
				disconnectedCtx, _ := workflow.NewDisconnectedContext(ctx)
				err := compensations[i]()
				if err != nil {
					logger.Error("Compensation activity failed.", "Error", err)
				}
			}
		}
	}()

	// 1. Reserve Inventory
	var reservationID string
	err := workflow.ExecuteActivity(ctx, ReserveInventoryActivity, orderDetails).Get(ctx, &reservationID)
	if err != nil {
		executionErr = fmt.Errorf("failed to reserve inventory: %w", err)
		return "", executionErr
	}
	// Add compensation for inventory reservation.
	compensations = append(compensations, func() error {
		return workflow.ExecuteActivity(ctx, ReleaseInventoryActivity, reservationID).Get(ctx, nil)
	})

	// 2. Process Payment
	var paymentID string
	err = workflow.ExecuteActivity(ctx, ProcessPaymentActivity, orderDetails).Get(ctx, &paymentID)
	if err != nil {
		executionErr = fmt.Errorf("failed to process payment: %w", err)
		return "", executionErr
	}
	// Add compensation for payment.
	compensations = append(compensations, func() error {
		return workflow.ExecuteActivity(ctx, RefundPaymentActivity, paymentID).Get(ctx, nil)
	})

	// 3. Update Inventory (confirm reservation)
	err = workflow.ExecuteActivity(ctx, UpdateInventoryActivity, reservationID).Get(ctx, nil)
	if err != nil {
		executionErr = fmt.Errorf("failed to update inventory: %w", err)
		return "", executionErr
	}
	// At this point, the financial transaction is committed. We remove the payment refund compensation.
	// If subsequent steps fail, we don't want to refund a successful order.
	compensations = compensations[:1] // Keep only the inventory release compensation (which is now a no-op but good practice)

	// 4. Notify Customer (non-critical step)
	notificationFuture := workflow.ExecuteActivity(ctx, NotifyCustomerActivity, orderDetails.OrderID)
	// We don't block on this. If it fails, we log it but don't fail the saga.
	// For a more robust implementation, this could trigger a separate workflow for retries.
	workflow.Go(ctx, func(ctx workflow.Context) {
		if err := notificationFuture.Get(ctx, nil); err != nil {
			logger.Error("Failed to notify customer, requires manual follow-up.", "OrderID", orderDetails.OrderID, "Error", err)
		}
	})

	logger.Info("Order saga completed successfully.", "OrderID", orderDetails.OrderID)
	return "ORDER_SUCCESSFUL", nil
}

// file: activities.go
package app

import (
	"context"
	"errors"
	"fmt"

	"go.temporal.io/sdk/activity"
)

// Mock dependencies
type PaymentService struct{}
func (ps *PaymentService) Charge(idempotencyKey, customerID string, amount int) (string, error) { 
    // In a real system, this would check a database table for the idempotencyKey
    // before processing the charge.
    fmt.Printf("Charging card with idempotency key: %s\n", idempotencyKey)
    if amount > 1000 {
        return "", errors.New("payment amount exceeds limit")
    }
    return fmt.Sprintf("payment-%s", idempotencyKey), nil
}
// ... other service methods

// ProcessPaymentActivity shows the idempotency pattern.
func ProcessPaymentActivity(ctx context.Context, orderDetails OrderDetails) (string, error) {
	logger := activity.GetLogger(ctx)
	info := activity.GetInfo(ctx)
	
	// Use the unique Workflow RunID and ActivityID as an idempotency key.
	// This key will be the same on retries of this specific activity execution.
	idempotencyKey := fmt.Sprintf("%s-%d", info.WorkflowExecution.RunID, info.ActivityID)
	logger.Info("Processing payment", "idempotencyKey", idempotencyKey)

	// In a real implementation, you would inject this service client.
	paymentClient := PaymentService{}
	paymentID, err := paymentClient.Charge(idempotencyKey, orderDetails.CustomerID, orderDetails.Amount)
	if err != nil {
		return "", err
	}

	return paymentID, nil
}

// ... Other activity implementations (ReserveInventory, RefundPayment, etc.)
func ReserveInventoryActivity(ctx context.Context, orderDetails OrderDetails) (string, error) { /* ... */ return "res-123", nil }
func ReleaseInventoryActivity(ctx context.Context, reservationID string) error { /* ... */ return nil }
func RefundPaymentActivity(ctx context.Context, paymentID string) error { /* ... */ return nil }
func UpdateInventoryActivity(ctx context.Context, reservationID string) error { /* ... */ return nil }
func NotifyCustomerActivity(ctx context.Context, orderID string) error { /* ... */ return nil }

// file: long_running_activity.go

func ProvisionResourceActivity(ctx context.Context, resourceSpec string) (string, error) {
    logger := activity.GetLogger(ctx)
    logger.Info("Starting resource provisioning.")

    // Simulate a long process with multiple steps.
    for i := 0; i < 10; i++ {
        time.Sleep(2 * time.Minute) // Simulate work being done

        // Record a heartbeat. If the worker crashes, the cluster knows much sooner.
        // The HeartbeatTimeout is configured in ActivityOptions.
        activity.RecordHeartbeat(ctx, fmt.Sprintf("Step %d completed", i+1))
    }

    logger.Info("Resource provisioning complete.")
    return "resource-id-xyz", nil
}

// In the workflow...
ao := workflow.ActivityOptions{
    StartToCloseTimeout: 30 * time.Minute,
    HeartbeatTimeout:    3 * time.Minute, // If no heartbeat is received in 3 mins, timeout.
    RetryPolicy: &temporal.RetryPolicy{ ... },
}

// In OrderSagaWorkflow

// Normal activities
ao := workflow.ActivityOptions{ ... }
ctx = workflow.WithActivityOptions(ctx, ao)

// Payment activity with a specific task queue
paymentAO := workflow.ActivityOptions{
    TaskQueue:           "payment-processing-queue",
    StartToCloseTimeout: 30 * time.Second,
    ...
}
paymentCtx := workflow.WithActivityOptions(ctx, paymentAO)

// ...

err = workflow.ExecuteActivity(paymentCtx, ProcessPaymentActivity, orderDetails).Get(ctx, &paymentID)

// Main worker pool for general tasks
mainWorker := worker.New(c, "main-task-queue", worker.Options{})
mainWorker.RegisterWorkflow(OrderSagaWorkflow)
mainWorker.RegisterActivity(ReserveInventoryActivity)
// ... register other activities

// Dedicated worker pool for payment processing
paymentWorker := worker.New(c, "payment-processing-queue", worker.Options{
    // Limit concurrent activities to 10 to respect rate limits.
    MaxConcurrentActivityExecutionSize: 10, 
})
paymentWorker.RegisterActivity(ProcessPaymentActivity)

// Start both workers
errg := new(errgroup.Group)
errg.Go(func() error { return mainWorker.Run(worker.InterruptCh()) })
errg.Go(func() error { return paymentWorker.Run(worker.InterruptCh()) })

if err := errg.Wait(); err != nil {
    log.Fatalln("Unable to start workers", err)
}

// At the start of OrderSagaWorkflow

// Set search attributes for this workflow execution.
// These must be pre-registered on the Temporal namespace.
searchAttributes := map[string]interface{}{
    "CustomStringField_OrderID":    orderDetails.OrderID,
    "CustomStringField_CustomerID": orderDetails.CustomerID,
    "CustomIntField_Amount":        orderDetails.Amount,
}
err := workflow.UpsertSearchAttributes(ctx, searchAttributes)
if err != nil {
    logger.Error("Failed to set search attributes", "Error", err)
}

# Find all failed orders for a specific customer
tctl workflow list --query "CustomStringField_CustomerID='cust-456' and ExecutionStatus='Failed'"