Saga Sql persistence Concurrency

,
1

I have a Saga that processes batch work. My solution is based on the article referenced in the documentation. Thus I wanted to mitigate optimistic concurrency errors. But unfortunately, even though I don’t update saga data when batch completed event is handled, saga data is persisted anyway. This leads to numbers of concurrency errors when many workers finished. I use SQL Persistence.

The code looks like:

public class InvoiceAndReportSaga : Saga<InvoiceAndReportSagaData>,
                                        IAmStartedByMessages<InboundShipmentUpdated>,
                                        IHandleMessages<SyncSkusWithItemMasterComplete>

public async Task Handle(ShipmentProcessingCompleted message, IMessageHandlerContext context)
{
    foreach (SeasonalProduct seasonalProduct in seasonalProducts)
    {
        await context.Send(new SyncSkusWithItemMaster
        {
            SagaCorrelationId = this.Data.SagaCorrelationId
        }).ConfigureAwait(false);
    }
}

public Task Handle(SyncSkusWithItemMasterComplete message, IMessageHandlerContext context)
{
    return Task.CompletedTask;
}
2

Are you using the latest version of SQL Persistence? That uses pessimistic locking and should solve concurrency issues on the same saga instance.

3

As an aside. The handle can be simplified to avoid the state machine using a Task.WhenAll. eg

public Task Handle(ShipmentProcessingCompleted message, IMessageHandlerContext context)
{
    return Task.WhenAll(seasonalProducts
        .Select(x => context.Send(
            new SyncSkusWithItemMaster
            {
                SagaCorrelationId = Data.SagaCorrelationId
            })));
}
4

@Simon_Cropp, my 2 cents:

In general, I would recommend to not elide the async/await keyword unless you are on a hot path. When going with the async/await keywords you avoid introducing subtle bugs like not awaiting a task. Eliding the keyword in the business code is most of the time not worth the risk you are exposing yourself.

5

Nice alternative implementation. Sends are buffered so I guess this would still run completely on one thread if I’m not mistaken.

6

We are using 3.2.0. The pessimistic concurrency not acceptable for us

7

You mean, not available? Enhancements are not backported and only added in new major or minor versions. Only critical bugs are backported.

I advise you to upgrade to NServiceBus 7. Migrating from 6 to 7 is a simple task if atleast already using all v6 API’s.