M.EventBrokerSlim.PersistentEvents.MongoDb 1.0.0

M.EventBrokerSlim.PersistentEvents.MongoDb

MongoDB storage backend for EventBrokerSlim persistent events - durable, at-least-once event delivery that survives process restarts, with support for horizontal scale-out.

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

Prerequisites

• .NET 8.0 or later
• M.EventBrokerSlim (pulled automatically as a dependency)
• A running MongoDB server (4.2 or later)

Installation

dotnet add package M.EventBrokerSlim.PersistentEvents.MongoDb

Database Setup

No manual setup is required. On first use, the storage creates the configured collection and ensures the necessary indexes. The default collection name is events - use a unique collection name per event broker instance when sharing the same database.

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

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)
        {
            retryPolicy.RetryAfter(TimeSpan.FromSeconds(5));
        }
    })
    .Execute(static async (OrderPlaced @event, ISomeService service, CancellationToken ct) =>
    {
        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 the database - 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 participate in fan-out:

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

Or using the options API:

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

Alternative: delegate handler

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

4. Configure the event broker with MongoDB persistence

serviceCollection.AddEventBroker(x => x
    .WithMaxConcurrentHandlers(3)
    .WithMongoDbPersistence((db, settings) =>
    {
        db.ConnectionString = "mongodb://localhost:27017";
        db.DatabaseName = "myapp";
        db.CollectionName = "events";
        settings.PollingInterval = TimeSpan.FromSeconds(10);
        settings.ProcessingTimeout = TimeSpan.FromMinutes(5);
    }));

Alternative: supply an existing IMongoDatabase instance

When the host application already manages a MongoClient or IMongoDatabase (e.g. registered via AddMongoClient), pass the database directly. ConnectionString and DatabaseName are ignored when an instance is provided:

IMongoDatabase mongoDatabase = mongoClient.GetDatabase("myapp");

serviceCollection.AddEventBroker(x => x
    .WithMongoDbPersistence((db, settings) =>
    {
        db.MongoDatabase = mongoDatabase;
        db.CollectionName = "events";
    }));

5. Start the persistent event broker

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

6. Publish events

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

Configuration Reference

DatabaseSettings

PersistentEventBrokerSettings

Important Considerations

Multi-instance horizontal scale-out. MongoDB is a server-based database suitable for multi-instance deployments. The claiming mechanism uses findOneAndUpdate with optimistic concurrency to ensure at most one instance processes a given record under normal operation.

MongoDB 4.2 or later required. The maintenance loop (RescheduleClaimedExceedingProcessingTimeoutAsync) uses aggregation pipeline updates introduced in MongoDB 4.2.

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. 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 handler 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 MongoDB is not atomic with the caller's own database transaction.

Database Schema

The package stores event documents in a single MongoDB collection (default name events). The collection and its indexes are created automatically on first use.

Event Document

Indexes

License

MIT