Merge pull request #309 from Vulthil/fix/outbox-configuration-guards

Fail fast on three silent outbox misconfigurations

Author: Vulthil

Vulthil.SharedKernel.slnx

@@ -108,6 +108,9 @@
     <Project Path="tests/Vulthil.SharedKernel.Outbox.Tests/Vulthil.SharedKernel.Outbox.Tests.csproj" />
     <Project Path="tests/Vulthil.SharedKernel.Outbox.EntityFrameworkCore.Tests/Vulthil.SharedKernel.Outbox.EntityFrameworkCore.Tests.csproj" />
   </Folder>
+  <Folder Name="/tests/Infrastructure/">
+    <Project Path="tests/Vulthil.SharedKernel.Infrastructure.Relational.Tests/Vulthil.SharedKernel.Infrastructure.Relational.Tests.csproj" />
+  </Folder>
   <Folder Name="/tests/Messaging/">
     <Project Path="tests/Vulthil.Messaging.Tests/Vulthil.Messaging.Tests.csproj" />
     <Project Path="tests/Vulthil.Messaging.Outbox.Tests/Vulthil.Messaging.Outbox.Tests.csproj" />

docs/articles/outbox-pattern.md

@@ -54,6 +54,14 @@ public sealed class AppDbContext(DbContextOptions<AppDbContext> options)
 
 `BaseDbContext` owns the `OutboxMessages` `DbSet`; apply the provider-optimized mapping in `OnModelCreating` by calling your provider's extension — `ApplyNpgsqlOutbox()`, `ApplyMySqlOutbox()`, or `ApplyCosmosOutbox()` (as shown above). The agnostic `ApplyOutbox()` is available for custom providers.
 
+Only one outbox-enabled `DbContext` is supported per host: the relay and retention background services resolve a
+single `IOutboxStore`, so calling `EnableOutboxProcessing()` for a second context throws an
+`InvalidOperationException` at startup instead of silently leaving the first context's messages unrelayed. The
+Npgsql and MySQL stores also require your context to derive from `BaseDbContext` (or otherwise implement
+`IUnitOfWork`) — without a transaction their `FOR UPDATE SKIP LOCKED` fetch would release its locks immediately,
+letting concurrent relay instances double-dispatch the same messages, so `RelationalOutboxStore` throws instead of
+running unprotected.
+
 ## Outbox Processing Options
 
 | Property | Default | Description |
@@ -166,6 +174,13 @@ directly. The transaction is established by one of:
   transaction and the consume filter joins it rather than nesting.
 - **Anything else** — wrap the work in `IUnitOfWork.ExecuteInTransactionAsync(...)`.
 
+Ambient `System.Transactions.TransactionScope` transactions are **not supported**: the capture gate checks for an
+Entity Framework Core transaction specifically, and EF Core does not surface an ambient scope as one. Publishing
+inside a `TransactionScope` with no EF Core transaction throws `NotSupportedException` rather than silently
+publishing directly while the scope is still uncommitted (a ghost message on rollback) — establish the transaction
+with one of the options above instead. Full ambient-scope support may be added in the future; it is deferred over
+provider enlistment edge cases.
+
 ## Typical Flow
 
 ```

docs/articles/packages/vulthil-sharedkernel-infrastructure.md

@@ -49,6 +49,10 @@ builder.AddDbContext<AppDbContext>(config => config
     }));
 ```
 
+Only one outbox-enabled `DbContext` is supported per host: the relay and retention background services resolve a
+single `IOutboxStore`, so a second `EnableOutboxProcessing()` call (on a different `DbContext`) throws an
+`InvalidOperationException` at startup instead of silently leaving the first context's messages unrelayed.
+
 ### Generic repository
 
 ```csharp

src/Vulthil.Messaging.Outbox/README.md

@@ -35,3 +35,8 @@ builder.AddMessaging(messaging =>
 
 The application's `DbContext` must implement `ISaveOutboxMessages` (a `BaseDbContext` already does). Capture is
 relational-only (it needs a transaction to enlist in).
+
+Ambient `System.Transactions.TransactionScope` transactions are not supported: publishing inside one with no active
+Entity Framework Core transaction throws `NotSupportedException` instead of silently publishing directly while the
+scope is uncommitted. Use `IUnitOfWork.ExecuteInTransactionAsync`, a transactional command, or a transactional
+consumer to establish the transaction instead.

src/Vulthil.Messaging.Outbox/TransactionalPublishFilter.cs

@@ -1,5 +1,6 @@
 using System.Diagnostics;
 using System.Text.Json;
+using System.Transactions;
 using Microsoft.Extensions.Options;
 using Vulthil.Messaging.Transport;
 using Vulthil.SharedKernel.Outbox;
@@ -9,18 +10,33 @@ namespace Vulthil.Messaging.Outbox;
 /// <summary>
 /// Publish/send filter that captures an outgoing message into the shared outbox table when a database transaction
 /// is active, so it is persisted atomically with the business changes and relayed to the broker only after the
-/// transaction commits. When no transaction is active the publish proceeds directly.
+/// transaction commits. When no transaction is active the publish proceeds directly. Ambient
+/// <see cref="System.Transactions.TransactionScope"/> transactions are not supported (see <see cref="PublishAsync"/>).
 /// </summary>
 internal sealed class TransactionalPublishFilter(
     IOutboxStore outboxStore,
     IMessageConfigurationProvider messageConfigurationProvider,
     IOptions<OutboxProcessingOptions> outboxProcessingOptions,
     TimeProvider timeProvider) : IPublishFilter
 {
+    /// <exception cref="NotSupportedException">
+    /// An ambient <see cref="System.Transactions.TransactionScope"/> is active but the outbox store reports no
+    /// Entity Framework Core transaction, so capture cannot enlist and a direct publish would escape the scope.
+    /// </exception>
     public async Task PublishAsync(PublishFilterContext context, PublishFilterDelegate next)
     {
         if (!outboxStore.IsInTransaction)
         {
+            if (Transaction.Current is not null)
+            {
+                throw new NotSupportedException(
+                    "TransactionScope-based ambient transactions are not supported by the transactional outbox: the " +
+                    "outbox store reports no active Entity Framework Core transaction, so this message cannot be " +
+                    "captured and would otherwise publish directly while the scope is uncommitted. Use " +
+                    "IUnitOfWork.ExecuteInTransactionAsync, a transactional command, or a transactional consumer to " +
+                    "establish the transaction instead.");
+            }
+
             await next(context);
             return;
         }

src/Vulthil.SharedKernel.Infrastructure.Relational/PublicAPI.Unshipped.txt

@@ -1 +1,2 @@
 #nullable enable
+override Vulthil.SharedKernel.Infrastructure.Relational.OutboxProcessing.RelationalOutboxStore<TContext>.BeginTransactionAsync(System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task<Vulthil.SharedKernel.Application.Data.IDbTransaction?>!

src/Vulthil.SharedKernel.Infrastructure.Relational/RelationalOutboxStore.cs

@@ -1,6 +1,7 @@
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Vulthil.SharedKernel.Application.Data;
 using Vulthil.SharedKernel.Outbox;
 using Vulthil.SharedKernel.Outbox.EntityFrameworkCore;
 
@@ -16,6 +17,25 @@ public class RelationalOutboxStore<TContext>(TContext dbContext, TimeProvider ti
     : EntityFrameworkOutboxStore<TContext>(dbContext, timeProvider, options)
     where TContext : DbContext, ISaveOutboxMessages
 {
+    /// <summary>
+    /// Opens the transaction for the relay batch, requiring <typeparamref name="TContext"/> to support one.
+    /// </summary>
+    /// <param name="cancellationToken">A token to observe for cancellation.</param>
+    /// <returns>The transaction to commit on success.</returns>
+    /// <exception cref="InvalidOperationException">
+    /// <typeparamref name="TContext"/> does not implement <see cref="IUnitOfWork"/>, so no transaction could be opened.
+    /// </exception>
+    protected override async Task<IDbTransaction?> BeginTransactionAsync(CancellationToken cancellationToken)
+    {
+        var transaction = await base.BeginTransactionAsync(cancellationToken);
+
+        return transaction ?? throw new InvalidOperationException(
+            $"RelationalOutboxStore could not open a transaction because '{typeof(TContext).Name}' does not implement " +
+            "IUnitOfWork. Without a transaction, provider row-locking (e.g. FOR UPDATE SKIP LOCKED) releases immediately " +
+            "after the fetch statement, so concurrent relay instances can double-dispatch the same messages. Derive " +
+            $"'{typeof(TContext).Name}' from BaseDbContext or implement IUnitOfWork so a transaction can be opened.");
+    }
+
     /// <inheritdoc />
     protected override async Task UpdateMessagesAsync(IReadOnlyList<Guid> successIds, IReadOnlyList<OutboxMessageFailure> failures, int maxRetries, DateTimeOffset processedOnUtc, CancellationToken cancellationToken)
     {

src/Vulthil.SharedKernel.Outbox/OutboxEngineServiceCollectionExtensions.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Hosting;
 
 namespace Vulthil.SharedKernel.Outbox;
 
@@ -15,7 +16,10 @@ public static class OutboxEngineServiceCollectionExtensions
     /// Registers the outbox engine: <see cref="OutboxProcessingOptions"/> (validated on start), the commit-time
     /// <see cref="IOutboxSignal"/>, the relay <see cref="OutboxProcessor"/> and its background service, and the
     /// default in-process <see cref="IOutboxDispatcher"/> for domain events. An <see cref="IOutboxStore"/> must be
-    /// registered separately by the persistence provider.
+    /// registered separately by the persistence provider. The engine's own services are registered idempotently, so
+    /// calling this repeatedly (e.g. once per <c>EnableOutboxProcessing</c> call) never duplicates them; only one
+    /// <see cref="IOutboxStore"/> is supported per host, because the relay and retention services resolve a single
+    /// instance, so a second one being already registered fails fast instead of orphaning the first context's outbox.
     /// </summary>
     /// <param name="services">The service collection.</param>
     /// <param name="configureOptions">An optional action to configure <see cref="OutboxProcessingOptions"/>.</param>
@@ -24,13 +28,29 @@ public static class OutboxEngineServiceCollectionExtensions
     /// <c>DbContext</c> lifetime so they resolve the store in the same scope.
     /// </param>
     /// <returns>The same service collection, for chaining.</returns>
+    /// <exception cref="InvalidOperationException">
+    /// An <see cref="IOutboxStore"/> is already registered for a different context.
+    /// </exception>
     public static IServiceCollection AddOutboxEngine(
         this IServiceCollection services,
         Action<OutboxProcessingOptions>? configureOptions = null,
         ServiceLifetime processorLifetime = ServiceLifetime.Scoped)
     {
         ArgumentNullException.ThrowIfNull(services);
 
+        var existingStore = services.FirstOrDefault(static descriptor => descriptor.ServiceType == typeof(IOutboxStore));
+        if (existingStore is not null)
+        {
+            var existingContextName = existingStore.ImplementationType?.IsGenericType == true
+                ? existingStore.ImplementationType.GetGenericArguments()[0].Name
+                : existingStore.ImplementationType?.Name ?? "an existing DbContext";
+
+            throw new InvalidOperationException(
+                $"Outbox processing is already enabled for '{existingContextName}'. Only one outbox-enabled DbContext " +
+                "is supported per host, because the outbox relay and retention services resolve a single IOutboxStore. " +
+                "Enabling it again for a second context would leave that context's outbox messages unrelayed with no diagnostic.");
+        }
+
         services.AddOptions<OutboxProcessingOptions>()
             .Configure(configureOptions ?? (static _ => { }))
             .Validate(
@@ -65,14 +85,14 @@ public static IServiceCollection AddOutboxEngine(
         services.TryAddSingleton(TimeProvider.System);
         services.TryAddSingleton<IOutboxSignal, OutboxSignal>();
 
-        services.Add(new ServiceDescriptor(typeof(OutboxProcessor), typeof(OutboxProcessor), processorLifetime));
+        services.TryAdd(new ServiceDescriptor(typeof(OutboxProcessor), typeof(OutboxProcessor), processorLifetime));
         services.TryAddEnumerable(new ServiceDescriptor(typeof(IOutboxDispatcher), typeof(DomainEventOutboxDispatcher), processorLifetime));
 
-        services.AddHostedService<OutboxBackgroundService>();
+        services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService, OutboxBackgroundService>());
 
         if (options.Retention.Enabled)
         {
-            services.AddHostedService<OutboxRetentionBackgroundService>();
+            services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService, OutboxRetentionBackgroundService>());
         }
 
         return services;

tests/Vulthil.Messaging.Outbox.Tests/TransactionalPublishFilterTests.cs

@@ -0,0 +1,101 @@
+using System.Text.Json;
+using System.Transactions;
+using Microsoft.Extensions.Options;
+using Vulthil.Messaging.Transport;
+using Vulthil.SharedKernel.Outbox;
+using Vulthil.xUnit;
+
+namespace Vulthil.Messaging.Outbox.Tests;
+
+public sealed class TransactionalPublishFilterTests : BaseUnitTestCase
+{
+    [Fact]
+    public async Task PublishingInsideAnAmbientTransactionScopeWithNoEntityFrameworkTransactionThrowsNotSupportedException()
+    {
+        // Arrange
+        GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(false);
+        var filter = CreateInstance<TransactionalPublishFilter>();
+        var context = NewContext();
+        var nextInvoked = false;
+        PublishFilterDelegate next = _ =>
+        {
+            nextInvoked = true;
+            return Task.CompletedTask;
+        };
+
+        // Act
+        NotSupportedException exception;
+        using (new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
+        {
+            exception = await Should.ThrowAsync<NotSupportedException>(() => filter.PublishAsync(context, next));
+        }
+
+        // Assert
+        exception.Message.ShouldContain("TransactionScope");
+        exception.Message.ShouldContain("IUnitOfWork.ExecuteInTransactionAsync");
+        nextInvoked.ShouldBeFalse();
+        GetMock<IOutboxStore>().Verify(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()), Times.Never);
+    }
+
+    [Fact]
+    public async Task PublishingWithNoTransactionAndNoAmbientScopePublishesDirectly()
+    {
+        // Arrange
+        GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(false);
+        var filter = CreateInstance<TransactionalPublishFilter>();
+        var context = NewContext();
+        var nextInvoked = false;
+        PublishFilterDelegate next = _ =>
+        {
+            nextInvoked = true;
+            return Task.CompletedTask;
+        };
+
+        // Act
+        await filter.PublishAsync(context, next);
+
+        // Assert
+        nextInvoked.ShouldBeTrue();
+        GetMock<IOutboxStore>().Verify(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()), Times.Never);
+    }
+
+    [Fact]
+    public async Task PublishingInsideAnEntityFrameworkTransactionCapturesTheMessageInsteadOfThrowing()
+    {
+        // Arrange
+        GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(true);
+        Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions()));
+        Use(TimeProvider.System);
+        GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions());
+        var filter = CreateInstance<TransactionalPublishFilter>();
+        var context = NewContext();
+        var nextInvoked = false;
+        PublishFilterDelegate next = _ =>
+        {
+            nextInvoked = true;
+            return Task.CompletedTask;
+        };
+
+        // Act
+        using (new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
+        {
+            await filter.PublishAsync(context, next);
+        }
+
+        // Assert
+        nextInvoked.ShouldBeFalse();
+        GetMock<IOutboxStore>().Verify(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()), Times.Once);
+        GetMock<IOutboxStore>().Verify(store => store.SaveChangesAsync(It.IsAny<CancellationToken>()), Times.Once);
+    }
+
+    private static PublishFilterContext NewContext() => new()
+    {
+        Message = new TestMessage("hello"),
+        MessageType = typeof(TestMessage),
+        Context = new PublishContext(),
+        Kind = PublishKind.Publish,
+        CancellationToken = CancellationToken,
+    };
+
+    private sealed record TestMessage(string Value);
+}