Theory of CQRS Command Handlers: Sagas, ARs and Event Subscriptions

[Rinat Abdullin]

Hi All,

I've just published my latest thoughts that seem to bring together
theory of CQRS, DDD, ES, my limited experience and the logic of
scalable/cloud systems in a consistent and non-contradictory manner.

http://abdullin.com/journal/2010/9/26/theory-of-cqrs-command-handlers-sagas-ars-and-event-subscrip.html

Just a quick but important note. Among other things I highlighted
differences between accepted usage of Sagas (they consume events and
publish commands) and the one that comes out logically (saga is an
entity and can't consume events directly). Probably this has effect on
less than 5% systems (with extreme scaling or complexity), but I still
think this to be important. After all if a theory is true, it should
cover all known practical cases.

I'm probably reinventing the bike here. So please feel free to post
feedback, criticism and share experience that breaks this logic.


Best regards,
Rinat

[Udi Dahan]

Rinat,

Nice post.

On the topic of partitioning, not just in your post but this list in
general, I think that too much emphasis has been put on partitioning based
on entity id.

I recommend that top level partitioning be done with SOA services, then
inside them, with Business Components (similar to DDD bounded contexts),
then inside them, dividing up into single user and collaborative domains,
where the single user world is partitioned inherently by user ID, and in the
collaborative domain we use CQRS. If partitioning is indeed needed in given
BC's command-side of CQRS, then one might look at additional partitioning,
but this is really only needed in extremely high throughput + low latency
domains.

Cheers,

-- Udi Dahan

Hi All,

http://abdullin.com/journal/2010/9/26/theory-of-cqrs-command-handlers-sagas-
ars-and-event-subscrip.html

Best regards,
Rinat
No virus found in this incoming message.
Checked by AVG - www.avg.com
Version: 9.0.856 / Virus Database: 271.1.1/3160 - Release Date: 09/26/10
09:01:00

[Rinat Abdullin]

Udi,

Thanks.

I agree about the partitioning. 95% of the real-world systems will
probably never need it along with some other specific benefits of
CQRS. Yet as long as the partitioning logic is enforced and not
violated, it will be rather straightforward to scale the system out.

In my case the ratio is more like 50%-50% due to the cloud computing.
There is an overall tendency in projects to use it's capabilities and
resources to build cost-effective (really cheap) systems. This
requires granularity to be thought through right from the start, if we
want to keep architectures simple and consistent. Hence my attempts to
align CQRS, DDD and cloud programmability to make life easier.

Best regards,
Rinat

[Nuno Lopes]

Udi,

On the topic of partitioning, not just in your post but this list in
general, I think that too much emphasis has been put on partitioning based
on entity id.

I agree with you.

Yet the AC solution is a case where our Repository Partitions (repository nodes), Domain Model partitions, and Views (processing nodes)  are collocated (same cluster of nodes). Views are read only data spaces intertwining the domain model state with external state.

This solves the problem of unavailability by dependency or the domino effect at the service level, but not at the app level. 

Still, may be due to my lack of understanding, intertwined with all this talk about composite UIs. I assume a composite UI is one where each part is severed with data from different ACs. In particular your statement here:

"UI’s from the CQRS found in each BC can be “mashed up” in a single application, providing users a single composite view on all parts of the problem domain. Composite UI frameworks are very useful for these cases."

http://www.udidahan.com/2009/12/09/clarified-cqrs/

That's very nice and dandy, but if the CRM (AC) is down, including the views served by it, parts of the UI will never function (such as the one showing the customer name), as such the Order will never be submitted. So there goes availability down the drain.

On another note your AC can be achieved with other means but events. For instance with bulk data transfers (DTS) over a fast pipe and denormalise on views. This IMHO is still CQRS yet with no events. We to this at a slice level. Not so much due to availability but performance.

The reason why companies choose not to use events is because "off the shelf" products such as SAP, Seibel and many others, are close products using complex API's ending up with difficult to trace changes. 

At a global level it is very, very difficult to achieve using ad-hoc analysis schemes based on mechanisms such as the ones being proposed. We need a fact/event engineering as well as science.

When enterprise products start indeed publishing events something like this will be easier to achieve. Added to this we have Cloud Services such as Sales Force where we don't have the concept of Bus.

Now in custom made solutions we can use events all the way and be more proactive about it. At least from the inside say out and within.

Does this make sense?

Nuno

[Rinat Abdullin]

Sorry for barging into the conversation, but what's "AC"?

> That's very nice and dandy, but if the CRM (AC) is down, including the views served by it, parts of the UI will never function (such as the one showing the customer name), as such the Order will never be submitted. So there goes availability down the drain.

Hm. You can merely keep denormalized views in the cloud (or CDNs). It
will cost pennies (in dev and maintenance) but will guarantee that
views will never be down. Same is with the queues (in the inverse).

Best regards,
Rinat

[Nuno Lopes]

AC = Autonomous Component (Check the article Clarified CQRS by Udi Dahan).

> Hm. You can merely keep denormalized views in the cloud (or CDNs). It
> will cost pennies (in dev and maintenance) but will guarantee that
> views will never be down. Same is with the queues (in the inverse).

Will it? How do we guarantee that views are consistent with the events received?

Granted we can relax our consistency here (consistency amongst the views), but if we are using this data to feed back our commands, how relax can we be?

Since events can arrive out of order ....

Cheers,

Nuno

[Rinat Abdullin]

Nuno,

1. Ensuring ordering is the responsibility of the receiving entity (in
this case - message handler populating views). Based on the case
specifics, this could be delegated to the infrastructure or specific
messaging implementation that guarantees ordering for certain cases
(i.e.: AMQP implementation)

2. Where needed, absolute consistency can be ensured by leveraging
version number passed from the events to views and to commands (i.e.:
see version fields within Command Handling Logic in my post). This way
command handler can choose to reject changes that were made by user
based on the stale data (and publish a message about that). As I
recall, Greg's CQRS sample demonstrates this behavior.

Best regards,
Rinat

[Nuno Lopes]

Hi,

> Ensuring ordering is the responsibility of the receiving entity (in
> this case - message handler populating views).

That is easier said then done :)

How? I mean, the denormalizer might even detect that the event came out of order based on some versioning scheme that it keeps track off, how should it proceed in a view? Should it roll back the changes applied and apply the events all again in the proper order? Since we are talking about views (not aggregate roots) how is this done? Should we keep snapshots of views too?

How do we do this in highly de denormalized model where events can come from more then one Aggregate Root. Or are we using one View per Aggregate only (no cross Aggregate Views)?

That is the crux of the matter IMHO.

If you look at Greg's example the conflict resolution is done at the Repository level, in which case no events are published. Not at the view level. The view accepts whatever event version that is published, whatever is the order.

Keeping with this example suppose that events arrived out of order to the view. This means that the version in the view could be one value bellow the current version in the store. This would mean that no longer the system would be able to process commands (they use the version in the views) since the version would never match with the one in the event store. Hence we would have a deadlock! Furthermore, with a difficult way to rebuild views.

I'm not being alarmist. Just rising questions. Any ideas?

Cheers,

Nuno

[Rinat Abdullin]

In case of scenario that's not massively scaled I would let the AMQP
to ensure order of events coming from one AR. If there are cases where
order between the AR is really important for the cross-AR view, then
conversation numbers and explicit logic will be used to correlate
"time" between various partitions.

For the massively scaled scenario - it will be responsibility of the
Subscription to apply incrementing sequence number to events, as they
leave the partition. View, just like any other entity, can hold on to
the incoming messages that came out of the order (activity in
Helland's papers). For the cross-partition views in massively scaled
scenario where order of events is really important, we still manually
correlate events.

Best regards,
Rinat

[Greg Young]

There is ordering within the aggregate boundary but not outside. I believe there was a slight error in the code on this (that does not materialize do to usage patterns). In general you always support ordering within aggregate boundaries.

--
Les erreurs de grammaire et de syntaxe ont été incluses pour m'assurer de votre attention

[Rinat Abdullin]

Greg, even if there is ordering within AR, consuming view might be
located in the other partition. Who knows, what will happen to event
order, as they cross these boundaries.

Anybody who tried to use Azure queues learns this pretty fast.

Best regards,
Rinat

[Greg Young]

There is always ordering within an aggregate boundary, even if you cross partitions its trivial to provide ordering within an aggregate boundary. It is only a problem when you mix aggregate boundaries in which case the general case is to handle out of order possibilities

[Rinat Abdullin]

Say, events come from a single AR to a view within a different
partition. Message channel happens to be cloud-based. It promises to
do it's best to do FIFO and do not create any duplicates. Yet, it does
not guarantee this 100% due to it's scalable nature, dynamic
repartitioning or whatever excuse it finds.

So there is 0.1% probability (taken for Azure SLA), that events
Er1,Er2,Er3, Er4 from a single AR will come to view as Er1, Er3, Er2,
Er4. That's inconvenient, but this is the price you have to pay for
the extreme benefits cloud provides.

Unless you are willing to take chances, either ordering should be
explicit (again, precisely as Pat's activities) or view should stay
within the partition of AR (which is impossible for cross-AR views).

Best regards,
Rinat

[Greg Young]

There is a sequence number (version id) on the events ... re-ordering is relatively simple

[Rinat Abdullin]

Yes. And it also relatively simple even if view subscribes only to a
subset of events (which prevents version numbers to be used directly
for ordering).

The point I was trying to make - even if events come from a single AR,
recipient might still need to explicitly handle reordering and
idempotency.

Best regards,
Rinat

[Nuno Lopes]

Hi Greg, thank your for the feedback.

> It is only a problem when you mix aggregate boundaries in which case the general case is to handle out of order possibilities

Agreed. I've found that if aggregate boundaries complement each other rather then compete, in the case of views they can be treated separately while updating the same data set (same views).

Within the the denormalizer and for recovering from out of order events it is necessary to keep a buffer of the events received if rolling snapshots are being used in the event store.

To me, handling out of order events should still be handled by the infrastructure "auto-magically".

Hi Rinat, thank you for your feed back.

> Yes. And it also relatively simple even if view subscribes only to a
> subset of events (which prevents version numbers to be used directly
> for ordering).

Why does it prevent? All we need is to reprocess all events from the out of order event up out of the denormalizer buffer.

Idempotency on events over views is not a problem since there should be transformation on the values going on (say making and storing values derived from calculations in the process). Things get trickier when we start making business logic over views and store derived values.

Things get messier if the buffer goes ballistic and the event stores just rolls its stuff up loosing previous events. Just in case, maybe we should keep events for awhile even when the window slides.

Complexity starts with relatively simple things. The sum of simple things is what makes it complex.

Simple things should be taken care by the infrastructure and let the developer worry about the hard stuff, that is indeed the part of automating the business. I think a simple scheme to handle out of order events can be devised for that, and provided by the infrastructure
.
Still this is a bit off my point to Udi.

My question was if that if the Composite UI get's its data from views on multiple BCs, if one is offline how will the page be serviced?

Nuno

[Rinat Abdullin]

> Why does it prevent? All we need is to reprocess all events from the out of order event up out of the denormalizer buffer.

If a view subscribes only to a certain events within AR, this would mean that:

History: Er1, Er2, Er3, Er4, Er5
Subscribed history: Er1, Er3, Er5 (we are not interested in Er4, and Er2).

Unless you are willing to accept logically inconsistent views, there
is no way to tell (based on just version numbers) that Er3 actually
comes immediately after Er1 and immediately before Er5. Handling this
at the event subscription level is trivial, though.

Best regards,
Rinat

[Nuno Lopes]

> Unless you are willing to accept logically inconsistent views, there
> is no way to tell (based on just version numbers) that Er3 actually
> comes immediately after Er1 and immediately before Er5. Handling this
> at the event subscription level is trivial, though.

Notice that versions are established per AR, not per kind of event. In other words:

So Er1(version1), Er2(Version2), Er3(Version3), Er4(Version4), Er5(Version5). They are all sourced from the same Aggregate Root.

The purpose of ordering events in this context is to reconstruct the view. In this case, since Er2 and Er4 are irrelevant for such purpose as you defined.

For instance. Say that events arrive Er5, Er3, Er1. The correct order for processing this events for this purpose of reconstructing the view would be Er1, Er3, Er5 regardless of the other events.

So if these are buffered, all we need is to reorder the buffer per ARs.

As I mentioned before, if events are sourced from multiple Aggregate Root to reconstruct a view, if they don't compete with facts (they should not) rather then complement, then each can be de-normalized independently over the same view.

Best Regards,
Nuno

[Rinat Abdullin]

Imagine you are getting events from a single AR in order: Er1, Er5, Er9, Er10.

Can you tell which events are missing from the sequence?

Best regards,
Rinat

[Nuno Lopes]

Hi Rinat.

Suppose we are getting Er4.

We know that Er10 was applied but If we don't know that

> Er1, Er5, Er9.

were applied how does the de-normalizer know what needs to be reordered?

Can you provide an example of how would this be realized at the application layer level?

Thanks,

Nuno

[Rinat Abdullin]

In my case that's trivial by using the subscriptions and Rx (just as
the logic within the article).

On the conceptual level that's (skipping partitioning and identity
address part):

observable
.Where(e => e.NeededBy<TDenormalizerClazz>())
.Select((e,i) => new DenormalizeThisEvent
{
Event = e,
CurrentOrder = i
});

As long as ordering is provided at the same place where filtering
runs, we'll have incremental order that's logically safe for crossing
the consistency boundaries.

Best regards,
Rinat

[Nuno Lopes]

Hi,

> observable
> .Where(e => e.NeededBy<TDenormalizerClazz>())
> .Select((e,i) => new DenormalizeThisEvent
> {
> Event = e,
> CurrentOrder = i
> });

Can you explain why this code can't be in the infrastructure. The event dispatcher knows what was dispatched and to whom right?

I'm not that familiar with Rx APIs.

Nuno

[Rinat Abdullin]

> The event dispatcher knows what was dispatched and to whom right?

In my understanding PubSub does not work this way. I don't dispatch
events to identity. I publish them to the infrastructure.

* Events are published, commands are sent to recipient.
* Entities may subscribe to events, providing Subscription query (rx
is just an implementation detail, but I strongly recommend to check
out this brilliant framework). By this very act they provide an
identity (their identity) and turn events into
DearEntityXPleaseHandleThis<IEvent> command. Routing and partitioning
kicks in here. More complex query could be in place of the
subscription (that's just the implementation detail)
* Each event might be subscribed to by 0..N entities. Once events go
through the subscription, they are transformed into 0..N commands
directed to the subscribing entity and are handled separately.

Best regards,
Rinat

[Greg Young]

The channel can do ordering. You could have events then from the same aggregate on multiple channels but that would be pretty weird. A channel can also handle ordering with partitions etc.

[Rinat Abdullin]

If channel guarantees ordering in the first place, then yes.

BTW, could you, please, recommend any reads on repartitioning live systems?

I just can't figure out how to repartition a system with live streams
with the guarantee that there will be no duplicates/ordering problems.
Whenever we need to have messages chase relocated entities - there are
boundaries crossed and seems to be always a possibility of messing up
things slightly (at least once delivery)

Best regards,
Rinat

[Jimit]

@Rinat could you elaborate on how you'd define subscription queries
using Rx? Reading your posts you suggest using IQbservable<T> to
define the event subscription but wouldn't that require that the event
handler have access to (or be given) the event stream? I was hoping
the implementation could be a bit more declarative, i.e the event
handler declares the events that it is interested in some fashion that
allows defining filtering, transformation, throttling requirements,
etc. and simply exposes and a method accepting an instance from the
resulting observable sequence. Declaratively declaring filters is easy
- simply use the specification pattern in some form... I'm kinda
rambling a bit here so I'm gonna stop but would appreciate any light
you or the community could shed on such usage.

> >> On Wed, Sep 29, 2010 at 6:17 PM, Nuno Lopes <nbplo...@gmail.com> wrote:
> >>> Hi Rinat.
>
> >>> Suppose we are getting Er4.
>
> >>> We know that Er10 was applied but If we don't know that
>
> >>>> Er1, Er5, Er9.
>
> >>> were applied how does the de-normalizer know what needs to be reordered?
>
> >>> Can you provide an example of how would this be realized at the application layer level?
>
> >>> Thanks,
>
> >>> Nuno
>
> >>> On Sep 29, 2010, at 3:57 PM, Rinat Abdullin wrote:
>
> >>>> Imagine you are getting events from a single AR in order: Er1, Er5, Er9, Er10.
>
> >>>> Can you tell which events are missing from the sequence?
>
> >>>> Best regards,
> >>>> Rinat
>

[bodrin]

Hi Rinat,

Very nice ideas so far :)

So, EventHandler-s are notified about the AR Event-s from the local partition only and not routed anywhere. And as Event-s pass through the EventSubscription they are transformed into Command-s which are then routed to the right partition. This way some kinds of EventHandler-s may need to reside into each partition, but this is not a problem. Am I understanding you correctly?

Also I think that there is no problem for the infrastructure to do the ordering of the DearEntityXPleaseHandleThis<IEvent> commands on the recepient side even if you have EventFilter-s support into the EventSubscription-s. Do you? (.. sorry I'm not into .NET and this rx)

bodrin.

[bodrin]

just added something below:

On Thu, Sep 30, 2010 at 3:32 PM, bodrin <bod...@gmail.com> wrote:

Hi Rinat,

Very nice ideas so far :)

So, EventHandler-s are notified about the AR Event-s from the local partition only and not routed anywhere. And as Event-s pass through the EventSubscription they are transformed into Command-s which are then routed to the right partition. This way some kinds of EventHandler-s may need to reside into each partition, but this is not a problem. Am I understanding you correctly?

This is in case the EventSubscription is always towards/for a certain AR instance. But do you go further and say I want to subscribe to Events from all ARs of a given type (or even more general) and then the infrastructure replicates your EventSubscription to every partition that hosts ARs of this type, so that you get notified?

[Rinat Abdullin]

@bodrin,

Yes, that's the idea.

In essence logically our Subscription is separated into two parts:
"Map/Filter" (that runs close to the source partition and converts
events into commands targeting the subscribing entity) and "Reduce"
part that is within the entity and could aggregate events from
multiple ARs.

In cases that do not need extreme scaling, events could be just
replicated to all partitions (all subscriptions running at the
recipient side). Slightly higher scalability requirements -
subscriptions are handled by the message bus centrally. Event higher
requirements - Map/Filter parts of subscriptions are replicated and
executed closer to the source partitions.

Logically this so far fits all the production cases I've been working
with. Practice in the next few months will show how (if) this theory
really applies to the real world.

Best regards,
Rinat

[Nuno Lopes]

The way I do AR or Sagas don't subscribing directly to events ATM.

Instead, the I extend the BUS to handle the events and deliver them to ARs,  Sagas or other form of domain services. Much as Udi does it I guess.

Nuno

[Udi Dahan]

If the UI part of the Customer Care BC cannot contact the query part of that AC, then you're right, data won't be shown.

This is the responsibility of the IT/Ops service as they are responsible for all hosting of ACs.

All that being said, the ability to provide scale and HA for queries is relatively easy when they're serving stale data - as Rinat mentioned, cloud storage and CDNs are very good at that.

 

Finally, the "submit order" button is not part of the UI of the Customer Care BC, but rather the Sales BC, so it wouldn't necessarily be "down" and the user could still submit their order.

The generation of an invoice (that requires the customer name) can happen later when that query part is back up.

 

Cheers,

 

-- Udi Dahan

 

 

From: ddd...@googlegroups.com [mailto:ddd...@googlegroups.com] On Behalf Of Nuno Lopes
Sent: Monday, September 27, 2010 9:52 AM
To: ddd...@googlegroups.com
Subject: Re: [DDD/CQRS] Theory of CQRS Command Handlers: Sagas, ARs and Event Subscriptions

 

Udi,

 

On the topic of partitioning, not just in your post but this list in
general, I think that too much emphasis has been put on partitioning based
on entity id.

 

I agree with you.

 

Yet the AC solution is a case where our Repository Partitions (repository nodes), Domain Model partitions, and Views (processing nodes)  are collocated (same cluster of nodes). Views are read only data spaces intertwining the domain model state with external state.

 

This solves the problem of unavailability by dependency or the domino effect at the service level, but not at the app level. 

 

Still, may be due to my lack of understanding, intertwined with all this talk about composite UIs. I assume a composite UI is one where each part is severed with data from different ACs. In particular your statement here:

 

"UI’s from the CQRS found in each BC can be “mashed up” in a single application, providing users a single composite view on all parts of the problem domain. Composite UI frameworks are very useful for these cases."

 

http://www.udidahan.com/2009/12/09/clarified-cqrs/

 

That's very nice and dandy, but if the CRM (AC) is down, including the views served by it, parts of the UI will never function (such as the one showing the customer name), as such the Order will never be submitted. So there goes availability down the drain.

 

On another note your AC can be achieved with other means but events. For instance with bulk data transfers (DTS) over a fast pipe and denormalise on views. This IMHO is still CQRS yet with no events. We to this at a slice level. Not so much due to availability but performance.

 

The reason why companies choose not to use events is because "off the shelf" products such as SAP, Seibel and many others, are close products using complex API's ending up with difficult to trace changes. 

 

At a global level it is very, very difficult to achieve using ad-hoc analysis schemes based on mechanisms such as the ones being proposed. We need a fact/event engineering as well as science.

 

When enterprise products start indeed publishing events something like this will be easier to achieve. Added to this we have Cloud Services such as Sales Force where we don't have the concept of Bus.

 

Now in custom made solutions we can use events all the way and be more proactive about it. At least from the inside say out and within.

 

Does this make sense?

 

Nuno

No virus found in this incoming message.
Checked by AVG - www.avg.com
Version: 9.0.856 / Virus Database: 271.1.1/3161 - Release Date: 09/26/10 12:40:00

[Nuno Lopes]

Hi Udi,

The point of my observation is that even though Components can be autonomous in isolation the UI usually breaks the boundaries of autonomy.

So we are probably better designing towards autonomous UIs and respective views. In some cases they we can leverage on durable messaging for commands.

What do you think?

Cheers,

Nuno