04 - Event Sourcing + CQRS Together: Projection, Snapshot and Consistency

Where event sourcing and CQRS meet: in the projection

You have a system that must be able to prove what happened to every order, and that must answer lists and searches much more often than it writes. The first requirement suggests saving the events; the second suggests never reading from the events. The solution is to use them together: the write side saves events (event sourcing), the read side reads from tables built from those events (CQRS). The meeting point is called a projection, and it is where the real problems arise: the projection is always a little behind, it may receive the same event twice and it must be possible to throw it away and rebuild it.

You need the two preceding articles: Event Sourcing for the event store and optimistic concurrency, CQRS for the two sides. Here I avoid repeating snapshots (they are in article 02, with the count of applied events) and I concentrate on what belongs only to the combination, using a single in-memory Java 21 program: an order, its store, two projections and the three difficult situations.

The path of a command, step by step

1. The client sends a command, for example ConfirmPayment.
2. The handler loads the aggregate by rereading its events from the event store.
3. The aggregate validates the command and produces one or more events, or refuses it.
4. The events are appended to the store, with the expected version for optimistic concurrency.
5. The events reach the projections, directly or through a broker.
6. The projections update the read models.
7. The client's queries read from the read model.

Here the order is everything: if you swap the steps the sequence does not work. But the point to keep in mind is where the chain breaks. Steps 1 to 4 are a single synchronous operation that ends with the reply to the client; from step 5 on the work is asynchronous, and the client has already received its reply. All the “eventual consistency” of this pattern lives in that break, which is why the command must return something that lets the client know when the read side has caught up, as we will see.

Write side: events, store and aggregate in Java 21

The events are records that implement a sealed interface: the compiler knows the closed list of types and, when you write a switch on Event, it forces you to handle all of them, without default. This is the pattern matching for switch feature finalized in Java 21 (JEP 441), and it is useful here because adding an event breaks compilation at every point that does not handle it.

The amounts are in cents (long), not in BigDecimal, to keep the listing short and not to confuse the reader with rounding questions. The store keeps two numbers per event: the position, global, which gives a single order to all the events, and the version, per aggregate, used for concurrency.

    sealed interface Event permits OrderPlaced, PaymentConfirmed, OrderCancelled {}
    record OrderPlaced(String orderId, String customerId, long totalCents) implements Event {}
    record PaymentConfirmed(String orderId, String transactionId) implements Event {}
    record OrderCancelled(String orderId, String reason) implements Event {}

    // A stored event: position is global (write order), version is per aggregate.
    record Stored(long position, String orderId, long version, Event event) {}

    static class EventStore {
        private final List<Stored> log = new ArrayList<>();
        long head() { return log.size(); }

        long versionOf(String id) { return log.stream().filter(s -> s.orderId().equals(id)).count(); }

        void append(String id, long expectedVersion, List<Event> events) {
            long current = versionOf(id);
            if (current != expectedVersion)
                throw new IllegalStateException("Conflict on " + id + ": expected v" + expectedVersion + ", found v" + current);
            long v = expectedVersion;
            for (Event e : events) log.add(new Stored(log.size() + 1, id, ++v, e));
        }

        List<Event> load(String id) { return log.stream().filter(s -> s.orderId().equals(id)).map(Stored::event).toList(); }

        List<Stored> readAfter(long position, int max) {
            return log.stream().filter(s -> s.position() > position).limit(max).toList();
        }
    }

append compares the expected version with the current one and refuses if they differ: it is the optimistic concurrency of the previous article, stripped to the bone. In a real database the check and the writes must be in the same transaction, otherwise the check can be overtaken by a concurrent write. readAfter returns the events after a position: it is the only operation a projection needs, and we will see why.

    enum Status { AWAITING_PAYMENT, PAID, CANCELLED }
    record Order(Status status) {
        static Order evolve(Order o, Event e) {
            return switch (e) {
                case OrderPlaced x -> new Order(Status.AWAITING_PAYMENT);
                case PaymentConfirmed x -> new Order(Status.PAID);
                case OrderCancelled x -> new Order(Status.CANCELLED);
            };
        }
        static Order rehydrate(List<Event> events) {
            Order o = null;
            for (Event e : events) o = evolve(o, e);
            return o;
        }
        List<Event> confirmPayment(String orderId, String tx) {
            if (status != Status.AWAITING_PAYMENT) throw new IllegalStateException("Payment not accepted in status " + status);
            return List.of(new PaymentConfirmed(orderId, tx));
        }
    }

The aggregate is two functions, as in article 02. evolve applies an event and decides nothing; confirmPayment is the decision: it accepts only from AWAITING_PAYMENT and produces an event. There is no setStatus: the status changes only by producing an event that evolve knows how to read. The version of an aggregate is not a field that the aggregate increments: it is decided by the store, which is the only one that knows how many events really exist.

Read side: the projection with a checkpoint

A projection receives the events in order and translates them into rows. The two properties that make it safe are the checkpoint, that is, the last position processed, and the fact that it can restart from zero. The checkpoint makes it idempotent: if it receives an event with a position lower than or equal to it, it ignores it.

It is needed because almost all brokers guarantee at-least-once delivery, not exactly-once. The second model, Collected, did not exist when the orders were written; you build it by rereading the history, and you do not have to change anything on the write side. It is one of the concrete reasons why the cost of the model is accepted.

    interface Projection { void handle(Stored s); }

    // Read model 1: one row per order. The checkpoint makes the handler idempotent.
    record Row(String customerId, Status status, long totalCents, long version) {}
    static class OrderList implements Projection {
        final Map<String, Row> rows = new TreeMap<>();
        long checkpoint = 0;
        int applied = 0;
        public void handle(Stored s) {
            if (s.position() <= checkpoint) return;            // already seen: repeated delivery, ignored
            Row r = rows.get(s.orderId());
            switch (s.event()) {
                case OrderPlaced e -> rows.put(e.orderId(), new Row(e.customerId(), Status.AWAITING_PAYMENT, e.totalCents(), s.version()));
                case PaymentConfirmed e -> rows.put(e.orderId(), new Row(r.customerId(), Status.PAID, r.totalCents(), s.version()));
                case OrderCancelled e -> rows.put(e.orderId(), new Row(r.customerId(), Status.CANCELLED, r.totalCents(), s.version()));
            }
            checkpoint = s.position();
            applied++;
        }
    }

    // Read model 2, born later: amount collected per customer. No new events, just a replay.
    static class Collected implements Projection {
        final Map<String, Long> totalOf = new HashMap<>();
        final Map<String, String> customerOf = new HashMap<>();
        final Map<String, Long> perCustomer = new TreeMap<>();
        public void handle(Stored s) {
            switch (s.event()) {
                case OrderPlaced e -> { customerOf.put(e.orderId(), e.customerId()); totalOf.put(e.orderId(), e.totalCents()); }
                case PaymentConfirmed e -> perCustomer.merge(customerOf.get(e.orderId()), totalOf.get(e.orderId()), Long::sum);
                case OrderCancelled e -> {}
            }
        }
    }

    static int refresh(EventStore store, OrderList p, int maxEvents) {
        List<Stored> batch = store.readAfter(p.checkpoint, maxEvents);
        batch.forEach(p::handle);
        return batch.size();
    }

In OrderList.handle the first line is the idempotency check; the last updates the checkpoint after having applied the event. In a real database the two things go in the same transaction, otherwise a crash between the row update and the checkpoint update leaves you with an event applied twice or never. refresh reads at most maxEvents events after the checkpoint: in the program it serves to simulate a projection that is late.

What happens when you run it

The main part creates six events on three orders, then looks at the three difficult situations: the delay, the repeated delivery, the rebuild. Rereading events already projected and the rebuild are the tests that matter, because they are the ones a superficial test does not cover.

    public static void main(String[] args) {
        EventStore store = new EventStore();
        store.append("o-1", 0, List.of(new OrderPlaced("o-1", "cli-A", 4500)));
        store.append("o-2", 0, List.of(new OrderPlaced("o-2", "cli-A", 1200)));
        store.append("o-3", 0, List.of(new OrderPlaced("o-3", "cli-B", 9900)));
        store.append("o-1", 1, Order.rehydrate(store.load("o-1")).confirmPayment("o-1", "tx-1"));
        store.append("o-3", 1, List.of(new OrderCancelled("o-3", "customer")));
        store.append("o-2", 1, Order.rehydrate(store.load("o-2")).confirmPayment("o-2", "tx-2"));
        System.out.println("events in the store: " + store.head());

        OrderList list = new OrderList();
        System.out.println("list right after the commands: " + list.rows.size() + " rows, checkpoint " + list.checkpoint);

        // The command on o-1 produced version 2. The client waits for the list to reach it.
        int rounds = 0;
        while (list.rows.get("o-1") == null || list.rows.get("o-1").version() < 2) { refresh(store, list, 2); rounds++; }
        System.out.println("o-1 visible at version 2 after " + rounds + " refreshes, checkpoint " + list.checkpoint);

        // Double delivery: events already projected arrive a second time and change nothing.
        int before = list.applied;
        store.readAfter(0, (int) list.checkpoint).forEach(list::handle);
        System.out.println("rereading events already seen: applied " + before + " -> " + list.applied + " (checkpoint " + list.checkpoint + ")");
        refresh(store, list, 100);
        System.out.println("list aligned: " + list.rows);

        // A second read model built today on yesterday's history.
        Collected collected = new Collected();
        store.readAfter(0, 100).forEach(collected::handle);
        System.out.println("collected per customer: " + collected.perCustomer);

        // Rebuilding the list from scratch: same result as the one kept alive.
        OrderList rebuilt = new OrderList();
        while (refresh(store, rebuilt, 100) > 0) {}
        System.out.println("identical rebuild: " + rebuilt.rows.equals(list.rows));

        try { store.append("o-1", 1, List.of(new OrderCancelled("o-1", "double click"))); }
        catch (IllegalStateException e) { System.out.println("concurrency: " + e.getMessage()); }
        try { Order.rehydrate(store.load("o-3")).confirmPayment("o-3", "tx-3"); }
        catch (IllegalStateException e) { System.out.println("rule: " + e.getMessage()); }
    }
}

// Output (running the whole file):
// events in the store: 6
// list right after the commands: 0 rows, checkpoint 0
// o-1 visible at version 2 after 2 refreshes, checkpoint 4
// rereading events already seen: applied 4 -> 4 (checkpoint 4)
// list aligned: {o-1=Row[customerId=cli-A, status=PAID, totalCents=4500, version=2], o-2=Row[customerId=cli-A, status=PAID, totalCents=1200, version=2], o-3=Row[customerId=cli-B, status=CANCELLED, totalCents=9900, version=2]}
// collected per customer: {cli-A=5700}
// identical rebuild: true
// concurrency: Conflict on o-1: expected v1, found v2
// rule: Payment not accepted in status CANCELLED

The first output line shows the problem: six events in the store, zero rows in the list. The command on o-1 produced version 2 and the client waits for it: the while loop calls refresh twice, two events at a time, and the checkpoint reaches 4. Note that o-1 appears at version 2 when the checkpoint is 4, not 6: the list is not aligned with the whole store, only with what that client needs.

In the reread of events already seen, applied stays at 4, so the repeated delivery has no effect. The rebuild gives a list identical to the one kept up to date: if it were not so, the projection would have a hidden dependency on time or on arrival order, which is a bug. The last two lines are the refusals: concurrency (two commands on the same version, the second loses) and the rule (you do not pay a cancelled order). Neither wrote anything.

Three strategies for the inconsistency window

After a successful command the list is not up to date. The client has three roads, which can be combined:

• Return the version in the reply to the command, and make the read wait until that version, as in the loop above.
• Poll with retries the read side until it shows the change, with a limit on time and on attempts.
• Project synchronously the critical data, updating the read model in the same transaction as the writing of the events.

The first two are the same idea, and the first is preferable because it does not guess a time: it knows what it is waiting for. The third removes the window only for that model, but ties the read side and the write side to the same database and the same transaction, which cancels the advantage of having them separate. Use it for a single critical view (for example the receipt shown right after payment), not as a rule. An alternative that does not touch the backend is to show the expected result in the interface right away and reconcile it when the read side is ready, which is the topic of the article on eventual consistency strategies.

Mistakes that cost dearly

The first is modifying an event that has already been written: the event store is the source of truth and rewriting it breaks the replay of every projection and the audit. If the schema changes, write a new version of the event and adapt the old ones on read (upcasting); versioning is covered in the article on domain events. The second is making a command write several aggregates in the same transaction: the consistency boundary is the aggregate, and to coordinate more than one you need a saga.

The third is publishing the events to the broker after writing them without guarantees: if the process dies in between, the projections will never know about the event, and the remedy is the outbox. The fourth is the non-idempotent projection: it works in tests and produces duplicate rows the first time the broker redelivers.

What you can do now

You can explain where the combined pattern becomes asynchronous, you can write a projection with a checkpoint that withstands repeated deliveries and the rebuild, and you can choose how to handle the inconsistency window instead of ignoring it. The next step is coordination between services, with the saga.