CQRS-Event Sourcing: how to process events in the expected order inside the read model

Question

I'm looking for some insights about how to handle the read model in a CQRS Event Sourcing application, in order to do the best to ensure data consistency.

The missing point is how to be sure (actually not sure, but reasonably sure) that the projections inside the read model are able to process the events created from the write model in the expected order (which is the order in which the write model generated the events).

Let's try to clarify the scenario with an example.

Let's suppose of having an app with a write stack generating events for an aggregate called news and read stack made by a single projection, which is basically a denormalized table listing all the news with some properties (for instance the title, the summary and the author name).
This way, a client app is able to render a user interface listing all the latest news published by the editors.

The infrastructure for the event sourcing is an event store which saves the events as documents inside a MongoDB collection and then publish a corresponding message to a service bus, so that with a classic pub-sub pattern all the interested projections are able to subscribe the message and do the proper work in response to it.
For those who are familiar with RabbitMQ, this kind of stuff could be implemented by using a fanout exchange. With Azure service bus, instead, you can use a topic.
In this kind of scenario the write model and the read model are deployed to different machines which are free to be scaled independently of each other, based on the request load on both the write stack and the read stack.

Given such a scenario, it is entirely possible that at a given point in time there are two or more instances of the app hosting the read model of our system. These instances will be competing consumers over the events published by the write model.

Imagine now that in a brief time interval two events, say E1 and E2, are published to the service bus and that there are two running instances, say M1 and M2, of the app hosting the read model.
Given these scenario it is entirely possible that the event E1 will be processed by the machine M1 and the event E2 will be concurrently processed by the machine M2 (remember that the two instances are competing over the messages published to the service bus). At this point the final state of the projection is unpredictable, because each one of the instances could be faster than the other.

A typical example is when both the events are of type TitleSet, because an editor has decided to change the news title twice in a very brief time interval: at the end of the processing the projection will contain the wrong title for the news and the wrong title will be visible to the final users in the client application (which, of course, gets the data for the user interface from the projection).

What is the best way to handle these kind of scenario so that the best possible consistency is ensured in the read stack of the application ?

P.S.: the events E1 and E2 have been generated in the expected order by the write model and they have been stored properly inside the Event Store. The data inconsistency we are talking about is only for the read model.

Answer 1

The infrastructure for the event sourcing is an event store which saves the events as documents inside a MongoDB collection and then publish a corresponding message to a service bus, so that with a classic pub-sub pattern all the interested projections are able to subscribe the message and do the proper work in response to it.

(Emphasis added)

This assumption here is the one that you want to push back on. Pub/Sub can work for consumers that only care about a single message in isolation. Consumers that need state should be consuming histories, not events.

In the irredeemably non-optimized case, a consumer reads the entire history of ordered events each time it runs, and then processes them all.

An optimized version of this is that the consumer tracks where in the event history it left off, and runs an "all events since event X" query to find out what has happened. The irredeemably non-optimized case simply being the degenerate case of this: "all events since there were no events".

You might still see the pub/sub pattern applied, not to rebuild the read model, but to wake up the consumer to pull history as described above (in effect, it becomes a latency reduction mechanism).

There's nothing wrong with the consumer also having a bit of clever to recognize that the received event is the immediate successor to what is already known. This is normally accomplished with meta data attached to the event, indicating its position in the history.

So in your competing consumer scenario, you might see two different read behaviors at work. The receiver of E1 determines that E1 is the immediate successor of the previous state, and simply goes to work. The receiver of E2 sees from the metadata that at least one event is missing, so refreshes its copy of the event stream, receiving in return the sequence [E1,E2], which it then consumes.

Some references

At the DDD Europe conference, I realized that the speakers I talked with where avoiding Pub/Sub whenever possible. -- Raymond Rutjes, 2016

Greg Young, Polyglot Data (2014); Greg talks a bit about the benefits of pull.