From 7b2926511b0edacf7d512826f974dbc774a08682 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:17:21 +0000 Subject: [PATCH 1/6] [Mocha] Reproduce outbox signalling before ambient transaction commit Outbox writes inside a TransactionScope signal the outbox worker before the row is committed and never signal at commit, so the worker can park indefinitely with a committed message in the table. --- ...gresOutboxAmbientTransactionSignalTests.cs | 181 ++++++++++++++++++ 1 file changed, 181 insertions(+) create mode 100644 src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs diff --git a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs new file mode 100644 index 00000000000..a20e646d90c --- /dev/null +++ b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs @@ -0,0 +1,181 @@ +using System.Collections.Concurrent; +using System.Transactions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Mocha.EntityFrameworkCore.Postgres.Tests.Helpers; +using Mocha.Outbox; +using Mocha.Transport.InMemory; +using Npgsql; + +namespace Mocha.EntityFrameworkCore.Postgres.Tests; + +public sealed class PostgresOutboxAmbientTransactionSignalTests(PostgresFixture fixture) + : IClassFixture +{ + private static CancellationToken TestToken => TestContext.Current.CancellationToken; + + [Fact] + public async Task PersistAsync_Should_SignalOnlyAfterCommit_When_AmbientTransactionIsUsed() + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var probeConnectionString = new NpgsqlConnectionStringBuilder(connectionString) { Enlist = false } + .ConnectionString; + var signal = new VisibilityProbeSignal(probeConnectionString); + await using var provider = await CreateProviderAsync(connectionString, signal, addHandler: false); + await EnsureCreatedAsync(provider); + + // act + using (var transaction = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) + { + await using var scope = provider.CreateAsyncScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + await bus.PublishAsync(new TestEvent { Payload = "ambient" }, TestToken); + signal.Mark("before-complete"); + transaction.Complete(); + } + + signal.Mark("after-commit"); + + // assert + var events = signal.Events.ToArray(); + var setsBeforeComplete = events.TakeWhile(e => e != "before-complete").Count(e => e.StartsWith("set")); + var setsSeeingNoRows = events.Count(e => e == "set(visible=0)"); + var setsAfterComplete = events.SkipWhile(e => e != "before-complete").Count(e => e.StartsWith("set")); + Assert.Equal((0, 0), (setsBeforeComplete, setsSeeingNoRows)); + Assert.NotEqual(0, setsAfterComplete); + } + + [Fact] + public async Task Worker_Should_DeliverMessage_When_PublishedInsideAmbientTransaction() + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var recorder = new MessageRecorder(); + await using var provider = await CreateProviderAsync( + connectionString, + new ResilientOutboxSignal(), + addHandler: true, + recorder); + await EnsureCreatedAsync(provider); + var hostedServices = provider.GetServices().ToList(); + foreach (var service in hostedServices) + { + await service.StartAsync(TestToken); + } + + try + { + // worker drained the empty table and is parked on the signal + await PublishAsync(provider, "warm-up"); + Assert.True(await recorder.WaitAsync(TimeSpan.FromSeconds(10))); + + // act + using (var transaction = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) + { + await PublishAsync(provider, "ambient"); + + // give the prematurely woken worker time to find nothing and park again + await Task.Delay(TimeSpan.FromSeconds(1), TestToken); + transaction.Complete(); + } + + var delivered = await recorder.WaitAsync(TimeSpan.FromSeconds(10)); + + // assert + Assert.True(delivered, "message committed inside a TransactionScope was not delivered"); + } + finally + { + foreach (var service in hostedServices) + { + await service.StopAsync(CancellationToken.None); + } + + await Task.Delay(250, CancellationToken.None); + } + } + + private static async Task PublishAsync(IServiceProvider provider, string payload) + { + await using var scope = provider.CreateAsyncScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + await bus.PublishAsync(new TestEvent { Payload = payload }, TestToken); + } + + private static async Task EnsureCreatedAsync(IServiceProvider provider) + { + await using var scope = provider.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + await db.Database.EnsureCreatedAsync(TestToken); + } + + private static async Task CreateProviderAsync( + string connectionString, + IOutboxSignal signal, + bool addHandler, + MessageRecorder? recorder = null) + { + var services = new ServiceCollection(); + services.AddSingleton(recorder ?? new MessageRecorder()); + services.AddLogging(); + services.AddDbContext(o => o.UseTestNpgsql(connectionString)); + services.AddSingleton(signal); + + var builder = services.AddMessageBus(); + builder.AddEntityFramework(ef => ef.UsePostgresOutbox()); + if (addHandler) + { + builder.AddEventHandler(); + } + + builder.AddInMemory(); + + var provider = services.BuildServiceProvider(); + var runtime = (MessagingRuntime)provider.GetRequiredService(); + await runtime.StartAsync(TestToken); + return provider; + } + + public sealed class TestEvent + { + public required string Payload { get; init; } + } + + public sealed class TestEventHandler(MessageRecorder recorder) : IEventHandler + { + public ValueTask HandleAsync(TestEvent message, CancellationToken cancellationToken) + { + recorder.Record(message); + return default; + } + } + + public sealed class MessageRecorder + { + private readonly SemaphoreSlim _semaphore = new(0); + + public void Record(object message) => _semaphore.Release(); + + public Task WaitAsync(TimeSpan timeout) => _semaphore.WaitAsync(timeout); + } + + private sealed class VisibilityProbeSignal(string connectionString) : IOutboxSignal + { + public ConcurrentQueue Events { get; } = new(); + + public void Mark(string label) => Events.Enqueue(label); + + public void Set() + { + using var connection = new NpgsqlConnection(connectionString); + connection.Open(); + using var command = connection.CreateCommand(); + command.CommandText = "SELECT COUNT(*) FROM outbox_messages"; + Events.Enqueue($"set(visible={command.ExecuteScalar()})"); + } + + public Task WaitAsync(CancellationToken cancellationToken) + => Task.Delay(Timeout.Infinite, cancellationToken); + } +} From cffdfb5b17a44f1dfdcb324c3e5d5f3a26a4a7c7 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:20:59 +0000 Subject: [PATCH 2/6] [Mocha] Fix outbox signalling before ambient transaction commit Outbox writes and SaveChanges inside a TransactionScope now set the outbox signal when the ambient transaction completes instead of immediately, so the worker no longer wakes before the row is visible and parks with a committed message in the table. --- .../Outbox/PostgresMessageOutbox.cs | 12 +-- .../src/Mocha.EntityFrameworkCore/Assembly.cs | 1 + .../Outbox/OutboxSaveChangesInterceptor.cs | 2 +- .../Outbox/OutboxSignalExtensions.cs | 26 ++++++ ...gresOutboxAmbientTransactionSignalTests.cs | 80 ++++++++++++++++--- 5 files changed, 104 insertions(+), 17 deletions(-) create mode 100644 src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs diff --git a/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresMessageOutbox.cs b/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresMessageOutbox.cs index a320c65455b..83936e99d4a 100644 --- a/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresMessageOutbox.cs +++ b/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/PostgresMessageOutbox.cs @@ -3,6 +3,7 @@ using Microsoft.EntityFrameworkCore.Storage; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; +using Mocha.EntityFrameworkCore; using Mocha.Middlewares; using Mocha.Utils; using Npgsql; @@ -16,8 +17,8 @@ namespace Mocha.Outbox; /// /// /// When a database transaction is active on the DbContext, the insert participates in that transaction -/// and the outbox signal is deferred until commit. Without an active transaction, the signal fires -/// immediately after insert to wake the outbox processor. +/// and the outbox signal is deferred until commit. Inside an ambient , +/// the signal fires when that transaction completes. Otherwise, the signal fires immediately after insert. /// internal sealed class PostgresMessageOutbox : IMessageOutbox, IDisposable { @@ -64,8 +65,9 @@ public static PostgresMessageOutbox Create(Type contextType, string optionsName, /// Serializes the message envelope and inserts it into the Postgres outbox table. /// /// - /// If no database transaction is active, the outbox signal is raised immediately to wake the - /// processor. Otherwise, the signal is deferred to the transaction commit interceptor. + /// With an active database transaction, the signal is deferred to the transaction commit interceptor. + /// With an ambient , it is raised when that transaction completes. + /// Otherwise, it is raised immediately. /// /// The message envelope to persist in the outbox. /// A token to observe for cancellation. @@ -98,7 +100,7 @@ public async ValueTask PersistAsync(MessageEnvelope envelope, CancellationToken if (_originalDbContext.Database.CurrentTransaction?.GetDbTransaction() is not NpgsqlTransaction) { - _signal.Set(); + _signal.SetAfterAmbientTransaction(); } } finally diff --git a/src/Mocha/src/Mocha.EntityFrameworkCore/Assembly.cs b/src/Mocha/src/Mocha.EntityFrameworkCore/Assembly.cs index f4e32ba40b4..73cc367eae1 100644 --- a/src/Mocha/src/Mocha.EntityFrameworkCore/Assembly.cs +++ b/src/Mocha/src/Mocha.EntityFrameworkCore/Assembly.cs @@ -2,3 +2,4 @@ [assembly: InternalsVisibleTo("Mocha.Tests")] [assembly: InternalsVisibleTo("Mocha.EntityFrameworkCore.Postgres.Tests")] +[assembly: InternalsVisibleTo("Mocha.EntityFrameworkCore.Postgres")] diff --git a/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSaveChangesInterceptor.cs b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSaveChangesInterceptor.cs index b0c585d2dc5..2855bc19c3a 100644 --- a/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSaveChangesInterceptor.cs +++ b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSaveChangesInterceptor.cs @@ -28,7 +28,7 @@ public override int SavedChanges(SaveChangesCompletedEventData eventData, int re if (context.Database.CurrentTransaction is null) { - signal.Set(); + signal.SetAfterAmbientTransaction(); } return result; diff --git a/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs new file mode 100644 index 00000000000..20ce6c77f2c --- /dev/null +++ b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs @@ -0,0 +1,26 @@ +using System.Transactions; +using Mocha.Outbox; + +namespace Mocha.EntityFrameworkCore; + +internal static class OutboxSignalExtensions +{ + /// + /// Sets the signal when the ambient completes, or immediately when + /// there is no ambient transaction. + /// + /// The outbox signal to set. + public static void SetAfterAmbientTransaction(this IOutboxSignal signal) + { + if (Transaction.Current is { } transaction) + { + SetOnCompletion(signal, transaction); + return; + } + + signal.Set(); + } + + private static void SetOnCompletion(IOutboxSignal signal, Transaction transaction) + => transaction.TransactionCompleted += (_, _) => signal.Set(); +} diff --git a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs index a20e646d90c..28802ae1e50 100644 --- a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs +++ b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using System.Text.Json; using System.Transactions; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; @@ -19,11 +20,10 @@ public async Task PersistAsync_Should_SignalOnlyAfterCommit_When_AmbientTransact { // arrange var connectionString = await fixture.CreateDatabaseAsync(); - var probeConnectionString = new NpgsqlConnectionStringBuilder(connectionString) { Enlist = false } - .ConnectionString; - var signal = new VisibilityProbeSignal(probeConnectionString); + var signal = new VisibilityProbeSignal(connectionString); await using var provider = await CreateProviderAsync(connectionString, signal, addHandler: false); await EnsureCreatedAsync(provider); + signal.Events.Clear(); // act using (var transaction = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) @@ -35,15 +35,60 @@ public async Task PersistAsync_Should_SignalOnlyAfterCommit_When_AmbientTransact transaction.Complete(); } - signal.Mark("after-commit"); + // assert + AssertSignaledOnlyAfterCommit(signal); + } + + [Fact] + public async Task SaveChanges_Should_SignalOnlyAfterCommit_When_AmbientTransactionIsUsed() + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var signal = new VisibilityProbeSignal(connectionString); + await using var provider = await CreateProviderAsync(connectionString, signal, addHandler: false); + await EnsureCreatedAsync(provider); + signal.Events.Clear(); + + // act + using (var transaction = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) + { + await using var scope = provider.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + db.Add(new OutboxMessage(Guid.NewGuid(), JsonDocument.Parse("{}"))); + await db.SaveChangesAsync(TestToken); + signal.Mark("before-complete"); + transaction.Complete(); + } // assert - var events = signal.Events.ToArray(); - var setsBeforeComplete = events.TakeWhile(e => e != "before-complete").Count(e => e.StartsWith("set")); - var setsSeeingNoRows = events.Count(e => e == "set(visible=0)"); - var setsAfterComplete = events.SkipWhile(e => e != "before-complete").Count(e => e.StartsWith("set")); - Assert.Equal((0, 0), (setsBeforeComplete, setsSeeingNoRows)); - Assert.NotEqual(0, setsAfterComplete); + AssertSignaledOnlyAfterCommit(signal); + } + + [Fact] + public async Task PersistAsync_Should_SignalOnCompletion_When_AmbientTransactionRollsBack() + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var signal = new VisibilityProbeSignal(connectionString); + await using var provider = await CreateProviderAsync(connectionString, signal, addHandler: false); + await EnsureCreatedAsync(provider); + signal.Events.Clear(); + + // act + using (new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) + { + await using var scope = provider.CreateAsyncScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + await bus.PublishAsync(new TestEvent { Payload = "ambient" }, TestToken); + signal.Mark("before-dispose"); + } + + signal.Mark("after-rollback"); + + // assert + Assert.Equal( + ["before-dispose", "set(visible=0)", "after-rollback"], + signal.Events.Distinct().ToArray()); } [Fact] @@ -96,6 +141,16 @@ public async Task Worker_Should_DeliverMessage_When_PublishedInsideAmbientTransa } } + private static void AssertSignaledOnlyAfterCommit(VisibilityProbeSignal signal) + { + var events = signal.Events.ToArray(); + var setsBeforeComplete = events.TakeWhile(e => e != "before-complete").Count(e => e.StartsWith("set")); + var setsSeeingNoRows = events.Count(e => e == "set(visible=0)"); + var setsAfterComplete = events.SkipWhile(e => e != "before-complete").Count(e => e.StartsWith("set")); + Assert.Equal((0, 0), (setsBeforeComplete, setsSeeingNoRows)); + Assert.NotEqual(0, setsAfterComplete); + } + private static async Task PublishAsync(IServiceProvider provider, string payload) { await using var scope = provider.CreateAsyncScope(); @@ -162,13 +217,16 @@ public sealed class MessageRecorder private sealed class VisibilityProbeSignal(string connectionString) : IOutboxSignal { + private readonly string _connectionString = + new NpgsqlConnectionStringBuilder(connectionString) { Enlist = false }.ConnectionString; + public ConcurrentQueue Events { get; } = new(); public void Mark(string label) => Events.Enqueue(label); public void Set() { - using var connection = new NpgsqlConnection(connectionString); + using var connection = new NpgsqlConnection(_connectionString); connection.Open(); using var command = connection.CreateCommand(); command.CommandText = "SELECT COUNT(*) FROM outbox_messages"; From 57aaa6dcc1192f7ecebaad0ad6208c83b3bb1be8 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:23:53 +0000 Subject: [PATCH 3/6] [Mocha] Subscribe to ambient transaction completion once per transaction Persisting many outbox messages in one TransactionScope no longer adds a completion handler per message, and a signal disposed during shutdown no longer throws out of the committed scope's Dispose. --- .../Outbox/OutboxSignalExtensions.cs | 27 ++++++++- ...gresOutboxAmbientTransactionSignalTests.cs | 59 +++++++++++++++++++ 2 files changed, 84 insertions(+), 2 deletions(-) diff --git a/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs index 20ce6c77f2c..3c471e517bc 100644 --- a/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs +++ b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs @@ -1,3 +1,4 @@ +using System.Runtime.CompilerServices; using System.Transactions; using Mocha.Outbox; @@ -5,9 +6,11 @@ namespace Mocha.EntityFrameworkCore; internal static class OutboxSignalExtensions { + private static readonly ConditionalWeakTable s_subscriptions = new(); + /// /// Sets the signal when the ambient completes, or immediately when - /// there is no ambient transaction. + /// there is no ambient transaction. The signal is set at most once per ambient transaction. /// /// The outbox signal to set. public static void SetAfterAmbientTransaction(this IOutboxSignal signal) @@ -22,5 +25,25 @@ public static void SetAfterAmbientTransaction(this IOutboxSignal signal) } private static void SetOnCompletion(IOutboxSignal signal, Transaction transaction) - => transaction.TransactionCompleted += (_, _) => signal.Set(); + { + if (s_subscriptions.TryGetValue(transaction, out var subscribed) && ReferenceEquals(subscribed, signal)) + { + return; + } + + s_subscriptions.AddOrUpdate(transaction, signal); + transaction.TransactionCompleted += (_, _) => SetUnlessDisposed(signal); + } + + private static void SetUnlessDisposed(IOutboxSignal signal) + { + try + { + signal.Set(); + } + catch (ObjectDisposedException) + { + // the outbox worker was shut down before the transaction completed + } + } } diff --git a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs index 28802ae1e50..5c355424582 100644 --- a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs +++ b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs @@ -39,6 +39,51 @@ public async Task PersistAsync_Should_SignalOnlyAfterCommit_When_AmbientTransact AssertSignaledOnlyAfterCommit(signal); } + [Fact] + public async Task PersistAsync_Should_SignalOnce_When_ManyMessagesShareAmbientTransaction() + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var signal = new VisibilityProbeSignal(connectionString); + await using var provider = await CreateProviderAsync(connectionString, signal, addHandler: false); + await EnsureCreatedAsync(provider); + signal.Events.Clear(); + + // act + using (var transaction = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) + { + await using var scope = provider.CreateAsyncScope(); + var bus = scope.ServiceProvider.GetRequiredService(); + for (var i = 0; i < 5; i++) + { + await bus.PublishAsync(new TestEvent { Payload = $"ambient-{i}" }, TestToken); + } + + signal.Mark("before-complete"); + transaction.Complete(); + } + + // assert + Assert.Equal(["before-complete", "set(visible=5)"], signal.Events.ToArray()); + } + + [Fact] + public void SetAfterAmbientTransaction_Should_NotThrowOnCommit_When_SignalIsDisposed() + { + // arrange + var signal = new DisposedOutboxSignal(); + var transaction = new TransactionScope(); + signal.SetAfterAmbientTransaction(); + transaction.Complete(); + + // act + var exception = Record.Exception(transaction.Dispose); + + // assert + Assert.Null(exception); + Assert.Equal(1, signal.SetCallCount); + } + [Fact] public async Task SaveChanges_Should_SignalOnlyAfterCommit_When_AmbientTransactionIsUsed() { @@ -215,6 +260,20 @@ public sealed class MessageRecorder public Task WaitAsync(TimeSpan timeout) => _semaphore.WaitAsync(timeout); } + private sealed class DisposedOutboxSignal : IOutboxSignal + { + public int SetCallCount { get; private set; } + + public void Set() + { + SetCallCount++; + throw new ObjectDisposedException(nameof(DisposedOutboxSignal)); + } + + public Task WaitAsync(CancellationToken cancellationToken) + => throw new ObjectDisposedException(nameof(DisposedOutboxSignal)); + } + private sealed class VisibilityProbeSignal(string connectionString) : IOutboxSignal { private readonly string _connectionString = From e6a9915a6f8de8fbc0e78a9d697c923853b99779 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:29:38 +0000 Subject: [PATCH 4/6] [Mocha] Simplify ambient transaction outbox signalling Drop the per-transaction subscription tracking and the disposed-signal guard. Repeated signals collapse in the auto-reset event, so one completion handler per write is sufficient. --- .../Outbox/OutboxSignalExtensions.cs | 27 ++------------ ...gresOutboxAmbientTransactionSignalTests.cs | 35 ++----------------- 2 files changed, 4 insertions(+), 58 deletions(-) diff --git a/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs index 3c471e517bc..20ce6c77f2c 100644 --- a/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs +++ b/src/Mocha/src/Mocha.EntityFrameworkCore/Outbox/OutboxSignalExtensions.cs @@ -1,4 +1,3 @@ -using System.Runtime.CompilerServices; using System.Transactions; using Mocha.Outbox; @@ -6,11 +5,9 @@ namespace Mocha.EntityFrameworkCore; internal static class OutboxSignalExtensions { - private static readonly ConditionalWeakTable s_subscriptions = new(); - /// /// Sets the signal when the ambient completes, or immediately when - /// there is no ambient transaction. The signal is set at most once per ambient transaction. + /// there is no ambient transaction. /// /// The outbox signal to set. public static void SetAfterAmbientTransaction(this IOutboxSignal signal) @@ -25,25 +22,5 @@ public static void SetAfterAmbientTransaction(this IOutboxSignal signal) } private static void SetOnCompletion(IOutboxSignal signal, Transaction transaction) - { - if (s_subscriptions.TryGetValue(transaction, out var subscribed) && ReferenceEquals(subscribed, signal)) - { - return; - } - - s_subscriptions.AddOrUpdate(transaction, signal); - transaction.TransactionCompleted += (_, _) => SetUnlessDisposed(signal); - } - - private static void SetUnlessDisposed(IOutboxSignal signal) - { - try - { - signal.Set(); - } - catch (ObjectDisposedException) - { - // the outbox worker was shut down before the transaction completed - } - } + => transaction.TransactionCompleted += (_, _) => signal.Set(); } diff --git a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs index 5c355424582..03836fb2c4d 100644 --- a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs +++ b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxAmbientTransactionSignalTests.cs @@ -40,7 +40,7 @@ public async Task PersistAsync_Should_SignalOnlyAfterCommit_When_AmbientTransact } [Fact] - public async Task PersistAsync_Should_SignalOnce_When_ManyMessagesShareAmbientTransaction() + public async Task PersistAsync_Should_SignalOnlyAfterCommit_When_ManyMessagesShareAmbientTransaction() { // arrange var connectionString = await fixture.CreateDatabaseAsync(); @@ -64,24 +64,7 @@ public async Task PersistAsync_Should_SignalOnce_When_ManyMessagesShareAmbientTr } // assert - Assert.Equal(["before-complete", "set(visible=5)"], signal.Events.ToArray()); - } - - [Fact] - public void SetAfterAmbientTransaction_Should_NotThrowOnCommit_When_SignalIsDisposed() - { - // arrange - var signal = new DisposedOutboxSignal(); - var transaction = new TransactionScope(); - signal.SetAfterAmbientTransaction(); - transaction.Complete(); - - // act - var exception = Record.Exception(transaction.Dispose); - - // assert - Assert.Null(exception); - Assert.Equal(1, signal.SetCallCount); + AssertSignaledOnlyAfterCommit(signal); } [Fact] @@ -260,20 +243,6 @@ public sealed class MessageRecorder public Task WaitAsync(TimeSpan timeout) => _semaphore.WaitAsync(timeout); } - private sealed class DisposedOutboxSignal : IOutboxSignal - { - public int SetCallCount { get; private set; } - - public void Set() - { - SetCallCount++; - throw new ObjectDisposedException(nameof(DisposedOutboxSignal)); - } - - public Task WaitAsync(CancellationToken cancellationToken) - => throw new ObjectDisposedException(nameof(DisposedOutboxSignal)); - } - private sealed class VisibilityProbeSignal(string connectionString) : IOutboxSignal { private readonly string _connectionString = From aa5aeb8393fa871a7b34b67bfca9025231a8c594 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:32:52 +0000 Subject: [PATCH 5/6] Update docs --- website/content/docs/mocha/reliability.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/website/content/docs/mocha/reliability.md b/website/content/docs/mocha/reliability.md index e02c4fdd8ed..8d1a838b72d 100644 --- a/website/content/docs/mocha/reliability.md +++ b/website/content/docs/mocha/reliability.md @@ -779,7 +779,7 @@ builder.Services ## How the outbox processor works -The outbox processor is a background hosted service (`IHostedService`). When the EF Core interceptors detect a `SaveChanges` or transaction commit, they signal the processor through `IOutboxSignal`. The processor reads pending `OutboxMessage` rows, deserializes the envelope, and dispatches each message through the bus's dispatch pipeline. +The outbox processor is a background hosted service (`IHostedService`). When the EF Core interceptors detect a `SaveChanges` or transaction commit, they signal the processor through `IOutboxSignal`. Inside an ambient `TransactionScope`, the signal is raised when the ambient transaction completes. The processor reads pending `OutboxMessage` rows, deserializes the envelope, and dispatches each message through the bus's dispatch pipeline. The `TimesSent` column tracks dispatch attempts. If dispatch fails, the processor retries on the next signal. Messages are deleted from the outbox table after successful dispatch. From b29b13ccbc41bc3335f6fb0852a5336456e9b5e4 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:43:42 +0000 Subject: [PATCH 6/6] [Mocha] Fix PostgresQueueDescriptor resolving the transport ThrowHelper The Postgres transport's own ThrowHelper shadows Mocha.ThrowHelper, which defines BeforeAndAfterConflict, so the build failed. --- .../Descriptors/PostgresQueueDescriptor.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs index 7495893f83f..36ddcf89ed5 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs @@ -124,7 +124,7 @@ public IPostgresQueueDescriptor UseReceive( { if (before is not null && after is not null) { - throw ThrowHelper.BeforeAndAfterConflict(); + throw Mocha.ThrowHelper.BeforeAndAfterConflict(); } if (before is null && after is null)