Shuttle.Recall 22.0.1

Shuttle.Recall

Shuttle.Recall is an event-sourcing mechanism for .NET that provides a flexible way to persist and retrieve event streams.

Installation

dotnet add package Shuttle.Recall

Registration

To register Shuttle.Recall, use the AddRecall extension method, which returns a RecallBuilder used to configure projections:

services
    .AddRecall()
    .AddProjection("ProjectionName", projection =>
    {
        projection.AddEventHandler<SomeEvent>((context, evt) => 
        {
            // handle event
        });
    });

AddRecall also registers an IHostedService (EventProcessorHostedService) that automatically starts and stops the IEventProcessor along with the host, whenever at least one projection has been registered — see IEventProcessor Lifecycle.

The following types are registered:

• IEventStore (Scoped): Used to retrieve and save event streams.
• IEventProcessor (Singleton): Used to process projections.
• IEventMethodInvoker (Singleton): Invokes event handling methods on aggregate roots.
• IEventHandlerInvoker (Singleton): Invokes projection event handlers (types, instances, and delegates).
• ISerializer (Singleton): Serializes and deserializes events.
• IConcurrencyExceptionSpecification (Singleton): Determines whether an exception raised while saving should be treated as a concurrency conflict.

Configuration Options

services.AddRecall(options =>
{
    options.EventProcessing.ProjectionThreadCount = 5;
    options.EventProcessing.IncludedProjections.Add("ProjectionName");
    options.EventProcessing.ExcludedProjections.Add("ExcludeMe");

    options.EventProcessing.ImmediateConsistency.Enabled = true;
    options.EventProcessing.ImmediateConsistency.IncludedProjections.Add("ProjectionName");

    options.EventStore.EventHandlingMethodName = "On";
});

EventProcessing Options

ImmediateConsistency Options

By default, a projection only ever processes an event once the background IEventProcessor gets round to it. Immediate consistency lets specific projections handle an event synchronously, as part of the IEventStore.SaveAsync call that persisted it, so that a read model built from that projection is guaranteed to reflect the event by the time SaveAsync returns.

These IncludedProjections/ExcludedProjections only decide which projections are handled immediately — they are separate from, and have no effect on, the EventProcessing.IncludedProjections/ExcludedProjections above, which decide which projections the eventual IEventProcessor handles at all.

If a projection's handler throws while being invoked immediately, the event is not lost: EventProcessing.ImmediateConsistencyFailed is raised, and the eventual IEventProcessor will still pick up and retry the event on its next pass.

options.EventProcessing.ImmediateConsistencyFailed += (args, cancellationToken) =>
{
    _logger.LogWarning("Projection '{ProjectionName}' failed to handle event '{EventId}' immediately: {Exception}", args.ProjectionName, args.PrimitiveEvent.EventId, args.Exception);

    return Task.CompletedTask;
};

You can also request immediate consistency for a single SaveAsync call, regardless of whether Enabled is set — see Saving with Immediate Consistency.

EventStore Options

RecallOptions also exposes a top-level Operation (AsyncEvent<OperationEventArgs>), a general-purpose instrumentation hook raised around internal Shuttle.Recall operations, carrying an operation name and associated data.

There is currently no built-in compression or encryption of stored events — if you need either, apply it in a custom ISerializer implementation.

Usage

Saving an Event Stream

var eventStore = serviceProvider.GetRequiredService<IEventStore>();
var streamId = Guid.NewGuid();
var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream);

Saving with Headers

var stream = await eventStore.GetAsync(streamId, builder =>
{
    builder.AddHeader("key", "value");
});

stream.Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream);

Saving with Concurrency Check

var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

stream.ConcurrencyInvariant(5); // throws EventStreamConcurrencyException if version != 5

await eventStore.SaveAsync(stream);

Saving with Correlation ID

var stream = await eventStore.GetAsync(streamId);

stream
    .WithCorrelationId(correlationId)
    .Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream);

Saving with Immediate Consistency

var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

await eventStore.SaveAsync(stream, builder =>
{
    builder.WithImmediateConsistency();
});

This requests immediate consistency for this save only, even when EventProcessing.ImmediateConsistency.Enabled is false. Which projections actually run immediately is still governed by EventProcessing.ImmediateConsistency.IncludedProjections/ExcludedProjections — see ImmediateConsistency Options.

Retrieving an Event Stream

var stream = await eventStore.GetAsync(streamId);

// Apply committed events to an aggregate root or state object
stream.Apply(someAggregateRoot);

Apply invokes a matching event-handling method (On by default, see EventStoreOptions.EventHandlingMethodName) on the target object for each event. If no matching method is found for an event type, an UnhandledEventException is thrown.

Retrieving Events by Type

var stream = await eventStore.GetAsync(streamId);

// Get only appended (not-yet-committed) events -- this is the default
var appendedEvents = stream.GetEvents(EventStream.EventRegistrationType.Appended);

// Get only committed events
var committedEvents = stream.GetEvents(EventStream.EventRegistrationType.Committed);

// Get all events
var allEvents = stream.GetEvents(EventStream.EventRegistrationType.All);

Committing Events Before Apply

var stream = await eventStore.GetAsync(streamId);

stream.Add(new SomeEvent { Data = "example" });

// Events are only applied after commit
stream.Commit();

// Now Apply() will include the committed events
stream.Apply(someAggregateRoot);

Removing an Event Stream

var stream = await eventStore.GetAsync(streamId);

stream.Remove();

await eventStore.RemoveAsync(streamId);

Checking Whether a Stream Needs Saving

var stream = await eventStore.GetAsync(streamId);

if (stream.ShouldSave())
{
    await eventStore.SaveAsync(stream);
}

ShouldSave() returns true if the stream has any appended events that have not yet been committed/saved.

Add, Apply, Commit, ConcurrencyInvariant, Remove, and WithCorrelationId all return the EventStream itself, so calls may be chained, e.g. stream.WithCorrelationId(id).Add(eventA).Add(eventB).

Projections

Handler Implementation

Implement the IEventHandler<T> interface to handle events:

public class OrderProjection : IEventHandler<OrderPlaced>
{
    public async Task HandleAsync(IEventHandlerContext<OrderPlaced> context, CancellationToken cancellationToken = default)
    {
        var evt = context.Event;
        var projection = context.Projection;
        var primitiveEvent = context.PrimitiveEvent;

        // Process the event
        await SaveToReadModelAsync(evt.OrderId, evt.Amount, cancellationToken);

        // Optionally defer for retry
        // context.Defer(TimeSpan.FromSeconds(5));
    }
}

Registering Projections

services
    .AddRecall()
    .AddProjection("OrderProjection", projection =>
    {
        projection.AddEventHandler<OrderProjection>();
    });

Inline Projection Handlers

Delegate-based handlers must be async and return a Task; the event type is inferred from the single IEventHandlerContext<T> parameter:

services
    .AddRecall()
    .AddProjection("OrderProjection", projection =>
    {
        projection.AddEventHandler(async (IEventHandlerContext<OrderPlaced> context) =>
        {
            var evt = context.Event;
            // handle event inline
        });
    });

Delegate-based Handlers

services
    .AddRecall()
    .AddProjection("ProjectionName", async (IEventHandlerContext<SomeEvent> context) =>
    {
        // handle event
    });

Delegate handlers may declare additional parameters beyond the IEventHandlerContext<T>; these are resolved from the DI container for each invocation, which is useful for pulling in a scoped DbContext or similar without an explicit handler class:

.AddEventHandler(async (IEventHandlerContext<OrderPlaced> context, OrderDbContext dbContext) =>
{
    await dbContext.Orders.AddAsync(new OrderEntity { Id = context.PrimitiveEvent.Id });
    await dbContext.SaveChangesAsync();
});

AddProjection/AddEventHandler also have overloads accepting a handler Type (with an optional Func<Type, ServiceLifetime> to control its registered lifetime) or an existing handler instance — useful when you want to register a handler without a generic type parameter.

IEventProcessor Lifecycle

If at least one projection has been registered, AddRecall registers an IHostedService that automatically calls IEventProcessor.StartAsync/StopAsync as the host starts and stops — in a typical IHost/ASP.NET Core application you do not need to drive this manually.

If you are hosting Shuttle.Recall outside of the generic host (e.g. a plain console application), you can start and stop it yourself:

var processor = serviceProvider.GetRequiredService<IEventProcessor>();

await processor.StartAsync();

// ... application runs ...

await processor.StopAsync();

IEventProcessor also exposes a Started property, and implements IDisposable/IAsyncDisposable.

EventEnvelope Properties

The EventEnvelope class contains metadata about each event as it is persisted:

EventStream Properties

Exceptions

• EventStreamConcurrencyException: Thrown by EventStream.ConcurrencyInvariant (and by an IEventStore implementation on save) when concurrent modification is detected.
• EventProcessingException: Thrown during projection registration/event processing failures.
• UnhandledEventException: Thrown by EventStream.Apply when the target object has no matching event-handling method for an event in the stream.
• EventStreamException: General event-stream-related failures raised by an IEventStore implementation.
• AggregateConstructorException: Thrown when an aggregate root cannot be constructed as expected.
• ProcessEventMethodMissingException, DuplicateKeyException: Additional exception types used internally / by storage implementations.

Related Projects

• Shuttle.Recall.WebApi and Shuttle.Recall.Vue (in this repository): an authenticated, permissioned REST API (built on Shuttle.Access) and matching Vue 3/Vuetify admin UI for searching and pruning a SQL Server-backed event store — a reference example of an end-to-end deployment.
• Shuttle.Recall.SqlServer.Storage: SQL Server implementation of IEventStore.
• Shuttle.Recall.SqlServer.EventProcessing: SQL Server implementation of projection/event processing, including immediate consistency tracking.
• Shuttle.Recall.Testing: base fixtures for verifying IEventStore/IEventProcessor implementations.
• Shuttle.Recall.OpenTelemetry: OpenTelemetry metrics and tracing for Recall-domain events.
• Shuttle.Recall.Samples: sample applications demonstrating event sourcing and projections.

Documentation

Please visit the Shuttle.Recall documentation for more information.