Aggregates delegating to other Aggregates

[danieleades]

hi there! From my reading, it seems to be a reasonably common pattern to have one aggregate (the 'root aggregate') delegate to multiple other aggregates. Or to put in another way, a single aggregate is the public entrypoint, but is in practice a 'facade' over multiple other aggregates.

I'm not sure how this pattern is implemented in the wild. Certainly it doesn't seem very 'rusty' for each aggregate to manage its own access to the data layer.

How would one implement this pattern using this crate?

One thing i thought of was using an enum to multiplex/coalesce multiple event streams, and then decomposing that again in the aggregate to delegate to other aggregates. Seems a little convoluted though.

in pseudocode (don't try and compile this!)-

let event_stream1: EventStream<Event1> = ... ;
let event_stream2: EventStream<Event2> = ... ;

enum Event {
    event1(Event1),
    event2(Event2),
}

let multiplex_stream: EventStream<Event> = stream::select(
    event_stream1.map(Into::into)),
    event_stream2.map(Into::into))
);

then the root aggregate would match on the event types and delegate each to 'child' aggregates (possibly recursively).

Am i making any sense?

[ar3s3ru]

Hey! Can you clarify what are you trying to do specifically? 😄

In general, Aggregates are supposed to be used only to maintain business invariants and represent transactional boundaries. A great resource on Aggregate modeling here: https://www.dddcommunity.org/library/vernon_2011/

One common thing is for Aggregates to depend on changes from other Aggregates. Example: an Order (aggregate 1) gets paid, so an Invoice (aggregate 2) needs to be created. In Event Sourcing (not just about this crate) one handles these cases using Process Managers.

A Process Manager is a component that listens to Domain Events and emits commands.

From the case above:

OrderEvent::WasPaid(x) -> InvocingProcessManager -> Invoice::Create(order: x)

Then the command is either handled synchronously inside the Process Manager (it might have the Command Handler injected, so you have it accessible from the Process Manager) or asynchronously:

ProcessManager -> Command [process 1]
  -> Message Broker 
-> Consumer -> Command Handler -> Domain Events [process 2]

Process Managers are basically Stream Processors (big thing in event-driven architectures) and it's a rather broad concepts. Process Managers can have state (stateful stream processors) or be stateless (typically coordinators, like Sagas).

Other interesting reading about Process Managers: https://event-driven.io/en/saga_process_manager_distributed_transactions/

Does this answer your question to some extent? (If so, I can explain how you could implement one with this library ✌🏻 )

[danieleades]

i guess that example describes how a change in one domain could create events to be processed in another domain. I'm more thinking about a domain which is composed of sub-domains. For example a transactional boundary might be represented by an Order, which is in turn composed of a list of OrderItems and a Customer (and so on).
An item in the order can be modified, say. So either the Order is an aggregate which is solely responsible for managing all of this state, or it delegates to other Aggregates under the hood (though only the Order aggregate is exposed to the consumer). Thus the Order is a transactional boundary, but its implementation is able to delegate to other transactions boundaries.

here's an example i stole from https://stackoverflow.com/questions/1958621/whats-an-aggregate-root

---

I was wondering what sort of a construction you might use to support this patttern?

I had considered multiplexing commands and events, along these lines

enum OrderEvent {
   Customer(CustomerEvent),
   OrderItem(OrderItemEvent),
}

and then having the OrderAggregate match on the event type and delegate to respective Aggregates.

But then i considered you might want to store the different event types as separate streams, precisely because you may wish to process them separately (for example in a projection) so the repository would also need to do some kind of multiplexing over multiple sources.

[danieleades]

ok, i'm still playing with this idea.

my current thought is to split the Aggregate trait into an Entity trait, who's only purpose is to be delegated to by an aggregate.

something along these lines-

#[async_trait]
pub trait Entity {
    type State: Default + Send;
    type Event: Send;
    type Command;
    type DomainErr: std::error::Error;
    type ConflictErr: std::error::Error;

    /// Update the state with a new event
    ///
    /// # Errors
    ///
    /// This method can fail if the history of the events is incompatible with
    /// the domain-specific logic of this aggregate. This should only be
    /// possible if the stream of events has been populated from two different
    /// sources, and that history is somehow incompatible.
    fn apply_event(
        &self,
        state: Self::State,
        event: &Self::Event,
    ) -> Result<Self::State, Self::ConflictErr>;

    /// Apply a command and return a list of events
    ///
    /// # Errors
    ///
    /// This method will return a [`Entity::DomainErr`] if the command is
    /// invalid with respect to the given state.
    async fn handle_command(
        &self,
        state: &Self::State,
        command: Self::Command,
    ) -> Result<Vec<Self::Event>, Self::DomainErr>;
}

pub(crate) trait EntityExt: Entity {
    /// reconstitue the entity state from a list of events
    ///
    /// # Errors
    ///
    /// This method can fail if the history of the events is incompatible with
    /// the domain-specific logic of this aggregate. This should only be
    /// possible if the stream of events has been populated from two different
    /// sources, and that history is somehow incompatible.
    fn recover_state<'a>(
        &'a self,
        events: impl IntoIterator<Item = &'a Self::Event>,
    ) -> Result<Self::State, Self::ConflictErr> {
        let state = events
            .into_iter()
            .try_fold(Self::State::default(), |state, event| {
                self.apply_event(state, event)
            })?;
        Ok(state)
    }
}

#[async_trait]
pub trait Aggregate: Entity {
    type Id;

    /// Return the unique identifier associated with this [`Aggregate`]
    fn id(&self) -> Self::Id;
}

thus, using the above example, a customer and an order would be Entities, and the Order would be both an Entity and an Aggregate that would delegate to the other Entities. Thus an Aggregate is really a facade over a number of Entities which collectively make up a domain.

ends up looking a bit like this (i'm sure there's a cleverer way to reduce the boilerplate in the delegation!):

mod order {
    use async_trait::async_trait;
    use event_sourcing::Entity;

    use super::{customer, items};

    pub struct State {
        customer: customer::State,
        items: items::State,
    }

    pub enum Command {
        Customer(customer::Command),
        Items(items::Command),
    }

    pub enum Event {
        Customer(customer::Event),
        Items(items::Event),
    }

    pub struct Order {
        customer_entity: customer::Customer,
        items_entity: items::Items,
    }

    #[async_trait]
    impl Entity for Order {
        type Command = Command;
        type ConflictErr = ConflictErr;
        type DomainErr = DomainErr;
        type Event = Event;
        type State = State;

        fn apply_event(
            &self,
            state: Self::State,
            event: &Self::Event,
        ) -> Result<Self::State, Self::ConflictErr> {
            let State { customer, items } = state;

            match event {
                Event::Customer(event) => {
                    customer = self.customer_entity.apply_event(customer, event)?;
                }
                Event::Items(event) => {
                    items = self.items_entity.apply_event(items, event)?;
                }
            }

            Ok(State { customer, items })
        }

        async fn handle_command(
            &self,
            state: &Self::State,
            command: Self::Command,
        ) -> Result<Vec<Self::Event>, Self::DomainErr> {
            match command {
                Command::Customer(command) => {
                    let events = self
                        .customer_entity
                        .handle_command(state.customer)
                        .await?
                        .into_iter()
                        .map(Event::Items)
                        .collect();
                    Ok(events)
                }
                Command::Items(command) => {
                    let events = self
                        .items_entity
                        .handle_command(state.items)
                        .await?
                        .into_iter()
                        .map(Event::Items)
                        .collect();
                    Ok(events)
                }
            }
        }
    }

    pub struct DomainErr;
    pub struct ConflictErr;
}

mod customer {
    pub struct State;
    pub struct Command;
    pub struct Event;

    pub struct Customer;

    impl Entity for Customer {}
}

mod items {
    use event_sourcing::Entity;

    pub struct State;
    pub struct Command;
    pub struct Event;

    pub struct Items;

    impl Entity for Items {}
}

In this model (and I think in your current implementation) the 'AggregateRootis really anAggregate` "handler".

You'll notice that the apply_event method takes the event by reference, instead of moving. This allows me to avoid cloning in the recover_state method, however specific implementations can still choose to clone if they wish (provided Event: Clone)

The ConflictErr type is an interesting one. I'm imagining that since the stream of events was created by the Aggregate, then this should usually be infallible (this should be an invariant of the application). However, I'm envisioning a use case where event sourcing is used for an "eventually consistent" model- multiple agents in a system are generating events and those event streams are later combined (split brain). The combined event stream may or may not be reconcilable. think CouchDB-type replication. I'm sure you've thought a lot more than me about this problem!

[ar3s3ru]

i guess that example describes how a change in one domain could create events to be processed in another domain. I'm more thinking about a domain which is composed of sub-domains. For example a transactional boundary might be represented by an Order, which is in turn composed of a list of OrderItems and a Customer (and so on).

I'm not sure what you refer to by "domain" and "sub-domain", but just to clarify my understanding of them using the DDD terminology:

• Domain refers to the problem you're operating in, e.g. the E-commerce domain.
• Domains are usually very big and complex, thus can be broken down in many Sub-domains, which highlights a specific problem to solve for the Domain we're operating in. There are various kinds of sub-domains but in DDD we (should) focus on the Core Domains
• Sub-domains typically live in the "problem space" (i.e. they define a problem to overcome as part of the business you're working with) and are typically implemented as Bounded Contexts in the solution space (how you solve the problem).
• Bounded Contexts have a clear boundary (e.g. a microservice) and inside of them we model multiple Entities to help solve the problem. So enter Value Objects (obejcts with deep equivalence) and Entities (uniquely-identifiable objects).
• Aggregates are a specific kind of Entities, which are used to enforce business invariants and implement a transactional boundary.
• Bounded Contexts commonly use multiple Aggregates to solve the problem at hand, sharing some kind of relationship (Order -> OrderItem, Person -> Account).
• Bounded Contexts usually communicate with each other (see Context Mapping), and the communication is usually happening through Aggregate relationships.

So far we probably have the same understanding, just wanted to clarify the lingo (Ubiquitous Language right? 😄)

An item in the order can be modified, say. So either the Order is an aggregate which is solely responsible for managing all of this state, or it delegates to other Aggregates under the hood (though only the Order aggregate is exposed to the consumer). Thus the Order is a transactional boundary, but its implementation is able to delegate to other transactions boundaries.

This is important, because you're establishing correlation between Aggregates, whether they belong to the same Bounded Context or not it's the same thing.

I really encourage you to read the articles on Aggregate modeling by Vernon (I linked them above, here again: https://www.dddcommunity.org/library/vernon_2011/), especially Part II.

Rule: Reference Other Aggregates By Identity
Prefer references to external aggregates only by their globally unique identity, not by holding a direct object reference (or “pointer”)

Rule: Use Eventual Consistency Outside the Boundary
Any rule that spans AGGREGATES will not be expected to be up-to-date at all times.
Thus, if executing a command on one aggregate instance requires that additional business rules execute on one or more other aggregates, use eventual consistency. Accepting that all aggregate instances in a large-scale, high-traffic enterprise are never completely consistent helps us accept that eventual consistency also makes sense in the smaller scale where just a few instances are involved.

What it means, with an example: if changes to the OrderItem are affecting Order (just to quote your example), then we have two cases:

1. OrderItem is nested inside Order aggregate, thus it will belong to the Order event stream,
2. Order is eventually consistent with OrderItem; changes to OrderItem will emit a Domain Event that is listened by a Process, which will take care of reflecting the required change in state for Order. In this case, OrderItem and Order belong to completely separate event streams.

More generally speaking, multiple Aggregates should not be changed as part of a single Command (this is the transactional boundary part). The problem with optimistic concurrency should've been a giveaway 😉

So bottom line: you don't need all of this complexity (and might actually be the wrong way of implementing Event Sourcing), just use eventual consistency and model explicit processes to eventually reconciliate the state of your Aggregates.

[danieleades]

plenty of food for thought there!

I've been playing around with Rust for a while now, but I'm very new to event sourcing and domain-oriented programming. I really appreciate you taking the time to discuss, and to link resources.

2. Order is eventually consistent with OrderItem; changes to OrderItem will emit a Domain Event that is listened by a Process, which will take care of reflecting the required change in state for Order. In this case, OrderItem and Order belong to completely separate event streams.

do you have any examples of what this looks like with your library? (Perhaps the example app already covers this pattern)

[ar3s3ru]

@danieleades sorry to leave you hanging for a week, work has been crazy 😅

do you have any examples of what this looks like with your library? (Perhaps the example app already covers this pattern)

Unfortunately the only close example I can find of this in the example app is the TotalOrdersProjection.

Essentially, taking the example from the previous messages, you need the following:

1. Commands that modify an Order aggregate,
2. Commands that modify the OrderItem aggregate,
3. Say that changes to OrderItem must be reflected in an Order, then we need to process a Subscription of OrderItem events,
   1. This is the trait for a Subscription.
4. The component that consumes the OrderItem events off a Subscription must filter-map the interesting events to Order commands. This is generally called a Process Manager.
5. The Order commands are either handled synchronously (through a command handler) or asynchronously (using a separate thread, or on a different process/container/pod using an event bus, etc.)
6. When the command is handled, the state of the Order affected will change.

The synchronization is eventually consistent, because the original command -- the one that modifies the OrderItem aggregate -- does not directly modify the Order aggregate, but it needs to go through a Process Manager and a Command Handler first.

---

Please note: when I talk about an Order command, I'm not talking about the Aggregate::handle method.

Your application exposes certain "actions" as part of its public API; for some it's a REST API, for others is a gRPC API, etc. Say that one of these actions is a POST /order-item endpoint that creates a new OrderItem: the Command performed behind this action is a CreateOrderItem command. Commands are handled by Command Handlers (sometimes they're also referred to as Application Services).

In the case of CreateOrderItem, we can expect a component, called CreateOrderItemHandler that acts as the Command Handler for that type. Since we're using an Aggregate to model OrderItem, the command handler will:

1. Parse the CreateOrderItem into an OrderItem::Command enum variant,
2. Call OrderItem::handle(command) (called through the AggregateRoot::handle) and get a new instance of OrderItem,
3. Use a Repository instance to add the new OrderItem instance, which will internally commit the domain events returned by OrderItem::handle.

Hope this was clear 😄