NServiceBus 7.7.0多消息处理时出现CosmosDB Saga更新冲突问题

问题场景 #

使用NServiceBus 7.7.0搭配CosmosDB作为Saga持久化存储时，处理多条关联同一Saga实例的消息时，频繁触发NServiceBus.Persistence.CosmosDB.SagaUpdate.Conflict错误。

你的配置代码 #

UseNServiceBus(context =>
{
    var appInsightsOptions = configuration.GetRequiredSection("AppInsights").Get<AppInsightsOptions>();
    NServiceBus.Logging.ILoggerFactory nservicebusLoggerFactory = new ExtensionsLoggerFactory(loggerFactory: new SerilogLoggerFactory());
    NServiceBus.Logging.LogManager.UseFactory(loggerFactory: nservicebusLoggerFactory);
    var endpointConfiguration = new EndpointConfiguration(endpointName);
    endpointConfiguration.EnableApplicationInsights(new TelemetryConfiguration(appInsightsOptions.InstrumentationKey));
    
    var transport = endpointConfiguration.UseTransport<AzureServiceBusTransport>();
    transport.ConnectionString(GetTransportConnectionString(configuration));
    //transport.Transactions(TransportTransactionMode.TransactionScope);
    endpointConfiguration.UseSerialization<NewtonsoftSerializer>();
    endpointConfiguration.UsePersistence<CosmosPersistence>()
        .CosmosClient(new CosmosClient(cosmosConnectionString))
        .DatabaseName(configuration.GetConnectionString("DATABASENAME"))
        .DefaultContainer(containerName: "sagastore", partitionKeyPath: "/id")
        .DisableContainerCreation();

    endpointConfiguration.LimitMessageProcessingConcurrencyTo(1);

    endpointConfiguration.SendFailedMessagesTo("error");
    endpointConfiguration.AuditProcessedMessagesTo("audit");
    endpointConfiguration.AuditSagaStateChanges(serviceControlQueue: "audit");
    endpointConfiguration.EnableInstallers();

    return endpointConfiguration;
}

问题原因 #

这个错误是CosmosDB乐观并发控制的预期行为：CosmosDB通过ETag标识文档版本，当两个操作尝试更新同一Saga实例（同一文档）且ETag不匹配时，就会抛出冲突。结合你的配置，可能的触发点包括：

• 未启用事务导致消息处理中途失败重试，重复更新Saga状态
• 消息未按Saga ID分区，导致同一Saga的消息被多个并发线程/实例处理
• 默认的冲突重试次数不足，无法处理瞬时冲突
• Saga处理逻辑中触发了同一实例的并发操作

解决方案 #

1. 配置CosmosDB Saga冲突重试 #

显式设置冲突重试次数，让NServiceBus自动处理瞬时冲突：

endpointConfiguration.UsePersistence<CosmosPersistence>()
    .CosmosClient(new CosmosClient(cosmosConnectionString))
    .DatabaseName(configuration.GetConnectionString("DATABASENAME"))
    .DefaultContainer(containerName: "sagastore", partitionKeyPath: "/id")
    .DisableContainerCreation()
    .ConfigureSagaPersistence()
    .RetryOnConflict(5); // 根据业务场景调整重试次数

2. 启用事务保证一致性 #

取消注释事务配置，确保消息处理和Saga状态更新在同一事务中，避免部分更新引发的重试冲突：

transport.Transactions(TransportTransactionMode.TransactionScope);

3. 确保消息按Saga ID分区 #

配置Azure Service Bus使用Saga ID作为分区键，保证同一Saga的消息被投递到同一队列分区，配合LimitMessageProcessingConcurrencyTo(1)实现单线程处理：

transport.UsePartitionKeyStrategy(context =>
{
    context.MessageHeaders.TryGetValue(NServiceBus.Headers.SagaId, out var sagaId);
    return sagaId;
});

4. 检查Saga处理逻辑 #

避免在Saga的消息处理方法中立即发送关联同一Saga的后续消息，这类操作可能触发并发更新。如果需要触发后续流程，优先使用Reply或延迟发送，确保操作顺序。

5. 验证CosmosDB容器一致性级别 #

将CosmosDB容器的一致性级别设置为会话一致性（默认），既能保证读取的一致性，又不会过度影响性能，减少因读取旧状态导致的更新冲突。

内容的提问来源于stack exchange，提问作者Raju Donthula