CQRS command handlers don’t inherently manage transactions across themselves; instead, each command handler is responsible for its own transactional integrity.
Let’s see this in action. Imagine a simple e-commerce system with OrderSaga.
public class OrderSaga :
CorrelateBy<string>,
InitiatedBy<CreateOrderCommand>,
Orchestrates<OrderCreatedEvent>,
IHandle<PaymentProcessedEvent>,
IHandle<InventoryAllocatedEvent>,
IHandle<OrderShippedEvent>,
IHandle<OrderFailedEvent>
{
private Guid _orderId;
private bool _paymentProcessed = false;
private bool _inventoryAllocated = false;
private bool _orderShipped = false;
public string CorrelationId { get; private set; }
public Task Handle(CreateOrderCommand message)
{
_orderId = message.OrderId;
CorrelationId = message.CorrelationId;
// Publish OrderCreatedEvent without explicit transaction management here.
// The event bus or underlying infrastructure might handle this.
return Publish(new OrderCreatedEvent
{
OrderId = _orderId,
CorrelationId = CorrelationId,
CustomerId = message.CustomerId,
OrderTotal = message.OrderTotal
});
}
public Task Handle(OrderCreatedEvent message)
{
// Once order is created, initiate payment and inventory allocation.
// These are separate commands, each handled independently.
return Task.WhenAll(
Publish(new ProcessPaymentCommand
{
OrderId = _orderId,
CorrelationId = CorrelationId,
Amount = message.OrderTotal
}),
Publish(new AllocateInventoryCommand
{
OrderId = _orderId,
CorrelationId = CorrelationId,
Items = message.Items
})
);
}
public Task Handle(PaymentProcessedEvent message)
{
_paymentProcessed = true;
TryCompleteOrder();
return Task.CompletedTask;
}
public Task Handle(InventoryAllocatedEvent message)
{
_inventoryAllocated = true;
TryCompleteOrder();
return Task.CompletedTask;
}
public Task Handle(OrderShippedEvent message)
{
_orderShipped = true;
// Final state, saga completion.
return CompleteAsync();
}
public Task Handle(OrderFailedEvent message)
{
// Handle failure scenarios, potentially compensating actions.
return Task.CompletedTask;
}
private void TryCompleteOrder()
{
if (_paymentProcessed && _inventoryAllocated && !_orderShipped)
{
// Publish ShipOrderCommand only when both payment and inventory are ready.
// This command handler will manage its own transaction.
Publish(new ShipOrderCommand
{
OrderId = _orderId,
CorrelationId = CorrelationId
});
}
}
}
The OrderSaga orchestrates the workflow. When CreateOrderCommand arrives, it publishes an OrderCreatedEvent. This event then triggers two separate commands: ProcessPaymentCommand and AllocateInventoryCommand. Each of these commands, when handled by their respective command handlers, should be transactional.
The core problem CQRS solves here is separating the command (intent to change state) from the event (fact that state has changed). The saga is a pattern that observes these events and issues new commands to drive the process forward. Crucially, the saga itself doesn’t typically contain the transaction for the underlying state changes. Instead, it relies on the individual command handlers to ensure their atomic updates.
Think of it like this: the saga is the conductor, and each command handler is a musician playing their instrument. The conductor tells them when to play (issue commands), but each musician is responsible for playing their note correctly (executing their command within a transaction).
Here’s how a typical ProcessPaymentCommandHandler might look:
public class ProcessPaymentCommandHandler : IHandleCommand<ProcessPaymentCommand>
{
private readonly IPaymentGateway _paymentGateway;
private readonly IRepository<Order> _orderRepository; // For fetching Order aggregate
public ProcessPaymentCommandHandler(IPaymentGateway paymentGateway, IRepository<Order> orderRepository)
{
_paymentGateway = paymentGateway;
_orderRepository = orderRepository;
}
public async Task Handle(ProcessPaymentCommand command)
{
// Start a transaction for this specific command handler's work.
using (var transaction = BeginDatabaseTransaction())
{
try
{
var order = await _orderRepository.GetByIdAsync(command.OrderId);
// Perform payment processing logic
var paymentResult = await _paymentGateway.ChargeAsync(command.CustomerId, command.Amount);
if (paymentResult.Success)
{
order.MarkPaymentAsProcessed(); // Update aggregate state
await _orderRepository.SaveAsync(order, transaction); // Save within the transaction
// Publish PaymentProcessedEvent *after* successful commit
await Publish(new PaymentProcessedEvent
{
OrderId = command.OrderId,
CorrelationId = command.CorrelationId
});
transaction.Commit(); // Commit the transaction
}
else
{
// Handle payment failure - potentially mark order as failed
order.MarkPaymentAsFailed();
await _orderRepository.SaveAsync(order, transaction);
await Publish(new OrderFailedEvent
{
OrderId = command.OrderId,
CorrelationId = command.CorrelationId,
Reason = "Payment Failed"
});
transaction.Rollback(); // Rollback on failure
}
}
catch (Exception ex)
{
transaction.Rollback(); // Rollback on any exception
// Log exception and potentially publish a system-level error event
throw; // Re-throw to signal failure up the chain
}
}
}
// Placeholder for actual transaction management
private IDbTransaction BeginDatabaseTransaction() { /* ... */ return null; }
}
The ProcessPaymentCommandHandler explicitly starts a database transaction, performs its operations (charging the customer, updating the Order aggregate), and then either commits or rolls back the transaction based on the outcome. The OrderCreatedEvent is published only after the transaction is committed, ensuring that the event reflects a consistent state change.
The most surprising thing about managing transactions in CQRS is that the "transaction" often isn’t a single, monolithic database transaction spanning multiple command handlers. Instead, it’s a series of independent, atomic transactions, each managed by its own command handler. The saga pattern then coordinates these independent transactional units.
The key levers you control are:
- Command Handler Transaction Scope: Each command handler defines its own transactional boundary. This is where you ensure the atomicity of the operations within that handler.
- Event Publishing Timing: Events should only be published after the transaction they are related to has been successfully committed. This guarantees that any downstream consumers (like sagas or other handlers) react to a consistent state.
- Saga State Management: The saga itself maintains state (e.g.,
_paymentProcessed,_inventoryAllocated). This state isn’t directly transactional in the database sense but represents the progress of the overall workflow. If the saga crashes and restarts, it can rebuild its state by replaying events. - Idempotency: Command handlers must be idempotent. This means processing the same command multiple times should have the same effect as processing it once. This is crucial for handling retries and ensuring consistency when messages are delivered more than once.
A common misconception is that a saga orchestrates a distributed transaction. While sagas coordinate actions that might involve multiple services, they don’t typically use two-phase commit (2PC). Instead, they rely on compensating actions for rollback. If payment succeeds but inventory allocation fails, the saga would issue a RefundPaymentCommand to undo the payment. This is a form of eventual consistency, not strong transactional consistency across services.
The next concept you’ll likely encounter is handling failures and implementing compensating actions within sagas.