M.EventBrokerSlim.PersistentEvents.PostgreSql 1.0.0

M.EventBrokerSlim.PersistentEvents.PostgreSql

PostgreSQL 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
• PostgreSQL server
• M.EventBrokerSlim (pulled automatically as a dependency)

Installation

dotnet add package M.EventBrokerSlim.PersistentEvents.PostgreSql

Database Setup

The package requires a schema, table, sequence, and indexes in your PostgreSQL database. There are two ways to set them up:

Option 1: Programmatic (development / simple deployments)

Call CreateEventsTable() on a DatabaseSettings instance:

var databaseSettings = new DatabaseSettings
{
    ConnectionString = "Host=localhost;Database=mydb;Username=myuser;Password=mypassword",
    Schema = "ebs_0"
};

databaseSettings.CreateEventsTable();

The call is idempotent - it uses CREATE ... IF NOT EXISTS and is safe to run on every startup in development.

Option 2: Manual SQL script

Apply the initialize_db.sql script included in the package source. The default schema name is ebs_0 - replace it with a unique name if running multiple event broker instances against 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

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 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 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 PostgreSQL persistence

serviceCollection.AddEventBroker(x => x
    .WithMaxConcurrentHandlers(3)
    .WithPostgreSqlPersistence((db, settings) =>
    {
        db.ConnectionString = "Host=localhost;Database=mydb;Username=myuser;Password=mypassword";
        db.Schema = "ebs_0";
        settings.PollingInterval = TimeSpan.FromSeconds(10);
        settings.ProcessingTimeout = TimeSpan.FromMinutes(5);
    }));

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

DatabaseSettings

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 PostgreSQL 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.

Database Schema

The package creates the following schema (default name ebs_0):

Events Table

Indexes

License

MIT