M.EventBrokerSlim.PersistentEvents.Redis 1.0.0

M.EventBrokerSlim.PersistentEvents.Redis

Redis storage backend for EventBrokerSlim persistent events - durable, at-least-once event delivery that survives process restarts.

For the design rationale behind persistent events, see the architecture document and ADRs.

Prerequisites

• .NET 8.0 or later
• Redis server
• M.EventBrokerSlim (pulled automatically as a dependency)

Installation

dotnet add package M.EventBrokerSlim.PersistentEvents.Redis

Quick Start

1. Define events and handlers

public record OrderPlaced(string OrderId, decimal Amount);

public class OrderPlacedHandler : IEventHandler<OrderPlaced>
{
    public async Task Handle(OrderPlaced @event, IRetryPolicy retryPolicy, CancellationToken cancellationToken)
    {
        // process the event - must be idempotent
    }

    public async Task OnError(Exception exception, OrderPlaced @event, IRetryPolicy retryPolicy, CancellationToken cancellationToken)
    {
        // optionally retry
        retryPolicy.RetryAfter(TimeSpan.FromSeconds(5));
    }
}

Alternative: delegate handler

Instead of a class, define the handler as a delegate pipeline:

public record OrderPlaced(string OrderId, decimal Amount);

IPipeline pipeline = PipelineBuilder.Create()
    .NewPipeline()
    .Execute(static async (IRetryPolicy retryPolicy, INext next) =>
    {
        try
        {
            await next.RunAsync();
        }
        catch(Exception exception)
        {
            // optionally retry
            retryPolicy.RetryAfter(TimeSpan.FromSeconds(5));
        }
    })
    .Execute(static async (OrderPlaced @event, ISomeService service, CancellationToken ct) =>
    {
        // process the event - must be idempotent
        await service.ProcessOrder(@event, ct);
    })
    .Build()
    .Pipelines[0];

2. Register the event registry

Map each event type to a stable string name. The name is stored in Redis - it must not change between deployments:

var eventRegistry = new EventRegistry()
    .Add<OrderPlaced>("OrderPlaced");

serviceCollection.AddSingleton(eventRegistry);

3. Register handlers with persistent names

Only handlers registered with a handlerName have their names included in fan-out - when an event is published, one storage record is created per registered handler name:

serviceCollection.AddTransientEventHandler<OrderPlaced, OrderPlacedHandler>(
    handlerName: "OrderPlacedHandler");

Or using the options API:

serviceCollection.AddTransientEventHandler<OrderPlaced, OrderPlacedHandler>(o => o
    .WithHandlerName("OrderPlacedHandler"));

Alternative: delegate handler

Register the delegate pipeline with a persistent name:

serviceCollection.AddEventHandlerPipeline<OrderPlaced>(pipeline,
    handlerName: "OrderPlacedHandler");

Or using the options API:

serviceCollection.AddEventHandlerPipeline<OrderPlaced>(pipeline, o => o
    .WithHandlerName("OrderPlacedHandler"));

4. Configure the event broker with Redis persistence

serviceCollection.AddEventBroker(x => x
    .WithMaxConcurrentHandlers(3)
    .WithRedisPersistence((redis, settings) =>
    {
        redis.ConnectionString = "localhost:6379";
        redis.KeyPrefix = "ebs_0";
        settings.PollingInterval = TimeSpan.FromSeconds(10);
        settings.ProcessingTimeout = TimeSpan.FromMinutes(5);
    }));

Using a registered IConnectionMultiplexer

If you already have a IConnectionMultiplexer registered in the DI container, you can reuse it instead of providing a connection string:

serviceCollection.AddEventBroker(x => x
    .WithMaxConcurrentHandlers(3)
    .WithRedisPersistence((redis, settings) =>
    {
        redis.UseRegisteredMultiplexer = true;
        redis.KeyPrefix = "ebs_0";
    }));

5. Start the persistent event broker

var serviceProvider = serviceCollection.BuildServiceProvider();
serviceProvider.UsePersistentEventBroker(throwOnValidationErrors: true);

On startup, validation checks that:

• Every handler with a handlerName has its event type registered in EventRegistry
• Every event in EventRegistry has at least one named handler (including NullPipeline registrations)

Set throwOnValidationErrors: true for strict mode (default logs warnings).

6. Publish events

IEventBroker is registered in the DI container and can be injected where needed:

await eventBroker.Publish(new OrderPlaced("order-123", 49.99m));

Configuration Reference

RedisSettings

PersistentEventBrokerSettings

Important Considerations

At-least-once delivery. A crash after claiming a record but before completing it may cause duplicate processing. Handlers must be idempotent.

Escaped exceptions are dead-lettered. If an exception escapes the handler pipeline unhandled, the record is immediately dead-lettered - IRetryPolicy is not consulted. Handle exceptions inside the pipeline to use retries.

Name stability. Changing a handlerName or an EventRegistry name is a breaking change - in-flight records under the old name will never be claimed. Treat name changes as migrations.

Serialization. Events are serialized using System.Text.Json with camelCase property naming, no indentation, and null values omitted. Event types must be serializable under these settings.

Not event sourcing. The store is a delivery mechanism, not an event log. Completed records are deleted according to CompletedRecordTtl.

Not a transactional outbox. The event write to Redis is not atomic with the caller's own database transaction.

Dead-letter monitoring. Records land in a dead-letter state when the retry policy is exhausted, when a handler abandons the event, or when an exception escapes unhandled. Dead-lettered records are not retried automatically - monitoring and tooling for inspection and requeue is necessary for production use.

Redis Data Model

All data is stored under the configured KeyPrefix (default ebs_0). The {KeyPrefix} portion is wrapped in braces in the actual keys to enable Redis hash-tag-based slot routing for cluster compatibility.

Event Records

Each event record is stored as a Redis Hash at key {KeyPrefix}:evt:{id} where id is a GUID.

Sorted Set Indexes

Four sorted sets partition records by status, scored by timestamp for efficient range queries:

Atomicity

All state transitions (schedule, claim, complete, retry, dead-letter, timeout reschedule, cleanup) are executed as atomic Lua scripts, ensuring consistency even under concurrent access.

License

MIT