The actor model provides the guarantee that requests to a single instance are processed serially, while requests to different instances can be processed concurrently. Distributed Erlang allows these instances to be scaled out amongst a cluster of nodes with transparent routing of commands to the instance, regardless of which connected node it is running on.
In Elixir and Erlang, the OTP platform provides the building blocks to host an aggregate instance as a process (as a `GenServer`). Following the "functional core, imperative shell" style I model the domain code as pure functions with the host process taking care of any IO, such as reading and appending the aggregate's events.
I find this point surprising. I would say the exact opposite. I would say that pub/sub and event sourcing are two sides of the same coin: events.
> what to do if sub happens after pub
That should only ever be a problem with a non-durable transport that doesn't have serialized writes per topic. Which, admittedly, can be pretty common. But it's not so much an event sourcing or pub/sub issue as much as a choice of message transport issue.
> Concurrency. Ensuring aggregates are essentially single-threaded entities is a must. Having the same aggregate id running in multiple places can cause some really fun bugs. This usually requires a distributed lock of some sort.
Or it requires partitioning the queues and using an optimistic lock when writing (just to be on the safe side).
I meant in the context of getting it right. I didn't experiment with all the pub/sub systems at the time, but most I experimented with would lose data in a catastrophic event and cause inconsistencies. This was several years ago though.
Fair enough. Those are probably message buses or message queues that are ephemeral transports. Since event sourcing is predicated upon permanent storage of events, there's no way to lose events that have already been committed (unless someone actually physically deletes the events).
One common technique if you have commutative aggregates is to have each writer just write to their own spot and then do a range query on read to re-join. If your aggregates don’t commute this, of course, doesn’t work and you’re stuck in “single threaded” land. I do remember reading a paper that avoided this, but i can’t remember the implementation / trade offs (If I can find the paper I’ll post it here)
This isn't necessarily an abuse of terminology - I believe if we look at an operator that takes sequences of events and appends them, what we're talking about is in fact commutativity of that operator (in terms of the impact on the system).
Yeah sure, imagine you're trying to compute count of how many times you've seen something. This is a SUM aggregate.
I assume the reason the author said "Ensuring aggregates are essentially single-threaded entities is a must" was because they are mutating a single key, if two different processes try to change the same value, you get inconsistent state.
An example of this would be:
Current K/V Pair: "A" -> 2
Modifier 1: Reads "A", gets 2
Modifier 2: Reads "A", gets 2
Modifier 1: Writes "A" -> 3
Modifier 2: Writes "A" -> 3
This means our sum is now inconsistent, as we would expect the value to be 4 in a single-threaded system.
However, because we are doing a sum, we can use commutativity to remove this conflict. Instead of each writer trying to write to a single key, you might make a compound key (Id, <Unique Writer Identifier>)
Current Value Pairs: [("A", "Modifier 1") -> 1, ("A", "Modifier 2") -> 1]
Modifier 1: Reads ("A", "Modifier 1"), gets 1
Modifier 2: Reads ("A", "Modifier 2"), gets 1
Modifier 1: Writes ("A", "Modifier 1") -> 2
Modifier 2: Writes ("A", "Modifier 2") -> 2
Then, a reader could just ask for all the keys with the prefix "A" (This is the range query). So a reader gets back (2, 2), which then can now merge into 4 as SUM is a commutative operation. Because the reader is taking care of the final aggregation step, there's no concurrency conflict so you can get away with having any number of writers as long as the read and read-side computation is cheap enough.
Some aggregates are commutative only in certain forms, like AVERAGE. To explain further, if one writer says the AVERAGE is 5 and another says it is 7, I can't combine those to say that the global average is 6.
However, if each writer stores both SUM and COUNT, I can use (COUNT1 + COUNT2)/(SUM1 + SUM2). This is because division doesn't commute, so I have to delay the non-commutative operation until the reader if I want to be able to merge two data sources.
Edited: Formatting
In DDD, an "aggregate" is a collection of state that can only be changed atomically, as a whole, by the outside world. It helps to think concretely of an Actor, i.e. some entity that owns a collection of state, such that the only way to mutate that state is to send a message to the actor. No matter how many messages come in from the outside world, only one is being handled at a time by the entity, and there are no concerns of concurrent access. Necessarily though, all messages are handled in a linear order.
From Martin Fowler's article on aggregates [0]:
> Aggregates are the basic element of transfer of data storage - you request to load or save whole aggregates. Transactions should not cross aggregate boundaries.
I think you'll be interested in CRDTs. They're like DDD aggregates where mutation operations are essentially always commutative, so there's a bit more alignment between the senses of "aggregate" being confused here.
I tried doing something similar to this several years ago, and here's a few issues I ran into:
1. Pub/sub in Event Sourcing is a bad idea. It's really hard to get right. (what to do if sub happens after pub due to scaling issues/infrastructure, etc?) Instead it's better to push commands deliberately to a process manager that handles the inter-domain communication and orchestration.
2. Concurrency. Ensuring aggregates are essentially single-threaded entities is a must. Having the same aggregate id running in multiple places can cause some really fun bugs. This usually requires a distributed lock of some sort.
3. Error handling. I ended up never sending a command to a domain directly, instead I sent it to a process manager that could handle all the potential failure cases.