CQRS and Domain Events for integration

Giorgio Sironi

Giorgio Sironi (@giorgiosironi)

I'm a developer (writes code, design stuff)
Interested in
- Automated testing and TDD
- Object-oriented programming
- Distributed systems

If you are looking at these slides on your pc, press S to see the notes

Context

Implications

many different logical and physical databases
continuous need for integration e.g. for reporting needs
no single services fits the bill

I'd not got as far to say these are microservices, but they are definitely small services running on separate logical and physical databases (and so you can't run JOINs between them). There are all sort of advantages to this approach, like allowing different teams and services to move forward uncoupled to each other. However, when you need to do something that requires data from many different services, for example make a large report over payment failures and why they failed, no single service can satisfy your query as the data is separated into the abstraction-layer-for-mobile-carriers service (which contains the detailed errors) and the Core Domain that registers all transactions. Consider for example a Subscription: it's an Aggregate with different meanings in two of these service, one contains all the state transition (paying, insolvent, suspended during vacation, canceled) and one instead all payments that have been made, the same Payment that is used in other business models like a single, non-recurring purchase. When you want to know how far in the past was the last successful payment of canceled subscriptions...

Bloody solutions (1/2)

Mega SQL query

on which database, since they are all small?
how to avoid thrashing the database machines?

When the ETL job to pull payments into the data warehouse starts, we go down for 10 minutes

Bloody solutions (2/2)

Joining CSV files

they become very large with historical data
inefficient even on a single day as data could be filtered
process of joining and filtering them becomes fragile as a consequence

Every morning at 7am I run a few MongoDB `touch` commands to warm up the database in RAM before the circus starts

In Command-Query Responsibility Segregation, all the actions over a system are divided into Command and Queries. There are peculiarities to each of them but think mostly about writes and reads. Commands are received by a single Aggregate, which is a machine deciding what to do. As a consequence its state is modified and Domain Events recording what has happened are produced. Queries have to be satisfied whatever their target is (possibly millions of Aggregates). They are mostly precomputed into a separate Read Model through a mechanism called Projection, which is eventually consistent with the Command-side's Domain Model. Looking at the result of Queries a *human* decides what to do, probably issuing new Commands. The services we have been talking about are the "code" in this picture. They have their own Domain Models, but they are not Read Models: they are optimized to deal with Commands like a user wanting to complete a payment, not for large cross-Aggregate calculations to fill reports. We had a well-designed system for the Command side, with isolated Aggregates even residing in different services. But we were missing the Queries side.

The source-of-truth service: Demetra

It works! Why?

there is all the data there, forever
decoupling over delivery and calculation of Domain Events
consolidation of data processing infrastructure in a single codebase

How: schema


{
    "id": "20ec0521-d127-4038-92ed-a4196c33705b",
    "meta": {
        "service": "core",
        "type": "payment-completed",
        "created_at": "2016-01-01T12:00:00.000Z",
        "received_at": "2016-01-01T12:00:00.200Z",
        "aggregate_id": "payment/23"
        "correlation_id": "container/subscription/42",
    },
    "payload": {
        "amount": "3.0000/EUR",
        ...
    }
}

The only collection/table backing the Source Of Truth contains the Domain Events, and here's how we stored them. Each event has an id which identifies it and is generated at creation. It can be used for idempotent delivery. Each event also has a created_at time of creation, generated in the original Service; but also a received_at which is the time in which it was registered in the Source Of Truth. When you query from the human point of view you care about created_at, but Projections keep track of the last received_at so that at any crash or restart they resume from the checkpoint they saved onwards. With aggregate_id I mean the DDD Aggregate that generated the event, while with correlation_id a longer-lived "container" that is useful for debugging the history of a user. In this case the Payment, generated and registered in seconds, is the Aggregate; the Subscription, which may last years before it's cancelled, is the source of correlation between disparate events. The payload contains the actual data. You may have additional ids here for correlation, like the phone number of the user; but I only suggest to include what's already there because in this model everything additional can be joined in by projections later rather than be retrieved with difficulty by the original Service.

How: technologies

We used MongoDB for storing the events, in a single collection of its own logical and physical database. Events are mostly unstructured in their payload, although the `meta` section is always consistent. We used it also for the projections, although given their tabular nature they got ported to PostgreSQL later. In retrospect we could have used PostgreSQL also for the events, using the JSON type for the payload. We choose MongoDB because it was already everywhere at the time, and that consolidated a lot of the headaches of learning the bits and pieces of a new database. There's value in consistent tooling across a platform with many services. The general advice is to pull in whatever you need for projections; be aware that maintenance costs will go up as, for example, you have more EC2 instances to pay and more driver-facing code to maintain. Still better than hammering everything in a single database not fit for your workload and see queries time out again.

How: projections


{
    "meta.type": { "$in" : interesting_types },
    "received_at": { "$gt": last_checkpoint, "$lt": stable_timestamp }
}

How: processes


      * * * * * www-data /var/www/demetra/bin/demetra projection:my-report
      * * * * * www-data sleep 10 && /var/www/demetra/bin/demetra projection:my-report
      * * * * * www-data sleep 20 && /var/www/demetra/bin/demetra projection:my-report
      ...

This was a very ugly thing to do, but it worked for an year: projections as processes that startup, apply events newer than their checkpoint, save the last event as new checkpoint and exit. Due to clock jitter (events are registered by multiple web servers receiving them through HTTP with clocks that may not be perfectly in sync) we only access events that are at least 10 seconds hold. Loss of latency but much lower probability of losing an event, and we are eventually consistent anyway. These crons run the projection every 10 seconds, due to their sleep statements. After a while we introduced also long running processes, that are kept up by a deamon restarting them upon the possible crash. Still all the projections are polling the database, as we cannot easily get pushed new events. There are some architecture that could do that, like having some RabbitMQ queue be populated upon event receival, but we did not go down the route for simplicity over performance.

Issues: delivery


      Request::to('https://source-of-truth.com/events', 'POST')
          ->withTheUsualAuthentication(...)
          ->asOptimisticJob()
          ->ensureIs(200, 409)
          ->retryPolicyIs(ExponentialBackoff::upTo('P1Y'))
          ->send()

We deliver a Domain Event in-process from the other service that produced it. It's an optimistic delivery, which means that in the happy path nothing gets stored in queues or databases in that service. Any result that is not 200 or 409 (including timeout) leads to the request being repeated by a persistent queue out of the main process. The retry policy has an exponential backoff: retry first after 1 second, if still negative after 2s, 4s, 8s... for a very long time as it's not acceptable to los Domain Events. Notice this doesn't protect from crash of the process generating the event: the only way to get to that level of reliability is by transactionally persisting in the local database the events along with the change in state of the related Aggregate. However, that did not fit at all with the existing architecture so we stuck to this delivery pattern.

Issues: out-of-order

purchase-landed, created 14:00:00, received 14:02:08 after N retries
purchase-completed, created 14:01:50, received 14:01:51

Issues: events are forever

Issues: importing

Excavate dirt (data) and put it into a standard form

Issues: archival

Conclusions

Domain Events are non-intrusive
A Source Of Truth very often comes up
Be prepared

Thanks!

@giorgiosironi @eLife

g.sironi@elifesciences.org

elifesciences.org (of course, we're hiring)