How to recover from domain events written to the stream but not published due to failure?

[webdevilopers]

            function (ActionEvent $event) use ($eventStore): void {
                $recordedEvents = $event->getParam('streamEvents', new \ArrayIterator());

                if (! $this->inTransaction($eventStore)) {
                    if ($event->getParam('streamNotFound', false)
                        || $event->getParam('concurrencyException', false)
                    ) {
                        return;
                    }

                    foreach ($recordedEvents as $recordedEvent) {
                        $this->eventBus->dispatch($recordedEvent);
                    }
                } else {
                    $this->cachedEventStreams[] = $recordedEvents;
                }
            }

• https://github.com/prooph/event-store-bus-bridge/blob/master/src/EventPublisher.php#L54-L56

Regardless of what event-bus to use - at this point of the code the events have already been written to the stream, correct? What if dispatching fails here for some reason? How can an app recover from the domain event not being published?
For instance a console command that re-dispatches / publishes the events? Or am I thinking "too carefully"?!

/cc @bas @prolic

Discussion on Twitter:

• https://twitter.com/webdevilopers/status/1292824930241961986

[webdevilopers]

Here are some suggestions made by @Shuttle and @xpmatteo e.g. using an "Event outbox":

• https://stackoverflow.com/questions/43436458/is-it-safe-to-publish-domain-event-before-persisting-the-aggregate

[prolic]

You can subscribe from the event stream and only publish messages from there. Going this way also allows you to rebuild and start from event 1 again.

…

On Mon, Aug 10, 2020, 10:07 webDEVILopers ***@***.***> wrote: Here are some suggestions made by @Shuttle <https://github.com/Shuttle> and @xpmatteo <https://github.com/xpmatteo> e.g. using an "Event outbox": - https://stackoverflow.com/questions/43436458/is-it-safe-to-publish-domain-event-before-persisting-the-aggregate — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#46 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AADAJPHSZISWIPQQHZ6VJMLR775INANCNFSM4PZ7XQFA> .

[webdevilopers]

You can subscribe from the event stream and only publish messages from there. Going this way also allows you to rebuild and start from event 1 again.

Thank you for your feedback. Indeed @JulianMay suggested that approach too on Twitter:

Optimally, don't put yourself in a "two-phase commit" situation to start with. @EventStore or @SQLStreamStore
are great for this, pull events from the store rather than (hopefully) pushing them to a bus

• https://twitter.com/TheJulianMay/status/1292840465486827520

[webdevilopers]

@Federkun also suggested the "Outbox pattern" (for NServiceBus):

Outbox is the way

• https://twitter.com/Federkun/status/1292866987509665792

Thanks to twitter's algorithm one of the next tweets by @cer dealed with the same topic:

• https://twitter.com/crichardson/status/1292865296122994689

A thread about implementing the Transaction Outbox pattern:

• https://microservices.io/patterns/data/transactional-outbox.html

A service command typically needs to update the database and send messages/events. For example, a service that participates in a saga needs to atomically update the database and sends messages/events. Similarly, a service that publishes a domain event must atomically update an aggregate and publish an event. The database update and sending of the message must be atomic in order to avoid data inconsistencies and bugs. However, it is not viable to use a distributed transaction that spans the database and the message broker to atomically update the database and publish messages/events.

[webdevilopers]

If I understand the pattern correctly then I think it can be achieved in the PHP world using the symfony messenger and the doctrine transport / transaction middleware.

If using the DoctrineTransactionMiddleware and a dispatched message throws an exception, then any database transactions in the original handler will be rolled back.

Is that correct @weaverryan? From the symfony docs example:

Let’s take the example of an application with both a command and an event bus. The application dispatches a command named RegisterUser to the command bus. The command is handled by the RegisterUserHandler which creates a User object, stores that object to a database and dispatches a UserRegistered message to the event bus.

There are many handlers to the UserRegistered message, one handler may send a welcome email to the new user. We are using the DoctrineTransactionMiddleware to wrap all database queries in one database transaction.

Problem 1: If an exception is thrown when sending the welcome email, then the user will not be created because the DoctrineTransactionMiddleware will rollback the Doctrine transaction, in which the user has been created.

Problem 2: If an exception is thrown when saving the user to the database, the welcome email is still sent because it is handled asynchronously.

• https://symfony.com/doc/current/messenger/message-recorder.html#an-example-registeruser-process
• https://symfony.com/doc/current/messenger.html#doctrine-transport

[gquemener]

This outbox pattern is quite interesting.

However, in the nservicebus presentation, I don't quite see how, in the second transactional phase of the process, the step 2 and 3 could actually run in the same transaction.

Step 2 is most-likely achieved against a message broker and step 3 against a RDBMS. Once a message has been published, AFAIK it would be very painfull, if not impossible, to unpublish it if step 3 were failing.

What am I missing here, please?

[JulianMay]

It's been over 5 years since I worked with NServiceBus, but IIRC it was relying on guarantees on the underlying provider. In our case we were using MSSQL and MSMQ, so the MSDTC was responsible for enforcing atomic transactions.
In general, your persistence layer is going to be your transactional constraint for transitioning state.
As long as your persistence layer is different from you transport layer (and doesn't integrate with some DTC that become the new painpoint/bottleneck), you have a disjoint you need to consider. Technical guarantees aside, I think the important question is: What happens if I commit to a changed state but my transport has a problem - do I still want to commit to my change? Usually, the answer is "yes, because I can emit later, when the transport is up", which leads you to emitting changes asynchronously, which means you don't need to have this problem (as per my tweet).
Outbox-pattern (If my understanding is correct) works by "committing to the events I want to be emitted later" along with the respective state-change into the same persistence, so I can rely on transaction guarantees from my persistence-layer, and would still usually emit events to transport asynchronously from committing to persistence.
( https://microservices.io/patterns/data/transactional-outbox.html )

I hope this helps :)

[gquemener]

Thanks @JulianMay for your clarifications.

This pattern seems very technology-specific as it requires this special piece of software called the Distributed Transaction Coordinator.

Sounds like there is a couple of options:

Use a projector to dispatch events in the bus

Pros

• Recovering from dispatching failure is straightforward because the projector (running in a separate process) is keeping a reference of the latest dispatched event (allowing to start publishing from that point when process is being restarted).
• You have the ability to replay the event stream from whatever point in the past you desire, easing the process of recovering from a failure.
• Quite straightforward to setup if you're using an event store implementation that provides projection support (like Prooph).

Cons

• Architecture is slightly more complex due to the need to start/monitor a separate process (the projector).
• The listeners (for example, a process manager) have the responsibility to act in an idempotent way, preventing duplicated events to be processed multiple times without repeating side effects.
• If you're not using an event store, you have to insert events in a special place (the outbox), that will be used by the projector (which probably have a different name then). This should preferably be done in the same DB transaction as the one used to store aggregate state changes, slightly increasing its execution time/complexity.

Dispatch the events in the bus right after the aggregate transaction has been committed (example)

Pros

• Simpler to setup and maintain
• Not bound to any storage infrastructure choice (events don't require to be somehow stored, as they are fetched from memory)

Cons

• Increase process execution time
• Failure during dispatch either means that event is lost or that it must be manually recovered

Have I summed up correctly?

[JulianMay]

To be clear: NServiceBus is technology-specific, the Outbox Pattern is not

Good summation, but I think there's more to say on the cons regarding "Dispatch the events in the bus right after the aggregate transaction has been committed": Even if you manually recovered events, you might not be able to "actually recover", because ordering could matter to downstream consumers. Something like a "first one to finish gets a bonus"-policy could be problematic.
A race-condition actually also exist even when the transport suceeds.

Deterministic ordering is crucial if you want the option to rehydrate event-projected readmodels to the same state, in an event sourced system this is really important in order to make versioning and iterating less painful.

[webdevilopers]

Great feedback @gquemener and @JulianMay, thanks!

Dispatch the events in the bus right after the aggregate transaction has been committed - this was indeed our current approach so far.

Most of our process managers are simple event listeners that synchronously build a new command and send it to the bus.
The messages in this bus are then handled async. If they fail we can retry. In PHP we use the Symfony Messenger for this w/ RabbitMQ.

The only possible "error" can occur while building the new command. Normally this should not happen since the data from the original aggregate event should always have a "valid state" to process. But I must confess that we do not have tests for this part yet. So when adding new features we might miss something in the command and the error occurs.
For those cases we normally have a recovery console command e.g. "send-email-after-registration-for-account-id:...".
But of course there are more complex use cases. And we would prefer a "full-process-retry" instead of adding commands all the time.

[JulianMay]

We've spend soo much of 2018 writing and running "data-patching tools" in 2018 because of consistency issues, hence my (over) emphasis on that issue 😅

FWIW: I personally don't have async between handle(event) => dispatch (command), but prefer it "earlier",
My heuristic is something like "never argue with facts (committed events)".
Synchously, the subscription can handle it's own problems, like case-specific compensations, catching an email-exception in the event-handler and scheduling automatic retry later.

I wrote some stuff about what I learned so far regarding these kind of issues (
https://www.eventstore.com/blog/counterexamples-regarding-consistency-in-event-sourced-solutions-part-1 ) which might be a relevant reference for posterity of this thread

[gquemener]

Most of our process managers are simple event listeners that synchronously build a new command and send it to the bus. The messages in this bus are then handled async.

FWIW: I personally don't have async between handle(event) => dispatch (command), but prefer it "earlier",

Indeed, I've already read that, as a rule of thumb, not a doctrine, only events should be considered for asynchronous handling, not commands. AFAIR, it had to do with being able to immediately respond to error while handling commands.

here and here for instance.

But I digress, feel free to open an other issue on this specific topic if you think it's appropriate :)