Event Sourcing Series Part 4: Implementing Sagas with Durable Functions

This is Part 4 of a 5-part series on Event Sourcing and Saga Orchestration with Azure.

Building a saga orchestrator from scratch is painful. You need to handle state persistence, retries, timeouts, and compensation logic. Azure Durable Functions does all of this for you.

Why Durable Functions for Sagas?

Durable Functions provides:

• Automatic state persistence - Saga state survives function restarts
• Replay-safe execution - Functions replay from checkpoints, not from scratch
• Built-in retry policies - Configurable retry with backoff
• Timers and timeouts - Wait for events with deadlines
• Sub-orchestrations - Compose complex workflows

The orchestrator function looks like normal async code, but Durable Functions handles all the distributed systems complexity.

The Order Saga Implementation

Let's build the complete order saga from Part 3.

Project Structure

OrderSaga/
├── Orchestrators/
│   └── OrderSagaOrchestrator.cs
├── Activities/
│   ├── CreateOrderActivity.cs
│   ├── ReserveInventoryActivity.cs
│   ├── ProcessPaymentActivity.cs
│   ├── ConfirmOrderActivity.cs
│   └── Compensation/
│       ├── CancelOrderActivity.cs
│       ├── ReleaseInventoryActivity.cs
│       └── RefundPaymentActivity.cs
├── Models/
│   ├── OrderSagaInput.cs
│   └── OrderSagaState.cs
└── Triggers/
    └── StartOrderSagaTrigger.cs

The Saga Input and State

public record OrderSagaInput(
    string CustomerId,
    List<OrderItemInput> Items,
    PaymentDetails Payment,
    ShippingAddress ShippingAddress
);

public record OrderItemInput(
    string ProductId,
    string ProductName,
    decimal UnitPrice,
    int Quantity
);

public class OrderSagaState
{
    public string OrderId { get; set; }
    public string PaymentTransactionId { get; set; }
    public string InventoryReservationId { get; set; }
    public List<string> CompletedSteps { get; set; } = new();
}

The Orchestrator Function

This is where the magic happens:

public static class OrderSagaOrchestrator
{
    [Function(nameof(OrderSagaOrchestrator))]
    public static async Task<OrderSagaResult> RunOrchestrator(
        [OrchestrationTrigger] TaskOrchestrationContext context)
    {
        var input = context.GetInput<OrderSagaInput>();
        var state = new OrderSagaState();
        var logger = context.CreateReplaySafeLogger(nameof(OrderSagaOrchestrator));

        try
        {
            // Step 1: Create Order
            logger.LogInformation("Creating order for customer {CustomerId}", input.CustomerId);
            state.OrderId = await context.CallActivityAsync<string>(
                nameof(CreateOrderActivity),
                new CreateOrderInput(input.CustomerId, input.Items));
            state.CompletedSteps.Add("CreateOrder");

            // Step 2: Reserve Inventory
            logger.LogInformation("Reserving inventory for order {OrderId}", state.OrderId);
            state.InventoryReservationId = await context.CallActivityAsync<string>(
                nameof(ReserveInventoryActivity),
                new ReserveInventoryInput(state.OrderId, input.Items));
            state.CompletedSteps.Add("ReserveInventory");

            // Step 3: Process Payment
            logger.LogInformation("Processing payment for order {OrderId}", state.OrderId);
            state.PaymentTransactionId = await context.CallActivityAsync<string>(
                nameof(ProcessPaymentActivity),
                new ProcessPaymentInput(state.OrderId, input.Payment, CalculateTotal(input.Items)));
            state.CompletedSteps.Add("ProcessPayment");

            // Step 4: Confirm Order
            logger.LogInformation("Confirming order {OrderId}", state.OrderId);
            await context.CallActivityAsync(
                nameof(ConfirmOrderActivity),
                new ConfirmOrderInput(state.OrderId, state.PaymentTransactionId));
            state.CompletedSteps.Add("ConfirmOrder");

            return new OrderSagaResult(true, state.OrderId, null);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Saga failed at step, starting compensation");

            // Compensate in reverse order
            await CompensateAsync(context, state, logger);

            return new OrderSagaResult(false, state.OrderId, ex.Message);
        }
    }

    private static async Task CompensateAsync(
        TaskOrchestrationContext context,
        OrderSagaState state,
        ILogger logger)
    {
        // Compensate in reverse order of completion
        var stepsToCompensate = state.CompletedSteps.AsEnumerable().Reverse();

        foreach (var step in stepsToCompensate)
        {
            try
            {
                switch (step)
                {
                    case "ProcessPayment":
                        logger.LogInformation("Compensating: Refunding payment");
                        await context.CallActivityAsync(
                            nameof(RefundPaymentActivity),
                            new RefundPaymentInput(state.PaymentTransactionId));
                        break;

                    case "ReserveInventory":
                        logger.LogInformation("Compensating: Releasing inventory");
                        await context.CallActivityAsync(
                            nameof(ReleaseInventoryActivity),
                            new ReleaseInventoryInput(state.InventoryReservationId));
                        break;

                    case "CreateOrder":
                        logger.LogInformation("Compensating: Cancelling order");
                        await context.CallActivityAsync(
                            nameof(CancelOrderActivity),
                            new CancelOrderInput(state.OrderId, "Saga compensation"));
                        break;
                }
            }
            catch (Exception compEx)
            {
                // Log but continue - try to compensate as much as possible
                logger.LogError(compEx, "Compensation failed for step {Step}", step);
            }
        }
    }

    private static decimal CalculateTotal(List<OrderItemInput> items)
        => items.Sum(i => i.UnitPrice * i.Quantity);
}

Activity Functions

Each activity is a single step that calls an external service:

public static class CreateOrderActivity
{
    [Function(nameof(CreateOrderActivity))]
    public static async Task<string> Run(
        [ActivityTrigger] CreateOrderInput input,
        FunctionContext context)
    {
        var logger = context.GetLogger(nameof(CreateOrderActivity));
        var orderService = context.InstanceServices.GetRequiredService<IOrderService>();

        logger.LogInformation("Creating order for customer {CustomerId}", input.CustomerId);

        var orderId = await orderService.CreateOrderAsync(
            input.CustomerId,
            input.Items.Select(i => new OrderItem(i.ProductId, i.ProductName, i.UnitPrice, i.Quantity)).ToList());

        return orderId;
    }
}

public static class ReserveInventoryActivity
{
    [Function(nameof(ReserveInventoryActivity))]
    public static async Task<string> Run(
        [ActivityTrigger] ReserveInventoryInput input,
        FunctionContext context)
    {
        var logger = context.GetLogger(nameof(ReserveInventoryActivity));
        var inventoryService = context.InstanceServices.GetRequiredService<IInventoryService>();

        logger.LogInformation("Reserving inventory for order {OrderId}", input.OrderId);

        var reservationId = await inventoryService.ReserveAsync(input.OrderId, input.Items);

        return reservationId;
    }
}

public static class ProcessPaymentActivity
{
    [Function(nameof(ProcessPaymentActivity))]
    public static async Task<string> Run(
        [ActivityTrigger] ProcessPaymentInput input,
        FunctionContext context)
    {
        var logger = context.GetLogger(nameof(ProcessPaymentActivity));
        var paymentService = context.InstanceServices.GetRequiredService<IPaymentService>();

        logger.LogInformation("Processing payment for order {OrderId}, amount {Amount}",
            input.OrderId, input.Amount);

        var transactionId = await paymentService.ProcessAsync(
            input.OrderId,
            input.PaymentDetails,
            input.Amount);

        return transactionId;
    }
}

Compensation Activities

public static class RefundPaymentActivity
{
    [Function(nameof(RefundPaymentActivity))]
    public static async Task Run(
        [ActivityTrigger] RefundPaymentInput input,
        FunctionContext context)
    {
        var logger = context.GetLogger(nameof(RefundPaymentActivity));
        var paymentService = context.InstanceServices.GetRequiredService<IPaymentService>();

        logger.LogInformation("Refunding payment {TransactionId}", input.TransactionId);

        await paymentService.RefundAsync(input.TransactionId);
    }
}

public static class ReleaseInventoryActivity
{
    [Function(nameof(ReleaseInventoryActivity))]
    public static async Task Run(
        [ActivityTrigger] ReleaseInventoryInput input,
        FunctionContext context)
    {
        var logger = context.GetLogger(nameof(ReleaseInventoryActivity));
        var inventoryService = context.InstanceServices.GetRequiredService<IInventoryService>();

        logger.LogInformation("Releasing inventory reservation {ReservationId}", input.ReservationId);

        await inventoryService.ReleaseReservationAsync(input.ReservationId);
    }
}

Adding Retry Policies

Real services fail. Add retry policies:

private static readonly TaskOptions RetryOptions = new(
    new TaskRetryOptions(
        firstRetryInterval: TimeSpan.FromSeconds(5),
        maxNumberOfAttempts: 3)
    {
        BackoffCoefficient = 2.0,
        MaxRetryInterval = TimeSpan.FromMinutes(1),
        RetryTimeout = TimeSpan.FromMinutes(5)
    });

// In orchestrator:
state.PaymentTransactionId = await context.CallActivityAsync<string>(
    nameof(ProcessPaymentActivity),
    new ProcessPaymentInput(state.OrderId, input.Payment, total),
    RetryOptions);  // <-- Automatic retries!

Adding Timeouts

Don't wait forever for external services:

// Wait for payment with timeout
using var cts = new CancellationTokenSource();
var paymentTask = context.CallActivityAsync<string>(
    nameof(ProcessPaymentActivity),
    paymentInput,
    RetryOptions);

var timeoutTask = context.CreateTimer(TimeSpan.FromMinutes(5), cts.Token);

var winner = await Task.WhenAny(paymentTask, timeoutTask);

if (winner == timeoutTask)
{
    throw new TimeoutException("Payment processing timed out");
}

cts.Cancel(); // Cancel the timer
state.PaymentTransactionId = await paymentTask;

Human Intervention Pattern

Some failures need manual resolution:

// In orchestrator - wait for human approval after failure
if (requiresManualReview)
{
    logger.LogWarning("Order {OrderId} requires manual review", state.OrderId);

    // Wait for external event (human clicks "Approve" in admin UI)
    var approved = await context.WaitForExternalEvent<bool>(
        "ManualApproval",
        TimeSpan.FromHours(24));

    if (!approved)
    {
        throw new SagaRejectedException("Order rejected during manual review");
    }
}

Starting the Saga

HTTP trigger to kick off the orchestration:

public static class StartOrderSagaTrigger
{
    [Function("StartOrderSaga")]
    public static async Task<HttpResponseData> Run(
        [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequestData req,
        [DurableClient] DurableTaskClient client,
        FunctionContext context)
    {
        var input = await req.ReadFromJsonAsync<OrderSagaInput>();

        // Start the orchestration
        var instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
            nameof(OrderSagaOrchestrator),
            input);

        var response = req.CreateResponse(HttpStatusCode.Accepted);
        await response.WriteAsJsonAsync(new
        {
            InstanceId = instanceId,
            StatusUrl = $"/api/saga-status/{instanceId}"
        });

        return response;
    }

    [Function("GetSagaStatus")]
    public static async Task<HttpResponseData> GetStatus(
        [HttpTrigger(AuthorizationLevel.Function, "get", Route = "saga-status/{instanceId}")]
        HttpRequestData req,
        [DurableClient] DurableTaskClient client,
        string instanceId)
    {
        var metadata = await client.GetInstanceAsync(instanceId);

        var response = req.CreateResponse(HttpStatusCode.OK);
        await response.WriteAsJsonAsync(new
        {
            InstanceId = instanceId,
            Status = metadata?.RuntimeStatus.ToString(),
            CreatedAt = metadata?.CreatedAt,
            LastUpdatedAt = metadata?.LastUpdatedAt,
            Output = metadata?.ReadOutputAs<OrderSagaResult>()
        });

        return response;
    }
}

Testing the Saga

Unit test the orchestrator with the Durable Functions test framework:

[Fact]
public async Task OrderSaga_Success_CompletesAllSteps()
{
    // Arrange
    var context = new Mock<TaskOrchestrationContext>();
    var input = new OrderSagaInput(
        CustomerId: "cust-123",
        Items: new List<OrderItemInput>
        {
            new("prod-1", "Widget", 10.00m, 2)
        },
        Payment: new PaymentDetails("card", "4111111111111111"),
        ShippingAddress: new ShippingAddress("123 Main St", "City", "12345")
    );

    context.Setup(c => c.GetInput<OrderSagaInput>()).Returns(input);

    context.Setup(c => c.CallActivityAsync<string>(
        nameof(CreateOrderActivity), It.IsAny<CreateOrderInput>()))
        .ReturnsAsync("order-456");

    context.Setup(c => c.CallActivityAsync<string>(
        nameof(ReserveInventoryActivity), It.IsAny<ReserveInventoryInput>()))
        .ReturnsAsync("reservation-789");

    context.Setup(c => c.CallActivityAsync<string>(
        nameof(ProcessPaymentActivity), It.IsAny<ProcessPaymentInput>()))
        .ReturnsAsync("payment-abc");

    // Act
    var result = await OrderSagaOrchestrator.RunOrchestrator(context.Object);

    // Assert
    Assert.True(result.Success);
    Assert.Equal("order-456", result.OrderId);
}

[Fact]
public async Task OrderSaga_PaymentFails_CompensatesCorrectly()
{
    // Arrange
    var context = new Mock<TaskOrchestrationContext>();
    // ... setup similar to above ...

    context.Setup(c => c.CallActivityAsync<string>(
        nameof(ProcessPaymentActivity), It.IsAny<ProcessPaymentInput>()))
        .ThrowsAsync(new PaymentFailedException("Insufficient funds"));

    // Act
    var result = await OrderSagaOrchestrator.RunOrchestrator(context.Object);

    // Assert
    Assert.False(result.Success);
    Assert.Contains("Insufficient funds", result.FailureReason);

    // Verify compensation was called
    context.Verify(c => c.CallActivityAsync(
        nameof(ReleaseInventoryActivity), It.IsAny<ReleaseInventoryInput>()), Times.Once);
    context.Verify(c => c.CallActivityAsync(
        nameof(CancelOrderActivity), It.IsAny<CancelOrderInput>()), Times.Once);
}

Monitoring and Observability

Durable Functions Storage

Saga state is stored in Azure Storage by default. Check the DurableFunctionsHubHistory table for execution history.

Application Insights

Enable Application Insights for full tracing:

// host.json
{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": false // Disable sampling for saga tracing
      }
    }
  },
  "extensions": {
    "durableTask": {
      "tracing": {
        "traceInputsAndOutputs": true,
        "traceReplayEvents": false
      }
    }
  }
}

Key Takeaways

1. Durable Functions handles the hard parts - State persistence, retries, replay
2. Activities are your service calls - Keep them focused and idempotent
3. Compensation is explicit - Track completed steps, compensate in reverse
4. Add retries and timeouts - External services will fail
5. Test with mocks - The framework supports unit testing

Coming Up Next

In Part 5, we'll tie everything together with interview-ready knowledge: DDD + Event Sourcing + Sagas, common questions, and when to tell interviewers "we don't need this complexity."

---

This is Part 4 of a 5-part series on Event Sourcing and Saga Orchestration:

• Part 1: The Honest Truth About Event Sourcing
• Part 2: Event Sourcing with Azure - Building Blocks
• Part 3: Saga Orchestration - Distributed Transactions Done Right
• Part 4: Implementing a Saga Orchestrator with Azure Durable Functions (You are here)
• Part 5: Putting It All Together - Interview-Ready Knowledge