Merge pull request #291 from Vulthil/fix/inbox-ambient-conflict

Fix inbox marker conflicts committing duplicate work inside ambient transactions

Author: Vulthil

docs/articles/inbox-pattern.md

@@ -16,7 +16,9 @@ This is an *idempotent receiver*, not a store-and-forward inbox: it rides on top
 
 If the consumer throws, the transaction is rolled back (marker included) and the message is reprocessed cleanly on redelivery. The marker is written **on commit, not on receipt**, so an interrupted delivery never leaves a marker that would suppress reprocessing.
 
-The check (step 2) and the marker write (step 5) are serialized only at the **marker insert**, not across the whole unit. Two duplicates of the same key in flight *at the same time* — delivered to two consumers, or across two channels — can both pass the step-2 check and run the consumer body before either commits; the unique marker then lets only one commit, while the other fails the insert and is settled as a duplicate. So deduplication is exactly-once for *sequential* redelivery, but **concurrent** duplicates can each execute the handler body once. Keep side effects idempotent if concurrent duplicate delivery is possible.
+The check (step 2) and the marker write (step 5) are serialized only at the **marker insert**, not across the whole unit. Two duplicates of the same key in flight *at the same time* — delivered to two consumers, or across two channels — can both pass the step-2 check and run the consumer body before either commits; the unique marker then lets only one commit, while the other loses the insert race. So deduplication is exactly-once for *sequential* redelivery, but **concurrent** duplicates can each execute the handler body once. Keep side effects idempotent if concurrent duplicate delivery is possible.
+
+How the loser is settled depends on who owns the transaction. When the relational store opens its own transaction (no outer filter already started one), it rolls back, rechecks the marker, and returns `false` — the delivery is settled as a duplicate inline, without an exception. When the store instead joins a transaction an outer filter already opened (see [Filter Registration Order](#filter-registration-order)), it cannot commit or roll back a transaction it doesn't own, so the marker conflict is **rethrown**. The transaction's owner rolls back — discarding this delivery's business writes along with the marker — and the message is redelivered; the step-2 check then deduplicates it. Either way no duplicate side effect is ever committed, but the ambient case surfaces as an exception and a redelivery round-trip rather than a quiet skip.
 
 ## Guarantees
 
@@ -109,6 +111,19 @@ builder.Services.AddRelationalInbox<AppDbContext>();
 
 The consumer and the store must share the same scoped `DbContext` instance (the default with `AddDbContext`), so the consumer's writes and the marker enlist in the same transaction. The consumer keeps calling `SaveChanges` as usual — the store owns the transaction, not `SaveChanges`. Add an EF Core migration for the `InboxMessage` table as you would for any entity.
 
+### Filter Registration Order
+
+`IConsumeFilter<TMessage>` instances compose in plain DI registration order — the first one resolved becomes the outermost — and nothing enforces a particular order between the inbox and other transactional filters, such as `Vulthil.Messaging.Outbox`'s transactional consumer filter. Register the idempotency filter **before** (outside) a transactional consumer filter:
+
+```csharp
+messaging.AddIdempotentInbox<OrderPlaced>(context => context.Message.OrderId.ToString());
+messaging.AddTransactionalConsumer<OrderPlaced>();
+```
+
+With this order the inbox filter runs first and opens its own transaction (the relational store's `ProcessAsync` sees no ambient transaction yet), and the transactional consumer filter joins it. On a concurrent duplicate the loser rolls back and rechecks inline — the efficient path described in [How It Works](#how-it-works).
+
+Registering them in the opposite order still produces correct results — the relational store's ambient-conflict rethrow (above) guarantees a duplicate is never committed either way — but the transactional consumer filter becomes the transaction owner, so the inbox store always runs its ambient path. A concurrent duplicate then always surfaces as an exception and a redelivery, instead of the cheaper inline rollback-and-recheck. Prefer the order above.
+
 ### Cosmos store
 
 The Cosmos store is wired the same way — apply the Cosmos mapping with `ApplyCosmosInbox()` and call `AddCosmosInbox<AppDbContext>()`:

src/Vulthil.Messaging.Inbox.Relational/RelationalIdempotencyStore.cs

@@ -15,6 +15,11 @@ namespace Vulthil.Messaging.Inbox.Relational;
 /// (<c>Database.CreateExecutionStrategy().ExecuteAsync</c>), so it works whether or not a retrying execution strategy
 /// is configured — there is no need to disable EF Core retries. Under a retrying strategy a transient fault re-runs
 /// the unit (including the consumer) on a cleared change tracker, which is consistent with at-least-once redelivery.
+/// If an outer filter already opened the transaction, the store joins it instead of opening its own; in that case a
+/// concurrent marker conflict is rethrown rather than swallowed, since the store cannot commit or roll back a
+/// transaction it does not own. The exception propagates to the transaction's owner, which rolls back — including
+/// this delivery's business writes — and the transport redelivers the message, at which point the pre-check
+/// deduplicates it.
 /// </remarks>
 /// <typeparam name="TContext">The application's <see cref="DbContext"/> type, which must expose the inbox set.</typeparam>
 internal sealed class RelationalIdempotencyStore<TContext>(TContext dbContext, TimeProvider timeProvider)
@@ -97,21 +102,8 @@ private async Task<bool> ProcessInAmbientTransactionAsync(string idempotencyKey,
 
         await process(cancellationToken);
         AddMarker(idempotencyKey);
-
-        try
-        {
-            await dbContext.SaveChangesAsync(cancellationToken);
-            return true;
-        }
-        catch (DbUpdateException)
-        {
-            if (await HasProcessedAsync(idempotencyKey, cancellationToken))
-            {
-                return false;
-            }
-
-            throw;
-        }
+        await dbContext.SaveChangesAsync(cancellationToken);
+        return true;
     }
 
     private Task<bool> HasProcessedAsync(string idempotencyKey, CancellationToken cancellationToken) =>

src/Vulthil.Messaging.Inbox/IInboxKeySelector.cs

@@ -11,10 +11,10 @@ public interface IInboxKeySelector<in TMessage>
     where TMessage : notnull
 {
     /// <summary>
-    /// Returns the idempotency key for the given delivery, or <see langword="null"/> to fall back to
-    /// <see cref="IMessageContext.MessageId"/>.
+    /// Returns the idempotency key for the given delivery, or <see langword="null"/> or an empty string to fall
+    /// back to <see cref="IMessageContext.MessageId"/>.
     /// </summary>
     /// <param name="context">The message context for the current delivery.</param>
-    /// <returns>The idempotency key, or <see langword="null"/> to use the message id.</returns>
+    /// <returns>The idempotency key, or <see langword="null"/> or an empty string to use the message id.</returns>
     string? GetKey(IMessageContext<TMessage> context);
 }

src/Vulthil.Messaging.Inbox/IdempotentConsumeFilter.cs

@@ -32,7 +32,8 @@ internal sealed class IdempotentConsumeFilter<TMessage>(
     public async Task ConsumeAsync(IMessageContext<TMessage> context, ConsumeDelegate<TMessage> next)
     {
         var messageType = typeof(TMessage).FullName ?? typeof(TMessage).Name;
-        var key = _keySelector.GetKey(context) ?? context.MessageId;
+        var key = _keySelector.GetKey(context);
+        key = string.IsNullOrEmpty(key) ? context.MessageId : key;
 
         if (string.IsNullOrEmpty(key))
         {

tests/Vulthil.Messaging.Inbox.Relational.Tests/RelationalIdempotencyStoreConflictTests.cs

@@ -0,0 +1,142 @@
+using Microsoft.Data.Sqlite;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Storage;
+using Vulthil.Messaging.Abstractions.Consumers;
+using Vulthil.Messaging.Inbox.EntityFrameworkCore;
+using Vulthil.xUnit;
+
+namespace Vulthil.Messaging.Inbox.Relational.Tests;
+
+public sealed class RelationalIdempotencyStoreConflictTests : BaseUnitTestCase
+{
+    private readonly string _databasePath = Path.Combine(Path.GetTempPath(), $"vulthil-inbox-conflict-{Guid.NewGuid():N}.db");
+
+    protected override async ValueTask Initialize()
+    {
+        await using var context = NewContext();
+        await context.Database.EnsureCreatedAsync(CancellationToken);
+    }
+
+    protected override ValueTask Dispose()
+    {
+        SqliteConnection.ClearAllPools();
+        File.Delete(_databasePath);
+        return ValueTask.CompletedTask;
+    }
+
+    [Fact]
+    public async Task ConcurrentDeliveriesForTheSameKeyLetExactlyOneCommit()
+    {
+        // Arrange
+        await using var contextA = NewContext();
+        await using var contextB = NewContext();
+
+        // Act — SQLite serializes the two own-transaction attempts under the hood (a shared busy timeout
+        // resolves the lock contention instead of failing fast), so of two genuinely concurrent deliveries
+        // for the same key, exactly one commits its business write and marker; the loser observes the
+        // marker as already processed and never runs its consumer body.
+        var taskA = CreateStore(contextA).ProcessAsync("race-key", MessageContext, async token =>
+        {
+            contextA.Things.Add(new Thing { Id = 1, Name = "a" });
+            await contextA.SaveChangesAsync(token);
+        }, CancellationToken);
+
+        var taskB = CreateStore(contextB).ProcessAsync("race-key", MessageContext, async token =>
+        {
+            contextB.Things.Add(new Thing { Id = 2, Name = "b" });
+            await contextB.SaveChangesAsync(token);
+        }, CancellationToken);
+
+        var results = await Task.WhenAll(taskA, taskB);
+
+        // Assert
+        results.Count(processed => processed).ShouldBe(1);
+        await using var verify = NewContext();
+        (await verify.Things.CountAsync(CancellationToken)).ShouldBe(1);
+        (await verify.InboxMessages.CountAsync(CancellationToken)).ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task AmbientTransactionConflictPropagatesInsteadOfCommittingDuplicateWork()
+    {
+        // Act
+        var propagated = await RunAmbientConflictScenarioAsync();
+
+        // Assert — a marker conflict inside an ambient transaction must never let the caller commit this
+        // delivery's business write: it either surfaces so the caller rolls back (fixed) or is silently
+        // absorbed while the caller commits anyway (the defect this pins).
+        propagated.ShouldNotBeNull();
+        await using var verify = NewContext();
+        (await verify.Things.AnyAsync(thing => thing.Name == "losing-consumer-write", CancellationToken)).ShouldBeFalse();
+        (await verify.InboxMessages.AnyAsync(CancellationToken)).ShouldBeFalse();
+    }
+
+    private async Task<DbUpdateException?> RunAmbientConflictScenarioAsync()
+    {
+        await using var outerContext = NewContext();
+        await using var outerTransaction = await outerContext.Database.BeginTransactionAsync(CancellationToken);
+        var store = CreateStore(outerContext);
+
+        try
+        {
+            await store.ProcessAsync("race-key", MessageContext, async token =>
+            {
+                outerContext.Things.Add(new Thing { Id = 1, Name = "losing-consumer-write" });
+                await outerContext.SaveChangesAsync(token);
+
+                await InsertConflictingMarkerOnTheSameTransactionAsync(outerContext, outerTransaction, token);
+            }, CancellationToken);
+
+            await outerTransaction.CommitAsync(CancellationToken);
+            return null;
+        }
+        catch (DbUpdateException exception)
+        {
+            await outerTransaction.RollbackAsync(CancellationToken);
+            return exception;
+        }
+    }
+
+    private static async Task InsertConflictingMarkerOnTheSameTransactionAsync(
+        TestDbContext outerContext, IDbContextTransaction outerTransaction, CancellationToken cancellationToken)
+    {
+        await using var conflictContext = new TestDbContext(
+            new DbContextOptionsBuilder<TestDbContext>().UseSqlite(outerContext.Database.GetDbConnection()).Options);
+        await conflictContext.Database.UseTransactionAsync(outerTransaction.GetDbTransaction(), cancellationToken);
+        conflictContext.InboxMessages.Add(new InboxMessage { MessageId = "race-key", ProcessedOnUtc = TimeProvider.System.GetUtcNow() });
+        await conflictContext.SaveChangesAsync(cancellationToken);
+    }
+
+    private static IMessageContext MessageContext => Mock.Of<IMessageContext>();
+
+    private TestDbContext NewContext() =>
+        new(new DbContextOptionsBuilder<TestDbContext>()
+            .UseSqlite(new SqliteConnectionStringBuilder { DataSource = _databasePath, DefaultTimeout = 5 }.ToString())
+            .Options);
+
+    private static RelationalIdempotencyStore<TestDbContext> CreateStore(TestDbContext context) =>
+        new(context, TimeProvider.System);
+
+    public sealed class TestDbContext(DbContextOptions<TestDbContext> options) : DbContext(options), ISaveInboxMessages
+    {
+        public DbSet<InboxMessage> InboxMessages => Set<InboxMessage>();
+
+        public DbSet<Thing> Things => Set<Thing>();
+
+        protected override void OnModelCreating(ModelBuilder modelBuilder)
+        {
+            ArgumentNullException.ThrowIfNull(modelBuilder);
+            modelBuilder.ApplyRelationalInbox();
+            modelBuilder.Entity<InboxMessage>().Property(message => message.ProcessedOnUtc)
+                .HasConversion(value => value.UtcDateTime, value => new DateTimeOffset(value, TimeSpan.Zero));
+            modelBuilder.Entity<Thing>().HasKey(thing => thing.Id);
+        }
+    }
+
+    public sealed class Thing
+    {
+        public int Id { get; set; }
+
+        public required string Name { get; set; }
+    }
+}

tests/Vulthil.Messaging.Inbox.Tests/IdempotentConsumeFilterTests.cs

@@ -80,6 +80,24 @@ public async Task MissingKeyWithRejectDisabledInvokesConsumerWithoutStore()
         _store.Verify(store => store.ProcessAsync(It.IsAny<string>(), It.IsAny<IMessageContext>(), It.IsAny<Func<CancellationToken, Task>>(), It.IsAny<CancellationToken>()), Times.Never);
     }
 
+    [Fact]
+    public async Task EmptyStringKeySelectorResultFallsBackToMessageId()
+    {
+        // Arrange
+        _context.SetupGet(context => context.MessageId).Returns("message-1");
+        GetMock<IInboxKeySelector<TestMessage>>()
+            .Setup(selector => selector.GetKey(It.IsAny<IMessageContext<TestMessage>>()))
+            .Returns(string.Empty);
+        SetupStore("message-1", runConsumer: true, processed: true);
+
+        // Act
+        await Target.ConsumeAsync(_context.Object, Next);
+
+        // Assert
+        _consumerInvoked.ShouldBeTrue();
+        _store.Verify(store => store.ProcessAsync("message-1", It.IsAny<IMessageContext>(), It.IsAny<Func<CancellationToken, Task>>(), It.IsAny<CancellationToken>()), Times.Once);
+    }
+
     [Fact]
     public async Task CustomKeySelectorIsUsedForDeduplication()
     {