fix(outbox): discard clears incoming envelope after rolled-back outbox

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<IMessageStore>();
+        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<DocumentAlreadyExistsException>().Discard();
+
+    public static IMartenOp Handle(DoThing message)
+        => MartenOps.Insert(new Marker(message.Id));
+}

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<IDocumentStore>();
+        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<IMessageStore>();
+        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<DocumentAlreadyExistsException>().Discard();
+
+    public static IPolecatOp Handle(DoPcThing message)
+        => PolecatOps.Insert(new PcMarker(message.Id));
+}

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();
     }
 }

src/Persistence/Wolverine.Polecat/FlushOutgoingMessagesOnCommit.cs

@@ -18,6 +18,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;
@@ -26,6 +33,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)
         {
@@ -46,7 +55,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;
             }
         }
 
@@ -55,6 +67,13 @@ public Task BeforeSaveChangesAsync(IDocumentSession session, CancellationToken t
 
     public 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();
     }
 }
@@ -92,7 +111,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.
         }
     }
 }