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

Aaronontheweb · 2025-11-21T14:26:12Z

Summary

fixes #7935

This PR adds completion callback support and async handler support to Akka.NET persistence, enabling users to:

Detect when all events in a PersistAll/PersistAllAsync batch have been persisted and their handlers executed
Use async handlers (Func<TEvent, Task>) with all persist methods

Note: This is a proof of concept implementation.

New API Reference

Async Handler Overloads (NEW)

Single event persistence with async handlers:

// Stashing (blocks incoming commands until persisted)
void Persist<TEvent>(TEvent @event, Func<TEvent, Task> handler)

// Non-stashing (continues processing commands while persisting)
void PersistAsync<TEvent>(TEvent @event, Func<TEvent, Task> handler)

Batch Persistence with Async Handlers (NEW)

// Stashing variants
void PersistAll<TEvent>(IEnumerable<TEvent> events, Func<TEvent, Task> handler)
void PersistAll<TEvent>(IEnumerable<TEvent> events, Func<TEvent, Task> handler, Action onComplete)
void PersistAll<TEvent>(IEnumerable<TEvent> events, Func<TEvent, Task> handler, Func<Task> onCompleteAsync)

// Non-stashing variants  
void PersistAllAsync<TEvent>(IEnumerable<TEvent> events, Func<TEvent, Task> handler)
void PersistAllAsync<TEvent>(IEnumerable<TEvent> events, Func<TEvent, Task> handler, Action onComplete)
void PersistAllAsync<TEvent>(IEnumerable<TEvent> events, Func<TEvent, Task> handler, Func<Task> onCompleteAsync)

Completion Callbacks for Sync Handlers (NEW)

// Stashing variants with sync handler
void PersistAll<TEvent>(IEnumerable<TEvent> events, Action<TEvent> handler, Action onComplete)
void PersistAll<TEvent>(IEnumerable<TEvent> events, Action<TEvent> handler, Func<Task> onCompleteAsync)

// Non-stashing variants with sync handler
void PersistAllAsync<TEvent>(IEnumerable<TEvent> events, Action<TEvent> handler, Action onComplete)
void PersistAllAsync<TEvent>(IEnumerable<TEvent> events, Action<TEvent> handler, Func<Task> onCompleteAsync)

Usage Examples

Async handler with completion callback:

PersistAll(events, async evt => 
{
    // Async work for each event
    await UpdateProjectionsAsync(evt);
}, 
onComplete: () => 
{
    // Called once after ALL events persisted and handlers executed
    Sender.Tell(new BatchComplete());
});

Async handler with async completion callback:

PersistAll(events, async evt => 
{
    await UpdateProjectionsAsync(evt);
}, 
onCompleteAsync: async () => 
{
    // Async completion work
    await NotifySubscribersAsync();
    Sender.Tell(new BatchComplete());
});

Empty batch handling:

// Completion callback is invoked immediately for empty event lists
PersistAll(Array.Empty<MyEvent>(), _ => { }, () => 
{
    // This is called immediately
    Sender.Tell(new BatchComplete());
});

Changes

New Features

Completion Callbacks: PersistAll and PersistAllAsync now accept optional onComplete (sync) or onCompleteAsync (async) callbacks that execute after all events in the batch are persisted
Async Handlers: All 4 persist methods (Persist, PersistAsync, PersistAll, PersistAllAsync) now support Func<TEvent, Task> async handlers

Implementation Details

Added 8 new invocation classes with interface segregation pattern for sync/async handlers and sync/async completion callbacks
Added IStashingInvocation marker interface for type-safe detection of stashing invocations
Added 10 new method overloads providing all combinations of sync/async handlers with sync/async completion callbacks
Updated PeekApplyHandler to chain all async operations in a single RunTask call to avoid nested RunTask exceptions
Maintains backward compatibility - all existing persist method signatures unchanged
Preserves actor semantics (stashing vs non-stashing behavior)

Test Coverage

Added comprehensive test suite in PersistenceCompletionCallbackSpec.cs:

PersistAllAsync_should_invoke_completion_callback_after_all_events_persisted
PersistAll_should_invoke_completion_callback_after_all_events_persisted
PersistAllAsync_should_invoke_async_completion_callback
PersistAll_should_invoke_async_completion_callback
PersistAllAsync_should_support_async_handlers
PersistAsync_should_support_async_handlers
Completion_callback_should_fire_exactly_once_per_batch
Empty_event_list_should_invoke_completion_callback_immediately

All 8 new tests pass consistently (100% pass rate).

Test Results

Full test suite results match baseline stability (266-268 passing, 7-10 failing - matching pre-existing baseline variance). No new failures introduced by this implementation.

Add completion callback support to PersistAll and PersistAllAsync methods to detect when all events in a batch have been persisted and their handlers executed. Add async handler support (Func<TEvent, Task>) for all persist methods (Persist, PersistAsync, PersistAll, PersistAllAsync). Changes: - Add 8 new invocation classes with interface segregation for sync/async handlers and sync/async completion callbacks - Add IStashingInvocation marker interface for type-safe detection of stashing invocations - Add 10 new method overloads across all 4 persist methods supporting completion callbacks and async handlers - Update PeekApplyHandler to chain async operations in a single RunTask call to avoid nested RunTask exceptions - Maintain backward compatibility and preserve actor semantics Test coverage: - 8 new tests covering completion callbacks and async handlers - All new tests pass consistently (100% pass rate) - No new failures introduced in existing test suite

Aaronontheweb

Detailed my changes - IMHO, I think I can simplify this by transforming all of the Action<object> callbacks into Func<object,Task> callbacks by just wrapping the delegate. That will cut down on the number of internal classes and interfaces pretty significantly.

Plus I identified some inconsistencies in how we're handling empty / null event arrays.

Aaronontheweb · 2025-11-24T19:33:52Z