Eventuous.OptimisticConcurrencyException handling in streaming

[melhelbawy]

We deploy our production using Kubernate in some time we got the below case...
Error handling command: 'EventPlanned' Eventuous.OptimisticConcurrencyException1 .... Update of Entity with id 69589a2d-dde4-40fb-baa6-8999ca6f71cc failed due to the wrong version. Unable to append events to Entity-69589a2d-dde4-40fb-baa6-8999ca6f71cc: Append failed due to WrongExpectedVersion. Stream: Entity-69589a2d-dde4-40fb-baa6-8999ca6f71cc, Expected state: NoStream, Actual version: 0 Append failed due to WrongExpectedVersion. Stream: Entity-69589a2d-dde4-40fb-baa6-8999ca6f71cc, Expected state: NoStream, Actual version: 0 at Eventuous.AggregateStore.Store[T](T aggregate, CancellationToken cancellationToken) at Eventuous.ApplicationService3.Handle(Object command, CancellationToken cancellationToken)

How we can handle Eventuous.OptimisticConcurrencyException our eventuous V 0.05.02 using shovel in inegration

[alexeyzimarev]

Please provide a bit more context. I don't see the shovel involved here, it's a normal behaviour of the aggregate store.

[melhelbawy]

Assume we have the below scenarios.

We have Order microservices wihch have couple of event that create Order stream in Eventstore. The OrderStarted event should publich and all subscribers should intiated ex. Cart, Payment, Vendors, Deliver Streams with the same OrderId Aggregate Id, note every stream subscribe the same OrderStarted event.

How we did implemetation for that, create Integration genreric handler which have TAggregate as parameter which handle integrated event for OrderStarted, then we use this integrated service in other microservices.

The Target streams should be same as below :

Order-3d34443c-095d-47bd-a099-e573a490aa2e
Cart-3d34443c-095d-47bd-a099-e573a490aa2e
Payment-3d34443c-095d-47bd-a099-e573a490aa2e
Vendors-3d34443c-095d-47bd-a099-e573a490aa2e
Delivery-3d34443c-095d-47bd-a099-e573a490aa2e

We Use Eventuous V 0.06.0, As you gives a hint for that Eventuous.OptimisticConcurrencyException is Aggreagte store behavoir, so can give how and where we handle it for the above scenario.

OrderStarted1 Eventuous.OptimisticConcurrencyException1[Cart]: Update of Cart with id 3d34443c-095d-47bd-a099-e573a490aa2e failed due to the wrong version. Unable to append events to Cart-3d34443c-095d-47bd-a099-e573a490aa2e Append failed due to WrongExpectedVersion. Stream: Cart-3d34443c-095d-47bd-a099-e573a490aa2e, Expected state: NoStream, Actual version: 0 Append failed due to WrongExpectedVersion. Stream: Cart-3d34443c-095d-47bd-a099-e573a490aa2e, Expected state: NoStream, Actual version: 0 at Eventuous.AggregateStore.Store[T](T aggregate, CancellationToken cancellationToken) at Eventuous.ApplicationService3.Handle(Object command, CancellationToken cancellationToken)

The event presisted to the EventStoreDB, unfortunately still this Error comes on.

[shashikantpawar21]

Any help on above issue..
we have upgraded the eventuous to 0.10.2 version
Still the issue persist

[JulianMay]

I firmly beleven this is an implementation problem in your code. Some hints you might find useful though (all of which are documented in https://eventuous.dev/docs):

"Expected state: NoStream, Actual version: 0" is a concurrency-error. It's thrown when you expect no such stream to exist already (like using "OnNew" : https://eventuous.dev/docs/application/app-service/#registering-command-handlers ), but there is already a stream with one event.
Either you should be using "OnExisting" or "OnAny",
or catch the exception and retry, so your aggregate invariants can guarantee no new events are added - depending on your requirements.

Also: Remember that you are given "at least once" guarantees, not "exactly once", on event handling, you cannot ignore idempotence without encountering these problems.

Also: if you are running this subscription in several instances, concurrency-errors are bound to be a headache.

Good luck

[shashikantpawar21]

thank you for your guidance

Yes, we tried OnAnyAsync one instead of OnNewAsync function

and in command handler added this check

if(aggregate.Original.Length == 0)
{
//apply event
}

in rest case ignore the stream

This supress the error.. Hope it is good approach ??
Any side effect or performance issue with respect to this ?

[JulianMay]

Given if(aggregate.Original.Length == 0).. solves the problem, this implies a previous event was comitted before the aggregate was loaded for this transaction.

Given OnAnyAsync solves the problem (but not the former), this implies an event is written to the eventstore after this instance is loaded, but before the new event from this instance is written to the eventstore.

I don't see any side-effects, the primary effect would be as follows:

• if(aggregate.Original.Length == 0) will cause the transaction to write 1 or 0 events, leaving the stream untouched or incremented with the event(s) being added.
• OnAnyAsync (and no invariant-checks in aggregate) will cause the stream to always be incremented with the events added.

As for performance: I would still suspect you are running subscription in parallel, with two processes fighting for the same work (maybe double check your kubernetes setup as I understood you are using, or look into competing consumers if you really need several nodes to share the same subscription), but otherwise and always: measure first: Do you perceive a performance problem?