pkg

daniloneto/LockFree.EventStorev1.1.0

.NET 9.0

Event store em memória lock-free, genérico, com capacidade fixa, particionado e agregações funcionais.

License

MIT

Deps

2

Install Size

—

Vulns

✓ 0

Published

Sep 10, 2025

Get Started

$ dotnet add package LockFree.EventStore

Readme

LockFree.EventStore

An in-memory event store, running as a service, to synchronize and validate operations across multiple instances with high concurrency and no locks.

🚀 Get started in 3 steps

1. Run the server

docker run --rm -p 7070:7070 daniloneto/lockfree-eventstore:latest

2. Add the client

dotnet add package LockFree.EventStore

3. Write and read

var es = new EventStoreClient("http://localhost:7070");
await es.Append("gateway/orders", new OrderCreated { Id = "o-1", Valor = 123 });
await foreach (var ev in es.Read("gateway/orders", from: 0))
{
    /* handle event */
}

🔁 Client Sample

See samples/ClientSample for an example that:

Sends events in parallel to gateway/orders
Reads the events back
Computes local aggregations

To run:

docker run --rm -p 7070:7070 daniloneto/lockfree-eventstore:latest
cd samples/ClientSample
 dotnet run

🌐 Example with multiple Gateways (docker-compose)

Start 1 EventStore, 3 gateways and Nginx load balancing:

docker compose up --build

Test sending orders (load balanced across gateways):

curl -X POST http://localhost:8080/orders
curl -X POST 'http://localhost:8080/orders/bulk?n=50'

View statistics:

curl http://localhost:8080/stats/local    # stats of one gateway (one of the 3)
curl http://localhost:8080/stats/global   # global consolidation (via central reader)

💡 Why use it

Real concurrency: multiple writers without mutexes.
Guaranteed integrity: consistent ordering, conditional append, and idempotency.
Simple operation: no external coordination, no dependencies.

var store = new EventStore<Pedido>();
store.TryAppend(new Pedido { Id = 1, Valor = 10m, Timestamp = DateTime.UtcNow });

var total = store.Aggregate(() => 0m, (acc, e) => acc + e.Valor,
    from: DateTime.UtcNow.AddMinutes(-10));

// Explicit capacity
var store = new EventStore<Pedido>(capacity: 100_000);

// Capacity and partitions
var store = new EventStore<Pedido>(capacity: 50_000, partitions: 8);

// Advanced configuration
var store = new EventStore<Pedido>(new EventStoreOptions<Pedido>
{
    Capacity = 100_000,
    Partitions = 16,
    OnEventDiscarded = evt => Logger.LogTrace("Event discarded: {Event}", evt),
    OnCapacityReached = () => Metrics.IncrementCounter("eventstore.capacity_reached"),
    TimestampSelector = new PedidoTimestampSelector(),
    // RFC 002: disable window tracking for pure append/snapshot workloads
    EnableWindowTracking = false
});

// Fluent API
var store = new EventStoreBuilder<Pedido>()
    .WithCapacity(100_000)
    .WithPartitions(8)
    .OnDiscarded(evt => Log(evt))
    .OnCapacityReached(() => NotifyAdmin())
    .WithTimestampSelector(new PedidoTimestampSelector())
    // RFC 002: opt-out when window queries are not used
    .WithEnableWindowTracking(false)
    .Create();

store.Count          // Current number of events
store.Capacity       // Maximum configured capacity
store.IsEmpty        // Whether it's empty
store.IsFull         // Whether it reached maximum capacity
store.Partitions     // Number of partitions

// Count by time window
var count = store.Count(from: inicio, to: fim);

// Sum of values
var sum = store.Sum(evt => evt.Valor, from: inicio, to: fim);

// Average
var avg = store.Average(evt => evt.Valor, from: inicio, to: fim);

// Min and max
var min = store.Min(evt => evt.Pontuacao, from: inicio, to: fim);
var max = store.Max(evt => evt.Pontuacao, from: inicio, to: fim);

// With filters
var filteredSum = store.Sum(
    evt => evt.Valor, 
    filter: evt => evt.Tipo == "Pagamento",
    from: inicio, 
    to: fim
);

var store = new EventStore<Event>(new EventStoreOptions<Event>
{
    // existing core options
    CapacityPerPartition = 1_000_000,
    Partitions = Environment.ProcessorCount,
    EnableFalseSharingProtection = true
});

// Configure snapshot subsystem (one-time)
var snapshotter = store.ConfigureSnapshots(
    new SnapshotOptions
    {
        Enabled = true,
        Interval = TimeSpan.FromMinutes(2),            // Time-based trigger (can be combined with MinEventsBetweenSnapshots)
        MinEventsBetweenSnapshots = 50_000,            // Event-count trigger
        MaxConcurrentSnapshotJobs = 2,                 // Limit concurrent save jobs
        SnapshotsToKeep = 3,                           // Pruning window
        MaxSaveAttempts = 5,                           // Retry attempts for transient errors
        BackoffBaseDelay = TimeSpan.FromMilliseconds(100),
        BackoffFactor = 2.0,
        CompactBeforeSnapshot = true,                  // (Future compaction hook)
        EnableLocalTracing = false                     // Enables ActivitySource if true
    },
    serializer: new BinarySnapshotSerializer(),
    store: new FileSystemSnapshotStore("snapshots")
);

await store.RestoreFromSnapshotsAsync(); // Rebuild in-memory state before serving traffic

nextDelay = baseDelay * factor^(attempt-1) + small jitter

dotnet run --project samples/SnapshotSensors/SnapshotSensors.csproj

[BOOT] Partitions restauradas de snapshot: X

dotnet run --project samples/SnapshotSensorsApi/SnapshotSensorsApi.csproj

curl -X POST http://localhost:5000/sensor \
  -H "Content-Type: application/json" \
  -d '{"deviceId":"dev-1","temperature":22.5,"humidity":48.2}'

curl http://localhost:5000/state
curl http://localhost:5000/metrics

// Clear all events
store.Clear();
store.Reset(); // Alias for Clear()

// Purge old events (requires TimestampSelector)
store.Purge(olderThan: DateTime.UtcNow.AddHours(-1));

// Detailed statistics
store.Statistics.TotalAppended        // Total appended events
store.Statistics.TotalDiscarded       // Total discarded events
store.Statistics.AppendsPerSecond     // Current append rate
store.Statistics.LastAppendTime       // Timestamp of the last append

cd .\samples\MetricsDashboard
 dotnet run