How to deal with an event if the domain aggregate was not yet persisted?

Question

When I call paid() on my aggregate it sets all the properties internally that need to be changed based on this action. But at this time the change has not been saved to the DB, so I can't reliably trigger the Invoice.paid event yet.

Do I get this right that I need to use another object, like a domain service to manipulate and save the invoice and trigger the event from within the service because I can't do it in the aggregate? Is there any other solution to this? I'm not using an event store yet.

Answer 1

The problem you've encountered has a lot to do with messaging implementation.

The solution of the problem you're describing depends on many variables and it is very difficult to reliably take into account all of them and build a generally-applicable robust solution. In fact, the problem is so difficult to tackle, people like Udi Dahan have created entire companies to build a reliable solution for the problem and to provide the solution as a service.

The simplest version of the problem is: when a certain event happens within the system, I want to react to it in the same database transaction (i.e. also synchronously) and do some synchronous operation as a reaction. This situation is fairly common, when you have a project containing multiple bounded contexts, you want to do some operation on a certain aggregate in a context B when an event on a different aggregate happens in a context A but still want to retain the atomicity that either N operations will succeed or none. Thanks to this mechanism, the bounded contexts are not directly tied to each other which proves to be a great thing when you decide to split the BCs between multiple projects and/or applications.

Implementing this solution is as easy as having a single TransactionManager spanning your entire application and building a simple EventDispatcher class with a dispatch(event: Message): void method, which internally loops over all attached event handlers as listeners to the dispatched event passes the dispatched event to the handler as a parameter.

---

Unfortunately, things get much more complicated very quickly, once you start working with multiple threads, want to process events asynchronously and/or have bounded contexts as separate micro services, among which keeping transactional safety is pretty much impossible. Along with eventual consistency, this introduces another requirements your application now must fulfil in order for its messaging processes to be considered reliable.

Whether it's simple asynchronous processing within the same application or the asynchronous processing is handled by dispatching the call to another micro-service, you'll need to start tracking a lot of information about what's happening in your system(s) in a transactionally safe way.

An effort to try to implement a transactionally-safe message handling

^{Let's have a look at the modelling process, starting with the simplest solution and gradually move onto better and better implementation.}

You start with a simple MyEventHandler, handling MyEvent, which depends on an AsynchronousServiceBus and calls serviceBus.dispatch(new SomeCommand()) as a response to a dispatched MyEvent in your system. What the AsynchronousServiceBus service bus does is, it takes the message passed to the dispatch method and forwards it to RabbitMQ from which it is asynchronously consumed and processed by a worker.

You program your system in the following way:

1. a command is issued to create an entity,
2. you start a database transaction to make the creation transactionally safe,
3. before committing the transaction, you dispatch a MyEvent to the system,
4. committing the transaction fails (e.g. due to unavailability of the database, unique check violation,...).

You've just ended up in a situation, where there's an event flying through your system which shouldn't have been present at all, because the entity has not been completed. To fix this problem, you reprogram your application to dispatch events after you commit a database transaction. You're happy with your solution, because now when an event is fired by your application, you can be sure it makes sense to do it so. Unfortunately, soon enough you encounter a problematic situation which started happening quite often:

1. a command is issued to create an entity,
2. you start a database transaction,
3. you commit the transaction and it goes through without a problem,
4. you try to dispatch the event after commit, but RabbitMQ fails to respond.

Because of 4th step, the event which had to be dispatched got lost and your commited entity from 3rd step is in a waiting state and will never be completed, because there's no record of the event in your system.

You figure out the only way how to fix the problem you've just encountered is to store dispatched events in the same database using the same transaction as storing the affected entity. Because of that, you reprogram your system once again in the following way:

1. a command is issued to create an entity,
2. you start a database transaction,
3. instead of directly handling an event by sending a command to RabbitMQ, the event is now persisted to a new dispatched_event database table using an INSERT statement in a PREPARED state,
4. you commit the transaction and it goes through without a problem,
5. polling service checks the dispatched_event for PREPARED events, loads those and pushes them over to RabbitMQ one by one,
6. when RabbitMQ responds with message accepted, the polling service changes the state of the event to SENT.

Unfortunately what has happened, the polling service has started to fail from time to time between successfully sending the message to RabbitMQ and changing the message's state to SENT, and you find out a lot of duplicated messages have started to appear in your infrastructure. In order to fix this problem, you introduce a unique id in the form of GUID and append this attribute to your message. You also adapt the RabbitMQ workers to check whether a message with a specified id has been already processed or not (this is done by the worker storing the ids of messages which have already been processed), and your problem with duplicated messages is solved.

Another developer joins your project and you find out that when MyEvent is dispatched, two followup commands are supposed to be executed. Because your network connection is really unstable, what happens from time to time, the first of the two commands is dispatched to RabbitMQ and successfully processed, but the second never went through. Because of this, you have once again ended up in a situation where only a partial change has been applied to your system and your system starts to be out of sync. Unfortunately, thanks to your deduplication logic, your commands are sharing the same event id and are discarded as duplicates on retry and never repeated.

In order to fix this problem, you realise what you really need to store is commands and not events, so you remodel your dispatched_event table to dispatched_command, rework your polling service and now effectively store all commands to be dispatched as a reaction to a specific event within your system. Thanks to this new implementation, specific failing commands can now be retried multiple times.

---

Introducing other constraints, such as having messages to be sent at most once or messages to be sent at least once introduces further complexity and constraints on your system and further complicates the potential solution.

Since you're in a PHP, I assume what you're after is the initial-simple synchronous implementation. That alone is fairly simple to do, should you have a decent transaction manager ensuring a transaction spans across the desired change to your system as well as changes invoked by listeners attached to dispatched events. A concrete solution for asynchronous messaging is a topic for complex discussion and also depends on your business needs and a general solution cannot really be provided.

Answer 2

Reliable State Transition

As far as I can understand, you have an invoice which can be in one of two states: Paid, Not Paid.

While the invoice is Not Paid, then some follow up actions can not happen. When it transitions to Paid you want those actions to happen. But if you peek you don't want to see a Not Paid invoice and those actions having occurred.

What you are looking for here is a transaction. Use your database engine to ensure the invoice is either paid, or not paid. If your application is restarted it will always load that invoice in one state or the other.

Unfortunately you will still need to manage side-effects such as: changes to files, internal state, or events for external services.

Side-effects and State

If you need to update other information not in the database you will need to manage your own roll-forward, and roll-back behaviours.

A good way to encapsulate this is with a Task Queue, and Workflows.

When the invoice commits, also commit a task object to a queue. Have that queue polled regularly (or even signal it post-commit) to pick up the next task. This task performs the next action in the workflow.

• If it succeeds it removes itself from the queue and adds any further tasks.
• If it fails you have several choices:
  • try again, cleaning up on the retry.
  • undo, and stop
  • undo, and create an undo task for the workflow this far.
  • undo, and continue

This style has the benefit of forcing you to think through all the possible states of the system, and how to move through them.

Eventing

Eventing causes side-effects. These usually cannot easily be undone, though there are exceptions, so generally they are sent only after your application is in the new state. In the case of the invoice, this is after the transaction commits.

You can use the task abstraction above to attempt to deliver events in a push manner. Either as a once off attempt, or repeated some number of times. There is no guarantee that this event will be received (or possibly even sent). So relying on a purely push approach will fail. This also implies that any task which creates an event should not rely on the event having been received.

As pushing events may fail, the only other solution is to support polling. For this to work you will need to log each event that needs to be communicated with some unique id for each event. This is called event sourcing.

You can implement a purely polling strategy, but this is inefficient. A hybrid push/pull approach is usually best. Poll infrequently, and attempt to push only a limited number of times. If the receiver misses any event it can detect this on the next poll/received push.