Skip to content
OccurrentOccurrent

Sagas

Layer 1: Domain events

A “saga” represents and coordinates a long-lived business process (where “long-lived” is kind of arbitrary). This is an advanced subject, so try other means first. A policy is often enough, and DCB covers the case where two rules must hold in one append. When you do need a saga, use Occurrent’s Saga DSL below. It lets you write a process manager in plain code, with timers and correlation, and it has Spring Boot support.

If the DSL doesn’t fit your needs, Occurrent is a library, so you can also hook in an existing solution, for example:

To integrate one of these with Occurrent, create a subscription that forwards the events written to the event store to it.

Saga DSL (Added in 0.31.0)

Section titled “Saga DSL”

A saga (more precisely a process manager) reacts to events, and to their absence over time, by issuing commands. Use one for a process that spans more than one stream and unfolds over real time, such as “cancel the order if payment is not reserved within 30 minutes”. A Saga<E, S, C> is the mirror image of a decider: a decider turns commands into events, a saga turns events (and its own timeouts) into commands. Like a decider it is only data and functions, with no I/O, so a test asserts equality on the effects it returns and needs no infrastructure at all.

Here is the order-fulfillment process. OrderPlaced reserves payment and arms a 30-minute timer, PaymentReserved ships the order, PaymentFailed cancels it, and the timer firing (nobody reserved or failed the payment in time) also cancels it:

Saga<OrderEvent, FlowState<OrderEvent>, OrderCommand> orderFulfillment =
        FlowSaga.<OrderEvent, OrderCommand>builder()
                .correlateAll(OrderEvent::orderId)
                .startsOn(OrderPlaced.class,
                        order -> List.of(new ReservePayment(order.orderId(), order.amount())))
                .step("awaiting-payment", step -> step
                        .on(PaymentReserved.class, Continuation.end(),
                                payment -> List.of(new ShipOrder(payment.orderId())))
                        .on(PaymentFailed.class, Continuation.end(),
                                failure -> List.of(new CancelOrder(failure.orderId(), failure.reason())))
                        .timeout(Duration.ofMinutes(30), Continuation.end(),
                                received -> List.of(new CancelOrder(received.initiating(OrderPlaced.class).orderId(), "payment timeout"))))
                .build();

A saga is not a substitute for a Dynamic Consistency Boundary. When two rules must hold atomically in one append, express both as a single DcbCriteria and let one decider decide against them together, which is cheaper and stronger. Use a saga only when the process is genuinely cross-boundary and time-involving. The “cancel if payment is not reserved within 30 minutes” rule has no single append at which both facts are known, because the payment may simply never arrive and the deciding write has to be triggered by the passage of time rather than by an event. That is the gap a saga fills, and the only one.

There are two ways to write a saga, and both produce the same Saga<E, S, C>, so the runner only ever runs one kind of thing. The flow DSL above is the shorthand for the common case, a linear process moving through a few named steps. Underneath it is the core DSL, an explicit fold and reaction per event type. Use the flow DSL when your process is a small sequence of steps, and drop to the core DSL when it is not.

The Core DSL

Section titled “The Core DSL”

The core DSL is Saga.builder(...) in Java and saga(...) { } in Kotlin. Both take an initial state, or none when the state should start out as null. Without one, the state’s type is nullable, @Nullable S in Java and S? in Kotlin. You register, per event type, an evolve that applies the event to state and a react that decides what to do now that the event has been applied. Timers get their own evolveOnTimeout and reactOnTimeout, keyed by name. evolve and react are kept separate on purpose. Rehydrating an instance from history calls only evolve, so replay can never re-issue a command.

Here is the same order-fulfillment process as the flow example above, written against an explicit OrderSagaState:

Saga<OrderEvent, @Nullable OrderSagaState, OrderCommand> orderFulfillment =
        Saga.<OrderEvent, OrderSagaState, OrderCommand>builder()
                .correlateAll(OrderEvent::orderId)
                .startsOn(OrderPlaced.class)
                .evolve(OrderPlaced.class, (state, e) -> new AwaitingPayment(e.orderId()))
                .react(OrderPlaced.class, (state, e) -> List.of(
                        SagaEffect.issue(new ReservePayment(e.orderId(), e.amount())),
                        SagaEffect.startTimeout("payment", Duration.ofMinutes(30))))
                .evolve(PaymentReserved.class, (state, e) -> new Completed(e.orderId()))
                .react(PaymentReserved.class, (state, e) -> List.of(
                        SagaEffect.issue(new ShipOrder(e.orderId())),
                        SagaEffect.cancelTimeout("payment")))
                .evolve(PaymentFailed.class, (state, e) -> new Cancelled(e.orderId(), e.reason()))
                .react(PaymentFailed.class, (state, e) -> List.of(
                        SagaEffect.issue(new CancelOrder(e.orderId(), e.reason())),
                        SagaEffect.cancelTimeout("payment")))
                .evolveOnTimeout("payment", (state, t) -> new Cancelled(t.sagaId(), "payment timeout"))
                .reactOnTimeout("payment", (state, t) -> List.of(SagaEffect.issue(new CancelOrder(t.sagaId(), "payment timeout"))))
                .isTerminal(state -> state instanceof Completed || state instanceof Cancelled)
                .build();

startsOn names the event types that create a new instance, and correlateAll (or a per-type correlate) says which instance every other event belongs to, described under correlation below. isTerminal marks the states that end the process. A terminal instance ignores further input, and the runner cancels its outstanding timers.

The flow DSL cannot express everything, on purpose. A step can wait on a fixed combination of alternatives and counts, built once when the saga is defined, but a count decided at runtime rather than fixed at build time is still out of reach, and so are accumulators across steps and “this event is valid in every step” matching. A process that needs any of those drops to the core DSL, where evolve and react can express them directly.

The Flow DSL

Section titled “The Flow DSL”

The flow DSL describes a process as a linear sequence of named steps. Here is one branch from a step, reacting to PaymentReserved:

on<PaymentReserved>(then = end) {
    if (it.partial) issue(ReserveRemainder(it.orderId))
    issue(ShipOrder(it.orderId))
}

A step holds a list of branches like the one above, tried in the order you declared them, and the first one satisfied wins. Each branch waits for either a single event type or a condition over the events received since the step was entered. A step can also have a timeout(...). Each branch and timeout names where the saga goes next through a Continuation. end completes the saga, next advances to the following step, and transitionTo("step") jumps to a step you name, including one the flow has already been through, which is how you write a retry loop. The whole step graph is validated at build() time, so a transitionTo to a step that does not exist is a build error, not a run-time surprise.

The order-fulfillment example above is the shape to copy for a branch-and-timeout step. For a timeout on its own, here is the “close the game if no player joins within 10 minutes” case:

Saga<GameEvent, FlowState<GameEvent>, CloseGame> lobby =
        FlowSaga.<GameEvent, CloseGame>builder()
                .startsOn(GameCreated.class)
                .correlateAll(GameEvent::gameId)
                .step("awaiting-players", step -> step
                        .on(PlayerJoined.class, Continuation.end())
                        .timeout(Duration.ofMinutes(10), Continuation.end(),
                                received -> List.of(new CloseGame(received.initiating(GameCreated.class).gameId()))))
                .build();

on<PlayerJoined>(then = end) above has no reaction at all. A branch, timeout or start that issues nothing simply omits it, in Java through a StepBuilder overload that takes no reaction.

When there is a reaction, it returns what issue gives back rather than nothing. That is what makes the mistake below a compile error instead of a saga that silently does nothing at runtime, since a Kotlin lambda expecting Unit would have accepted the command and discarded it:

// Does not compile, the command is produced and never issued
on<PaymentReserved>(then = end) { ShipOrder(it.orderId) }

A reaction can still run several statements and branch on the event’s data, as long as its last expression is an issue.

In the Kotlin DSL, when a reaction issues a command only in some cases and none in the others, say so with nothing. In Java a reaction returns the list of commands to issue, so returning an empty list issues nothing. It means there is nothing to issue, not that nothing happens. The branch still fires and still follows its then, so the flow advances either way. An if without an else has type Unit and cannot close the lambda, so give it an else branch or end on nothing:

// Issue only when the payment was partial, otherwise nothing
on<PaymentReserved>(then = end) {
    if (it.partial) issue(ReserveRemainder(it.orderId)) else nothing
}


// The same thing when the conditional is the last of several statements
on<PaymentReserved>(then = end) {
    if (it.partial) issue(ReserveRemainder(it.orderId))
    nothing
}

Kotlin’s when expression works the same way as the if above. Every branch must end on a command or on nothing. Issuing no command is an outcome in its own right rather than an omission, which is what the word is for.

on<PaymentReserved>(then = end) {
    when {
        it.partial -> issue(ReserveRemainder(it.orderId))
        else -> nothing
    }
}

What a Reaction Reads

Section titled “What a Reaction Reads”

An on<T> branch’s reaction is handed the event that fired it, the way on<PaymentReserved>(then = end) { payment -> ... } is handed the PaymentReserved. A timeout has no such event, so its reaction takes a received parameter instead, the way the game lobby’s does:

timeout(after = Duration.ofMinutes(10), then = end) { received ->
    issue(CloseGame(received.initiating<GameCreated>().gameId))
}

received is a ReceivedEvents, the events the instance has kept with the initiating event first. received.initiating<GameCreated>() returns the start event, which is how the reaction above reaches a gameId that only GameCreated has. Java writes the same call as received.initiating(GameCreated.class).

The other accessors ask about an event type rather than about the start event. first<T>() returns the first event of that type, null in Kotlin and an empty Optional in Java when none has arrived. all<T>() returns every event of that type in arrival order, count<T>() returns how many of them have arrived, any<T>() and none<T>() answer whether any has, and asList() returns the kept events in arrival order. Java passes a Class to each of the type-based ones, received.count(PaymentFailed.class).

Not every callback reads the same amount of an instance’s history. What Each Callback Reads sets that out, once conditions and guards are in view.

Those reified accessors are Kotlin extensions, so a file outside the org.occurrent.dsl.saga.flow package imports each one it uses, import org.occurrent.dsl.saga.flow.initiating for the example above. Leaving that import out for initiating specifically does not report a missing extension. ReceivedEvents also has a no-arg initiating() member, and a member wins over an extension, so the compiler points at that member with “No type arguments expected” and sends you looking in the wrong place.

Loops and Absolute Timeouts

Section titled “Loops and Absolute Timeouts”

The timeouts so far have all been timeout(after = ...), which fires once a relative duration has elapsed. timeout(at = { received -> ... }) fires at an absolute Instant computed from the received events instead, an auction’s end time for example.

A transitionTo names any step, including the current one, which is how a flow expresses a loop. In the example below, an auction stays open as long as bids keep arriving. Each BidPlaced transitions the bidding step back to itself, and an absolute timeout closes it once its end time passes. Re-entering the step re-arms its timeout, but because the deadline comes from the initiating event, the timeout still fires at the auction’s original end time rather than sliding forward on every bid:

Saga<AuctionEvent, FlowState<AuctionEvent>, CloseAuction> auction =
        FlowSaga.<AuctionEvent, CloseAuction>builder()
                .startsOn(AuctionStarted.class)
                .correlate(AuctionStarted.class, AuctionStarted::auctionId)
                .correlate(BidPlaced.class, BidPlaced::auctionId)
                .step("bidding", step -> step
                        .on(BidPlaced.class, Continuation.transitionTo("bidding"))
                        .timeout(received -> received.initiating(AuctionStarted.class).endsAt(), Continuation.end(),
                                received -> List.of(new CloseAuction(received.initiating(AuctionStarted.class).auctionId()))))
                .build();