Spirantis.VectorClock.Redis 10.0.0

Spirantis.VectorClock

Causal consistency for eventually-consistent read models. When a write side produces changes and a separate read side (a projection) catches up asynchronously — the classic CQRS / event-sourcing split — a caller that just made a write can read back stale data. Spirantis.VectorClock gates those reads against per-entity version vectors, retrying until the read is at least as fresh as the writes already observed (read-your-writes / monotonic reads), and carries that causal context across services with a portable consistency token.

It generalizes a scalar high-water-mark mechanism proven in production at scale to genuine version vectors — so beyond read-your-writes it also provides multi-writer conflict detection and resolution: optimistic concurrency over dotted version vectors, with value-bearing sibling merge. The clock store is pluggable — in-process, Redis, or your own event log. See docs/DESIGN.md for the model and the capability ladder.

Targets .NET 10.

Packages

Concepts

Quick start

Register streams and wire the engine (DI):

using Spirantis.VectorClock;
using Spirantis.VectorClock.Abstraction;

public sealed class Order
{
    public string OrderId { get; set; } = "";
    public decimal Total { get; set; }
    public VersionVector Clock { get; set; } = VersionVector.Empty;
}

services.AddVectorClock(vc =>
{
    vc.Resource<Order>()
      .Stream("order", o => o.Clock, (o, clock) => o.Clock = clock, o => o.OrderId);
    vc.UseActor("command-svc-1");      // a stable writer identity
});

Write side — advance the version, then emit the event:

public sealed class OrderCommands(IVectorStore vectors, IEventBus events)
{
    public async Task RaiseTotal(Order order, decimal total, CancellationToken ct)
    {
        order.Total = total;
        await vectors.IncreaseVectorsAsync(order, cancellationToken: ct); // stamps order.Clock, raises the mark
        await events.PublishAsync(new OrderUpdated(order), ct);
    }
}

Read side — gate the projection read until it has caught up:

public sealed class OrderQueries(IVectorStore vectors, IOrderProjection projection)
{
    public Task<Order?> GetFresh(string orderId, CancellationToken ct) =>
        vectors.RetrieveValidAsync<Order>(
            async token =>
            {
                var order = await projection.GetAsync(orderId, token);
                return order is null
                    ? VectoredResult<Order, Order>.None(null!)   // nothing to gate
                    : Vectored.Resource(order);                  // gate this order's clock
            },
            cancellationToken: ct);
}

RetrieveValidAsync calls the retriever, and if the returned Order.Clock does not yet descend the recorded high-water mark, retries with backoff until it does (or the budget is spent). A built engine is reusable and thread-safe for reads.

Validation modes

A retriever returns a VectoredResult describing how to gate what it fetched:

• Vectored.Resource(resource) — gate one resource's clock.
• Vectored.List(result, resources) — gate every item; one stale item fails the batch.
• VectoredResult<…>.ForOverrides(result, VectorOverride.Absent("order", orderId)) — the negative-read case: there is no resource, but you assert a (stream, scope) high-water mark directly, so an empty result that should not be empty ("I just created this") is treated as stale and retried.
• VectoredResult<…>.None(result) — return without gating.

Streams and scopes

A resource can carry several streams at different scopes — there is nothing special about one:

vc.Resource<Order>()
  .Stream("order",    o => o.OrderClock,    (o, c) => o.OrderClock = c,    o => o.OrderId)     // per entity
  .Stream("customer", o => o.CustomerClock, (o, c) => o.CustomerClock = c, o => o.CustomerId); // per customer (group)

Each read or write names the subset of streams it cares about (streams: parameter), defaulting to all. A group-scoped stream lets a read gate on "anything happened for this customer" even when no specific order is in hand.

Attribute alternative

The same registrations can be declared on the type and scanned — both front-ends feed one registry:

using Spirantis.VectorClock.DataAnnotations;

[VectoredResource]
public sealed class Order
{
    public string OrderId { get; set; } = "";

    [VectorProperty(nameof(OrderId))]
    public VersionVector Clock { get; set; } = VersionVector.Empty;
}

// ...
vc.ScanAssembly(typeof(Order).Assembly);

[VectorProperty] can scope by a substring of the key property (e.g. a parent id embedded as a prefix of the entity id): [VectorProperty(nameof(Self), startIndex, length)].

Multi-master: conflict detection & resolution

When more than one writer can update the same entity concurrently, gating reads is not enough — two writers must be able to tell they diverged. IVersionedStore commits versions through the clock store's atomic compare-and-set, keeping a DottedVersionVectorSet per entity: two writers from the same base who never saw each other are retained as siblings rather than one silently overwriting the other.

services.AddVersionedStore();   // in-memory by default; reuses the IActorId from AddVectorClock

// Each writer commits with the causal context it observed.
var result = await versions.AppendAsync(key, context: observed, ct);
if (result.IsConflicted)
{
    // result.Versions holds the concurrent siblings.
}

To carry the values and reconcile them, use the value-bearing engine and a domain merge:

services.AddVersionedStore<Cart>();

await a.AppendAsync(key, context, cartFromA);     // concurrent writers...
await b.AppendAsync(key, context, cartFromB);     // ...diverge into two siblings

// Merge the sibling values into one superseding version.
var resolved = await versions.ResolveAsync(key, siblings => Cart.Merge(siblings));

Resolve writes the merged value with a context that has seen every sibling, collapsing the conflict; a lost compare-and-set race re-merges over the freshly-read siblings. Detection works on the in-memory and Redis stores and over an event log — anywhere IClockStore is backed.

Cross-service read-your-writes

A consistency token lets one service's writes enforce freshness when read back through another. Capture it after a write and carry it (e.g. as an HTTP header); use it as the floor of a later read so results must also descend the clocks it carries — even against a cold store:

// Producer service
await vectors.IncreaseVectorsAsync(order);
string header = vectors.Capture(order).Encode();    // hand this to the caller

// Any reader service
var floor = ConsistencyToken.Decode(header);
var fresh = await vectors.RetrieveValidAsync<Order>(
    token => RetrieveFromProjection(orderId, token),
    floor: floor);

ASP.NET Core

Spirantis.VectorClock.AspNetCore carries the token on an HTTP header automatically — middleware reads it from the incoming request into a per-request IConsistencyContext and writes the accumulated token back on the response:

services.AddConsistencyTokenPropagation();   // header defaults to X-Consistency-Token
app.UseConsistencyToken();

// In a handler (IConsistencyContext injected): gate against the caller's context,
// and fold in what you observe so the response carries it onward.
var fresh = await vectors.RetrieveValidAsync<Order>(
    token => RetrieveFromProjection(orderId, token),
    floor: consistency.Token);
consistency.Observe(vectors.Capture(fresh!));

Distributed storage (Redis)

The default store is in-process. For multiple instances — or to survive partition rebalancing, where a reassigned partition would otherwise start blind and accept stale reads — use the Redis store. It keeps each clock in a hash and performs the monotonic compare-and-set atomically via a Lua script, so concurrent writers never lose a raise:

using StackExchange.Redis;

var redis = await ConnectionMultiplexer.ConnectAsync("localhost:6379");
services.AddRedisClockStore(redis, o => o.Expiry = TimeSpan.FromHours(4)); // optional sliding expiry
services.AddVectorClock(vc => { /* ... */ });

AddRedisClockStore overrides the in-memory default regardless of call order. The generic AddRedisClockStore<TClock>(redis, codec) backs the conflict-detection stores too — pass DottedVersionVectorSetCodec.Default for metadata sibling sets, or new DottedVersionVectorSetCodec<TValue>(valueCodec) for value-bearing ones. Any backend with an atomic conditional-update primitive can implement IClockStore<TClock>.

Event-log backend

In an event-sourced deployment the event log is already the ordering authority. Implement IEventStreamStore over it (an expected-version append — the idiom of EventStoreDB, a SQL version column, Cosmos, …) and adapt it with EventStreamClockStore<T>; the stream version becomes the concurrency token, so every store above — read-gating high-water marks and the OCC version sets — commits through the log with no separate clock store:

IClockStore<VersionVector> marks =
    new EventStreamClockStore<VersionVector>(myLog, VersionVectorCodec.Default);

IClockStore<DottedVersionVectorSet> versions =
    new EventStreamClockStore<DottedVersionVectorSet>(myLog, DottedVersionVectorSetCodec.Default);

The clock algebra

The contracts and clocks are usable standalone:

var a = VersionVector.Empty.Increment("node-1");
var b = a.Increment("node-2");

b.Compare(a);     // After
a.Descends(b);    // false  (a has not seen node-2's event)
a.Merge(b);       // the join: {node-1:1, node-2:1}

• VersionVector — the map {actor → counter}; element-wise partial order, Concurrent is a first-class outcome. The single-actor case is a plain monotonic counter.
• DottedVersionVector — a single version's clock (a unique event dot kept separate from its causal context), so two writes coordinated by the same actor are correctly detected as concurrent rather than ordered. Used by the multi-writer conflict-detection path.

Both implement ILogicalClock<TSelf> (Compare / Descends / Increment); VersionVector also implements IMergeableClock<TSelf> (Merge).

Repository layout

src/
  Spirantis.VectorClock.slnx               # solution
  Directory.Build.props                    # shared build settings (versioning, CSharpier)
  Spirantis.VectorClock.Abstraction/       # algebra + contracts (NuGet package)
  Spirantis.VectorClock/                   # engine, registry, builder, DI (NuGet package)
  Spirantis.VectorClock.DataAnnotations/   # attribute front-end (NuGet package)
  Spirantis.VectorClock.Redis/             # Redis clock store (NuGet package)
  Spirantis.VectorClock.AspNetCore/        # consistency-token HTTP propagation (NuGet package)
  Spirantis.VectorClock.Test/              # xUnit v3 tests
  Spirantis.VectorClock.Benchmarks/        # BenchmarkDotNet (performance)
docs/
  DESIGN.md                                # architecture, capability ladder, roadmap

Build & test

dotnet build src/Spirantis.VectorClock.slnx
dotnet test  src/Spirantis.VectorClock.slnx

The Redis integration tests skip automatically unless a server is reachable at localhost:6379 (or REDIS_CONNECTION):

docker run -d --rm -p 6379:6379 redis:7-alpine
dotnet test src/Spirantis.VectorClock.slnx

Formatting is enforced at build time by CSharpier (CSharpier_Check). To format locally:

dotnet tool restore
dotnet csharpier format src

Benchmarks

Spirantis.VectorClock.Benchmarks (BenchmarkDotNet) measures the hot paths — run in Release:

dotnet run -c Release --project src/Spirantis.VectorClock.Benchmarks -- --filter '*'
# a subset, or a quick noisy pass while iterating:
dotnet run -c Release --project src/Spirantis.VectorClock.Benchmarks -- --filter '*ClockBenchmarks*'
dotnet run -c Release --project src/Spirantis.VectorClock.Benchmarks -- --job short

Results are written to BenchmarkDotNet.Artifacts/.

Status

The full capability ladder is implemented and tested: read-your-writes gating, the write side, multi-writer conflict detection (optimistic concurrency over dotted version vectors) and value-bearing resolution, the consistency token (in-process and over HTTP), and three interchangeable storage backends — in-memory, Redis, and an event log. See docs/DESIGN.md for the architecture and the capability ladder.

License

MIT