An in-memory event store, running as a service, to synchronize and validate operations across multiple instances with high concurrency and no locks.
Important: Prefer zero-allocation APIs introduced in vNext. Legacy Query-based APIs are kept for compatibility and marked as [Obsolete], delegating to zero-alloc implementations.
🚀 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.
📌 Typical scenario
Two (or more) gateways behind a load balancer need to record operations to the same stream. Lockfree.EventStore ensures order and integrity even under high parallelism, without relying on locks, keeping all state in memory.
📚 Complete documentation
Below is the full technical documentation of the API, advanced features, benchmarks, and usage examples.
Key Features
Lock-free MPMC writes with FIFO discard
Key-based partitioning for high concurrency
Consistent snapshots without blocking producers
Functional aggregations and time-window queries
Zero external dependencies, ready for AOT/Trimming
Fluent API for advanced configuration
Built-in metrics and observability
Specialized aggregations (Sum, Average, Min, Max)
Basic Usage Example
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));
New Constructors
// 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();
State Properties
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
Specialized Aggregations
// 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
);
Note: Time-filtered queries require EnableWindowTracking = true. When disabled, a clear InvalidOperationException is thrown: "Window tracking is disabled. EnableWindowTracking must be true to use window queries."
Snapshots with Filters
// Filtered snapshot
var eventosRecentes = store.Snapshot(
filter: evt => evt.Timestamp > DateTime.UtcNow.AddMinutes(-5)
);
// Snapshot by time window
var snapshot = store.Snapshot(from: inicio, to: fim);
// Snapshot with filter and time window
var filtrado = store.Snapshot(
filter: evt => evt.Valor > 100,
from: inicio,
to: fim
);
Cleanup and Maintenance
// Clear all events
store.Clear();
store.Reset(); // Alias for Clear()
// Purge old events (requires TimestampSelector)
store.Purge(olderThan: DateTime.UtcNow.AddHours(-1));
Metrics and Observability
// 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
Examples
MetricsDashboard
Fully featured web API for collecting and querying real-time metrics:
cd .\samples\MetricsDashboard
dotnet run
Available endpoints:
POST /metrics - Add metric
GET /metrics/sum?label=cpu_usage - Sum values by label
GET /metrics/top?k=5 - Top K metrics
See samples/MetricsDashboard/TESTING.md for a complete testing guide.
