Skip to content

Event Completion Awaiter

Verified by tests

EventCompletionAwaiterTests, SyncEventTrackerTests, DispatchOptionsTests, DispatcherPerspectiveSyncCoverageTests — library CI run #31657041675 (2026-08-13)

Event Completion Awaiter enables waiting for events to be fully processed by ALL perspectives before returning. This is essential for RPC-style calls where you need to ensure complete processing before responding to the caller.

The Problem

When using LocalInvokeAsync for RPC-style dispatching, the response returns immediately after cascade completes. However, perspectives may still be processing the cascaded events:

flowchart TD
    Invoke["LocalInvokeAsync(CreateOrderCommand)"]
    Handler["Handler executes<br/>Returns OrderId<br/>(Emits OrderCreatedEvent)"]
    Cascade["Cascade completes<br/>(Event sent to perspective worker)"]
    Response["Response returned!<br/>(But perspective hasn't processed yet!)"]
    Worker["Perspective Worker<br/>OrderPerspective<br/>ReportingPerspective<br/>(Processes OrderCreatedEvent async)"]

    Invoke --> Handler
    Handler --> Cascade
    Cascade --> Response
    Response -->|"gap - perspectives still processing"| Worker

    style Response fill:#f8d7da,stroke:#dc3545,stroke-width:2px

The solution: Use IEventCompletionAwaiter or DispatchOptions.WithPerspectiveWait() to wait for all perspectives to finish.

Event-Based vs Perspective-Based Waiting

Whizbang provides two distinct waiting semantics:

Approach Waits For Use Case
IPerspectiveSyncAwaiter One specific perspective Query consistency - ensure a handler sees its own changes in a specific perspective
IEventCompletionAwaiter All perspectives RPC completion - ensure all processing is complete before responding

When to Use Each

Use IPerspectiveSyncAwaiter when: - You need to query a specific perspective after emitting events - You want read-your-writes consistency for one perspective - You're using [AwaitPerspectiveSync] attribute on receptors

Use IEventCompletionAwaiter when: - Making RPC calls via LocalInvokeAsync - You need to ensure ALL perspectives have processed before responding - The caller needs a guarantee that all side effects are complete

Usage: DispatchOptions.WithPerspectiveWait()

The simplest way to wait for all perspectives is using DispatchOptions:

Usage: DispatchOptions.WithPerspectiveWait()

using Whizbang.Core.Dispatch;

public class OrderService {
    private readonly IDispatcher _dispatcher;

    public async Task<Guid> CreateOrderAsync(CreateOrderRequest request, CancellationToken ct) {
        var command = new CreateOrderCommand {
            CustomerId = request.CustomerId,
            Items = request.Items
        };

        // Wait for ALL perspectives to process cascaded events
        var options = new DispatchOptions()
            .WithCancellationToken(ct)
            .WithPerspectiveWait(timeout: TimeSpan.FromSeconds(30));

        var orderId = await _dispatcher.LocalInvokeAsync<Guid>(command, options);

        // At this point, ALL perspectives have processed the OrderCreatedEvent
        return orderId;
    }
}

Configuration Options

Configuration Options

public sealed class DispatchOptions {
    // When true, LocalInvokeAsync waits for all perspectives to finish
    public bool WaitForPerspectives { get; set; }

    // Timeout for waiting (default: 30 seconds)
    public TimeSpan PerspectiveWaitTimeout { get; set; } = TimeSpan.FromSeconds(30);

    // Cancellation for the dispatch operation (default: CancellationToken.None)
    public CancellationToken CancellationToken { get; set; } = CancellationToken.None;

    // Fluent API
    public DispatchOptions WithPerspectiveWait(TimeSpan? timeout = null);
    public DispatchOptions WithCancellationToken(CancellationToken token);
}

Dispatcher Integration

The Dispatcher integrates with event completion through the _waitForPerspectivesIfNeededAsync method, which is called after receptor invocation completes.

Integration Architecture

flowchart TD
    Invoke["LocalInvokeAsync with DispatchOptions"]
    Step1["1. Check for [AwaitPerspectiveSync] (if present)<br/>_awaitPerspectiveSyncIfNeededAsync()"]
    Step2["2. Invoke receptor<br/>var result = await invoker(message);"]
    Step3["3. Auto-cascade events from result<br/>_cascadeEventsFromResultAsync()"]
    Step4["4. Wait for ALL perspectives (if requested)<br/>_waitForPerspectivesIfNeededAsync(options)"]
    Step5["5. Return result"]

    Invoke --> Step1
    Step1 --> Step2
    Step2 --> Step3
    Step3 --> Step4
    Step4 --> Step5

Implementation

The dispatcher checks DispatchOptions.WaitForPerspectives after receptor execution:

Implementation

public async ValueTask<TResult> LocalInvokeAsync<TResult>(
    object message,
    DispatchOptions options) {

  // 1. Await perspective sync if receptor has [AwaitPerspectiveSync]
  await _awaitPerspectiveSyncIfNeededAsync(message, messageType, options.CancellationToken);

  // 2. Invoke receptor
  var result = await invoker(message);

  // 3. Cascade events from result
  await _cascadeEventsFromResultAsync(result, messageType, sourceEnvelope: envelope);

  // 4. Wait for ALL perspectives if requested
  await _waitForPerspectivesIfNeededAsync(options);

  return result;
}

_waitForPerspectivesIfNeededAsync Implementation

_waitForPerspectivesIfNeededAsync Implementation

private async ValueTask _waitForPerspectivesIfNeededAsync(DispatchOptions options) {
  // Short-circuit if not waiting for perspectives
  if (!options.WaitForPerspectives) {
    return;
  }

  // Short-circuit if no event completion awaiter available
  if (_eventCompletionAwaiter is null) {
    return;
  }

  // Get the scoped event tracker (field or from AsyncLocal accessor)
  var scopedTracker = _scopedEventTracker ?? ScopedEventTrackerAccessor.CurrentTracker;
  if (scopedTracker is null) {
    return;
  }

  // Get all events emitted during this invocation
  var emittedEvents = scopedTracker.GetEmittedEvents();
  if (emittedEvents.Count == 0) {
    return;  // No events to wait for
  }

  var eventIds = emittedEvents.Select(e => e.EventId).Distinct().ToList();

  // Wait for ALL perspectives to process these events
  var success = await _eventCompletionAwaiter.WaitForEventsAsync(
      eventIds,
      options.PerspectiveWaitTimeout,
      options.CancellationToken);

  if (!success) {
    throw new PerspectiveSyncTimeoutException(
        $"Timed out waiting for {eventIds.Count} events to be processed by all perspectives. " +
        $"Timeout: {options.PerspectiveWaitTimeout.TotalMilliseconds}ms");
  }
}

Key Features

1. Scoped Event Tracking

The dispatcher uses either: - Injected _scopedEventTracker (scoped DI) - ScopedEventTrackerAccessor.CurrentTracker (ambient access)

This enables event tracking even when the dispatcher is a singleton.

2. Automatic EventId Discovery

Events are automatically captured during receptor execution:

Key Features

// Receptor emits events
await _eventStore.AppendAsync(streamId, new OrderCreatedEvent());

// _scopedTracker automatically captures EventId
// (via event store decorator)

// Dispatcher queries tracker after receptor completes
var eventIds = scopedTracker.GetEmittedEvents()
    .Select(e => e.EventId)
    .Distinct()
    .ToList();

3. Timeout Handling

When timeout occurs, throws PerspectiveSyncTimeoutException:

Key Features (2)

try {
  var options = new DispatchOptions().WithPerspectiveWait();
  await _dispatcher.LocalInvokeAsync(command, options);
} catch (PerspectiveSyncTimeoutException ex) {
  _logger.LogWarning("Perspective processing timed out: {Message}", ex.Message);
  // Handle timeout
}

4. Zero Overhead When Disabled

Multiple short-circuit checks ensure zero overhead when WaitForPerspectives = false:

Key Features (3)

if (!options.WaitForPerspectives) return;        // First check
if (_eventCompletionAwaiter is null) return;     // Not registered
if (scopedTracker is null) return;               // No scope
if (eventIds.Count == 0) return;                 // No events

All LocalInvokeAsync Overloads

The integration works across all LocalInvokeAsync overloads that accept DispatchOptions:

All LocalInvokeAsync Overloads

// Receptor with typed business result
ValueTask<TResult> LocalInvokeAsync<TResult>(
    object message,
    DispatchOptions options);

// Void receptor
ValueTask LocalInvokeAsync(
    object message,
    DispatchOptions options);

// Receptor with typed business result AND delivery receipt
ValueTask<InvokeResult<TResult>> LocalInvokeWithReceiptAsync<TResult>(
    object message,
    DispatchOptions options);

All paths call _waitForPerspectivesIfNeededAsync(options) after receptor execution.

Usage: IEventCompletionAwaiter

For more control, inject IEventCompletionAwaiter directly:

Usage: IEventCompletionAwaiter

using Whizbang.Core.Perspectives.Sync;

public class OrderOrchestrator {
    private readonly IDispatcher _dispatcher;
    private readonly IEventCompletionAwaiter _completionAwaiter;
    private readonly IScopedEventTracker _scopedTracker;

    public async Task<OrderResult> ProcessOrderAsync(CreateOrderCommand cmd, CancellationToken ct) {
        // Execute command
        var receipt = await _dispatcher.SendAsync(cmd, new DispatchOptions().WithCancellationToken(ct));

        // Get all event IDs emitted in this scope
        var emittedEvents = _scopedTracker.GetEmittedEvents();
        var eventIds = emittedEvents.Select(e => e.EventId).Distinct().ToList();

        if (eventIds.Count > 0) {
            // Wait for ALL perspectives to finish processing
            var success = await _completionAwaiter.WaitForEventsAsync(
                eventIds,
                timeout: TimeSpan.FromSeconds(30),
                ct);

            if (!success) {
                throw new PerspectiveSyncTimeoutException(
                    $"Timed out waiting for {eventIds.Count} events to be processed");
            }
        }

        return new OrderResult { OrderId = receipt.StreamId ?? Guid.Empty, FullyProcessed = true };
    }
}

API Reference

API Reference

public interface IEventCompletionAwaiter : IAwaiterIdentity {
    /// <summary>
    /// Unique identity for per-awaiter tracking and cleanup.
    /// Inherited from IAwaiterIdentity.
    /// </summary>
    Guid AwaiterId { get; }

    /// <summary>
    /// Waits for events to be processed by ALL perspectives.
    /// Returns when no perspectives are still tracking any of the specified events.
    /// </summary>
    Task<bool> WaitForEventsAsync(
        IReadOnlyList<Guid> eventIds,
        TimeSpan timeout,
        CancellationToken cancellationToken = default);

    /// <summary>
    /// Checks if events have been fully processed by all perspectives.
    /// </summary>
    bool AreEventsFullyProcessed(IReadOnlyList<Guid> eventIds);
}

Updated

IEventCompletionAwaiter now extends IAwaiterIdentity. The AwaiterId is passed to ISyncEventTracker.WaitForAllPerspectivesAsync() for per-awaiter cleanup on cancellation. See Awaiter Identity.

How It Works

The event completion system uses per-perspective tracking:

flowchart TD
    Emit["Event emitted (EventId = abc123)"]

    subgraph Tracker["SyncEventTracker"]
        Tracked["Tracked Events (per-perspective)<br/>(abc123, OrderPerspective)<br/>(abc123, ReportingPerspective)"]
    end

    Mark1["OrderPerspective calls<br/>MarkProcessedByPerspective(abc123, #quot;OrderPerspective#quot;)<br/><br/>Removes only: (abc123, OrderPersp.)<br/>Still tracked: (abc123, Reporting)"]
    Mark2["ReportingPerspective calls<br/>MarkProcessedByPerspective(abc123, #quot;ReportingPerspective#quot;)<br/><br/>No more entries for abc123<br/>→ Signals WaitForAllPerspectivesAsync"]

    Emit --> Tracker
    Tracker -->|"Perspective workers process event"| Mark1
    Mark1 -->|"When ALL perspectives done"| Mark2

Key Methods

Method Purpose
MarkProcessedByPerspective(eventIds, perspectiveName) Called by perspective worker when done processing
WaitForPerspectiveEventsAsync(eventIds, perspectiveName, timeout) Wait for ONE perspective (used by IPerspectiveSyncAwaiter)
WaitForAllPerspectivesAsync(eventIds, timeout) Wait for ALL perspectives (used by IEventCompletionAwaiter)

Timeout Handling

When perspectives don't complete within the timeout:

Timeout Handling

var options = new DispatchOptions()
    .WithPerspectiveWait(timeout: TimeSpan.FromSeconds(5));

try {
    await _dispatcher.LocalInvokeAsync(command, options);
} catch (PerspectiveSyncTimeoutException ex) {
    // Handle timeout - perspectives are still processing
    _logger.LogWarning("Perspective processing timed out: {Message}", ex.Message);

    // Options:
    // 1. Return partial success with warning
    // 2. Queue for retry
    // 3. Return error to caller
}

Best Practices

Do: Use for External API Responses

Do: Use for External API Responses

[HttpPost]
public async Task<IActionResult> CreateOrder(CreateOrderRequest request) {
    var options = new DispatchOptions()
        .WithCancellationToken(HttpContext.RequestAborted)
        .WithPerspectiveWait();
    var orderId = await _dispatcher.LocalInvokeAsync<Guid>(
        new CreateOrderCommand(request), options);

    // Safe to query any perspective - all are up to date
    return Ok(orderId);
}

Do: Set Appropriate Timeouts

Do: Set Appropriate Timeouts

// Short timeout for real-time APIs
.WithPerspectiveWait(TimeSpan.FromSeconds(5))

// Longer timeout for batch operations
.WithPerspectiveWait(TimeSpan.FromSeconds(60))

Don't: Use When Not Needed

Don't: Use When Not Needed

// Fire-and-forget scenarios don't need to wait
await _dispatcher.SendAsync(command); // No waiting

// Only wait when caller needs complete processing
var options = new DispatchOptions().WithPerspectiveWait();
await _dispatcher.LocalInvokeAsync(command, options); // Waits

Don't: Confuse with Perspective-Specific Sync

Don't: Confuse with Perspective-Specific Sync

// WRONG: Using event completion when you only need one perspective
var options = new DispatchOptions().WithPerspectiveWait();
await _dispatcher.LocalInvokeAsync(command, options);
var order = await _orderLens.GetByIdAsync(orderId, ct);

// RIGHT: Use IPerspectiveSyncAwaiter for single-perspective consistency
await _syncAwaiter.WaitAsync(typeof(OrderPerspective),
    SyncFilter.CurrentScope().Build(), ct);
var order = await _orderLens.GetByIdAsync(orderId, ct);

DI Registration

IEventCompletionAwaiter is automatically registered by AddWhizbang():

DI Registration

services.AddWhizbang(options => {
    // Configuration
});

// Resolving
var awaiter = serviceProvider.GetRequiredService<IEventCompletionAwaiter>();