Projections and Reactors

A projection and a reactor both follow an aggregate's committed events and return nothing. A projection writes a read model, such as a table a page reads. A reactor acts on the outside world: it sends a notification or calls another system. A processor that sends commands is a saga instead: Write a Saga. This page writes one of each for the eShop's order. Subscriptions, Failures and Replay explains how they are delivered.

Do you need a projection?

Every aggregate already keeps a snapshot of its state, and the snapshot is queryable. Write a projection when the read model needs another shape than the state (a row per buyer, a summary across aggregates), or another store (a search index, a table another system reads).

Write a projection

A projection is a class annotated @Projection. Each handler, annotated @OnEvent, takes a committed event; its dependencies are the constructor's parameters, bound from the application's beans:

@Blocking
@OnEvent
fun onEvent(event: StateEvent<OrderState>) {
    val state = event.state
    summaries.save(
        OrderSummary(
            tenantId = event.ref.tenantId,
            orderId = event.ref.id,
            buyerId = event.scope.ownerId,
            status = state.status,
            totalAmount = state.totalAmount,
            units = state.quantities().values.sum(),
            updatedAt = event.commitTime,
        ),
        version = event.version,
    )
}

• StateEvent<OrderState> calls the handler once for each commit of every order, with state the order's state after the commit. Take StateTypedEvent<S, E> instead to be called only for the events of type E, once per such event.
• scope.ownerId is the aggregate's owner after the commit, here the buyer; commitTime is when it committed.
• @Blocking runs the handler on a virtual thread, since it writes to a store outside Wow. A handler may also be suspend, or return a Reactor Mono.
• The handler is a subscription of its own, OrderSummaryProjection.onEvent.

Keep it idempotent

A commit may be handled more than once: after a failed attempt, a replay, or a lease handover between nodes. A projection's write is therefore guarded by the version it writes, so handling a commit again, or an older commit late, never replaces a newer row. The order summaries keep the newest version of each order:

override fun save(summary: OrderSummary, version: Int) {
    summaries.merge(summary.tenantId to summary.orderId, Versioned(version, summary)) { saved, newer ->
        if (newer.version > saved.version) newer else saved
    }
}

In SQL the same guard is an upsert that only updates a row whose version is lower than the commit's.

Write a reactor

A reactor is a class annotated @Reactor. Its handlers take the same events, and the context gives each call an idempotency key for the outside call:

/**
 * Tells the buyer of a shipped order that it is on its way. The notification service is a remote call that may be
 * slow or briefly down, so the reactor gives it its own budget: five attempts, from one second apart up to thirty, and
 * ten seconds for each, before the order is parked for the handler in the failure ledger. A buyer the service does not
 * know is a failure no retry can fix, so it parks the order at once.
 *
 * A reactor starts at the end of the log: deploying it does not notify the buyers of the orders shipped before.
 */
@Reactor
@Retry(
    maxAttempts = 5,
    initialBackoff = "1s",
    maxBackoff = "30s",
    timeout = "10s",
    unrecoverable = [UnknownBuyerException::class],
)
class ShipmentNotifier(private val notifications: BuyerNotifications) {
    @OnEvent
    suspend fun onOrderShipped(event: StateTypedEvent<OrderState, OrderShipped>, context: SubscriptionContext) {
        val buyer = event.scope.ownerId ?: return
        notifications.orderShipped(buyer, event.ref.id, key = context.invocationKey)
    }
}

context.invocationKey is the same every time the same commit is handled by the same handler: executionId (subscription:ref@version, the ref as context.aggregate/tenant/id) for a handler of the whole stream, executionId:eventIndex for a handler of one event type. A % or : in the subscription's name, and a % or / in the tenant or the id, is percent-escaped (%25, %3A, %2F), so no two executions share a key; a name without them reads as it is. Pass it to the outside system as its idempotency key, so a notification asked for twice is sent once. @Retry gives the reactor its own retry budget: Failures and Retries.

Where they start

A subscription that has no checkpoint yet in a partition starts by its kind:

Kind	Starts	So that
Projection	At the beginning of the log	Deploying it fills its read model with the history
Projection, on another context's stream	At the beginning of the topic	It fills its read model with what was published before
Reactor	At the end of the log	Deploying it does not notify anyone about the past
Reactor, on another context's stream	Where its service joins the stream: at the beginning when no other subscription of the service has read it yet, else at the end (JOIN in the Manifest)	A service that joins later still gets what was published before, and a reactor added later does not act on the history again

@OnEvent(start = ...) overrides it, and so does the host's wow.subscriptions.[<context>.<processor.function>].start, beginning, end or an ISO-8601 instant. An instant starts at the first entry of each partition, in log order, committed at or after it, and every entry after that one is handled, whatever its commit time: commit times are the decisions' times, not ordered along the log. When none is committed at or after it yet, the subscription starts at the end. To find that entry, its first start in a partition reads the partition from the beginning (it handles none of the entries before it), so on a long log that first start takes a while; the node logs a line when the read begins and one when it ends. Either way the start applies only until the subscription has a checkpoint; after that it goes on from there.

Rebuild a read model

Because a projection's position is its checkpoint, its read model can always be rebuilt from the log. To change a read model's shape, deploy the new projection under a new name: it starts at the beginning, catches up, and the reads switch to its store once it has. Then remove the old one. Subscriptions, Failures and Replay describes the options.

Test them

projectionSpec and reactorSpec hand the generated handlers committed events, and the test checks what the fake port then holds. Handing the same commit twice is how a test shows that a projection is idempotent:

@Test
fun `the same commit handled again should save the same summary`() = projectionSpec(handlers) {
    submitted()
    whenEvent(OrderAwaitingValidation, by = AwaitStockValidation::class)
    // A subscription hands a commit at least once: the same commit again saves the same summary, at its version.
    whenRedelivered()
    saved.distinct().size.assert().isEqualTo(1)
}

Test Domain Code describes the specs.