Hi Alex,
the main concern you express is that messages sent from an
EventsourcedProcessor are eventually delivered to their recipients, which is
what a durable message queue (formerly called PersistentChannel) is for. The
main use of the non-persistent Channel is to deduplicate messages, but it
fundamentally cannot do that reliably in any case (since a confirmation may
have been on its way when it crashed).
In the solution you describe, if your Earnings must make sure to send to the
Aggregator, then they will in either case need to resend the message
periodically until confirmed, not only during restart, otherwise the message
can be deferred indefinitely; this is the same regardless of which transport
helper is used. My concrete proposal would be to store unconfirmed outbound
messages in the Earnings’ state and resend based on a timer tick—this extends
naturally to recovery after a machine crash.
I think we’ll see this resending functionality factored out in a common trait
soon as more people start building software this way.
Regards,
Roland
21 maj 2014 kl. 15:30 skrev ahjohannessen <ahjohannes...@gmail.com>:
> Hi Roland,
>
> You state the following:
>
> "...talks about Channel since that was only needed to contain the
> side-effecting replay nature of command sourced processors."
>
> A channel is also needed when sending events from an eventsourced processor
> to another actor when one needs at-least-once delivery rather
> than at-most-once:
>
> def receiveRecover: Receive = {
> case event: String => handleEvent(event)
> }
>
> def handleEvent(event: String) = {
> // update state
> // ...
> // reliably deliver events
> channel ! Deliver(Persistent(event), destination.path)
> // alternatively something that encapsulates this pair
> }
>
>
> We are dependent on this very functionality for reliable coordination in our
> apps and removing Channel leaves us worried.
>
>
> "...Your description below is rather terse, so it is not fully clear to me
> how you are using Channel in this case and what a replacement should be, can
> you elaborate?"
>
> As stated earlier, we have many eventsourced processors of same type, e.g.
> 10000 instances of "Earnings", that are loaded on demand by a supervisor
> when a command enters into the domain. When an instance is loaded the command
> is forwarded to this guy and when the command
> results in domain events we acknowledge to sender in the persist callback
> *but* also forward the persistent message to another actor, let's call it
> "stream",
> that needs to react to this. It ensures that a calculation is initiated by
> delivering a message to a "change reactor" via a unique channel.
> Furthermore "stream" also ensures that a message is delivered, via another
> unique channel, to an EarningsAggregator that essentially replicates all of
> those 10000 instances' events, because we need a global view in order to
> maintain read models.
>
> In receiveRecover we *also* forward replayed events to "stream" in order to
> ensure that messages get delivered, via channels, to the respective
> destinations,
> in the case of a JVM crash.
>
> During system start up we start all recently active (say last 24 hours)
> Earnings instances to ensure that all calculations are triggered and that our
> EarningsAggregator
> gets the event, again in the name of crash paranoia. These instances shut
> themselves down after a reasonable receive timeout in order to keep memory
> footprint low.
> This roughly happens by sending a "passivate me" message to their supervisor
> which in turn ensures a proper shutdown and all of the instance's messages
> are
> processed by using a combination of poison pill, become, stash and listening
> to terminated.
>
> In the above I mention Earnings eventsourced processor, however we have
> around 10 similar types under same supervisor that are similarly structured
> and have the same
> need of what a channel gives us.
>
> Other usages of channels that we have are communication among processes.
> Often here it is perfectly fine for us to use channels because it ensures
> that something eventually
> will happen, some eventsourced processor emits an event to reactor (ordinary
> actor) via a channel (also during receiveRecover) that transforms the event
> to a command in the
> language of another eventsourced processor. All of such processes are
> idempotent and maintaining an queue on both sides is just a lot of useless
> busywork in most of our
> use cases.
>
> I have no idea of what a replacement of Channel would be because I do not
> have an issue with it and think it is a good primitive that solves many of
> our use cases in ways
> that are satisfying and saves us a lot of work trying to re-implement the
> same functionality.
>
> "...My motivation here is to not remove needed functionality without improved
> replacement."
>
> That is good to know :)
>
>
> On Tuesday, May 20, 2014 7:32:38 PM UTC+1, rkuhn wrote:
> Hi Alex,
>
> I have filed the ticket for Processor’s removal
> (https://github.com/akka/akka/issues/15230), which also talks about Channel
> since that was only needed to contain the side-effecting replay nature of
> command sourced processors. Your description below is rather terse, so it is
> not fully clear to me how you are using Channel in this case and what a
> replacement should be, can you elaborate?
>
> There is also the discussion ticket for reinventing PersistentChannel
> (https://github.com/akka/akka/issues/15231) which might be of interest in
> this context. My motivation here is to not remove needed functionality
> without improved replacement.
>
> Regards,
>
> Roland
>
> 9 maj 2014 kl. 12:15 skrev ahjohannessen <ahjoha...@gmail.com>:
>
>> Hi Roland,
>>
>> We use Channel in conjunction with Eventsourced Processor (EP) in our
>> applications in receiveRecover.
>> It would be sad to see it go away without a reasonable alternative.
>>
>> One scenario in our apps is that we use DDD/ES and have a lot of EPs of same
>> type, e.g. 10000 instances,
>> that are loaded on demand by a supervisor.
>>
>> In order to have a single view on all of these, we inject an actor that
>> wraps a single channel / aggregator EP combo
>> into these instances on creation. This makes it possible to react to
>> changes, even in case of JVM crashes,
>> in that family of EPs as well as having a global view of that family.
>>
>>
>> On Friday, May 9, 2014 9:13:17 AM UTC+1, rkuhn wrote:
>>
>> 9 maj 2014 kl. 09:58 skrev Martin Krasser <kras...@googlemail.com>:
>>
>>>
>>> On 09.05.14 09:25, Roland Kuhn wrote:
>>>>
>>>> 9 maj 2014 kl. 09:08 skrev Martin Krasser <kras...@googlemail.com>:
>>>>
>>>>>
>>>>> On 09.05.14 08:41, Roland Kuhn wrote:
>>>>>> Hi Martin,
>>>>>>
>>>>>> 9 maj 2014 kl. 08:05 skrev Martin Krasser <kras...@googlemail.com>:
>>>>>>
>>>>>>> Hi Roland,
>>>>>>>
>>>>>>> thanks for starting a discussion on this. Here are some initial
>>>>>>> thoughts on your proposal:
>>>>>>>
>>>>>>> "... very same throughput optimization by applying the state changes
>>>>>>> before persisting them ..."
>>>>>>>
>>>>>>> I think we agree that whatever changes are going to be made in the
>>>>>>> future, we must keep the throughput optimizations (by batching
>>>>>>> writes/updates). As you said, with an EP, this can only be achieved by
>>>>>>> applying events to current state *before* persisting them. Furthermore,
>>>>>>> to enable batching, an EP must therefore be able to process new
>>>>>>> commands while (previous) events are about to be persisted. This
>>>>>>> however has a very important consequence for commands that read current
>>>>>>> state. If we allow events to be applied to current state *before*
>>>>>>> persisting them, we allow clients to read state from that EP that may
>>>>>>> not be re-readable after a crash. For example:
>>>>>>>
>>>>>>> - EP receives update command, derives event and applies it immediately
>>>>>>> to current state
>>>>>>> - EP (asynchronously) persists event
>>>>>>> - EP receives a read command (while event persistence is in progress)
>>>>>>> - EP (successfully) returns read response to requestor
>>>>>>> - EP JVM crashes before event was successfully persisted
>>>>>>> - EP state cannot be reconstructed i.e. previous read cannot be
>>>>>>> repeated.
>>>>>>
>>>>>> This is only true if the recovery is incomplete: the update command will
>>>>>> not have been acknowledged at this point, so if someone cared about it
>>>>>> they will send it again during recovery and the EP will eventually end
>>>>>> up in a state where the read will return the same value again. If this
>>>>>> type of consistency is not good enough, then you can always defer reads
>>>>>> within the write model until after persistence is completed, meaning
>>>>>> that the read is only performed once a corresponding read “event” has
>>>>>> gone through the journal. We could allow events that are only looped
>>>>>> through to make this work, just like non-Persistent commands are looped
>>>>>> today (and for the same reason).
>>>>>
>>>>> Delaying reads is only an option when reads are made via messages to a
>>>>> (E)P. If my processor manages state via an STM ref where only the
>>>>> processor updates the STM ref but reads go directly to the STM ref, then
>>>>> you cannot delay reads.
>>>>
>>>> In this scenario you would delay updating the STM ref until after the
>>>> persistence loop, which is exactly the same as for a current
>>>> command-sourced Processor: the read gets delayed until after the writes
>>>> are processed, in the same way the STM ref update gets delayed by the
>>>> write having to go through the journal. Effects, consistency and latency
>>>> are the same in both implementations.
>>>
>>> That's true. So, to achieve
>>>
>>> - repeatable reads
>>> - low read latency and
>>> - high write throughput
>>>
>>> reads can go to the STM refs directly and EP must update the STM ref only
>>> after having persisted the events. If one *additionally* wants to achieve
>>>
>>> - read-your-own-write consistency (assuming a client issues an update
>>> command, immediately followed by a read command)
>>>
>>> one would need a way to loop read commands through the journal as well
>>> before serving them (which probably requires an addition to the API then).
>>> Alternatively, a client only issues a read after having received a
>>> write-ack (at the cost of an additional roundtrip).
>>
>> This is an interesting remark: normally read-your-writes is only guaranteed
>> for reads submitted after having received the ACK for the write, so what we
>> are providing here is actually a qualitative improvement on that status quo
>> that is only possible in Reactive systems (normally the ACK is signaled by a
>> synchronous non-exceptional method return).
>>
>>> Anyway, I think you convinced me, as usual :) Great proposal, Dr Kuhn!
>>
>> And as usual you helped in refining the proposal: the addition of looping
>> non-persistent events through the journal is an important one, thanks for
>> providing the use-case!
>>
>> So, to summarize, we can incorporate all current functionality provided by
>> Processor and Channel into EventsourcedProcessor by adding the following two
>> features:
>>
>> the ability to opt out of stashing everything while waiting for persist()ing
>> the ability to loop non-persistent events through the journal
>>
>> Everyone, please consider what this would mean for your code base and
>> comment, now is the right time to speak up! The same goes for opinions on
>> whether PersistentChannel pulls its weight or not (as argued earlier in this
>> thread).
>>
>> Regards,
>>
>> Roland
>>
>>>
>>> Cheers,
>>> Martin
>>>
>>> --
>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>> >>>>>>>>>> Check the FAQ:
>>> >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
>>> ---
>>> You received this message because you are subscribed to the Google Groups
>>> "Akka User List" group.
>>> To unsubscribe from this group and stop receiving emails from it, send an
>>> email to akka-user+...@googlegroups.com.
>>> To post to this group, send email to akka...@googlegroups.com.
>>> Visit this group at http://groups.google.com/group/akka-user.
>>> For more options, visit https://groups.google.com/d/optout.
>>
>>
>>
>> Dr. Roland Kuhn
>> Akka Tech Lead
>> Typesafe – Reactive apps on the JVM.
>> twitter: @rolandkuhn
>>
>>
>>
>> --
>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>> >>>>>>>>>> Check the FAQ:
>> >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
>> ---
>> You received this message because you are subscribed to the Google Groups
>> "Akka User List" group.
>> To unsubscribe from this group and stop receiving emails from it, send an
>> email to akka-user+...@googlegroups.com.
>> To post to this group, send email to akka...@googlegroups.com.
>> Visit this group at http://groups.google.com/group/akka-user.
>> For more options, visit https://groups.google.com/d/optout.
>
>
>
> Dr. Roland Kuhn
> Akka Tech Lead
> Typesafe – Reactive apps on the JVM.
> twitter: @rolandkuhn
>
>
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ:
> >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
Dr. Roland Kuhn
Akka Tech Lead
Typesafe – Reactive apps on the JVM.
twitter: @rolandkuhn
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
