From a25486d255abf520404a79d8fea6a8805dce4b62 Mon Sep 17 00:00:00 2001 From: Robert Dusek Date: Mon, 6 Jul 2026 14:39:40 -0500 Subject: [PATCH] fix(outbox): discard clears incoming envelope after rolled-back outbox commit The Marten and Polecat FlushOutgoingMessagesOnCommit listeners flipped Envelope.Status to Handled in BeforeSaveChangesAsync, before the commit. On rollback (e.g. a duplicate document insert violating the unique PK) the DB row stayed 'Incoming' but the stale in-memory flag made DurableReceiver._markAsHandled skip the real UPDATE, so a .Discard() policy left the envelope stuck as 'Incoming' and it was reprocessed on every InboxStaleTime reclaim / restart. Defer the in-memory status flip to AfterCommitAsync, which only runs once the commit is durable. Polecat's ITransactionParticipant path has no after-commit hook, so it simply no longer sets the premature flag (the idempotent _markAsHandled UPDATE covers the success path). Adds red-green regression tests for both Marten and Polecat. Co-Authored-By: Claude Opus 4.8 --- .../Bug_discard_after_failed_outbox_commit.cs | 106 ++++++++++++++++++ .../Bug_discard_after_failed_outbox_commit.cs | 98 ++++++++++++++++ .../FlushOutgoingMessagesOnCommit.cs | 24 +++- .../FlushOutgoingMessagesOnCommit.cs | 30 ++++- 4 files changed, 255 insertions(+), 3 deletions(-) create mode 100644 src/Persistence/MartenTests/Bugs/Bug_discard_after_failed_outbox_commit.cs create mode 100644 src/Persistence/PolecatTests/Bugs/Bug_discard_after_failed_outbox_commit.cs diff --git a/src/Persistence/MartenTests/Bugs/Bug_discard_after_failed_outbox_commit.cs b/src/Persistence/MartenTests/Bugs/Bug_discard_after_failed_outbox_commit.cs new file mode 100644 index 000000000..785cf6168 --- /dev/null +++ b/src/Persistence/MartenTests/Bugs/Bug_discard_after_failed_outbox_commit.cs @@ -0,0 +1,106 @@ +using IntegrationTests; +using JasperFx; +using JasperFx.Core; +using JasperFx.Resources; +// JasperFx 2.0 lifted DocumentAlreadyExistsException out of Marten.Exceptions +// into the JasperFx namespace; Marten throws the lifted type. +using DocumentAlreadyExistsException = JasperFx.DocumentAlreadyExistsException; +using Marten; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Npgsql; +using Shouldly; +using Weasel.Postgresql; +using Wolverine; +using Wolverine.ErrorHandling; +using Wolverine.Marten; +using Wolverine.Persistence.Durability; +using Wolverine.Runtime.Handlers; +using Wolverine.Tracking; + +namespace MartenTests.Bugs; + +// When a durable-local-queue handler enlists the Marten outbox and SaveChangesAsync +// rolls back (here: a duplicate document insert violates the unique PK), a +// .Discard() error policy must still clear the incoming envelope. The bug was that +// FlushOutgoingMessagesOnCommit.BeforeSaveChangesAsync flipped the in-memory +// Envelope.Status to Handled BEFORE the commit; on rollback the DB row stayed +// 'Incoming' but the stale in-memory flag made DurableReceiver's _markAsHandled +// optimization skip the real UPDATE, stranding the row as 'Incoming' forever. +public class Bug_discard_after_failed_outbox_commit : PostgresqlContext, IAsyncLifetime +{ + private const string Schema = "discard_failed_commit"; + private IHost _host = null!; + + public async Task InitializeAsync() + { + await using (var conn = new NpgsqlConnection(Servers.PostgresConnectionString)) + { + await conn.OpenAsync(); + await conn.DropSchemaAsync(Schema); + await conn.CloseAsync(); + } + + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery() + .IncludeType(typeof(DoThingHandler)); + + opts.Durability.Mode = DurabilityMode.Solo; + opts.Policies.AutoApplyTransactions(); + opts.Policies.UseDurableLocalQueues(); + + opts.Services.AddMarten(m => + { + m.Connection(Servers.PostgresConnectionString); + m.DatabaseSchemaName = Schema; + m.DisableNpgsqlLogging = true; + }).IntegrateWithWolverine(); + + opts.Services.AddResourceSetupOnStartup(); + }).StartAsync(); + } + + public async Task DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + [Fact] + public async Task discarded_duplicate_should_not_be_stuck_in_incoming() + { + var store = _host.Services.GetRequiredService(); + await store.Admin.ClearAllAsync(); + + var id = Guid.NewGuid().ToString(); + + // 1st send: inserts Marker(id), commits, envelope marked Handled. + await _host.TrackActivity().Timeout(30.Seconds()) + .ExecuteAndWaitAsync(c => c.PublishAsync(new DoThing(id))); + + // 2nd send (duplicate): MartenOps.Insert hits the unique PK inside the outbox + // SaveChangesAsync => DocumentAlreadyExistsException => .Discard(). + await _host.TrackActivity().Timeout(30.Seconds()).DoNotAssertOnExceptionsDetected() + .ExecuteAndWaitAsync(c => c.PublishAsync(new DoThing(id))); + + var counts = await store.Admin.FetchCountsAsync(); + + counts.Incoming.ShouldBe(0); + } +} + +public record DoThing(string Id); + +// Marten document; `Id` is the identity => unique PK. A 2nd Insert of the same Id throws. +public record Marker(string Id); + +public static class DoThingHandler +{ + public static void Configure(HandlerChain chain) + => chain.OnException().Discard(); + + public static IMartenOp Handle(DoThing message) + => MartenOps.Insert(new Marker(message.Id)); +} diff --git a/src/Persistence/PolecatTests/Bugs/Bug_discard_after_failed_outbox_commit.cs b/src/Persistence/PolecatTests/Bugs/Bug_discard_after_failed_outbox_commit.cs new file mode 100644 index 000000000..9079807d0 --- /dev/null +++ b/src/Persistence/PolecatTests/Bugs/Bug_discard_after_failed_outbox_commit.cs @@ -0,0 +1,98 @@ +using IntegrationTests; +using JasperFx; +using JasperFx.Core; +using JasperFx.Resources; +// JasperFx 2.0 lifted DocumentAlreadyExistsException into the JasperFx namespace. +using DocumentAlreadyExistsException = JasperFx.DocumentAlreadyExistsException; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Polecat; +using Shouldly; +using Wolverine; +using Wolverine.ErrorHandling; +using Wolverine.Persistence.Durability; +using Wolverine.Polecat; +using Wolverine.Runtime.Handlers; +using Wolverine.Tracking; + +namespace PolecatTests.Bugs; + +// Polecat counterpart of the Wolverine.Marten regression: when a durable-local-queue +// handler enlists the Polecat outbox and SaveChangesAsync rolls back (duplicate document +// insert), a .Discard() policy must still clear the incoming envelope. The bug was that +// FlushOutgoingMessagesOnCommit flipped Envelope.Status to Handled before the commit; on +// rollback the DB row stayed 'Incoming' but the stale in-memory flag made DurableReceiver's +// _markAsHandled optimization skip the real UPDATE, stranding the row as 'Incoming'. +public class Bug_discard_after_failed_outbox_commit : IAsyncLifetime +{ + private const string Schema = "discard_failed_commit_pc"; + private IHost _host = null!; + + public async Task InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery() + .IncludeType(typeof(DoPcThingHandler)); + + opts.Services.AddPolecat(m => + { + m.ConnectionString = Servers.SqlServerConnectionString; + m.DatabaseSchemaName = Schema; + }) + .IntegrateWithWolverine(); + + opts.Durability.Mode = DurabilityMode.Solo; + opts.Policies.AutoApplyTransactions(); + opts.Policies.UseDurableLocalQueues(); + + opts.Services.AddResourceSetupOnStartup(); + }).StartAsync(); + + var store = (DocumentStore)_host.Services.GetRequiredService(); + await store.Database.ApplyAllConfiguredChangesToDatabaseAsync(); + } + + public async Task DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + [Fact] + public async Task discarded_duplicate_should_not_be_stuck_in_incoming() + { + var store = _host.Services.GetRequiredService(); + await store.Admin.ClearAllAsync(); + + var id = Guid.NewGuid().ToString(); + + // 1st send: inserts PcMarker(id), commits, envelope marked Handled. + await _host.TrackActivity().Timeout(30.Seconds()) + .ExecuteAndWaitAsync(c => c.PublishAsync(new DoPcThing(id))); + + // 2nd send (duplicate): PolecatOps.Insert hits the unique PK inside the outbox + // SaveChangesAsync => DocumentAlreadyExistsException => .Discard(). + await _host.TrackActivity().Timeout(30.Seconds()).DoNotAssertOnExceptionsDetected() + .ExecuteAndWaitAsync(c => c.PublishAsync(new DoPcThing(id))); + + var counts = await store.Admin.FetchCountsAsync(); + + counts.Incoming.ShouldBe(0); + } +} + +public record DoPcThing(string Id); + +// Polecat document; `Id` is the identity => unique PK. A 2nd Insert of the same Id throws. +public record PcMarker(string Id); + +public static class DoPcThingHandler +{ + public static void Configure(HandlerChain chain) + => chain.OnException().Discard(); + + public static IPolecatOp Handle(DoPcThing message) + => PolecatOps.Insert(new PcMarker(message.Id)); +} diff --git a/src/Persistence/Wolverine.Marten/FlushOutgoingMessagesOnCommit.cs b/src/Persistence/Wolverine.Marten/FlushOutgoingMessagesOnCommit.cs index 989b4249a..107a1cddf 100644 --- a/src/Persistence/Wolverine.Marten/FlushOutgoingMessagesOnCommit.cs +++ b/src/Persistence/Wolverine.Marten/FlushOutgoingMessagesOnCommit.cs @@ -13,6 +13,16 @@ internal class FlushOutgoingMessagesOnCommit : DocumentSessionListenerBase private readonly MessageContext _context; private readonly PostgresqlMessageStore _messageStore; + // Tracks whether BeforeSaveChangesAsync queued the "mark incoming handled" + // UPDATE into this batch. The in-memory Envelope.Status flag is only flipped + // once the commit actually succeeds (in AfterCommitAsync). Flipping it before + // the commit left it stale on rollback (e.g. a duplicate document insert), so + // DurableReceiver's _markAsHandled optimization — which skips the real UPDATE + // when Status == Handled — would strand the row as 'Incoming' forever and it + // would be reprocessed on every reclaim/restart. See + // Bug_discard_after_failed_outbox_commit. + private bool _queuedHandledUpdate; + public FlushOutgoingMessagesOnCommit(MessageContext context, PostgresqlMessageStore messageStore) { _context = context; @@ -21,6 +31,8 @@ public FlushOutgoingMessagesOnCommit(MessageContext context, PostgresqlMessageSt public override Task BeforeSaveChangesAsync(IDocumentSession session, CancellationToken token) { + _queuedHandledUpdate = false; + // No need to do anything for HTTP requests if (_context.Envelope == null) { @@ -85,7 +97,10 @@ public override Task BeforeSaveChangesAsync(IDocumentSession session, Cancellati var keepUntil = DateTimeOffset.UtcNow.Add(_context.Runtime.Options.Durability.KeepAfterMessageHandling); session.QueueSqlCommand($"update {incomingTableName} set {DatabaseConstants.Status} = '{EnvelopeStatus.Handled}', {DatabaseConstants.KeepUntil} = ? where id = ?", keepUntil, _context.Envelope.Id); - _context.Envelope.Status = EnvelopeStatus.Handled; + + // Defer the in-memory status flip to AfterCommitAsync — the UPDATE + // above is only durable if this batch commits. See _queuedHandledUpdate. + _queuedHandledUpdate = true; } // This was buggy in real usage. @@ -103,6 +118,13 @@ public override Task BeforeSaveChangesAsync(IDocumentSession session, Cancellati public override Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token) { + // The queued mark-handled UPDATE is only durable now that the commit + // succeeded, so it's safe to trust the in-memory optimization flag. + if (_queuedHandledUpdate && _context.Envelope != null) + { + _context.Envelope.Status = EnvelopeStatus.Handled; + } + return _context.FlushOutgoingMessagesAsync(); } } \ No newline at end of file diff --git a/src/Persistence/Wolverine.Polecat/FlushOutgoingMessagesOnCommit.cs b/src/Persistence/Wolverine.Polecat/FlushOutgoingMessagesOnCommit.cs index bd5697e2a..aa56d7d90 100644 --- a/src/Persistence/Wolverine.Polecat/FlushOutgoingMessagesOnCommit.cs +++ b/src/Persistence/Wolverine.Polecat/FlushOutgoingMessagesOnCommit.cs @@ -17,6 +17,13 @@ internal class FlushOutgoingMessagesOnCommit : IDocumentSessionListener private readonly MessageContext _context; private readonly SqlServerMessageStore _messageStore; + // Only flip the in-memory Envelope.Status to Handled once the commit actually + // succeeds (AfterCommitAsync). Setting it pre-commit left the flag stale on + // rollback, so DurableReceiver's _markAsHandled optimization (which skips the + // real UPDATE when Status == Handled) would strand the row as 'Incoming'. + // Mirrors the Wolverine.Marten FlushOutgoingMessagesOnCommit fix. + private bool _queuedHandledUpdate; + public FlushOutgoingMessagesOnCommit(MessageContext context, SqlServerMessageStore messageStore) { _context = context; @@ -25,6 +32,8 @@ public FlushOutgoingMessagesOnCommit(MessageContext context, SqlServerMessageSto public Task BeforeSaveChangesAsync(IDocumentSession session, CancellationToken token) { + _queuedHandledUpdate = false; + // No need to do anything for HTTP requests if (_context.Envelope == null) { @@ -45,7 +54,10 @@ public Task BeforeSaveChangesAsync(IDocumentSession session, CancellationToken t // Use ITransactionParticipant to execute the SQL in the same transaction session.AddTransactionParticipant(new MarkIncomingAsHandledParticipant( _messageStore.IncomingFullName, _context.Envelope.Id, keepUntil)); - _context.Envelope.Status = EnvelopeStatus.Handled; + + // Defer the in-memory status flip to AfterCommitAsync — the UPDATE + // above is only durable if this batch commits. See _queuedHandledUpdate. + _queuedHandledUpdate = true; } } @@ -54,6 +66,13 @@ public Task BeforeSaveChangesAsync(IDocumentSession session, CancellationToken t public Task AfterCommitAsync(IDocumentSession session, CancellationToken token) { + // The queued mark-handled UPDATE is only durable now that the commit + // succeeded, so it's safe to trust the in-memory optimization flag. + if (_queuedHandledUpdate && _context.Envelope != null) + { + _context.Envelope.Status = EnvelopeStatus.Handled; + } + return _context.FlushOutgoingMessagesAsync(); } } @@ -91,7 +110,14 @@ public async Task BeforeCommitAsync(SqlConnection connection, SqlTransaction tra cmd.Parameters.AddWithValue("@keepUntil", keepUntil); cmd.Parameters.AddWithValue("@id", _context.Envelope.Id); await cmd.ExecuteNonQueryAsync(token); - _context.Envelope.Status = EnvelopeStatus.Handled; + + // Deliberately do NOT flip _context.Envelope.Status to Handled here: this + // runs inside BeforeCommitAsync, before the transaction commits, and + // ITransactionParticipant exposes no after-commit hook to defer to. Setting + // it pre-commit left a stale flag on rollback that made DurableReceiver skip + // the real mark-handled UPDATE, stranding the row as 'Incoming'. The + // (idempotent) UPDATE via DurableReceiver._markAsHandled covers the success + // path instead. } } }