mookid8000
commented
3 years ago Hi, I've tried reproducing the issue here.
I think your experience might have been caused by a missing await in this block of code
public async Task Handle(KickoffSagaMessages message)
{
_logger.Info("Processing Kickoff - {id}", this.Data.SagaInstanceId);
this.Data.SagaInstanceId = message.SagaInstanceId;
this.Data.stuffDone += "Initiated;";
_bus.Publish(new SagaMessageEarth()
{
SagaInstanceId = message.SagaInstanceId
});
_logger.Info("Published Earth....Done Processing Kickoff - {id}", this.Data.SagaInstanceId);
}which should have read
public async Task Handle(KickoffSagaMessages message)
{
_logger.Info("Processing Kickoff - {id}", this.Data.SagaInstanceId);
this.Data.SagaInstanceId = message.SagaInstanceId;
this.Data.stuffDone += "Initiated;";
// 👇😅
await _bus.Publish(new SagaMessageEarth()
{
SagaInstanceId = message.SagaInstanceId
});
_logger.Info("Published Earth....Done Processing Kickoff - {id}", this.Data.SagaInstanceId);
}Could you try again with that added to your code?
(closing for now – feel free to pick it up again, if it still doesn't behave as expected)
mkoziel2000
glazkovalex
I noticed in Rebus 6.5.5 that when I define a straight forward Saga with 4 messages fired in a chain, the messages can get processed before the SagaData has a chance to get updated and passed along. This basically makes the SagaData as a holder of state between messages not very useful. I don't remember seeing this problem in earlier versions of Rebus, although the transport layer we used in the earlier version was SQL Server. I'm currently testing out the Kafka transport layer in the 6.5.5 version where I see the issue (the Saga storage is still in SQL Server).
Here is the basic Saga setup
What I should expect to see is that all the Sagas I pump into the system should make it to the MarkAsComplete() command. Instead, its about 50% of the Sagas that get marked and 50% that never satisfy the IF condition when processing "Msg 3". It seems to be timing related around how long it takes to consume and process the message relative to Rebus getting the SagaData internally updated after a previous message is done processing. In the cases where I get a stale SagaData, there is a Conflict Resolution invoked by Rebus as the result of a Concurrency collision. It seems the only way to overcome this problem is to add an artificial delay to the publishing of the next message in the Saga using deferLocal().
Its an arbitrary test, but I was basically verifying that Kafka as a transport (and subscriber) with SQL Server as the storage for Saga would work together and function properly when I ran into the odd behavior. I don't remember seeing this issue when I used SQLServer for the transport (maybe because the mechanism is slow enough that the SagaData can get processed in time before the next message gets handled) - Or it might have still happened but it was very infrequent.
Is there a way to guarantee that the SagaData is not stale when doing chain-based message sequencing? I suspect that a command&control pattern might be better suited to leveraging SagaData (A central handler that deals with the sequencing logic to determine which message to publish next based on the last message published)...but then, thinking about that pattern, it could also suffer from the same stale SagaData scenario since the worker handler has to publish a message to return control back to the controller handler before it exits its handler method.
Please advise. I have to imagine people must be running into this issue when they are hooked up to fast transport systems like Kafka or Rabbit.
Here is an example Saga that I was using for the test: