State transfer and async processing

[gbragi]

This is really interesting work you have here! I have a couple of questions

1. Regarding asynchronous processing of large data. Let's say that you had some type of data that requires some processing, but it wouldn't be feasible to load all that data into memory and pass it on to the pure decide function or store it as the state for the decider. Where would such processing fit into this model?
2. Regarding event driven state transfer. Lets say one had multiple (distributed) deciders and that the decision of one decider depended on state that is derived from the events from another decider. How would one go about shipping the events between the two deciders. Could the events from decider B be passed directly to the evolve function of decider A? Or would there have to be some read model and commands put in between?

Thanks for what you have done here!

[idugalic]

Hi @gbragi

1. The decider is a pure logic/computation, it is not aware of any side-effects like storing and fetching the data. How do you RUN your deciders is actually the concern of the application layer. At the moment we have two flavors of how the data is saved (not in memory, but in some durable storage like SQL):
   • (event sourcing) https://github.com/fraktalio/fmodel/blob/main/application-vanilla/src/commonMain/kotlin/com/fraktalio/fmodel/application/EventSourcingAggregateExtension.kt
     • one of the options there is async/reactive Flow operator: fun <C, E> Flow<C>.publishTo(aggregate: EventSourcingAggregate<C, *, E>): Flow<E> = aggregate.handle(this)
   • (state stored) https://github.com/fraktalio/fmodel/blob/main/application-vanilla/src/commonMain/kotlin/com/fraktalio/fmodel/application/StateStoredAggregateExtension.kt
     • one of the options there is async/reactive Flow operator: fun <C, S> Flow<C>.publishTo(aggregate: StateStoredAggregate<C, S, *>): Flow<S> = aggregate.handle(this)

So, you are not keeping this Flow of commands in memory. It will sequentally handle commands one by one. The Aggregate(Decider+EventRepository) will take that command message, fetch the current state (or events) and store the new state (or events) for every command. You can further optimize this by handling commands concurrently (not sequentally): https://github.com/fraktalio/fmodel/blob/main/application-vanilla/src/jvmMain/kotlin/com/fraktalio/fmodel/application/EventSourcingAggregateActorExtension.kt
I wrote a blog series about this: https://fraktalio.com/blog/side-effects-storing-and-fetching-the-data.html

2. Distributing your components will bring additional complexity. You now need to take care of tools/brokers to distribute events (or commands) across your deciders. I wrote a post that relates to this subject: https://fraktalio.com/blog/able-to-be-aggregated.html
   If two Deciders need to communicate with each other it is always via the Saga component, so the Saga will react on the event from Decider1 and send Command to Decider2.

• If you are in the monolith situation you could combine these two deciders and the saga into one aggregate. Check the EventSourcingOrchestratingAggregate.
• If you are in the microservices (distributed) scenario and you have your deciders as independent apps you could create individual aggregates, per Decider and let your broker do the job.

BTW, It should be very difficult to design your Decider components to misuse other Decider events. We can use the type system of Kotlin to take care of that: https://fraktalio.com/blog/types-and-functions.html

I am also sharing a Demo project that demonstrates different strategies (distributed/monolith):https://github.com/fraktalio/fmodel-demos

[idugalic]

Kotlin Flow and coroutines are important part of this lib. It depends on it to support async and non-blocking streaming with backpressure.

[gbragi]

Hey @idugalic
thanks for the reply. It clarifies a lot. I'm still not sure if I understand bullet 1 completely or maybe my first question above wasn't clear. Let me show an example of what I was thinking about. This is the PodcastTranscriptionDecider (guess this could apply to a View as well).

1. Command: HighlightPodcastRange { startTime: 10s, endTime: 300s }
2. Decider/Application layer needs to stream the selected podcast episode range and transcribe the words from that range (it would be unfeasible to load and pass the entire episode to the decider).
3. Event: PodcastRangeHighlighted { startTime: 10s, endTime: 300s, data: "Something being talked about on a podcast" }

Is there a place for such a procedure in fmodel? Would the only way be to make a custom StateRepository that loads the range into memory and sends it to the decider?

[idugalic]

Hi,
There is a place :) There are couple of options that I can think of, lets collaborate on the first :

1. Command: HighlightPodcastRange { startTime: 10s, endTime: 300s, audioChank: partOfAStream } Please notice, command has a part of the audio stream as the part of the payload.
2. Decider<HighlightPodcastRange, PodcastHighlighted, PodcastRangeHighlighted> Please, notice that PodcastHighlighted represents the state (S), it is a whole Podcast text/transcript.
3. PodcastRangeHighlighted is an Event (E) and it represents just a part/range of the Podcast text/transcript.

You can use this Decider and create a StateStoredAggregate<HighlightPodcastRange, PodcastHighlighted, PodcastRangeHighlighted>:

val myStateStoredAggregate = stateStoredAggregate(
        decider = myDeciderInstance,
        stateRepository = myRepositoryInstance
    )

and use that Aggregate.

myStateStoredAggregate.handle(flowOf(command1, command2....))

or more conveniently:

flowOf(command1, command2....).publishTo(myStateStoredAggregate)

Please notice that this aggregate is only fetching and storing/updating the final Podcast trascript/text. Audio is not part of this state. Audio ranges are rather part of the Command payload, and you have to came up with the process of reading that audio file and emit commands in a kotlin Flow:

flow {
   emit ( /* create a Command by reading Audio in chanks/ranges */ )
}.publishTo(myStateStoredAggregate)

[idugalic]

A second option is what you mentioned: Would the only way be to make a custom StateRepository that loads the range into memory and sends it to the decider?

In both options, you need to implement StateRepository

• in option 1 (I described) that state is just TEXT, no audio. Command is sending the audio. StateRepository is simple.
• in option 2 (you described) that state is the whole AUDIO and the TEXT (Pair< PodcastAudio, PodcastHighlighted >). Command is NOT sending the audio. StateRepository is more complex and you have to implement it in a way it returns regions of the Audio as specified in the command itself {startTime: 10s, endTime: 300s}.

Option 2 seems to be more robust because it includes the AUDIO fetching effects/logic. But, It could be that this PodcastTranscriptionDecider is doing two different things in this case: 1 splitting the audio and 2 transcription.

Option 1 tends to separate these two concerns (splitting, and transcription) better. In this case, the splitting process is done by another component (it can be a new PodcastSpliterDecider that is responsible for this or any component/service that you might already have in your system). Deciders can be combined, FYI: podcastSpliterDecider.combine(podcastTranscriptionDecider) under one big aggregate. https://fraktalio.com/blog/able-to-be-aggregated.html