Session 17: Saga, Outbox & Distributed Transactions

Mục tiêu: Implement Saga pattern, Transactional Outbox, và giải quyết consistency trong distributed systems.

1. Vấn Đề: Distributed Transactions

Monolith:
  BEGIN TRANSACTION
    INSERT INTO Orders ...
    UPDATE Inventory ...
    INSERT INTO Payments ...
  COMMIT
  → ACID guaranteed ✅

Microservices:
  Order Service: INSERT INTO Orders ...     ✅
  Inventory Service: UPDATE Stock ...       ✅
  Payment Service: Charge credit card ...   ❌ FAILED!
  → Order created, stock reserved, nhưng payment failed!
  → Inconsistent state! 💥

2PC (Two-Phase Commit) - Tại Sao KHÔNG Dùng?

❌ Coordinator là single point of failure
❌ Blocking: tất cả participants lock resources
❌ Không scale trong distributed systems
❌ Latency cao

→ Microservices dùng SAGA pattern thay thế

2. Saga Pattern

Orchestration (Central coordinator)

┌────────────┐
│ Saga       │
│Orchestrator│
│            │
│ 1. Create  │──→ Order Service ──→ Order Created
│ 2. Reserve │──→ Inventory Service ──→ Stock Reserved
│ 3. Charge  │──→ Payment Service ──→ Payment OK
│ 4. Confirm │──→ Order Service ──→ Order Confirmed
│            │
│ On failure:│
│ Compensate │──→ Undo previous steps trong reverse order
└────────────┘

csharp
// MassTransit Saga State Machine
public class OrderSagaState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; } = null!;
    public Guid OrderId { get; set; }
    public Guid CustomerId { get; set; }
    public decimal TotalAmount { get; set; }
    public DateTime CreatedAt { get; set; }
    public string? FailureReason { get; set; }
}

public class OrderSaga : MassTransitStateMachine<OrderSagaState>
{
    // States
    public State OrderCreated { get; } = null!;
    public State StockReserved { get; } = null!;
    public State PaymentProcessed { get; } = null!;
    public State Completed { get; } = null!;
    public State Faulted { get; } = null!;

    // Events
    public Event<OrderSubmitted> OrderSubmitted { get; } = null!;
    public Event<StockReserved> StockReserved_ { get; } = null!;
    public Event<StockReservationFailed> StockReservationFailed { get; } = null!;
    public Event<PaymentCompleted> PaymentCompleted_ { get; } = null!;
    public Event<PaymentFailed> PaymentFailed_ { get; } = null!;

    public OrderSaga()
    {
        InstanceState(x => x.CurrentState);

        Event(() => OrderSubmitted, x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => StockReserved_, x => x.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentCompleted_, x => x.CorrelateById(ctx => ctx.Message.OrderId));

        Initially(
            When(OrderSubmitted)
                .Then(ctx =>
                {
                    ctx.Saga.OrderId = ctx.Message.OrderId;
                    ctx.Saga.CustomerId = ctx.Message.CustomerId;
                    ctx.Saga.TotalAmount = ctx.Message.TotalAmount;
                    ctx.Saga.CreatedAt = DateTime.UtcNow;
                })
                .Publish(ctx => new ReserveStock
                {
                    OrderId = ctx.Saga.OrderId,
                    Items = ctx.Message.Items
                })
                .TransitionTo(OrderCreated)
        );

        During(OrderCreated,
            When(StockReserved_)
                .Publish(ctx => new ProcessPayment
                {
                    OrderId = ctx.Saga.OrderId,
                    Amount = ctx.Saga.TotalAmount,
                    CustomerId = ctx.Saga.CustomerId
                })
                .TransitionTo(StockReserved),
            When(StockReservationFailed)
                .Then(ctx => ctx.Saga.FailureReason = ctx.Message.Reason)
                .Publish(ctx => new CancelOrder { OrderId = ctx.Saga.OrderId })
                .TransitionTo(Faulted)
                .Finalize()
        );

        During(StockReserved,
            When(PaymentCompleted_)
                .Publish(ctx => new ConfirmOrder { OrderId = ctx.Saga.OrderId })
                .TransitionTo(Completed)
                .Finalize(),
            When(PaymentFailed_)
                .Then(ctx => ctx.Saga.FailureReason = ctx.Message.Reason)
                // Compensate: release reserved stock
                .Publish(ctx => new ReleaseStock { OrderId = ctx.Saga.OrderId })
                .Publish(ctx => new CancelOrder { OrderId = ctx.Saga.OrderId })
                .TransitionTo(Faulted)
                .Finalize()
        );
    }
}

Choreography (Event-driven, no coordinator)

Order Service ──publish──→ OrderSubmitted
                                │
Inventory Service ──listen──→ ReserveStock ──publish──→ StockReserved
                                                             │
Payment Service ──listen──→ ProcessPayment ──publish──→ PaymentCompleted
                                                             │
Order Service ──listen──→ ConfirmOrder

On failure: mỗi service tự compensate khi nhận failure event

Orchestration vs Choreography

3. Transactional Outbox Pattern

Vấn Đề: Dual Write

❌ Dual Write Problem:
  1. Save order to DB          ✅ Success
  2. Publish event to broker   ❌ Failed (broker down)
  → DB có data, nhưng event KHÔNG được publish!
  → Downstream services KHÔNG biết order mới!

Outbox Pattern Solution

┌─────────────────────────────────────────┐
│              Order Service               │
│                                          │
│  1. BEGIN TRANSACTION                    │
│     INSERT INTO Orders (...)             │
│     INSERT INTO OutboxMessages (...)     │ ← Same transaction!
│  COMMIT                                 │
│                                          │
│  2. Background Worker:                   │
│     SELECT FROM OutboxMessages           │
│     WHERE ProcessedAt IS NULL            │
│     → Publish to Message Broker          │
│     → UPDATE OutboxMessages              │
│       SET ProcessedAt = NOW()            │
└─────────────────────────────────────────┘

csharp
// Outbox Message entity
public class OutboxMessage
{
    public Guid Id { get; set; }
    public string Type { get; set; } = null!;     // Event type name
    public string Payload { get; set; } = null!;   // JSON serialized event
    public DateTime CreatedAt { get; set; }
    public DateTime? ProcessedAt { get; set; }
    public int RetryCount { get; set; }
    public string? Error { get; set; }
}

// Save entity + outbox message trong cùng transaction
public class OrderService(AppDbContext db)
{
    public async Task SubmitOrderAsync(Order order)
    {
        order.Submit();

        // Save entity
        db.Orders.Add(order);

        // Save outbox message (same transaction!)
        var integrationEvent = new OrderSubmittedIntegrationEvent
        {
            OrderId = order.Id,
            TotalAmount = order.Total.Amount
        };

        db.OutboxMessages.Add(new OutboxMessage
        {
            Id = Guid.NewGuid(),
            Type = nameof(OrderSubmittedIntegrationEvent),
            Payload = JsonSerializer.Serialize(integrationEvent),
            CreatedAt = DateTime.UtcNow
        });

        await db.SaveChangesAsync(); // Atomic!
    }
}

// Background worker: publish outbox messages
public class OutboxProcessor(
    IServiceProvider serviceProvider,
    ILogger<OutboxProcessor> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try
            {
                using var scope = serviceProvider.CreateScope();
                var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
                var publisher = scope.ServiceProvider.GetRequiredService<IPublishEndpoint>();

                var messages = await db.OutboxMessages
                    .Where(m => m.ProcessedAt == null && m.RetryCount < 5)
                    .OrderBy(m => m.CreatedAt)
                    .Take(100)
                    .ToListAsync(ct);

                foreach (var message in messages)
                {
                    try
                    {
                        var eventType = Type.GetType(message.Type)!;
                        var @event = JsonSerializer.Deserialize(message.Payload, eventType)!;
                        await publisher.Publish(@event, ct);

                        message.ProcessedAt = DateTime.UtcNow;
                    }
                    catch (Exception ex)
                    {
                        message.RetryCount++;
                        message.Error = ex.Message;
                        logger.LogError(ex, "Failed to publish outbox message {Id}", message.Id);
                    }
                }

                await db.SaveChangesAsync(ct);
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "Outbox processor error");
            }

            await Task.Delay(TimeSpan.FromSeconds(5), ct);
        }
    }
}

4. Inbox Pattern (Idempotent Consumer)

csharp
// Đảm bảo mỗi message chỉ process 1 lần
public class InboxConsumer<TMessage>(AppDbContext db) where TMessage : class
{
    public async Task<bool> TryProcessAsync(
        Guid messageId, Func<Task> handler, CancellationToken ct)
    {
        // Check if already processed
        var exists = await db.InboxMessages
            .AnyAsync(m => m.MessageId == messageId, ct);

        if (exists) return false; // Already processed

        // Process
        await handler();

        // Record as processed
        db.InboxMessages.Add(new InboxMessage
        {
            MessageId = messageId,
            ProcessedAt = DateTime.UtcNow
        });

        await db.SaveChangesAsync(ct); // Atomic
        return true;
    }
}

5. Compensating Transactions

csharp
// Ví dụ: Payment processing flow
// 
// Happy path:
// 1. Reserve Stock ✅
// 2. Charge Credit Card ✅
// 3. Confirm Order ✅
//
// Compensation khi step 2 fails:
// 1. Release Stock (undo step 1)
// 2. Cancel Order (undo order creation)
//
// ⚠️ Compensations PHẢI idempotent!
// ⚠️ Compensations có thể fail → cần retry + dead letter

public class ReleaseStockConsumer(InventoryDbContext db) : IConsumer<ReleaseStock>
{
    public async Task Consume(ConsumeContext<ReleaseStock> context)
    {
        var reservation = await db.StockReservations
            .FirstOrDefaultAsync(r => r.OrderId == context.Message.OrderId);

        if (reservation is null) return; // Already released (idempotent)

        reservation.Status = ReservationStatus.Released;
        await db.SaveChangesAsync();
    }
}

Tổng Kết Session 17

Bài tập: Implement Order Saga với MassTransit State Machine: CreateOrder → ReserveStock → ProcessPayment → ConfirmOrder. Add compensating actions khi payment fails.