Concurrency Management with DDD in C#: Optimistic Locking, Versioning, and ETags

-- EF Core generated SQL:
UPDATE [Orders]
SET [Status] = @p0, [CustomerId] = @p1
WHERE [Id] = @p2 AND [RowVersion] = @p3;

-- If @@ROWCOUNT = 0, another session modified the row → conflict
-- If @@ROWCOUNT = 1, update succeeded

public class Order
{
    public int Id { get; set; }
    public uint Version { get; set; }
}

// PostgreSQL exposes xmin through the standard concurrency token mechanism.
modelBuilder.Entity<Order>()
    .Property(o => o.Version)
    .IsRowVersion();

// Domain model (simplified)
public class Order // Aggregate Root
{
    public Guid Id { get; private set; }
    public Guid CustomerId { get; private set; }
    public string Status { get; private set; } = "Draft";
    private readonly List<OrderLine> _lines = new();
    public IReadOnlyCollection<OrderLine> Lines => _lines.AsReadOnly();

    // Concurrency token (EF Core)
    public byte[] RowVersion { get; private set; } = Array.Empty<byte>();

    public void AddLine(Guid productId, int quantity, decimal unitPrice)
    {
        if (quantity <= 0) throw new ArgumentOutOfRangeException(nameof(quantity));
        _lines.Add(new OrderLine(productId, quantity, unitPrice));
    }

    public void Submit()
    {
        if (!_lines.Any()) throw new InvalidOperationException("Cannot submit an empty order.");
        Status = "Submitted";
    }
}

public record OrderLine(Guid ProductId, int Quantity, decimal UnitPrice);

public sealed class OrderingDbContext : DbContext
{
    public DbSet<Order> Orders => Set<Order>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        var order = modelBuilder.Entity<Order>();
        order.HasKey(o => o.Id);
        order.Property(o => o.RowVersion)
             .IsRowVersion() // marks as concurrency token
             .IsRequired();
        order.OwnsMany(o => o.Lines, b =>
        {
            b.WithOwner();
            b.Property<int>("Id");
            b.HasKey("Id");
        });
    }
}

public class OrdersRepository
{
    private readonly OrderingDbContext _db;
    public OrdersRepository(OrderingDbContext db) => _db = db;

    public async Task SaveAsync(Order order, CancellationToken ct = default)
    {
        try
        {
            await _db.SaveChangesAsync(ct);
        }
        catch (DbUpdateConcurrencyException ex)
        {
            throw new ConcurrencyException("Order was modified by another process.", ex);
        }
    }
}

public sealed class ConcurrencyException : Exception
{
    public ConcurrencyException(string message, Exception? inner = null) : base(message, inner) {}
}

-- Step 1: UPDATE with concurrency check
SET NOCOUNT ON;
UPDATE [Orders]
SET [Status] = @p0
WHERE [Id] = @p1 AND [RowVersion] = @p2;

-- Step 2: Check affected rows
SELECT [RowVersion]
FROM [Orders]
WHERE @@ROWCOUNT = 1 AND [Id] = @p1;

-- If @@ROWCOUNT = 0 → DbUpdateConcurrencyException

catch (DbUpdateConcurrencyException ex)
{
    foreach (var entry in ex.Entries)
    {
        var clientValues = (Order)entry.Entity;
        var databaseValues = await entry.GetDatabaseValuesAsync();

        if (databaseValues == null)
        {
            // Row was deleted by another user between read and write
            Console.WriteLine("The order was deleted by another user.");
        }
        else
        {
            // Row was modified — show differences to the user
            var currentStatus = databaseValues.GetValue<string>(nameof(Order.Status));
            var attemptedStatus = clientValues.Status;
            Console.WriteLine($"Status was '{currentStatus}', you attempted to set it to '{attemptedStatus}'");
        }
    }
}

using Polly;
using Polly.Retry;

public sealed class SubmitOrderHandler
{
    private readonly OrdersRepository _repo;
    private readonly OrderingDbContext _db;
    private static readonly ResiliencePipeline Retry = new ResiliencePipelineBuilder()
        .AddRetry(new RetryStrategyOptions
        {
            MaxRetryAttempts = 3,
            Delay = TimeSpan.FromMilliseconds(50),
            BackoffType = DelayBackoffType.Exponential,
            ShouldHandle = new PredicateBuilder().Handle<ConcurrencyException>(),
        })
        .Build();

    public SubmitOrderHandler(OrdersRepository repo, OrderingDbContext db)
    {
        _repo = repo; _db = db;
    }

    public Task HandleAsync(Guid orderId, CancellationToken ct)
        => Retry.ExecuteAsync(async _ =>
        {
            var order = await _db.Orders.Include(o => o.Lines).FirstAsync(o => o.Id == orderId, ct);
            order.Submit();
            await _repo.SaveAsync(order, ct);
        }, ct);
}

[ApiController]
[Route("api/orders")]
public class OrdersController : ControllerBase
{
    private readonly OrderingDbContext _db;
    private readonly OrdersRepository _repo;

    public OrdersController(OrderingDbContext db, OrdersRepository repo)
    { 
        _db = db; 
        _repo = repo; 
    }

    [HttpGet("{id}")]
    public async Task<IActionResult> Get(Guid id, CancellationToken ct)
    {
        var order = await _db.Orders.Include(o => o.Lines).FirstOrDefaultAsync(o => o.Id == id, ct);
        if (order is null) return NotFound();
        Response.Headers.ETag = $"\"{Convert.ToBase64String(order.RowVersion)}\""; // quoted per RFC
        return Ok(order);
    }

    [HttpPut("{id}/submit")]
    public async Task<IActionResult> Submit(Guid id, CancellationToken ct)
    {
        if (!Request.Headers.TryGetValue("If-Match", out var etagHeader))
            return StatusCode(StatusCodes.Status428PreconditionRequired, "If-Match header required");

        var expected = ParseEtag(etagHeader!);
        var order = await _db.Orders.Include(o => o.Lines).FirstOrDefaultAsync(o => o.Id == id, ct);
        if (order is null) return NotFound();

        // Compare expected RowVersion with current
        if (!order.RowVersion.AsSpan().SequenceEqual(expected))
            return StatusCode(StatusCodes.Status412PreconditionFailed);

        order.Submit();
        try
        {
            await _repo.SaveAsync(order, ct);
            return NoContent();
        }
        catch (ConcurrencyException)
        {
            return Conflict(); // 409
        }

        static byte[] ParseEtag(string etag)
        {
            var trimmed = etag.Trim('"');
            return Convert.FromBase64String(trimmed);
        }
    }
}

public class Product
{
    public Guid Id { get; private set; }
    public string Name { get; private set; } = string.Empty;

    [ConcurrencyCheck]
    public int Version { get; private set; }

    public void Rename(string name)
    {
        Name = name;
    }
}

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<Product>()
        .Property(p => p.Version)
        .IsConcurrencyToken();
}

CREATE TRIGGER [dbo].[Products_VERSION]
ON [dbo].[Products]
AFTER UPDATE
AS
BEGIN
    SET NOCOUNT ON;

    UPDATE p
    SET p.Version = p.Version + 1
    FROM [dbo].[Products] p
    INNER JOIN INSERTED i ON p.Id = i.Id;
END

-- PostgreSQL equivalent
CREATE OR REPLACE FUNCTION update_version()
RETURNS TRIGGER AS $$
BEGIN
    NEW.version := OLD.version + 1;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER products_version_trigger
    BEFORE UPDATE ON products
    FOR EACH ROW
    EXECUTE FUNCTION update_version();

// Data Annotations equivalent:
// [ConcurrencyCheck] marks the property as a concurrency token
// [DatabaseGenerated(Computed)] tells EF Core the DB generates the value on INSERT and UPDATE
public class Product
{
    public Guid Id { get; private set; }
    public string Name { get; private set; } = string.Empty;

    [ConcurrencyCheck]
    [DatabaseGenerated(DatabaseGeneratedOption.Computed)]
    public int Version { get; private set; }
}

// Fluent API equivalent — ValueGeneratedOnAddOrUpdate tells EF Core
// the database controls value generation (trigger)
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<Product>()
        .Property(p => p.Version)
        .IsConcurrencyToken()
        .ValueGeneratedOnAddOrUpdate(); // DB handles value generation (trigger)
}

// Marker interface — entities with application-managed version
public interface IVersionedEntity
{
    int Version { get; set; }
}

// Domain entity implements the interface
public class Product : IVersionedEntity
{
    public Guid Id { get; private set; }
    public string Name { get; private set; } = string.Empty;

    [ConcurrencyCheck] // marks as concurrency token via Data Annotations
    public int Version { get; set; } // public setter for interceptor

    public void Rename(string name)
    {
        Name = name;
    }
}

// Fluent API equivalent — no ValueGeneratedOnAddOrUpdate because
// the application controls the value
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<Product>()
        .Property(p => p.Version)
        .IsConcurrencyToken();
}

// Interceptor — increments Version on every modified entity before save
public sealed class ConcurrencyVersionInterceptor : SaveChangesInterceptor
{
    public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
        DbContextEventData eventData,
        InterceptionResult<int> result,
        CancellationToken cancellationToken = default)
    {
        if (eventData.Context is null) return ValueTask.FromResult(result);

        foreach (var entry in eventData.Context.ChangeTracker.Entries<IVersionedEntity>())
        {
            if (entry.State == EntityState.Modified)
                entry.Entity.Version++;
        }

        return ValueTask.FromResult(result);
    }
}

// Register in DbContext
dbContext.AddInterceptor(new ConcurrencyVersionInterceptor());

-- SQL Server: pessimistic lock via table hint
SELECT * FROM [Orders] WITH (UPDLOCK, ROWLOCK) WHERE [Id] = @id;

-- PostgreSQL: pessimistic lock via SELECT ... FOR UPDATE
SELECT * FROM orders WHERE id = @id FOR UPDATE;

// Domain layer — aware of concurrency as a business concept
public class Order
{
    public Guid Id { get; private set; }
    public byte[] RowVersion { get; private set; } = Array.Empty<byte>();
    private readonly List<OrderLine> _lines = new();

    public void Submit()
    {
        if (!_lines.Any()) throw new InvalidOperationException("Cannot submit an empty order.");
        Status = "Submitted";
    }
}

// Infrastructure layer — configures how the token is persisted
modelBuilder.Entity<Order>()
    .Property(o => o.RowVersion)
    .IsRowVersion();

public class Order // Aggregate Root
{
    public Guid Id { get; private set; }
    public Guid CustomerId { get; private set; }
    public string Status { get; private set; } = "Draft";
    private readonly List<OrderLine> _lines = new();
    public IReadOnlyCollection<OrderLine> Lines => _lines.AsReadOnly();

    // Concurrency token — EF Core manages the value on INSERT and UPDATE
    public byte[] RowVersion { get; private set; } = Array.Empty<byte>();

    public void AddLine(Guid productId, int quantity, decimal unitPrice)
    {
        if (quantity <= 0) throw new ArgumentOutOfRangeException(nameof(quantity));
        _lines.Add(new OrderLine(productId, quantity, unitPrice));
    }

    public void Submit()
    {
        if (!_lines.Any()) throw new InvalidOperationException("Cannot submit an empty order.");
        Status = "Submitted";
    }
}

public record OrderLine(Guid ProductId, int Quantity, decimal UnitPrice);

public sealed class OrderingDbContext : DbContext
{
    public DbSet<Order> Orders => Set<Order>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        var order = modelBuilder.Entity<Order>();
        order.HasKey(o => o.Id);
        order.Property(o => o.RowVersion)
             .IsRowVersion()  // marks as concurrency token
             .IsRequired();
        order.OwnsMany(o => o.Lines, b =>
        {
            b.WithOwner();
            b.Property<int>("Id");
            b.HasKey("Id");
        });
    }
}

public class OrdersRepository
{
    private readonly OrderingDbContext _db;
    public OrdersRepository(OrderingDbContext db) => _db = db;

    public async Task SaveAsync(Order order, CancellationToken ct = default)
    {
        try
        {
            await _db.SaveChangesAsync(ct);
        }
        catch (DbUpdateConcurrencyException ex)
        {
            throw new ConcurrencyException("Order was modified by another process.", ex);
        }
    }
}

public sealed class ConcurrencyException : Exception
{
    public ConcurrencyException(string message, Exception? inner = null) : base(message, inner) {}
}

[ApiController]
[Route("api/orders")]
public class OrdersController : ControllerBase
{
    private readonly OrderingDbContext _db;
    private readonly OrdersRepository _repo;

    public OrdersController(OrderingDbContext db, OrdersRepository repo)
    {
        _db = db;
        _repo = repo;
    }

    [HttpGet("{id}")]
    public async Task<IActionResult> Get(Guid id, CancellationToken ct)
    {
        var order = await _db.Orders.Include(o => o.Lines)
            .FirstOrDefaultAsync(o => o.Id == id, ct);
        if (order is null) return NotFound();

        // Expose the concurrency token as an ETag (quoted per RFC 7232)
        Response.Headers.ETag = $"\"{Convert.ToBase64String(order.RowVersion)}\"";
        return Ok(order);
    }

    [HttpPut("{id}")]
    public async Task<IActionResult> Update(Guid id, UpdateOrderDto dto, CancellationToken ct)
    {
        // Enforce If-Match — return 428 if missing
        if (!Request.Headers.TryGetValue("If-Match", out var etagHeader))
            return StatusCode(StatusCodes.Status428PreconditionRequired,
                "Include If-Match header with the current ETag.");

        var expected = ParseEtag(etagHeader!);
        var order = await _db.Orders.Include(o => o.Lines)
            .FirstOrDefaultAsync(o => o.Id == id, ct);
        if (order is null) return NotFound();

        // Return 412 if the ETag is stale
        if (!order.RowVersion.AsSpan().SequenceEqual(expected))
            return StatusCode(StatusCodes.Status412PreconditionFailed);

        order.Submit();

        try
        {
            await _repo.SaveAsync(order, ct);
            return NoContent();
        }
        catch (ConcurrencyException)
        {
            return Conflict(); // 409
        }

        static byte[] ParseEtag(string etag)
        {
            var trimmed = etag.Trim('"');
            return Convert.FromBase64String(trimmed);
        }
    }
}

using Polly;
using Polly.Retry;

public sealed class SubmitOrderHandler
{
    private readonly OrdersRepository _repo;
    private readonly OrderingDbContext _db;

    private static readonly ResiliencePipeline Retry = new ResiliencePipelineBuilder()
        .AddRetry(new RetryStrategyOptions
        {
            MaxRetryAttempts = 3,
            Delay = TimeSpan.FromMilliseconds(50),
            BackoffType = DelayBackoffType.Exponential,
            ShouldHandle = new PredicateBuilder().Handle<ConcurrencyException>(),
            OnRetry = args =>
            {
                Console.WriteLine($"Retry {args.AttemptNumber} after {args.RetryDelay.TotalMilliseconds}ms");
                return ValueTask.CompletedTask;
            },
        })
        .Build();

    public SubmitOrderHandler(OrdersRepository repo, OrderingDbContext db)
    {
        _repo = repo;
        _db = db;
    }

    public Task HandleAsync(Guid orderId, CancellationToken ct)
        => Retry.ExecuteAsync(async _ =>
        {
            // IMPORTANT: Detach stale entity before re-reading
            var existing = _db.Orders.Local.FirstOrDefault(o => o.Id == orderId);
            if (existing != null)
                _db.Entry(existing).State = EntityState.Detached;

            var order = await _db.Orders.Include(o => o.Lines)
                .FirstAsync(o => o.Id == orderId, ct);

            order.Submit();
            await _repo.SaveAsync(order, ct);
        }, ct);
}