Dynamic Consistency Boundaries

[cdegroot]

Please add to the material. My starting point is:

• Sarah Pellegrini's Talk (italian with subtitles);
• AxonIQ blog post;
• Another Talk by Sarah, with less Italian and subtitles.

From my org notes, summarizing:

The basic deal is to have the concept of aggregate (with an id) but only have message handlers.

They build the subset of aggregate root state they need by selecting and replaying events

Command -> MessageHandler

MessageHandler <-> EventLog(query events of type A, B, C, for aggregate id X)

MessageHandler -> Event

Note that multiple MessageHandlers can run in parallel. Therefore, they need to optimistically lock
on the query. Sara explains the details at around 38m30s, but essentially the event set returned in the
query contains a last version number, and the append sends the query + the last version number seen. If the
same query matches a later version number, the transaction is retried.

[cdegroot]

Mapping state building could be done through streams, but that requires a more premeditated approach. Another idea is to slap an index on the log and do the queries ad-hoc.

The optimistic query wouldn't be too hard.

Smells, superficially, like two changes to the EventStore API and then there should be enough meat to e.g. implement a sample system. Maybe based on https://github.com/bwaidelich/dcb-example-courses?

[jwilger]

The first thing we need to do, IMO, is align on the "Context and Problem Statement" part of the ADR. Otherwise we'll all end up talking past each other. And it's possible there are alternatives to DCBs that would also address said problems that should be considered.

[jwilger]

So, what, specifically, is the problem we are hoping to address for users of Commanded?

[cdegroot]

To me, "god aggregates" + "you can only design once". A lot of stuff that is mildly related has to sit together because the aggregate needs to know about them, but there are logical interconnected subsets of related events/commands in there. And once events "belong" to an aggregate, that is it, all you can do for future stuff is add more there. I think there's pretty widespread agreement that "Aggregate Root" is not a super nice pattern in CQRS/ES and that most systems grow these huge aggregates that control too much (at my current job, a patient visit is one of these - we need to run pretty much all events through them, turning them into processing bottlenecks and it's also our only aggregate root that's now spread over multiple modules, each operating on a subset of the commands and events; if the subsets were disjoint, then that would indicate a clear path forward, but they overlap, of course :-))

[drteeth]

I think there are paths for splitting aggregates/streams, and ways to make smaller aggregates that work together. I have a Device aggregate and a separate DeviceBattery aggregate which handles only the battery updates. I think this weakly demonstrates that you can split them given enough work, possibly introducing an EventHandler to transform 1 stream into 2 streams.

I think the talks point to the real stumbling block: I have an Event that seems to strongly pulled towards two aggregates/streams that are otherwise separate (Course vs Student). I can fix this by running the command through the course and then use an EventHandler (or the dreaded ProcessManager) to let the correlated Student aggregate know about the change to the course, but by then we've lost any way to reject the initial command, and have a bunch of connecting cruft to wrangle.

So, to re-frame:
We are hoping to provide Commanded users with a way of considering events from multiple streams in CommandHandlers, and to emit events back to one or more of those streams. Handling commands this way should be protected from dirty reads and writes and should be atomic. This new path should be opt-in and should not disrupt the existing "single aggregate/stream" implementation.

[cdegroot]

Much better than my scribbles. I guess I shouldn't be jotting down notes during work time in public spaces ;-)

[drteeth]

Event-modeling-ish diagram of the traditional "one-stream-one-aggregate" way:

The complexity of coordinating between streams here is obvious, half of the work done here is spent marshalling commands and events to communicate what's happening between aggregates/streams. Worse, we have to implement compensating or rollback events to undo the enrollment if we exceed the student's course enrollment limit.

It is worth pointing out that in a traditional, non-reactive UI (a "Dead View", or across an API), we cannot let the user know if the student's course enrollment limit is exceeded. This is due to the asynchronous nature of the data flow here.

The nice thing here is that since each stream is fully independent, there is no contention between then. The OCC (Optimistic concurrency control) check can be handled simply by remembering the last position in the stream as an integer. Likewise, we only need to append events to a single stream.

When appending, Commanded creates the events and the links them first to the aggregate's stream (ex: course-123 or student-234). It then links those events in the "$all" stream which provides global ordering of events.

[drteeth]

Event-modeling-ish diagram of how it would be modeled with DCB:

[drteeth]

The conceptual model here is much much simpler, and matches what how the business thinks about it too.

We gain the ability to give the caller immediate feedback on both errors, and there are fewer commands and events to write and maintain.

We need a way to consider events from multiple streams when deciding to accept or reject a command.

[drteeth]

Here's a view of two courses and two students, and the events a command handler would need to consider in order to enroll Bob to the English Literature course.

Notice that we need to consider only certain events from the Lit course, as well as the creation events from Bob's stream. We will also need to consider all Student Enrollment events that relate to bob from all other course streams.

After loading these events, and folding over them to reach some internal state, we decide that we're going to accept Bob's enrollment, and we've also noticed that we've reached the maximum capacity of the literature class. In order to persist these events, we need to check that:

• Bob's stream has no new events that we care about (there aren't any in this case)
• The literature stream has not changed capacity, and there have been no new enrollments
• Bob has not been enrolled in any other courses

As we're dealing with multiple streams, we can no longer have a single integer and the UUID of a stream to make these checks.

The query presented in the talks includes might look like

query = %StreamQuery{
  ids: [
    %{course: "course-lit"},
    %{student: "student-bob"}
  ],
  events: [
    "CourseCreated",
    "StudentCreated",
    "StudentEnrolled",
    "CourseCapacityChanged"
  ]
}

That would net us:

1. CourseCreated from Lit (at stream version 1)
2. StudentCreated from Bob (at stream version 1)
3. StudentEnrollment from Algebrea for Bob (at stream version 3)
4. CourseCapacityChanged from Lit (at stream version 3)
5. StudentEnrollment from Alice for Lit (at stream version 4)

When we accept the command, we need to emit two events for the Lit stream at versions 4, and 5 respectively. When we try to append the events, the event store needs to confirm that:

1. There are no CourseCreated, StudentEnrollment, CourseCapacityChanged events in the Lit stream with sequence numbers higher than 3.
2. There are no StudentCreated events in the Bob stream with sequence numbers higher than 1.
3. There are no StudentEnrollment events in the Algebra stream that reference Bob with a sequence number higher than 3
4. There are no StudentEnrollment events in the Physics stream that reference Bob with a sequence number higher than 0

[drteeth]

As a possible solution, if we were to link StudentEnrollment events to the $all stream, course's stream, and the student's stream then we could gather the events we need just be reading from student and course streams.

I guess this is like manually indexing, so maybe there is also an argument for letting Postgres do it against the $all stream with database indexing.

[drteeth]

I spent some time documenting insert_events.sql.eex so we can all grok how it works:
https://github.com/commanded/eventstore/blob/master/lib/event_store/sql/statements/insert_events.sql.eex

We could loop over the last bit where it links the events from the "source stream" to the $all stream. We'd do this for every other stream in the query. This does imply that we'd need to identify one of the streams as "primary" and the others as... others.

[drteeth]

It looks like I misunderstood the OCC check, you re-run the StreamQuery, but you only need to check the position of the most recent event in the results against what it was when the handler loaded the events. So, despite what I said, it is enough to store the StreamQuery and a simple integer representing the sequence of the most recent even in the $all stream.

[drteeth]

I think we can see here that appending events to the stream is not going to be terribly hard, what is going to be harder is:

1. Collecting events from across the streams in an efficient way
2. Doing the Optimistic concurrency control

[drteeth]

There is also the question of what to do with Commanded's Aggregates.

Aggregates are two things in Commanded:

1. The code used to re-hydrate some state by folding over a single stream's events
2. An OTP process that caches this state

If we're going to move from 1 aggregate module with 1 process to 1 command handler per workflow are we also going to create processes for these command handlers?

[cdegroot]

My hunch is that aggregates will become a sort of special case of command handler and that yes, you want processes for them just the same.

[jwilger]

I already do this with my projects in a way. For every MyCommand I have a MyCommand.Aggregate module. The reason I do this now is so that I avoid building god-aggregates. Each aggregate has fields only for the data that is actually needed by the command to make a decision. The apply function is set up to noop on events the aggregate doesn't care about (and log a debug log that the event was skipped).

Not only does this reduce the cognitive load of looking at the aggregate code, it also cuts down on memory usage (assuming you manage aggregate lifetimes appropriately.)

I have confirmed that Commanded will support multiple aggregates reading a single event stream without issue; the process is identified by both aggregate module name and stream id together.

[jwilger]

I would go as far as to say that this should be the recommended approach, given that it requires no changes to the library, has pretty clear performance and modeling benefits, and has no effect on anyone who wants to keep doing things in the current manner.

In particular, this change is what finally broke me of some dysfunctional (pun intended) OO thinking in my modeling process. Since I suspect a lot of folks come to Elixir from an OO background, the more guidance we give toward functional design thinking the better (as a language community.)

[jwilger]

As an aside, for anyone curious, this is what I'm using to implement this pattern (and cut down on some of the repetitive boilerplate) in my Commanded apps. https://github.com/jwilger/commanded_boilerplate

[drteeth]

I'm imagining that this would cause the various Aggregate to contend over the expected stream version?

If we imagine (a contrived example) of dispatching two different commands:

dispatch(%Withdraw{account: 1, amount: 100}) 
# starts Withdraw.Aggregate process
# read 0 events from disk
# appends `Withdrawn` event to stream as v1 

dispatch(%Deposit{account: 1, amount: 100})
# starts Deposit.Aggregate process
# read 1 events
# appends `Deposited` event to stream as v2 

dispatch(%Withdraw{account: 1, amount: 100})
# Withdraw.Aggregate is already running with state at v1
# try to append `Withdrawn` event 
# expected version fails as the stream has moved on to v2
# reloads the events from disk
# retries and appends `Withdrawn` events to stream as v3

dispatch(%Deposit{account: 1, amount: 100})
# Deposit.Aggregate is already running with state at v2
# try to append `Deposited` event 
# expected version fails as the stream has moved on to v3
# reloads the events from disk
# retries and appends `Deposited` events to stream as v4

I don't have a good grasp of how much of an actual problem this would be in a production system, but it does seem to defeat some of the purpose of having processes in front of the event store. I'm thinking of it in terms of a cache miss.

You're right that this is very similar to how it might work for DCB if we assume that we'll always (mostly?) end up with 1 command per command handler / StreamQuery / Process. DCB relaxes some of the constraints at the cost of an increase in potential contention. If you advance any of the streams that a command handler depends on, then it will have to reload from disk.

I think this is probably just fine as people outside of Elixir, and people like @yordis, always read events from disk and that's good enough.

If you can model ontop of a single stream, you can avoid contention, if you can't, you trade a little bit of contention for avoiding the quagmire of juggling commands and events with ProcessManagers. I think that's a win.

[drteeth]

Here's the structure of the streams & events tables. I've omitted FK constraints and References.

                         Table "event_store.events"
     Column     |           Type           | Collation | Nullable | Default 
----------------+--------------------------+-----------+----------+---------
 event_id       | uuid                     |           | not null | 
 event_type     | text                     |           | not null | 
 causation_id   | uuid                     |           |          | 
 correlation_id | uuid                     |           |          | 
 data           | bytea                    |           | not null | 
 metadata       | bytea                    |           |          | 
 created_at     | timestamp with time zone |           | not null | now()
Indexes:
    "events_pkey" PRIMARY KEY, btree (event_id)

                 Table "event_store.stream_events"
         Column          |  Type  | Collation | Nullable | Default 
-------------------------+--------+-----------+----------+---------
 event_id                | uuid   |           | not null | 
 stream_id               | bigint |           | not null | 
 stream_version          | bigint |           | not null | 
 original_stream_id      | bigint |           |          | 
 original_stream_version | bigint |           |          | 
Indexes:
    "stream_events_pkey" PRIMARY KEY, btree (event_id, stream_id)
    "ix_stream_events" UNIQUE, btree (stream_id, stream_version)

                                                Table "event_store.streams"
     Column     |           Type           | Collation | Nullable |                        Default                         
----------------+--------------------------+-----------+----------+--------------------------------------------------------
 stream_id      | bigint                   |           | not null | nextval('event_store.streams_stream_id_seq'::regclass)
 stream_uuid    | text                     |           | not null | 
 stream_version | bigint                   |           | not null | 0
 created_at     | timestamp with time zone |           | not null | now()
 deleted_at     | timestamp with time zone |           |          | 
Indexes:
    "streams_pkey" PRIMARY KEY, btree (stream_id)
    "ix_streams_stream_uuid" UNIQUE, btree (stream_uuid)

[drteeth]

Notably missing is an index on event_type

[cdegroot]

Somewhat off-topic, but I would not mind having at least the option to put more indexes on events. I keep seeing people doing ad-hoc queries - for debugging, troubleshooting, or simply some ad-hoc aggregation - and these queries are invariably slow. Optional, because an index will make writes slower and increase disk space, but having event type, and even causation and correlation and creation date indexed would come in handy

[yordis]

I've just realized there are some misunderstandings surrounding this topic, prompting me to ask a question:

How would it work when student streams are in database1 and course streams are in database2?

[drteeth]

It would not work.

I think we have to assume a single database, probably even a single commanded application.

[yordis]

I'm sorry for the question. I just wanted to document this publicly in case others ask me if my "one-aggregate-one-module" approach fixes the same problem as DCB, which it doesn't. That is an entirely different conversation.

Also, because of that limitation, those Processors are invaluable in the system and architecture design. It would be critical to document and make it extremely clear.

[drteeth]

Always worth asking!

I hadn't thought about it honestly, I just assumed we'd be doing this in one database. Can you think of a way that it could work across databases?

If we bring in a second database, I think we'd have to deal with the dual-write problem no?

[yordis]

Before we proceed, let me clarify: my focus is on system design and architecture, and the potential impact of neglecting these considerations.

If a single database meets your needs, that's perfectly fine. From the Commanded’s perspective, what matters most is ensuring DCB does not disrupt existing reliability while continuing to improve the system. I advocate for experimentation and exploring different approaches tailored to specific needs. I am not discouraging anyone from exploring DCB or related aspects.

---

Even using the student-course enrollment example, assuming this setup matches "how business thinks" requires specific contextual alignment. This assumption often ignores scaling, physical constraints, or integration with existing systems. Acknowledging these assumptions is very important.

In an ideal scenario, it is all about Domain. In that case, we would have only one stream and be fine.

However, we are building software and can not ignore its limitations. Streams are there to help with spacetime.

Regarding the dual-write problem. Replacing these components often requires streams to function as a Process Manager (conceptually, not Commanded's implementation). New streams, such as "Enrollment," emerge, interacting with others (e.g., student, course) and coordinating processes previously handled by processors. This architecture can accommodate routing to multiple databases when necessary.

That is what those removed Processors could be doing there.

Adding coordination streams acknowledges the limitations of physical constraints, recognizing eventual consistency as acceptable until resolution is achieved. Those "duplicates" events are not duplicates, just different information (and that's where the focus should lie).

I emphasize that most Process Managers in Commanded could be better modeled as independent streams. These new streams would naturally emerge to manage the state without relying on "Commanded magic," so we would see such information clearer.

Research Greg Young’s opposition to multi-stream commits, as noted in this GitHub issue. Multi-stream transactions are often a shortcut for system design, relying heavily on the database. This approach locks systems into single-database architectures, complicating any transition from this dependency—similar to Figma’s experience.

The same problem, different perspectives, and multi-stream commit could be the solution.

Addressing these issues is more straightforward than it seems, provided we shift our focus to tooling rather than database constraints. Improved tooling could resolve many challenges without requiring significant rewrites. I acknowledge that we do not have the proper tools and that some people prefer easy things (including myself); I focus on maintaining the simplicity of understanding physics.

---

I think it's worth clarifying that whether you use stream-based architecture or not, routing data to its destination remains a problem. The key is designing the system to work with physical constraints. By focusing on routing, you avoid the need for extensive redesigns and architectural overhauls when scaling.

While no solution is perfect, thoughtful design minimizes the impact of these inevitable trade-offs.

That is why I keep asking, what is the cost associated with such system design?

We understand the limitations, trade-offs, and costs associated with single-database-centric architectures and the risks of relying on such components. For example, How Figma's Database Team Lived to Tell the Scale demonstrates the challenges of scaling with such assumptions.