Handling asynchronous domain events with multiple handlers & allowing them to fail independently

Question

We're building a service using DDD, CQRS, & Hexagonal Architecture that allows the user to upload a CSV feed for which every row will be transformed and sent to a third party over HTTP.

We have two aggregates: Feed & Delivery. Feed contains original entries from CSV and Delivery contains payload, status, and fail count.

Both Feed & Delivery generate domain events which we then insert to a DB in the same transaction as the aggregate that emitted it ("transactional outbox" pattern). This way we keep events consistent with their aggregate states and handle eventual consistency well, i.e. aggregates are always consistent, but overall application state will be eventually consistent because the whole processing chain might take a while.

Finally, we have an out-of-process listener ("transaction log tailer") that reads inserted events and asynchronously feeds them back to our app so that we can build complex workflows. Once event is processed, its processed value is updated in DB (to prevent it from being processed again in case our listener delivers it more than once). This allows us to keep all our use cases small and let them fail as much as they want: every event will be retried independently. Our application layer acts as a "router" for asynchronous domain events, and also handles commands.

Our events are FeedCreated, DeliveryCreated, DeliveryScheduled, & DeliverySucceeded. Our commands are CreateFeed, ScheduleDelivery, & ExecuteDelivery.

The flow is as follows (format is event->command->event):

• [User] -> CreateFeed -> FeedCreated
• FeedCreated -> CreateDeliveries -> DeliveryCreated[]
• DeliveryCreated -> ScheduleDelivery -> DeliveryScheduled
• DeliveryScheduled -> ExecuteDelivery -> DeliverySucceeded (on success) || DeliveryScheduled (on failure)

All of this works pretty well so far, but with that come few questions.

Let's say we add a new command RegisterAttempt which writes delivery information to another aggregate, and we want to call it every time DeliveryScheduled happens. Our options are:

• Do this in the same handler (and same transaction) where we already handle DeliveryScheduled (which runs ExecuteDelivery). Cons: we'll violate SRP because now our handler runs two commands (ExecuteDelivery and RegisterAttempt), so failure from one of them will fail both.
• Make 2 separate handlers. Cons: one of them will mark the event record as "processed" and thus prevent second handler from commiting on completion.
• Insert one copy of each event for every handler (aggregate) interested in processing it. Cons: this will require the publisher (use case) to know about all stakeholders ahead of time and will polute DB with duplicate events which doesn't sound right since only one event happened, so we're mixing concepts of topic & subscriptions here.
• Make event handler behave like a multiplexer: upon receiving 1 event, create N "EventInstance" records (one for each handler) and put them on a bus so that they can then be processed by actual handlers (aggregates) independently. Cons: we now have events and event instances, which sounds confusing.
• Store a list of "succeeded" handlers within the event record, populating it after every successful handler so that if one fails, we retry the whole event, but skip handlers that previously succeeded. Cons: this again violates SRP because our handler is operating on multiple aggregates.
• Have only one handler for any event and do only one thing (command) there. This is probably the best and easiest solution. It also respects SRP really well at several levels and makes handlers (and surrounding code) very simple. Cons: this requires us to unwind flows like Event1->(Command1,Command2,Command3) into Event1->Command1->Event2->Command2->Event3->Command3 which increases overall latency, makes parallelization harder and introduces surrogate events that might make no sense from the domain's point of view.

TL;DR: When using Transactional Outbox pattern in DDD, what if we want to have multiple handlers (i.e. multiple aggregates that need to react to some event) for the same domain event that need to be retried independently, but there's only a single instance of the event? Where to store the state of each handler?

Even though our architecture works well so far (because we currently only have 1 handler for any given event), I feel like I may be getting some concepts fundamentally wrong. Any thoughts & advices would be greatly appreciated!

Answer 1

TL;DR: Do not use the processed field used by the listener of the transactional outbox pattern to detect failures in downstream handlers/subscribers.

Use a publish/subscribe structure together with a 'supervisor'-service (or multiple of them). The publish/subscribe structure is as @ScottMiddenbergen suggests. It decouples event producers/publisher from the consumers/handlers/subscribers. Multiple handlers can subscribe to a single event. You can still apply the transactional outbox pattern and have the listener send the events/messages to the broker. Use the processed field only to recover from crashes of the listener. The broker has mechanisms to ensure guaranteed delivery. The guaranteed delivery can safeguard against some forms of failure.

But we want the producer to be independant from the handler(s) and we want to be able to subscribe multiple handlers to a single event. Therefore, a handler cannot wait for confirmation from all its subscribers before it confirms receiving and handling the event it received. So guaranteed delivery by the broker doesn't solve the problem when a handler crashes after receiving and confirming the message. This is where the 'supervisor' comes in.

The supervisor is responsible for ensuring that the handlers run as expected. The supervisor takes one of the responsibilities OP @AndrewDunai has given to the listener in their current solution: namely recovering from failures in downstream handlers. This makes that the listener is only responsible for sending events to the broker when a transaction completes.
In OP's solution the downstream handlers are responsible for detecting errors in the handling of the event, we shift this responsibility to the supervisor. This makes that the handlers are only responsible for handling the event and only for their part of the handling. (When a downstream handler detects a problem itself, it could still notify the supervisor of this problem.)
I named the supervisor after a concept from erlang-otp. (There probably is a pattern in event-driven architecture for this concept. I'm not currently aware of that pattern, so I like to learn.):

1. The supervisor is configured with requirements of the software system.
   The requirements that the supervisor supports are of the form "when event e₁ is sent, event e₂ should occur within time t.
   (Perhaps you can tie in the representation of the requirements the in supervisor with your integration tests.)

2. We subscribe the supervisor with the producers/event-streams where it can capture the events it's interested in.
   For example:

   *    eDelivery_created({correlation_id})
   |\
   | \
   |  \
   *   |  HandlerSchedule_delivery
   |   *  Supervisor::Expectation_creator
   |   *  eExpect_event(eDelivery_succeeded({correlation_id}) || eDelivery_failed{correlation_id}, {expiration_time})
   *   |  eDelivery_scheduled
   *   |  HandlerExecute_delivery
   *   |  eDelivery_succeeded({correlation_id})
    \  |
     \ |
      \|
       *  Supervisor::Event_awaiter // Yay! all went successfully, no action needed 
   

   or

   *    eDelivery_created({correlation_id})
   |\
   | \
   |  \
   *   |  HandlerSchedule_delivery
   |   *  Supervisor::Expectation_creator
   |   *  eExpect_event(eDelivery_succeeded({correlation_id}) || eDelivery_failed{correlation_id}, {expiration_time})
   *   |  eDelivery_scheduled
   X   |  CRASHED!! HandlerExecute_delivery
       |
     * |  eTimeout
      \|
       *  Supervisor::Event_awaiter
       *  eRequired_event_did_not_occur({correlation_id}) || eDelivery_failed{correlation_id}, {expiration_time})
       *  Supervisor::Correction_mechanism // starts repairs on the events/handlers involved
   

   (Note: I introduced the Delivery_failed-event to show that we can configure the supervisor to ignore intermediate events and handlers, considering them implementation details. Checking for Delivery_succeeded || Delivery_scheduled would catch the Delivery_scheduled from the Schedule-delivery-handler and thereby missing crashing Executor.)

3. When the supervisor receives the antecedent-event, it registers in its internal store that it should await the expected events.
   (This differs from the listener approach of OP in that we not mark the antecedent event as processed, the supervisor only cares for the result).

   • When the supervisor receives the required consequence event, it removes the consequence event from the set of events it is waiting for.
   • When the supervisor does not receive the expected event within the expected timeframe, the supervisor has sever options to intervene: re-insert the antecedent event into its queue (or a fresh copy of the antecedent event), restart the failing service (or if you are using a cluster with failover capability, let mark the service as failing and let the cluster handle it), notify the devops team/administrators, etc.

By explicitly monitoring the requirements of our system (in terms of events) the system is self-correcting. And we can easily support events that should have more than one effect. We can also distinguish between required event chains and those that are optional or nice to have. When introducing new functionality, we first build the chain of events and handlers and only when the chain is working we add a rule to the supervisor. When removing functionality we do this vice versa: first remove the requirement from the supervisor, then dismantle the chain of events and handlers.

There are some caveats.

• The supervisor itself can crash: heartbeat can help to detect and recover from this situation and the system will continue working as required without a supervisor until a critical handler crashes.
• Events may arrive out of order at the supervisor. (This is unlikely, since the transactional outbox pattern often ensures message are sent in the order in which they were sent by the application.) This might cause the supervisor to miss expected events and cause damage by its attempts to repair. Either we configure and test the messaging network carefully or we use a window within which we capture and temporarily preserve events that might be expected later.