CQRS Eventual Consistency Event Dependencies

Question

I'm new to the CQRS and eventual consistency models, so forgive me if this is a stupid question.

Given that I'm just getting started, I have a local in memory CommandBus and EventPublisher. My events are persisted to a RavenDB database for replay purposes, but events are published and handlers are invoked in locally (not queued externally via NServiceBus et al). The EventPublisher does publish events asynchronously (a la Task.Factory.StartNew).

Sometimes my events have dependencies (eg. the OrderReceived event must be processed into the ReadModel before the OrderShipmentStatusUpdated event can be correctly processed into the ReadModel).

How can I handle this kind of scenario? Using Sagas? How would one use Sagas in a simple in memory model like described above? Is it considered acceptable to "defer" an event (perhaps mark it as deferred and simply try to reprocess all deferred events "once in a while")?

What are some strategies for dealing with this?

Thank you.

Answer 1

A queue with retries most of the time solves such issues. If your handler/aggregate/denormalizer can't process a message, because preconditions are not met - fail it hard. Then you'll process some more messages from the queue until this one becomes visible again. If message fails more than 3 times - discard it into error queue for further analysis.

If it's an expected workflow and you actually have to wait - create Saga if not modeling with DDD/Event Sourcing. If modeling with DDD/Event Sourcing - Aggregates will cover such functionality most of the time.

Answer 2

The question is why the events are arriving out of order?

If it’s because of "technical forces" you can look at your infrastructure addressing this but this normally would require a full-blown message queue.

If messages/events are out of sync because of "business processes" then you can look at Sagas/ProcessManager. Have a look at Greg Young’s approach of using the event store as queue.

To get to your question about deferring events: What you can consider is to use a simplified approach by modeling the required ordering using Aggregates. It will consume events and notify the read side only when conditions are met. The view will only project events that have been "validated" by the aggregate. You have the benefit of making implicit concepts explicit in your model. The down side is that you’ll have more events generated. (Yes, nothing is for free)

Have a look at Rinat Abdullin’s approach on this. It's an old but interesting post and perhaps refer to the Being The Worst podcast which also covers some of this.

Answer 3

I agree with ILICH.

Events happened and the read model should just report on that. The read model should not be in a position to validate events.

It may be easier to have the shipping logic in the domain model rather then create a new saga.

Another way to 'cheat' on a smaller project is to use the event as a trigger to read from the eventstore. So when your read model gets an event notification it should load all the "new" events from the eventstore. This ensures that nothing is lost in transit and simplifies what you use the bus for.

In the prototyping stage it also lets you blow away your read models at will and have them automatically rebuild.

I keep a little table of "job history" that reflects the most recent event consumed to process a particular read model. If I want to change the schema of the read model I just delete the appropriate record.

Answer 4

It's better to use data from Events Store instead of data from Read Model for processing OrderShipmentStatusUpdated event, this will solve the issues connected with eventual consistency.

Events store is the only source of truth.

Answer 5

On the problems of sync/async event publishing:

By wrapping the given DomainEvents with an EventCommit object having an own commit id persisted into the event store, we could asynchronously process the EventCommit objects, while the DomainEvents within the EventCommit were synchronously published. The wrapping happens in our UnitOfWork which tracks our in-memory aggregates when the changes to a given aggregate are requested.

On the read side the commit-id can be used to verify correct ordering directly in the dequeing process before dispatching the events to the handlers.