Stateful Saga or event-handling aggregate

[kobasad]

I'm trying to model a workflow where an event triggers a command with attributes depending on the current workflow state. Basically, the Workflow represents the Automation pattern in Event Modeling. I went through the existing examples, but still can't figure out how to do that, as I need to react on an event, but stateless Saga is not suitable for this use case. I feel an Aggregate is what I really need, but I'm not sure how to model it properly.

[idugalic]

Hi @kobasad,

FModel is not promoting ProcessManagers ("Stateful Saga"). We had it a long time ago in FModel, but we removed it:

1. All we do is: modeling workflows and processes (verbs, not nouns). Decider/Aggregate is the component that should be used for that. It will properly model all the events that are needed for this process to work, and that important state is now stored as a series of events (in case of event sourced system).
2. Your hunch is correct, Decider/Aggregate should model this flow, SagaManager could simply subscribe to these events and integrate/automate. There is no important state in Saga, all important stuff is in Decider. Logic does not leak to Saga. Saga is a dumb pipe...

If you can share more about your specific case, I will be glad to elaborate.

I am not sure how and if event modeling community is discussing implementation details of Automation pattern.
It would be helpful if you would explain your understanding of Automation pattern (as defined by Event Modeling).
I can imagine a regular View (some table in your DB) / ToDo list that you scan from time to time to send commands to another system. But this is not supported by FModel, out of the box. Or, a process manager (linked above) that will support this type of Automation more robustly.

[kobasad]

Hi @idugalic,
I understand Automation like this: there's some state (view/read model) and associated business logic. Whenever there is a state change due to an event, a command is sent based on the current state.

Here's an example scenario:

Workflow manager requests one or more reports based on its progress. When a report is fetched, data points are extracted and the workflow checks whether all required data points have been collected. If not, it requests another report, and so on until all required data points are collected. So, each new command depends on the progress (what data points were already fetched).

So DataPointsAdded event may result in a FetchReport command, but the exact command depends on the state.

Event modeling does not prescribe how implementation should be done, as far as I know.

[idugalic]

OK.
This workflow progress is a View. So, in Fmodel you could create a Materialized View for it.
The gear at the top is your custom implementation that can pull this View/Database with some frequency and go through the ToDo List check what is currently pending and publish commands based on the state. Potentially you could make this reactive and eliminate the pooling. However, this gear is not encoded as a formal component in FModel.

How could we encode this in FModel:

/**
 * A general model of the Process/StatefulSaga that emits its own Events (Ep).
 * (S, Ep) -> S function/relation allows this Process to either store the State/S or list of Events/Ep
 * 
 * @param S State of the process
 * @param E Domain Event that indirectly evolves internal state (S) of this process
 * @param Ep Internal Events of the process that directly evolves internal state (S) of this process
 * @param C Commands that will be published
 */
data class Process<E, S, Ep, C>(
    val ingest: (E, S) -> Iterable<Ep>,
    val evolve: (S, Ep) -> S,
    val react: (S, Ep) -> Iterable<C>,
    val pending: (S) -> Iterable<C>,
    val initialState: S,
)

// Specialization of the Process that stores S/State durably.
// This model fits the definition of `Automation` by Event Modeling (E -> S -> C). This S/State is called `ToDo` List in Event Modeling.
data class StateStoredProcess<E, S, C>(
    val evolve: (S, E) -> S,
    val react: (S, E) -> Iterable<C>,
    val pending: (S) -> Iterable<C>,
    val initialState: S,
)

Should we add this along Decider, Saga and View?

[kobasad]

Nice! Looks like exactly what I was looking for. One question: pending is a function for polling if react is not used for, well, reactive processing, right?
I don't think I'm in a position to suggest whether or not this should be added to Fmodel, as I'm still learning event modeling and Fmodel. But it seems it's not the first question regarding a stateful saga or process manager, so I think there's some demand in this type.

[idugalic]

To better relate with Event Modeling definition of integration patterns:

• Translation
• Automation

// Previously called Saga
data class Translation<E, C> (
    val react: (E) -> Iterable<C>,
)

data class Automation<E, S, C>(
    val evolve: (S, E) -> S,
    val react: (S, E) -> Iterable<C>,
    val pending: (S) -> Iterable<C>,
    val initialState: S,
)

[idugalic]

Nice! Looks like exactly what I was looking for. One question: pending is a function for polling if react is not used for, well, reactive processing, right? YES.

We can replace react and pending with only one function: val react: (S, E?) -> Iterable<C>, in where E is optional.

[kobasad]

@idugalic What do you think about a Reactor with an optional state? Stateless Reactor is a Saga, stateful one - a Process. Then we'd have a nice symmetric pair of types: Decider and Reactor, one reacting to commands and emitting events, and another one reacting to events and emitting commands.

[idugalic]

/**
 * Event Modeling - Translation pattern
 *
 * @param C Command of my system
 * @param AR Action Result / Event of other system
 */
data class Translation<AR, C> (
    val react: (AR) -> Sequence<C>,
)

/**
 * Event Modeling - Automation pattern
 *
 * @param E Event of my system
 * @param S State of my system
 * @param A Action / Command of other system
 * @param AR Action Result / Event of other system
 */
data class Automation<E, S, A, AR>(
    val evolve: (S, E) -> S,
    val ack: (S, AR) -> S,
    val pending: (S) -> Sequence<A>,
    val initialState: S,
)

• The S/View is updated/accumulated by your Events: evolve: (S, E) -> S
• You can pool the S/View in some frequency to check if there are pending tasks: pending: (S) -> Sequence<A>
• Alternatively, you can replace pooling with a reactive approach by composing evolve and pending: pending: (evolve (S, E)) -> Sequence
• Update the S/View with the result of the Action/Command: ack: (S, AR) -> S

[kobasad]

Found this blog post which contains a really nice model. Essentially: (S, E/AR) -> C/A.

[idugalic]

OK.

This is E -> S -> C.
Effectively, the author introduced S/State as a param to react method/function.
Once we introduce that S/State formally in our domain model, the next best step is to model managing that state as well. When will that S/State trigger the next A/Action/Command? Should we update that S/State based on the AR/Action result, and how?

So simple react(e)->c is not good enough and that is why we would like to discuss this new type (Automation):

/**
 * Event Modeling - Automation pattern
 *
 * @param E Event of my system
 * @param S State of my system
 * @param A Action / Command of other system
 * @param AR Action Result / Event of other system
 */
data class Automation<E, S, A, AR>(
    val evolve: (S, E) -> S,
    val ack: (S, AR) -> S,
    val pending: (S) -> Sequence<A>,
    val initialState: S,
)

• The S/View is updated/accumulated by your Events: evolve: (S, E) -> S
• You can pool the S/View in some frequency to check if there are pending tasks: pending: (S) -> Sequence<A>
• Alternatively, you can replace pooling with a reactive approach by composing evolve and pending: pending: (evolve (S, E)) -> Sequence
• Update the S/View with the result of the Action/Command: ack: (S, AR) -> S

How would you implement Event Streaming to evolve your Views is an orthogonal topic to this. This will open new topics: the ordering of events, and delivery guarantee (ExactlyOnce, AtLeastOnce, AtMostOnce). BTW, AtLeastOnce is a reality!
Most of the EventStores support robust streaming already. You can have concurrent consumers/views evolving the state robustly.
Streaming platforms (Kafka) or Queue brokers(RabbitMQ) could be used to deliver these events. We have an open-sourced fstore-sql Postgres powered Event Store, that can do robust event streaming as well (no Kafka ;) ) - https://github.com/fraktalio/fstore-sql

These are examples that are using fstore-sql :

• https://github.com/fraktalio/fmodel-ktor-demo/blob/main/src/main/kotlin/com/fraktalio/application/SagaManager.kt
• https://github.com/fraktalio/fmodel-spring-demo

[kobasad]

I'm not really 100% agree with that article: I think it uses evolve/decide/react in a wrong way, especially in the part where the same events are fed to both evolve and react. I think evolve should consume internal events (event sourcing), while react should consume external events (AR).
I think something is missing in your proposed Automation: where are the internal events (E) coming from? None of the functions produce them. It feels like the state of Automation should be driven by external events and its own decisions. Should it consume its own commands/actions and produce internal events? This gets too complicated.

[idugalic]

I'm not really 100% agree with that article: -> I do not fully agree as well.

I think something is missing in your proposed Automation: where are the internal events (E) coming from? -> Events are coming from your system (Deciders will produce and store them). These Events will trigger Automation (Gear at the top of EventModeling blueprint) to start the process:

• The S/View is updated/accumulated by your Events: evolve: (S, E) -> S
• You can pool the S/View in some frequency to check if there are pending tasks: pending: (S) -> Sequence
• Alternatively, you can replace pooling with a reactive approach by composing evolve and pending: pending: (evolve (S, E)) -> Sequence
• Update the S/View with the result of the Action/Command: ack: (S, AR) -> S

It is important to differentiate events/commands per system. Automation as described by Event Modeling is only reacting on its own events and publish commands to other systems, effectively integrating two systems over the command / command is sent over the wire.

So,

/**
 * Event Modeling - Automation pattern
 *
 * @param C Command of my system
 * @param E Event of my system
 * @param S State of my system / View
 * @param A Action / Command of other system
 * @param AR Action Result / Event of other system
 */
data class Automation<C, E, S, A, AR>(
    val evolve: (S, E) -> S,
    val ack: (S, AR) -> S,
    val pendingActions: (S) -> Sequence<A>,
    val pendingActionResults: (S) -> Sequence<C>,
    val initialState: S,
)

1. The S/View is updated/accumulated by your Events: evolve: (S, E) -> S
2. You can pool the S/View in some frequency to check if there are pending actions/tasks: pendingActions: (S) -> Sequence , and send these Actions/Commands to that system
3. Alternatively, you can replace pooling with a reactive approach by composing evolve and pendingActions: (evolve (S, E)) -> Sequence
4. Update the S/View with the Result of the Action/Command: ack: (S, AR) -> S
5. You can pool the S/View in some frequency to check if there are pending action results/tasks: pendingActionResults: (S) -> Sequence , and send these Commands to your system
6. Alternatively, you can replace pooling with a reactive approach by composing ack and pendingActionResults: (ack (S, AR)) -> Sequence