feat(persistence): completion callbacks via Defer - simplified alternative to #7937 #7954

Aaronontheweb · 2025-11-28T17:16:07Z

Summary

This is a simplified alternative implementation to #7937, which adds completion callback and async handler support to Akka.NET persistence.

Fixes #7935

What's Different From #7937?

The Original Approach (#7937)

The original PR embedded completion callbacks directly in invocation objects with an IsLastInBatch flag to track when the final event handler completes. This required:

8 invocation classes (4 base + 4 with completion callbacks)
IsLastInBatch tracking logic to determine when to invoke completion callbacks
Complex handler chaining in PeekApplyHandler to chain completion callbacks

This Simplified Approach

After discussing with @gregorius, we realized that DeferAsync already provides most of what completion callbacks need - it queues a handler to run after all pending invocations complete.

The key insight: instead of embedding completion callbacks in invocation objects, we can simply call Defer/DeferAsync after PersistAll/PersistAllAsync:

public void PersistAll<TEvent>(IEnumerable<TEvent> events, Action<TEvent> handler, Action onComplete)
{
    PersistAll(events, handler);  // Existing method
    if (onComplete != null)
        Defer<object>(null, _ => onComplete());  // Queue completion after all handlers
}

Why This Is Better

Aspect	#7937 (Original)	This PR (Simplified)
Invocation classes	8 classes	4 classes
Completion tracking	`IsLastInBatch` flag	Uses existing `Defer`/`DeferAsync`
Code complexity	Complex handler chaining	Simple method delegation
Lines of code	~700+ new lines	~470 new lines

Critical Design Detail: Stashing Semantics

There's one subtlety that required a new internal stashing Defer method:

PersistAll (stashing) must use stashing Defer for completion callbacks
PersistAllAsync (non-stashing) uses DeferAsync for completion callbacks

Using DeferAsync for PersistAll would break stashing - commands would unstash before the completion callback runs. The internal Defer method increments _pendingStashingPersistInvocations to maintain correct stashing behavior.

Changes

New Interfaces

IPendingHandlerInvocation - base interface for handler invocations
ISyncHandlerInvocation - sync handler with Action<object> Handler
IAsyncHandlerInvocation - async handler with Func<object, Task> AsyncHandler
IStashingInvocation - marker interface for stashing invocations

New Invocation Classes (4 total, vs 8 in #7937)

StashingHandlerInvocation : ISyncHandlerInvocation, IStashingInvocation
StashingAsyncHandlerInvocation : IAsyncHandlerInvocation, IStashingInvocation
AsyncHandlerInvocation : ISyncHandlerInvocation (non-stashing)
AsyncAsyncHandlerInvocation : IAsyncHandlerInvocation (non-stashing)

New Public Methods

Persist<TEvent>(TEvent, Func<TEvent, Task>) - async handler
PersistAsync<TEvent>(TEvent, Func<TEvent, Task>) - async handler
PersistAll<TEvent>(events, handler, Action onComplete) - sync completion callback
PersistAll<TEvent>(events, handler, Func<Task> onCompleteAsync) - async completion callback
PersistAll<TEvent>(events, Func<TEvent, Task>, Action) - async handler + sync completion
PersistAll<TEvent>(events, Func<TEvent, Task>, Func<Task>) - async handler + async completion
PersistAllAsync<TEvent>(events, handler, Action onComplete) - sync completion callback
PersistAllAsync<TEvent>(events, handler, Func<Task> onCompleteAsync) - async completion callback
PersistAllAsync<TEvent>(events, Func<TEvent, Task>, Action) - async handler + sync completion
PersistAllAsync<TEvent>(events, Func<TEvent, Task>, Func<Task>) - async handler + async completion
DeferAsync<TEvent>(TEvent, Func<TEvent, Task>) - async handler

Internal Methods

Defer<TEvent>(TEvent, Action<TEvent>) - stashing variant (internal)
Defer<TEvent>(TEvent, Func<TEvent, Task>) - stashing async variant (internal)

Updated Recovery Logic

PeekApplyHandler now handles both sync and async handlers
PersistingEvents uses IStashingInvocation marker interface

Test Plan

9 new tests in PersistenceCompletionCallbackSpec.cs - all passing
99 core persistence tests (PersistentActorSpec, PersistentActorStashingSpec) - all passing
Full test suite maintains baseline pass rate
Stashing behavior verified: commands remain stashed until completion callbacks finish for PersistAll
Non-stashing behavior verified: commands can interleave with PersistAllAsync handlers

Add completion callback overloads for PersistAll and PersistAllAsync that invoke a callback after all events have been persisted and their handlers executed. Also add async handler support (Func<TEvent, Task>) to all persist methods. Key changes: - Add IPendingHandlerInvocation, ISyncHandlerInvocation, IAsyncHandlerInvocation, and IStashingInvocation interfaces for type-safe handler invocation - Add StashingHandlerInvocation, StashingAsyncHandlerInvocation, AsyncHandlerInvocation, and AsyncAsyncHandlerInvocation classes - Add Persist<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add PersistAsync<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add PersistAll overloads with completion callbacks (sync and async) - Add PersistAllAsync overloads with completion callbacks (sync and async) - Add DeferAsync<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add internal stashing Defer methods for completion callback support - Update PeekApplyHandler to handle async handlers via RunTask - Update PersistingEvents to use IStashingInvocation marker interface Stashing semantics are preserved: PersistAll completion callbacks use internal stashing Defer (increments _pendingStashingPersistInvocations), while PersistAllAsync uses non-stashing DeferAsync.

Update verified API file to reflect: - New public methods: Persist/PersistAsync/PersistAll/PersistAllAsync async handler overloads and completion callback overloads - New public method: DeferAsync with async handler - Internal invocation classes: AsyncHandlerInvocation, StashingHandlerInvocation, and IPendingHandlerInvocation are now internal (implementation detail)

…callbacks

Aaronontheweb

Detailed my changes - pushing up some more test coverage but otherwise this looks good

Aaronontheweb · 2025-12-01T17:36:38Z

src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistence.DotNet.verified.txt

 [assembly: System.Runtime.Versioning.TargetFrameworkAttribute(".NETCoreApp,Version=v6.0", FrameworkDisplayName=".NET 6.0")]
 namespace Akka.Persistence
 {
-    public sealed class AsyncHandlerInvocation : Akka.Persistence.IPendingHandlerInvocation


Marked as internal now

Aaronontheweb · 2025-12-01T17:36:47Z

src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistence.DotNet.verified.txt

        public override void AroundPreStart() { }
        protected override bool AroundReceive(Akka.Actor.Receive receive, object message) { }
        public void DeferAsync<TEvent>(TEvent evt, System.Action<TEvent> handler) { }
+        public void DeferAsync<TEvent>(TEvent evt, System.Func<TEvent, System.Threading.Tasks.Task> handler) { }


Now supports Task

Aaronontheweb · 2025-12-01T17:38:42Z

src/core/Akka.API.Tests/verify/CoreAPISpec.ApprovePersistence.DotNet.verified.txt

        protected virtual void OnRecoveryFailure(System.Exception reason, object message = null) { }
        protected virtual void OnReplaySuccess() { }
        public void Persist<TEvent>(TEvent @event, System.Action<TEvent> handler) { }
+        public void Persist<TEvent>(TEvent @event, System.Func<TEvent, System.Threading.Tasks.Task> handler) { }


Same sets of overloads as #7937 - we now support Task in all persistence callbacks AND an optional OnComplete handler which can be an Action or a Func<Task>

Aaronontheweb · 2025-12-01T17:40:52Z

src/core/Akka.Persistence.Tests/PersistenceCompletionCallbackSpec.cs

+                            _events.Add(evt.Data);
+                            _completionOrder.Add($"handler:{evt.Data}");
+                        }, () =>
+                        {


Dedicated OnComplete handler

Aaronontheweb · 2025-12-01T17:41:05Z

src/core/Akka.Persistence.Tests/PersistenceCompletionCallbackSpec.cs

+                            _events.Add(evt.Data);
+                            _completionOrder.Add($"handler:{evt.Data}");
+                        }, async () =>
+                        {


Async completion handler

Aaronontheweb · 2025-12-01T17:50:59Z

src/core/Akka.Persistence/Eventsourced.cs

+            }
+
+            PersistAll(events, handler);
+            if (onCompleteAsync != null)


We now compose the onComplete handler to use the appropriate Defer / DeferAsync overload.

Aaronontheweb · 2025-12-01T17:55:28Z

src/core/Akka.Persistence/Eventsourced.cs

+        /// <typeparam name="TEvent">The event type.</typeparam>
+        /// <param name="evt">The event to pass to the handler.</param>
+        /// <param name="handler">The handler to invoke.</param>
+        internal void Defer<TEvent>(TEvent evt, Action<TEvent> handler)


New internal method - Defer differs from DeferAsync in that it is a stashing invocation handler, meaning that it wants to be treated like a Persist / PersistAll call and interleaved with them, rather than running only once all of them have been completed.

Aaronontheweb · 2025-12-01T17:55:58Z

src/core/Akka.Persistence/Eventsourced.Recovery.cs

            {
-                _pendingInvocations.First.Value.Handler(payload);
+                // Async handler - run via RunTask
+                RunTask(async () =>


Run async handlers inside RunTask

Aaronontheweb · 2025-12-01T17:56:12Z

src/core/Akka.Persistence/Eventsourced.Recovery.cs

+                // Sync handler - invoke directly
+                try
+                {
+                    syncInv.Handler(payload);


Otherwise, run other handlers synchronously

Aaronontheweb · 2025-12-01T17:57:24Z

src/core/Akka.Persistence/Eventsourced.Recovery.cs

                    if (m1.ActorInstanceId == _instanceId)
                    {
                        UpdateLastSequenceNr(m1.Persistent);
-                        try


Moved all of the try...catch behavior to inside the PeekApplyHandler method.

…ethods - Convert all tests to use async/await with ExpectMsgAsync instead of sync-over-async ExpectMsg calls - Add tests for PersistAll/PersistAllAsync with empty events to verify completion callbacks are invoked immediately for all overloads: - PersistAll with sync completion callback (existing) - PersistAll with async completion callback (new) - PersistAllAsync with sync completion callback (new) - PersistAllAsync with async completion callback (new) - Update EmptyEventsWithCompletionActor to support all four scenarios

Arkatufus

I have some concern about this PR, mostly regarding the "in order processing/invocation guarantee" promise

Arkatufus · 2025-12-01T18:59:31Z

src/core/Akka.Persistence/Eventsourced.cs

+        {
+            if (events == null || !events.Any())
+            {
+                onComplete?.Invoke();


I don't think this is quite right, need to think that these Persist, PersistAll, and Defer as "fluent compositing" the method callbacks inside the _pendingInvocation list and they still need to be invoked in sequence. If another async operation was queued before this PersistAll call, the operations would be invoked out of order, which breaks the "in order execution guarantee" promise.

I think this should be

if (events == null || !events.Any()) { if (onComplete != null) Defer<object>(null, _ => onComplete()); return; }

src/core/Akka.Persistence/Eventsourced.cs

Arkatufus · 2025-12-01T19:03:01Z

src/core/Akka.Persistence/Eventsourced.cs

+
+            if (_pendingInvocations.Count == 0)
+            {
+                RunTask(() => handler(evt));


I'm not sure that invoking this asynchronously in a different thread as a detached task is the right move here

It's not though - that's how await is run inside the actor's context

i.e. ReceiveAsync is implemented on top of the RunTask handler from the UntypedActor

…maintain ordering When PersistAll/PersistAllAsync is called with empty events, the completion callback must still be queued through Defer/DeferAsync to maintain the in-order execution guarantee. Previously, the callback was invoked immediately which could cause out-of-order execution if there were pending invocations from prior Persist/PersistAll calls. Changes: - Replace immediate invocation with Defer/DeferAsync for all 8 overloads that have completion callbacks when events collection is null or empty - Add SequentialPersistOrderingActor test actor for ordering verification - Add test: Persist followed by empty PersistAll maintains execution order - Add test: Sequential PersistAll with empty in middle maintains order

Arkatufus

LGTM

…ative to akkadotnet#7937 (akkadotnet#7954) * feat(persistence): add completion callbacks and async handler support Add completion callback overloads for PersistAll and PersistAllAsync that invoke a callback after all events have been persisted and their handlers executed. Also add async handler support (Func<TEvent, Task>) to all persist methods. Key changes: - Add IPendingHandlerInvocation, ISyncHandlerInvocation, IAsyncHandlerInvocation, and IStashingInvocation interfaces for type-safe handler invocation - Add StashingHandlerInvocation, StashingAsyncHandlerInvocation, AsyncHandlerInvocation, and AsyncAsyncHandlerInvocation classes - Add Persist<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add PersistAsync<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add PersistAll overloads with completion callbacks (sync and async) - Add PersistAllAsync overloads with completion callbacks (sync and async) - Add DeferAsync<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add internal stashing Defer methods for completion callback support - Update PeekApplyHandler to handle async handlers via RunTask - Update PersistingEvents to use IStashingInvocation marker interface Stashing semantics are preserved: PersistAll completion callbacks use internal stashing Defer (increments _pendingStashingPersistInvocations), while PersistAllAsync uses non-stashing DeferAsync. * chore: update API approval for persistence completion callbacks Update verified API file to reflect: - New public methods: Persist/PersistAsync/PersistAll/PersistAllAsync async handler overloads and completion callback overloads - New public method: DeferAsync with async handler - Internal invocation classes: AsyncHandlerInvocation, StashingHandlerInvocation, and IPendingHandlerInvocation are now internal (implementation detail) * chore: update .NET Framework API approval for persistence completion callbacks * fixed API approvals * test(persistence): add empty events tests and convert to async test methods - Convert all tests to use async/await with ExpectMsgAsync instead of sync-over-async ExpectMsg calls - Add tests for PersistAll/PersistAllAsync with empty events to verify completion callbacks are invoked immediately for all overloads: - PersistAll with sync completion callback (existing) - PersistAll with async completion callback (new) - PersistAllAsync with sync completion callback (new) - PersistAllAsync with async completion callback (new) - Update EmptyEventsWithCompletionActor to support all four scenarios * fix(persistence): use Defer for empty events completion callbacks to maintain ordering When PersistAll/PersistAllAsync is called with empty events, the completion callback must still be queued through Defer/DeferAsync to maintain the in-order execution guarantee. Previously, the callback was invoked immediately which could cause out-of-order execution if there were pending invocations from prior Persist/PersistAll calls. Changes: - Replace immediate invocation with Defer/DeferAsync for all 8 overloads that have completion callbacks when events collection is null or empty - Add SequentialPersistOrderingActor test actor for ordering verification - Add test: Persist followed by empty PersistAll maintains execution order - Add test: Sequential PersistAll with empty in middle maintains order

…ative to #7937 (#7954) (#7957) * feat(persistence): add completion callbacks and async handler support Add completion callback overloads for PersistAll and PersistAllAsync that invoke a callback after all events have been persisted and their handlers executed. Also add async handler support (Func<TEvent, Task>) to all persist methods. Key changes: - Add IPendingHandlerInvocation, ISyncHandlerInvocation, IAsyncHandlerInvocation, and IStashingInvocation interfaces for type-safe handler invocation - Add StashingHandlerInvocation, StashingAsyncHandlerInvocation, AsyncHandlerInvocation, and AsyncAsyncHandlerInvocation classes - Add Persist<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add PersistAsync<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add PersistAll overloads with completion callbacks (sync and async) - Add PersistAllAsync overloads with completion callbacks (sync and async) - Add DeferAsync<TEvent>(TEvent, Func<TEvent, Task>) async handler overload - Add internal stashing Defer methods for completion callback support - Update PeekApplyHandler to handle async handlers via RunTask - Update PersistingEvents to use IStashingInvocation marker interface Stashing semantics are preserved: PersistAll completion callbacks use internal stashing Defer (increments _pendingStashingPersistInvocations), while PersistAllAsync uses non-stashing DeferAsync. * chore: update API approval for persistence completion callbacks Update verified API file to reflect: - New public methods: Persist/PersistAsync/PersistAll/PersistAllAsync async handler overloads and completion callback overloads - New public method: DeferAsync with async handler - Internal invocation classes: AsyncHandlerInvocation, StashingHandlerInvocation, and IPendingHandlerInvocation are now internal (implementation detail) * chore: update .NET Framework API approval for persistence completion callbacks * fixed API approvals * test(persistence): add empty events tests and convert to async test methods - Convert all tests to use async/await with ExpectMsgAsync instead of sync-over-async ExpectMsg calls - Add tests for PersistAll/PersistAllAsync with empty events to verify completion callbacks are invoked immediately for all overloads: - PersistAll with sync completion callback (existing) - PersistAll with async completion callback (new) - PersistAllAsync with sync completion callback (new) - PersistAllAsync with async completion callback (new) - Update EmptyEventsWithCompletionActor to support all four scenarios * fix(persistence): use Defer for empty events completion callbacks to maintain ordering When PersistAll/PersistAllAsync is called with empty events, the completion callback must still be queued through Defer/DeferAsync to maintain the in-order execution guarantee. Previously, the callback was invoked immediately which could cause out-of-order execution if there were pending invocations from prior Persist/PersistAll calls. Changes: - Replace immediate invocation with Defer/DeferAsync for all 8 overloads that have completion callbacks when events collection is null or empty - Add SequentialPersistOrderingActor test actor for ordering verification - Add test: Persist followed by empty PersistAll maintains execution order - Add test: Sequential PersistAll with empty in middle maintains order

Aaronontheweb changed the title ~~feat(persistence): add completion callbacks and async handler support~~ feat(persistence): completion callbacks via Defer - simplified alternative to #7937 Nov 28, 2025

Aaronontheweb added akka-persistence api-change labels Nov 28, 2025

Aaronontheweb added 3 commits November 28, 2025 13:13

chore: update .NET Framework API approval for persistence completion …

7fc9cbf

…callbacks

fixed API approvals

22651df

Aaronontheweb commented Dec 1, 2025

View reviewed changes

Arkatufus reviewed Dec 1, 2025

View reviewed changes

Arkatufus approved these changes Dec 1, 2025

View reviewed changes

Aaronontheweb merged commit e582943 into akkadotnet:dev Dec 1, 2025
11 checks passed

Aaronontheweb deleted the feature/persistence-completion-callbacks-v2 branch December 1, 2025 21:36

Aaronontheweb mentioned this pull request Dec 1, 2025

feat(persistence): add completion callbacks and async handler support #7937

Closed

Aaronontheweb added this to the 1.6.0 milestone Dec 1, 2025

Aaronontheweb mentioned this pull request Dec 2, 2025

Backport: feat(persistence): completion callbacks via Defer #7957

Merged

feat(persistence): completion callbacks via Defer - simplified alternative to #7937 #7954

feat(persistence): completion callbacks via Defer - simplified alternative to #7937 #7954

Conversation

Aaronontheweb commented Nov 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

What's Different From #7937?

The Original Approach (#7937)

This Simplified Approach

Why This Is Better

Critical Design Detail: Stashing Semantics

Changes

New Interfaces

New Invocation Classes (4 total, vs 8 in #7937)

New Public Methods

Internal Methods

Updated Recovery Logic

Test Plan

Uh oh!

Aaronontheweb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Arkatufus left a comment

Choose a reason for hiding this comment

Uh oh!

Arkatufus Dec 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Arkatufus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Aaronontheweb commented Nov 28, 2025 •

edited

Loading

Arkatufus Dec 1, 2025 •

edited

Loading