Rebus Saga配置异常求助：并发消息与延迟消息数据问题

Hey there! Let's tackle those two Rebus Saga problems you're facing—they're common pitfalls, but easy to fix once you know what's going on.

1. IsNew Property Not Updating When Sending Concurrent Shipping Orders #

What's Happening #

When you send two ShippingOrder messages for the same RouteId at the same time:

• The first message creates the Saga instance, so IsNew is true during its handler execution.
• The second message immediately loads the already-created Saga (thanks to the RouteId association), so IsNew is false for its handler.

If your logic was tying order updates to the IsNew flag, you might be missing entries because you only run that logic when IsNew is true.

The Fix #

Stop linking your order-adding logic to IsNew. Instead, always add the incoming order to your Saga's data collection, regardless of whether the Saga is new or existing. Use IsNew only for one-time setup (like scheduling the delayed ProcessRoute message):

public async Task Handle(ShippingOrder message)
{
    // Always add the order to the Saga's data—no IsNew check needed here
    Data.ShippingOrders.Add(message);

    // Only schedule the delayed message if this is the first order for the RouteId
    if (IsNew)
    {
        await Bus.DelaySend(TimeSpan.FromSeconds(30), new ProcessRoute { RouteId = message.RouteId });
    }
}

Also, double-check your Saga mapping configuration to ensure RouteId is correctly linked between messages and Saga data:

protected override void ConfigureHowToFindSaga(SagaPropertyMapper<ShippingOrderSagaData> mapper)
{
    mapper.MapSaga(saga => saga.RouteId)
          .ToMessage<ShippingOrder>(msg => msg.RouteId)
          .ToMessage<ProcessRoute>(msg => msg.RouteId);
}

Rebus automatically locks Saga instances during processing (database locks for distributed storage, in-memory locks for local testing), so you don't have to worry about concurrent writes corrupting your order list.

2. Delayed Message Handler Gets Stale Saga Data #

What's Happening #

If you're seeing old data in your ProcessRoute handler, it's likely because either:

• You accidentally included a snapshot of the Saga data when sending the delayed message (instead of just the RouteId), or
• Your Saga storage isn't configured to load the latest version when the delayed message triggers.

The Fix #

First, make sure your delayed message only carries the RouteId—no extra Saga data. Rebus will use this RouteId to fetch the latest version of the Saga data from storage when the delayed message is processed.

Then, in your handler, just use the Data property directly—it's already the fresh, up-to-date Saga data:

public async Task Handle(ProcessRoute message)
{
    // Data contains all ShippingOrders added since the Saga was created
    var ordersGroupedByAddress = Data.ShippingOrders.GroupBy(order => order.Address);

    // Transform and send to your web service
    foreach (var addressGroup in ordersGroupedByAddress)
    {
        var shipmentGroup = new ShipmentGroup
        {
            RouteId = message.RouteId,
            Address = addressGroup.Key,
            Orders = addressGroup.ToList()
        };
        await SendToYourWebService(shipmentGroup);
    }

    // Mark the Saga as complete once processing is done
    MarkAsComplete();
}

Extra Checks #

• Ensure your ShippingOrderSagaData class is properly serializable (public properties, no non-serializable fields) so Rebus can save and load it correctly.
• Verify your Saga storage configuration—for example, if using SQL Server:

  Configure.With(yourContainerAdapter)
      .Transport(t => t.UseSqlServer("your-connection-string", "rebus_messages"))
      .Sagas(s => s.StoreInSqlServer("your-connection-string", "rebus_sagas", "rebus_saga_indexes"))
      .Start();
  

That should resolve both issues! Let me know if you run into any edge cases.

内容的提问来源于stack exchange，提问作者Gerson Dias