ZeroAlloc.EventSourcing.PostgreSql

ZeroAlloc.EventSourcing

A high-performance, zero-allocation event sourcing library for .NET with streaming capabilities and production-grade reliability features.

Key Features

• Zero-Allocation Design: Optimized for performance-critical applications with minimal garbage collection
• Event Sourcing: Full event sourcing support with append-only event store
• Stream Consumers: Production-grade consumers for reliable event consumption
• Projections: Multiple projection types for denormalized views
• Snapshots: Optimize aggregate loading with configurable snapshot strategies
• SQL Adapters: SQL Server and PostgreSQL support with Testcontainers testing
• Checkpoint Tracking: Automatic position tracking with recovery capabilities
• Comprehensive Testing: Extensive test suite and integration testing patterns

Quick Start

Installation

dotnet add package ZeroAlloc.EventSourcing

Basic Event Sourcing

var adapter = new SqlEventStoreAdapter(connectionString);
var serializer = new JsonEventSerializer();
var typeRegistry = new EventTypeRegistry();

var eventStore = new EventStore(adapter, serializer, typeRegistry);

// Append events
var streamId = new StreamId("order-123");
await eventStore.AppendAsync(
    streamId,
    new[] { (object)new OrderPlacedEvent { Id = "123", Amount = 100m } },
    StreamPosition.Start
);

// Read events
var envelope = await eventStore.ReadAsync(streamId);
foreach (var evt in envelope.Events)
{
    Console.WriteLine($"Event: {evt.Event}");
}

Packages

All packages follow zero-allocation principles and are optimized for high-throughput scenarios.

Stream Consumers

ZeroAlloc.EventSourcing includes production-grade stream consumers for reliable event consumption with automatic position tracking, retry logic, and configurable error handling.

Key Features

• Position Tracking: Resume consumption from exact point after restart
• Batch Processing: Configurable batch sizes (1-10,000 events) for optimal throughput
• Retry Logic: Exponential backoff with configurable max retries
• Error Handling: FailFast, Skip, or DeadLetter strategies
• Commit Strategies: AfterEvent, AfterBatch, or Manual control
• Production Ready: SQL checkpoint store with atomic upsert, tested with Testcontainers

Quick Start

var consumer = new StreamConsumer(eventStore, checkpointStore, "my-consumer");
await consumer.ConsumeAsync(async (envelope, ct) => {
    // Process event
    Console.WriteLine(envelope.Event);
});

See Stream Consumers Documentation for complete guide.

Kafka Integration

Consume events directly from Kafka topics with the same reliability features:

var options = new KafkaConsumerOptions
{
    BootstrapServers = "localhost:9092",
    Topic = "my-events",
    GroupId = "my-service"
};

var consumer = new KafkaStreamConsumer(options, checkpointStore, serializer, registry);
await consumer.ConsumeAsync(async (envelope, ct) => {
    // Process event from Kafka
    await handler.ProcessAsync(envelope, ct);
});

See Kafka Consumer Documentation for complete guide.

Projections

Build denormalized views of your event data with multiple projection types:

var projection = new Projection(eventStore, "order-summaries");
await projection.ProjectAsync(async (envelope, state, ct) => {
    if (envelope.Event is OrderPlacedEvent ope)
    {
        state["total"] = (decimal)(state["total"] ?? 0m) + ope.Amount;
    }
    return state;
});

Snapshots

Optimize aggregate loading with snapshots:

var options = new SnapshotOptions
{
    Strategy = SnapshotStrategy.EveryNEvents(100)
};

var snapshot = await eventStore.GetSnapshotAsync(streamId, options);

OpenTelemetry Instrumentation

ZeroAlloc.EventSourcing.Telemetry adds a hand-rolled decorator around IAggregateRepository<TAggregate, TId> that records Activity spans and metrics for every aggregate LoadAsync and SaveAsync — without taking a dependency on the OTel SDK.

dotnet add package ZeroAlloc.EventSourcing.Telemetry

services
    .AddEventSourcing()
    .UseInMemoryEventStore()
    .AddAggregate<OrderAggregate, Guid>()
    .WithTelemetry();   // call after each aggregate registration, before Build()

The decorator emits, under the ZeroAlloc.EventSourcing activity-source/meter name:

• Spans — aggregate.load and aggregate.save, tagged with aggregate.type (= typeof(TAggregate).Name); status set to Error on exception
• Counters — aggregate.loads_total and aggregate.saves_total, incremented only when Result.IsSuccess
• Histograms — aggregate.load_duration_ms and aggregate.save_duration_ms, recorded for both success and failure paths

Any OpenTelemetry SDK wired to the process picks up all three instruments automatically.

Breaking change in v2.0: WithTelemetry() now decorates IAggregateRepository<,> instead of IEventStore. The old event_store.append / event_store.read / event_store.subscribe spans no longer exist; the deleted InstrumentedEventStore and its [Instrument] attribute on IEventStore are gone. Existing dashboards must be re-pointed at aggregate.load / aggregate.save. See docs/telemetry.md for the full migration table. The legacy UseEventSourcingTelemetry() extension remains as [Obsolete] and now delegates to WithTelemetry().

Documentation

Complete documentation available at /docs:

• Getting Started
• Core Concepts
• Usage Guides
• Testing Strategies
• Performance & Benchmarks
• Advanced Topics

Development

Build

dotnet build ZeroAlloc.EventSourcing.slnx --configuration Release

Tests

dotnet test ZeroAlloc.EventSourcing.slnx --configuration Release

Benchmarks

dotnet run --project benchmarks/ZeroAlloc.EventSourcing.Benchmarks -c Release

License

See LICENSE file for details.

Contributing

See CONTRIBUTING.md for guidelines.