Skip to content

Commit e8d85bb

Browse files
committed
fix: fail fast on duplicate outbox registrations, missing relay transactions, and ambient TransactionScopes
1 parent 8c588a5 commit e8d85bb

12 files changed

Lines changed: 394 additions & 5 deletions

File tree

Vulthil.SharedKernel.slnx

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,9 @@
108108
<Project Path="tests/Vulthil.SharedKernel.Outbox.Tests/Vulthil.SharedKernel.Outbox.Tests.csproj" />
109109
<Project Path="tests/Vulthil.SharedKernel.Outbox.EntityFrameworkCore.Tests/Vulthil.SharedKernel.Outbox.EntityFrameworkCore.Tests.csproj" />
110110
</Folder>
111+
<Folder Name="/tests/Infrastructure/">
112+
<Project Path="tests/Vulthil.SharedKernel.Infrastructure.Relational.Tests/Vulthil.SharedKernel.Infrastructure.Relational.Tests.csproj" />
113+
</Folder>
111114
<Folder Name="/tests/Messaging/">
112115
<Project Path="tests/Vulthil.Messaging.Tests/Vulthil.Messaging.Tests.csproj" />
113116
<Project Path="tests/Vulthil.Messaging.Outbox.Tests/Vulthil.Messaging.Outbox.Tests.csproj" />

docs/articles/outbox-pattern.md

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,14 @@ public sealed class AppDbContext(DbContextOptions<AppDbContext> options)
5454

5555
`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.
5656

57+
Only one outbox-enabled `DbContext` is supported per host: the relay and retention background services resolve a
58+
single `IOutboxStore`, so calling `EnableOutboxProcessing()` for a second context throws an
59+
`InvalidOperationException` at startup instead of silently leaving the first context's messages unrelayed. The
60+
Npgsql and MySQL stores also require your context to derive from `BaseDbContext` (or otherwise implement
61+
`IUnitOfWork`) — without a transaction their `FOR UPDATE SKIP LOCKED` fetch would release its locks immediately,
62+
letting concurrent relay instances double-dispatch the same messages, so `RelationalOutboxStore` throws instead of
63+
running unprotected.
64+
5765
## Outbox Processing Options
5866

5967
| Property | Default | Description |
@@ -166,6 +174,13 @@ directly. The transaction is established by one of:
166174
transaction and the consume filter joins it rather than nesting.
167175
- **Anything else** — wrap the work in `IUnitOfWork.ExecuteInTransactionAsync(...)`.
168176

177+
Ambient `System.Transactions.TransactionScope` transactions are **not supported**: the capture gate checks for an
178+
Entity Framework Core transaction specifically, and EF Core does not surface an ambient scope as one. Publishing
179+
inside a `TransactionScope` with no EF Core transaction throws `NotSupportedException` rather than silently
180+
publishing directly while the scope is still uncommitted (a ghost message on rollback) — establish the transaction
181+
with one of the options above instead. Full ambient-scope support may be added in the future; it is deferred over
182+
provider enlistment edge cases.
183+
169184
## Typical Flow
170185

171186
```

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,10 @@ builder.AddDbContext<AppDbContext>(config => config
4949
}));
5050
```
5151

52+
Only one outbox-enabled `DbContext` is supported per host: the relay and retention background services resolve a
53+
single `IOutboxStore`, so a second `EnableOutboxProcessing()` call (on a different `DbContext`) throws an
54+
`InvalidOperationException` at startup instead of silently leaving the first context's messages unrelayed.
55+
5256
### Generic repository
5357

5458
```csharp

src/Vulthil.Messaging.Outbox/README.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,3 +35,8 @@ builder.AddMessaging(messaging =>
3535

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

src/Vulthil.Messaging.Outbox/TransactionalPublishFilter.cs

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
using System.Diagnostics;
22
using System.Text.Json;
3+
using System.Transactions;
34
using Microsoft.Extensions.Options;
45
using Vulthil.Messaging.Transport;
56
using Vulthil.SharedKernel.Outbox;
@@ -9,18 +10,33 @@ namespace Vulthil.Messaging.Outbox;
910
/// <summary>
1011
/// Publish/send filter that captures an outgoing message into the shared outbox table when a database transaction
1112
/// is active, so it is persisted atomically with the business changes and relayed to the broker only after the
12-
/// transaction commits. When no transaction is active the publish proceeds directly.
13+
/// transaction commits. When no transaction is active the publish proceeds directly. Ambient
14+
/// <see cref="System.Transactions.TransactionScope"/> transactions are not supported (see <see cref="PublishAsync"/>).
1315
/// </summary>
1416
internal sealed class TransactionalPublishFilter(
1517
IOutboxStore outboxStore,
1618
IMessageConfigurationProvider messageConfigurationProvider,
1719
IOptions<OutboxProcessingOptions> outboxProcessingOptions,
1820
TimeProvider timeProvider) : IPublishFilter
1921
{
22+
/// <exception cref="NotSupportedException">
23+
/// An ambient <see cref="System.Transactions.TransactionScope"/> is active but the outbox store reports no
24+
/// Entity Framework Core transaction, so capture cannot enlist and a direct publish would escape the scope.
25+
/// </exception>
2026
public async Task PublishAsync(PublishFilterContext context, PublishFilterDelegate next)
2127
{
2228
if (!outboxStore.IsInTransaction)
2329
{
30+
if (Transaction.Current is not null)
31+
{
32+
throw new NotSupportedException(
33+
"TransactionScope-based ambient transactions are not supported by the transactional outbox: the " +
34+
"outbox store reports no active Entity Framework Core transaction, so this message cannot be " +
35+
"captured and would otherwise publish directly while the scope is uncommitted. Use " +
36+
"IUnitOfWork.ExecuteInTransactionAsync, a transactional command, or a transactional consumer to " +
37+
"establish the transaction instead.");
38+
}
39+
2440
await next(context);
2541
return;
2642
}
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,2 @@
11
#nullable enable
2+
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

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
using Microsoft.EntityFrameworkCore;
22
using Microsoft.Extensions.Logging;
33
using Microsoft.Extensions.Options;
4+
using Vulthil.SharedKernel.Application.Data;
45
using Vulthil.SharedKernel.Outbox;
56
using Vulthil.SharedKernel.Outbox.EntityFrameworkCore;
67

@@ -16,6 +17,25 @@ public class RelationalOutboxStore<TContext>(TContext dbContext, TimeProvider ti
1617
: EntityFrameworkOutboxStore<TContext>(dbContext, timeProvider, options)
1718
where TContext : DbContext, ISaveOutboxMessages
1819
{
20+
/// <summary>
21+
/// Opens the transaction for the relay batch, requiring <typeparamref name="TContext"/> to support one.
22+
/// </summary>
23+
/// <param name="cancellationToken">A token to observe for cancellation.</param>
24+
/// <returns>The transaction to commit on success.</returns>
25+
/// <exception cref="InvalidOperationException">
26+
/// <typeparamref name="TContext"/> does not implement <see cref="IUnitOfWork"/>, so no transaction could be opened.
27+
/// </exception>
28+
protected override async Task<IDbTransaction?> BeginTransactionAsync(CancellationToken cancellationToken)
29+
{
30+
var transaction = await base.BeginTransactionAsync(cancellationToken);
31+
32+
return transaction ?? throw new InvalidOperationException(
33+
$"RelationalOutboxStore could not open a transaction because '{typeof(TContext).Name}' does not implement " +
34+
"IUnitOfWork. Without a transaction, provider row-locking (e.g. FOR UPDATE SKIP LOCKED) releases immediately " +
35+
"after the fetch statement, so concurrent relay instances can double-dispatch the same messages. Derive " +
36+
$"'{typeof(TContext).Name}' from BaseDbContext or implement IUnitOfWork so a transaction can be opened.");
37+
}
38+
1939
/// <inheritdoc />
2040
protected override async Task UpdateMessagesAsync(IReadOnlyList<Guid> successIds, IReadOnlyList<OutboxMessageFailure> failures, int maxRetries, DateTimeOffset processedOnUtc, CancellationToken cancellationToken)
2141
{

src/Vulthil.SharedKernel.Outbox/OutboxEngineServiceCollectionExtensions.cs

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
using Microsoft.Extensions.DependencyInjection;
22
using Microsoft.Extensions.DependencyInjection.Extensions;
3+
using Microsoft.Extensions.Hosting;
34

45
namespace Vulthil.SharedKernel.Outbox;
56

@@ -15,7 +16,10 @@ public static class OutboxEngineServiceCollectionExtensions
1516
/// Registers the outbox engine: <see cref="OutboxProcessingOptions"/> (validated on start), the commit-time
1617
/// <see cref="IOutboxSignal"/>, the relay <see cref="OutboxProcessor"/> and its background service, and the
1718
/// default in-process <see cref="IOutboxDispatcher"/> for domain events. An <see cref="IOutboxStore"/> must be
18-
/// registered separately by the persistence provider.
19+
/// registered separately by the persistence provider. The engine's own services are registered idempotently, so
20+
/// calling this repeatedly (e.g. once per <c>EnableOutboxProcessing</c> call) never duplicates them; only one
21+
/// <see cref="IOutboxStore"/> is supported per host, because the relay and retention services resolve a single
22+
/// instance, so a second one being already registered fails fast instead of orphaning the first context's outbox.
1923
/// </summary>
2024
/// <param name="services">The service collection.</param>
2125
/// <param name="configureOptions">An optional action to configure <see cref="OutboxProcessingOptions"/>.</param>
@@ -24,13 +28,29 @@ public static class OutboxEngineServiceCollectionExtensions
2428
/// <c>DbContext</c> lifetime so they resolve the store in the same scope.
2529
/// </param>
2630
/// <returns>The same service collection, for chaining.</returns>
31+
/// <exception cref="InvalidOperationException">
32+
/// An <see cref="IOutboxStore"/> is already registered for a different context.
33+
/// </exception>
2734
public static IServiceCollection AddOutboxEngine(
2835
this IServiceCollection services,
2936
Action<OutboxProcessingOptions>? configureOptions = null,
3037
ServiceLifetime processorLifetime = ServiceLifetime.Scoped)
3138
{
3239
ArgumentNullException.ThrowIfNull(services);
3340

41+
var existingStore = services.FirstOrDefault(static descriptor => descriptor.ServiceType == typeof(IOutboxStore));
42+
if (existingStore is not null)
43+
{
44+
var existingContextName = existingStore.ImplementationType?.IsGenericType == true
45+
? existingStore.ImplementationType.GetGenericArguments()[0].Name
46+
: existingStore.ImplementationType?.Name ?? "an existing DbContext";
47+
48+
throw new InvalidOperationException(
49+
$"Outbox processing is already enabled for '{existingContextName}'. Only one outbox-enabled DbContext " +
50+
"is supported per host, because the outbox relay and retention services resolve a single IOutboxStore. " +
51+
"Enabling it again for a second context would leave that context's outbox messages unrelayed with no diagnostic.");
52+
}
53+
3454
services.AddOptions<OutboxProcessingOptions>()
3555
.Configure(configureOptions ?? (static _ => { }))
3656
.Validate(
@@ -65,14 +85,14 @@ public static IServiceCollection AddOutboxEngine(
6585
services.TryAddSingleton(TimeProvider.System);
6686
services.TryAddSingleton<IOutboxSignal, OutboxSignal>();
6787

68-
services.Add(new ServiceDescriptor(typeof(OutboxProcessor), typeof(OutboxProcessor), processorLifetime));
88+
services.TryAdd(new ServiceDescriptor(typeof(OutboxProcessor), typeof(OutboxProcessor), processorLifetime));
6989
services.TryAddEnumerable(new ServiceDescriptor(typeof(IOutboxDispatcher), typeof(DomainEventOutboxDispatcher), processorLifetime));
7090

71-
services.AddHostedService<OutboxBackgroundService>();
91+
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService, OutboxBackgroundService>());
7292

7393
if (options.Retention.Enabled)
7494
{
75-
services.AddHostedService<OutboxRetentionBackgroundService>();
95+
services.TryAddEnumerable(ServiceDescriptor.Singleton<IHostedService, OutboxRetentionBackgroundService>());
7696
}
7797

7898
return services;
Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
using System.Text.Json;
2+
using System.Transactions;
3+
using Microsoft.Extensions.Options;
4+
using Vulthil.Messaging.Transport;
5+
using Vulthil.SharedKernel.Outbox;
6+
using Vulthil.xUnit;
7+
8+
namespace Vulthil.Messaging.Outbox.Tests;
9+
10+
public sealed class TransactionalPublishFilterTests : BaseUnitTestCase
11+
{
12+
[Fact]
13+
public async Task PublishingInsideAnAmbientTransactionScopeWithNoEntityFrameworkTransactionThrowsNotSupportedException()
14+
{
15+
// Arrange
16+
GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(false);
17+
var filter = CreateInstance<TransactionalPublishFilter>();
18+
var context = NewContext();
19+
var nextInvoked = false;
20+
PublishFilterDelegate next = _ =>
21+
{
22+
nextInvoked = true;
23+
return Task.CompletedTask;
24+
};
25+
26+
// Act
27+
NotSupportedException exception;
28+
using (new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
29+
{
30+
exception = await Should.ThrowAsync<NotSupportedException>(() => filter.PublishAsync(context, next));
31+
}
32+
33+
// Assert
34+
exception.Message.ShouldContain("TransactionScope");
35+
exception.Message.ShouldContain("IUnitOfWork.ExecuteInTransactionAsync");
36+
nextInvoked.ShouldBeFalse();
37+
GetMock<IOutboxStore>().Verify(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()), Times.Never);
38+
}
39+
40+
[Fact]
41+
public async Task PublishingWithNoTransactionAndNoAmbientScopePublishesDirectly()
42+
{
43+
// Arrange
44+
GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(false);
45+
var filter = CreateInstance<TransactionalPublishFilter>();
46+
var context = NewContext();
47+
var nextInvoked = false;
48+
PublishFilterDelegate next = _ =>
49+
{
50+
nextInvoked = true;
51+
return Task.CompletedTask;
52+
};
53+
54+
// Act
55+
await filter.PublishAsync(context, next);
56+
57+
// Assert
58+
nextInvoked.ShouldBeTrue();
59+
GetMock<IOutboxStore>().Verify(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()), Times.Never);
60+
}
61+
62+
[Fact]
63+
public async Task PublishingInsideAnEntityFrameworkTransactionCapturesTheMessageInsteadOfThrowing()
64+
{
65+
// Arrange
66+
GetMock<IOutboxStore>().Setup(store => store.IsInTransaction).Returns(true);
67+
Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions()));
68+
Use(TimeProvider.System);
69+
GetMock<IMessageConfigurationProvider>().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions());
70+
var filter = CreateInstance<TransactionalPublishFilter>();
71+
var context = NewContext();
72+
var nextInvoked = false;
73+
PublishFilterDelegate next = _ =>
74+
{
75+
nextInvoked = true;
76+
return Task.CompletedTask;
77+
};
78+
79+
// Act
80+
using (new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
81+
{
82+
await filter.PublishAsync(context, next);
83+
}
84+
85+
// Assert
86+
nextInvoked.ShouldBeFalse();
87+
GetMock<IOutboxStore>().Verify(store => store.AddOutboxMessage(It.IsAny<OutboxMessage>()), Times.Once);
88+
GetMock<IOutboxStore>().Verify(store => store.SaveChangesAsync(It.IsAny<CancellationToken>()), Times.Once);
89+
}
90+
91+
private static PublishFilterContext NewContext() => new()
92+
{
93+
Message = new TestMessage("hello"),
94+
MessageType = typeof(TestMessage),
95+
Context = new PublishContext(),
96+
Kind = PublishKind.Publish,
97+
CancellationToken = CancellationToken,
98+
};
99+
100+
private sealed record TestMessage(string Value);
101+
}

0 commit comments

Comments
 (0)