Ama.CRDT 1.0.250915-ci1717

Ama.CRDT

A .NET library for achieving eventual consistency in distributed systems using Conflict-free Replicated Data Types (CRDTs). It provides a simple, high-level API to compare, patch, and merge POCOs (Plain Old C# Objects), with merge behavior controlled by attributes.

Features

• Attribute-Driven Strategies: Define conflict resolution logic directly on your POCO properties using a rich set of attributes like [CrdtLwwStrategy], [CrdtCounterStrategy], [CrdtOrSetStrategy], and [CrdtStateMachineStrategy].
• POCO-First: Work directly with your C# objects. The library handles recursive diffing and patching seamlessly.
• Clean Data/Metadata Separation: Keeps your data models pure by storing CRDT state (timestamps, version vectors) in a parallel CrdtMetadata object.
• Extensible: Easily create and register your own custom CRDT strategies, element comparers, and timestamp providers.
• Multi-Replica Support: Designed for scenarios with multiple concurrent writers, using a factory pattern to create services for each unique replica.
• Dependency Injection Friendly: All services are designed to be registered and resolved through a standard DI container.

Installation

You can install Ama.CRDT via the .NET CLI or the NuGet Package Manager in Visual Studio.

.NET CLI

dotnet add package Ama.CRDT

NuGet Package Manager

In Visual Studio, open the NuGet Package Manager Console and run:

Install-Package Ama.CRDT

Or, you can search for Ama.CRDT in the NuGet Package Manager UI and install it from there.

Getting Started

1. Setup

In your Program.cs or service configuration file, register the CRDT services.

using Ama.CRDT.Extensions;

var builder = WebApplication.CreateBuilder(args);

// Add CRDT services to the DI container.
// This registers all core services, including strategies, providers,
// and the ICrdtScopeFactory for creating replica-specific scopes.
builder.Services.AddCrdt();

// ... other service registrations

var app = builder.Build();

2. Define Your Model

Decorate your POCO properties with the desired CRDT strategy attributes. These attributes tell the library how to handle concurrent changes.

using Ama.CRDT.Attributes;

public class UserStats
{
    // LwwStrategy is the default, so this attribute is optional.
    // It's shown here for clarity.
    [CrdtLwwStrategy]
    public string LastSeenLocation { get; set; } = string.Empty;

    [CrdtCounterStrategy]
    public long LoginCount { get; set; }

    [CrdtOrSetStrategy] // Use OR-Set to allow badges to be re-added after removal.
    public List<string> Badges { get; set; } = [];
}

Basic Usage

The core workflow involves creating a patch from a change and applying it to another replica.

using Ama.CRDT.Models;
using Ama.CRDT.Services;
using Microsoft.Extensions.DependencyInjection;

// Assume 'serviceProvider' is your configured IServiceProvider

// 1. Get the scope factory. This is typically a singleton.
var scopeFactory = serviceProvider.GetRequiredService<ICrdtScopeFactory>();

// 2. Create a scope for a specific replica (e.g., a user session)
using var userScope = scopeFactory.CreateScope("user-session-abc");

// 3. Resolve replica-specific services from the scope.
// These services are configured with the ReplicaId "user-session-abc".
var patcher = userScope.ServiceProvider.GetRequiredService<ICrdtPatcher>();
var applicator = userScope.ServiceProvider.GetRequiredService<ICrdtApplicator>();
var metadataManager = userScope.ServiceProvider.GetRequiredService<ICrdtMetadataManager>();

// 4. Establish an initial state
var originalState = new UserStats { LoginCount = 5, Badges = ["newcomer"] };
var originalMetadata = metadataManager.Initialize(originalState);
var originalDocument = new CrdtDocument<UserStats>(originalState, originalMetadata);

// 5. Modify the state
var modifiedState = new UserStats { LoginCount = 6, Badges = ["newcomer", "explorer"] };

// 6. Create a patch.
var patch = patcher.GeneratePatch(originalDocument, modifiedState);

// 7. On another replica, apply the patch.
// First, create a scope and get services for the other replica.
using var serverScope = scopeFactory.CreateScope("server-node-xyz");
var serverApplicator = serverScope.ServiceProvider.GetRequiredService<ICrdtApplicator>();
var serverMetadataManager = serverScope.ServiceProvider.GetRequiredService<ICrdtMetadataManager>();

var serverState = new UserStats { LoginCount = 5, Badges = ["newcomer"] };
var serverMetadata = serverMetadataManager.Initialize(serverState);
var serverDocument = new CrdtDocument<UserStats>(serverState, serverMetadata);

serverApplicator.ApplyPatch(serverDocument, patch);

// 8. Assert the new state is correct
// serverDocument.Data.LoginCount is now 6
// serverDocument.Data.Badges now contains ["newcomer", "explorer"]

Controlling Merge Behavior with CRDT Strategies

This library uses attributes on your POCO properties to determine how to merge changes. You can control the conflict resolution logic for each property individually.

Advanced Usage: Multi-Replica Synchronization

For distributed systems with multiple writers, you need a unique set of services for each replica. The ICrdtScopeFactory is the recommended way to create these. This example shows two replicas modifying the same object concurrently and converging to a consistent state.

using Ama.CRDT.Models;
using Ama.CRDT.Services;
using Microsoft.Extensions.DependencyInjection;

// 1. Get the scope factory from the root DI container.
var scopeFactory = serviceProvider.GetRequiredService<ICrdtScopeFactory>();

// 2. Create scopes and resolve services for each replica.
// All CRDT services (patcher, applicator, metadata manager) must be
// resolved from a replica-specific scope to be configured correctly.
using var scopeA = scopeFactory.CreateScope("replica-A");
var patcherA = scopeA.ServiceProvider.GetRequiredService<ICrdtPatcher>();
var applicatorA = scopeA.ServiceProvider.GetRequiredService<ICrdtApplicator>();
var metadataManagerA = scopeA.ServiceProvider.GetRequiredService<ICrdtMetadataManager>();

using var scopeB = scopeFactory.CreateScope("replica-B");
var patcherB = scopeB.ServiceProvider.GetRequiredService<ICrdtPatcher>();
var applicatorB = scopeB.ServiceProvider.GetRequiredService<ICrdtApplicator>();
var metadataManagerB = scopeB.ServiceProvider.GetRequiredService<ICrdtMetadataManager>();

// 3. Establish a base state
var baseState = new UserStats { LastSeenLocation = "Lobby", LoginCount = 10, Badges = ["welcome"] };
var baseMetadata = metadataManagerA.Initialize(baseState); // Use any manager for initialization

// 4. Create two replicas from the base state (deep cloning data and metadata)
var docA = new CrdtDocument<UserStats>(
    new UserStats { LastSeenLocation = "Lobby", LoginCount = 10, Badges = ["welcome"] },
    metadataManagerA.Clone(baseMetadata)
);

var docB = new CrdtDocument<UserStats>(
    new UserStats { LastSeenLocation = "Lobby", LoginCount = 10, Badges = ["welcome"] },
    metadataManagerB.Clone(baseMetadata)
);

// 5. Modify both replicas independently
// Replica A: User logs in again and earns a new badge
var modifiedAState = docA.Data; 
modifiedAState.LoginCount++; // 11
modifiedAState.Badges.Add("veteran");

// Replica B: User changes location and also logs in
var modifiedBState = docB.Data; 
modifiedBState.LastSeenLocation = "Marketplace";
modifiedBState.LoginCount++; // 11

// 6. Generate patches
var patchFromA = patcherA.GeneratePatch(
    new CrdtDocument<UserStats>(baseState, baseMetadata), // Compare against original base state
    modifiedAState
);
var patchFromB = patcherB.GeneratePatch(
    new CrdtDocument<UserStats>(baseState, baseMetadata),
    modifiedBState
);

// 7. Synchronize: Cross-apply patches
// Apply A's patch to B's document
applicatorB.ApplyPatch(docB, patchFromA);

// Apply B's patch to A's document
applicatorA.ApplyPatch(docA, patchFromB);

// 8. Assert Convergence
// Both replicas now have the same converged state.
// docA.Data and docB.Data are now identical.
// - LastSeenLocation: "Marketplace" (LWW from B wins, assuming later timestamp)
// - LoginCount: 12 (Counter incremented by both, 10 + 1 + 1)
// - Badges: ["veteran", "welcome"] (OR-Set merge adds "veteran")

Serializing and Transmitting Patches

Once a CrdtPatch is generated, it needs to be sent to other replicas. This is typically done by serializing the patch to JSON.

The library provides pre-configured JsonSerializerOptions for this purpose through the CrdtJsonContext class. Using CrdtJsonContext.DefaultOptions is the recommended way to serialize patches, as it automatically handles:

• Polymorphic ICrdtTimestamp types.
• Polymorphic object payloads in CrdtOperation.Value.
• Dictionaries with non-string keys.

using Ama.CRDT.Models;
using Ama.CRDT.Models.Serialization;
using System.Text.Json;

// Assume 'patch' is the CrdtPatch you generated.
CrdtPatch patch = ...; 

// 1. Serialize the patch to a JSON string using the recommended options.
string jsonPayload = JsonSerializer.Serialize(patch, CrdtJsonContext.DefaultOptions);

// This 'jsonPayload' can now be sent across the network.

// --- On the receiving replica ---

// 2. Deserialize the JSON string back into a CrdtPatch object.
CrdtPatch receivedPatch = JsonSerializer.Deserialize<CrdtPatch>(jsonPayload, CrdtJsonContext.DefaultOptions);

// 3. Apply the received patch.
// var applicator = ...;
// var document = ...;
// applicator.ApplyPatch(document, receivedPatch);

Important: If you have created a custom ICrdtTimestamp or a custom operation payload type, you must register it with the serialization system. See the section on "Extensibility" for details.

Managing Metadata Size

The CrdtMetadata object stores the necessary state to resolve conflicts and ensure eventual consistency. Over time, this metadata can grow. The library provides tools to help you keep it compact.

Pruning LWW Tombstones

When you remove a property, its timestamp in the Lww metadata is kept as a "tombstone" to prevent old versions from re-introducing the deleted value. You can periodically prune old tombstones using ICrdtMetadataManager.

using Ama.CRDT.Services;
using Ama.CRDT.Models;
using Ama.CRDT.Services.Providers;

// 1. Resolve services from a replica scope
// var scope = ...;
// var metadataManager = scope.ServiceProvider.GetRequiredService<ICrdtMetadataManager>();
// var timestampProvider = scope.ServiceProvider.GetRequiredService<ICrdtTimestampProvider>();

// 2. Assume 'myMetadata' is the CrdtMetadata for a document
CrdtMetadata myMetadata = ...;

// 3. Define a threshold (e.g., prune everything older than 30 days).
// (This example assumes the default EpochTimestamp, based on milliseconds)
long thirtyDaysInMillis = 30L * 24 * 60 * 60 * 1000;
var nowTimestamp = (EpochTimestamp)timestampProvider.Now();
var thresholdTimestamp = new EpochTimestamp(nowTimestamp.Value - thirtyDaysInMillis);

// 4. Prune the metadata
metadataManager.PruneLwwTombstones(myMetadata, thresholdTimestamp);

// 5. Save the compacted metadata
// ...

Version Vector Compaction

The library uses a version vector (CrdtMetadata.VersionVector) and a set of seen exceptions (SeenExceptions) to provide idempotency. This process is managed automatically by the CrdtApplicator for strategies marked with [IdempotentWithContinuousTimeAttribute] when using a continuous timestamp provider (like SequentialTimestampProvider), keeping metadata size under control without manual intervention.

Serializing Metadata Efficiently

To avoid bloated JSON output from empty collections in the metadata object, use the pre-configured CrdtJsonContext.MetadataCompactOptions. It automatically omits empty collections and handles all necessary custom converters.

using Ama.CRDT.Models;
using Ama.CRDT.Models.Serialization;
using System.Text.Json;

// Assume 'metadata' is your CrdtMetadata object.
CrdtMetadata metadata = ...; 

// 1. Serialize using the compact options.
string jsonPayload = JsonSerializer.Serialize(metadata, CrdtJsonContext.MetadataCompactOptions);

// This 'jsonPayload' is now compact and can be stored or sent.

// --- When reading the metadata back ---

// 2. Deserialize using the same options.
CrdtMetadata deserializedMetadata = JsonSerializer.Deserialize<CrdtMetadata>(jsonPayload, CrdtJsonContext.MetadataCompactOptions);

Extensibility: Creating Custom Strategies

You can extend the library with your own conflict resolution logic by creating a custom strategy.

1. Create a Custom Attribute

Create an attribute inheriting from CrdtStrategyAttribute.

public sealed class MyCustomStrategyAttribute() : CrdtStrategyAttribute(typeof(MyCustomStrategy));

2. Implement ICrdtStrategy

Create a class that implements ICrdtStrategy, using the context objects for parameters.

using Ama.CRDT.Services.Strategies;

public sealed class MyCustomStrategy : ICrdtStrategy
{
    public void GeneratePatch(GeneratePatchContext context)
    {
        // Add custom diffing logic here
        // var (patcher, operations, path, ...) = context;
    }

    public void ApplyOperation(ApplyOperationContext context)
    {
        // Add custom application logic here
        // var (root, metadata, operation) = context;
    }
}

3. Register in the DI Container

Register your new strategy with a scoped lifetime.

// In Program.cs
// ...
builder.Services.AddCrdt();

// Register the custom strategy with a scoped lifetime
builder.Services.AddScoped<MyCustomStrategy>();

// Make it available to the strategy provider
builder.Services.AddScoped<ICrdtStrategy>(sp => sp.GetRequiredService<MyCustomStrategy>());

You can now use [MyCustomStrategy] on your POCO properties.

Advanced Extensibility

Customizing Array Element Comparison

By default, collection strategies use deep equality. To identify complex objects by a unique property (like an Id), implement IElementComparer.

1. Implement IElementComparer

Example: A comparer for User objects that uses the Id property.

Services/UserComparer.cs

using Ama.CRDT.Services.Providers;
using System.Diagnostics.CodeAnalysis;
using YourApp.Models; 

public class UserComparer : IElementComparer
{
    public bool CanCompare(Type type) => type == typeof(User);

    public new bool Equals(object? x, object? y)
    {
        if (x is User userX && y is User userY)
        {
            return userX.Id == userY.Id;
        }
        return object.Equals(x, y);
    }

    public int GetHashCode([DisallowNull] object obj)
    {
        return (obj is User user) ? user.Id.GetHashCode() : obj.GetHashCode();
    }
}

2. Register the Comparer

Use the AddCrdtComparer<TComparer>() extension method.

// In Program.cs
builder.Services.AddCrdt();
builder.Services.AddCrdtComparer<UserComparer>();

Providing a Custom Timestamp

You can replace the default timestamping mechanism by implementing ICrdtTimestampProvider.

1. Implement ICrdtTimestampProvider

The provider must be thread-safe if used concurrently and should be registered with a scoped lifetime.

using Ama.CRDT.Services.Providers;
using Ama.CRDT.Models;
using System.Threading;

public sealed class LogicalClockProvider : ICrdtTimestampProvider
{
    private long counter = 0;
    public bool IsContinuous => true;

    public ICrdtTimestamp Now() => new SequentialTimestamp(Interlocked.Increment(ref counter));
    public ICrdtTimestamp Init() => new SequentialTimestamp(0);
    public ICrdtTimestamp Create(long value) => new SequentialTimestamp(value);

    public IEnumerable<ICrdtTimestamp> IterateBetween(ICrdtTimestamp start, ICrdtTimestamp end)
    {
        if (start is not SequentialTimestamp s || end is not SequentialTimestamp e) yield break;
        for (var i = s.Value + 1; i < e.Value; i++) yield return new SequentialTimestamp(i);
    }
}

2. Register the Provider

Use AddCrdtTimestampProvider<TProvider>() to replace the default.

// In Program.cs
builder.Services.AddCrdt();
builder.Services.AddCrdtTimestampProvider<LogicalClockProvider>();

3. Register a Custom Timestamp Type for Serialization

If you create your own ICrdtTimestamp implementation, you must register it for serialization.

Example: A custom VectorClockTimestamp.

Models/VectorClockTimestamp.cs

public readonly record struct VectorClockTimestamp(long Ticks) : ICrdtTimestamp
{
    public int CompareTo(ICrdtTimestamp? other)
    {
        if (other is null) return 1;

        if (other is not VectorClockTimestamp otherClock)
        {
            throw new ArgumentException("Cannot compare VectorClockTimestamp to other timestamp types.");
        }
        
        return Ticks.CompareTo(otherClock.Ticks);
    }
}

In Program.cs

builder.Services.AddCrdt();
builder.Services.AddCrdtTimestampProvider<VectorClockProvider>(); 

// Register the custom timestamp type with a unique discriminator string.
builder.Services.AddCrdtTimestampType<VectorClockTimestamp>("vector-clock");

Now, when a patch with a VectorClockTimestamp is serialized, the JSON will include a "$type": "vector-clock" discriminator.

How It Works

• ICrdtScopeFactory: A singleton factory for creating isolated IServiceScope instances. Each scope is configured for a specific ReplicaId and provides its own set of scoped services. This is the primary entry point for working with multiple replicas.
• ICrdtPatcher: A scoped service that generates a CrdtPatch by comparing two versions of a document. It is a stateless service that only reads the from document's metadata to understand the state before the change. It uses the ICrdtStrategyProvider to find the correct strategy for each property.
• ICrdtApplicator: A scoped service that applies a CrdtPatch to a document. It uses the ICrdtStrategyProvider to find the correct strategy to modify the POCO and its metadata. It also handles idempotency checks using version vectors for supported strategies.
• ICrdtStrategyProvider: A service that inspects a property's attributes (e.g., [CrdtCounterStrategy]) to resolve the appropriate ICrdtStrategy from the DI container. It provides default strategies (LWW for simple types, ArrayLcs for collections) if no attribute is present.
• ICrdtMetadataManager: A scoped helper service for managing the CrdtMetadata object. It can initialize metadata from a POCO, compact it to save space, and perform other state management tasks.

Building and Testing

To build the project, simply run:

dotnet build

To run the unit tests:

dotnet test

License

The code is licensed under MIT.