GH-5395 A dead letter marked replayable mid-move is no longer deleted without reaching the inbox

src/Persistence/MySql/MySqlTests/dead_letter_failure_group_compliance.cs

@@ -3,6 +3,7 @@
 using Wolverine;
 using Wolverine.ComplianceTests;
 using Wolverine.MySql;
+using Wolverine.RDBMS;
 
 namespace MySqlTests;
 
@@ -17,4 +18,10 @@ public override async Task<IHost> BuildCleanHost()
                 opts.Durability.Mode = DurabilityMode.Solo;
             }).StartAsync();
     }
+
+    // LOCK TABLES is not transactional, so it needs its own release
+    protected override Task<IAsyncDisposable?> blockDeadLetterHistoryWritesAsync() =>
+        holdLockAsync(
+            $"lock tables {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} read",
+            "unlock tables");
 }

src/Persistence/Oracle/OracleTests/dead_letter_failure_group_compliance.cs

@@ -3,6 +3,7 @@
 using Wolverine;
 using Wolverine.ComplianceTests;
 using Wolverine.Oracle;
+using Wolverine.RDBMS;
 
 namespace OracleTests;
 
@@ -17,4 +18,8 @@ public override async Task<IHost> BuildCleanHost()
                 opts.Durability.Mode = DurabilityMode.Solo;
             }).StartAsync();
     }
+
+    protected override Task<IAsyncDisposable?> blockDeadLetterHistoryWritesAsync() =>
+        holdLockAsync(
+            $"lock table {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} in share mode");
 }

src/Persistence/PostgresqlTests/dead_letter_failure_group_compliance.cs

@@ -3,6 +3,7 @@
 using Wolverine;
 using Wolverine.ComplianceTests;
 using Wolverine.Postgresql;
+using Wolverine.RDBMS;
 
 namespace PostgresqlTests;
 
@@ -17,4 +18,8 @@ public override async Task<IHost> BuildCleanHost()
                 opts.Durability.Mode = DurabilityMode.Solo;
             }).StartAsync();
     }
+
+    protected override Task<IAsyncDisposable?> blockDeadLetterHistoryWritesAsync() =>
+        holdLockAsync(
+            $"lock table {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} in share mode");
 }

src/Persistence/SqlServerTests/Persistence/dead_letter_failure_group_compliance.cs

@@ -3,6 +3,7 @@
 using Wolverine;
 using Wolverine.ComplianceTests;
 using Wolverine.SqlServer;
+using Wolverine.RDBMS;
 
 namespace SqlServerTests.Persistence;
 
@@ -20,4 +21,8 @@ public override async Task<IHost> BuildCleanHost()
                 opts.Durability.DeadLetterQueueExpirationEnabled = true;
             }).StartAsync();
     }
+
+    protected override Task<IAsyncDisposable?> blockDeadLetterHistoryWritesAsync() =>
+        holdLockAsync(
+            $"select count(*) from {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} with (tablock, holdlock)");
 }

src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs

@@ -31,28 +31,38 @@ public void ConfigureCommand(DbCommandBuilder builder)
         // semicolon is enough for the providers that concatenate, but Oracle needs a real split
         builder.StartNewCommand();
 
+        // GH-5395. At READ COMMITTED (the PostgreSQL, SQL Server and Oracle default) each statement below takes a
+        // snapshot of its own, so "replayable = true" alone can match a dead letter marked after the insert above
+        // already ran -- and the delete would then drop a row that never reached the inbox. Everything from here on is limited to the dead letters that
+        // insert copied, found by their uncommitted inbox row, which no other transaction can add or remove.
+        var incoming = _database.TableNameFor(DatabaseConstants.IncomingTable);
+        var deadLetters = _database.TableNameFor(DatabaseConstants.DeadLetterTable);
+        var copied =
+            $"{DatabaseConstants.Replayable} = {builder.ParameterPrefix}replayable and exists (select 1 from {incoming} i " +
+            $"where i.{DatabaseConstants.Id} = {deadLetters}.{DatabaseConstants.Id} " +
+            $"and (i.{DatabaseConstants.ReceivedAt} = {deadLetters}.{DatabaseConstants.ReceivedAt} " +
+            $"or (i.{DatabaseConstants.ReceivedAt} is null and {deadLetters}.{DatabaseConstants.ReceivedAt} is null)))";
+
         // GH-5209. The delete below loses the dead letter row, so record its replay history first. The dead letter
         // row already carries any earlier history (the dead letter insert read it), so the new history row simply
         // replaces the old one: delete-then-insert rather than a provider specific upsert. Grouped by id because
         // MessageIdentity.IdAndDestination allows one id per destination.
         var history = _database.TableNameFor(DatabaseConstants.DeadLetterHistoryTable);
-        var deadLetters = _database.TableNameFor(DatabaseConstants.DeadLetterTable);
         builder.Append(
-            $"delete from {history} where {DatabaseConstants.Id} in (select {DatabaseConstants.Id} from {deadLetters} where {DatabaseConstants.Replayable} = {builder.ParameterPrefix}replayable);");
+            $"delete from {history} where {DatabaseConstants.Id} in (select {DatabaseConstants.Id} from {deadLetters} where {copied});");
 
         builder.StartNewCommand();
 
         builder.Append(
             $"insert into {history} ({DatabaseConstants.Id}, {DatabaseConstants.ReplayCount}, {DatabaseConstants.LastReplayedAt}, {DatabaseConstants.FirstFailedAt}) " +
             $"select {DatabaseConstants.Id}, max(coalesce({DatabaseConstants.ReplayCount}, 0)) + 1, {builder.ParameterPrefix}replayed_at, " +
             $"min(coalesce({DatabaseConstants.FirstFailedAt}, {DatabaseConstants.FailedAt}, {DatabaseConstants.SentAt})) " +
-            $"from {deadLetters} where {DatabaseConstants.Replayable} = {builder.ParameterPrefix}replayable group by {DatabaseConstants.Id};");
+            $"from {deadLetters} where {copied} group by {DatabaseConstants.Id};");
         builder.AddNamedParameter("replayed_at", DateTimeOffset.UtcNow);
 
         builder.StartNewCommand();
 
-        builder.Append(
-            $"delete from {_database.TableNameFor(DatabaseConstants.DeadLetterTable)} where {DatabaseConstants.Replayable} = {builder.ParameterPrefix}replayable;");
+        builder.Append($"delete from {deadLetters} where {copied};");
     }
 
     public Task ReadResultsAsync(DbDataReader reader, IList<Exception> exceptions, CancellationToken token)

src/Testing/Wolverine.ComplianceTests/DeadLetterFailureGroupCompliance.cs

@@ -1,3 +1,4 @@
+using System.Data.Common;
 using FailureGroupSamples;
 using JasperFx.Core;
 using JasperFx.Resources;
@@ -605,4 +606,93 @@ public async Task old_replay_history_is_swept_by_the_expiration()
         await thePersistence.Inbox.MoveToDeadLetterStorageAsync(envelope, Failures.CardDeclined("again"));
         (await theDeadLetters.DeadLetterEnvelopeByIdAsync(envelope.Id))!.ReplayCount.ShouldBe(0);
     }
+
+    /// <summary>
+    /// GH-5395. Hold a lock that blocks writes to the dead letter history table until the result is disposed,
+    /// or null for a store that cannot open the race window this way
+    /// </summary>
+    protected virtual Task<IAsyncDisposable?> blockDeadLetterHistoryWritesAsync() =>
+        Task.FromResult<IAsyncDisposable?>(null);
+
+    /// <summary>
+    /// Run <paramref name="lockSql"/> on a connection of its own and keep it open until disposed. A
+    /// transactional lock is released by the commit, any other by <paramref name="releaseSql"/>
+    /// </summary>
+    protected async Task<IAsyncDisposable?> holdLockAsync(string lockSql, string? releaseSql = null)
+    {
+        var database = (IMessageDatabase)thePersistence;
+        var conn = await database.DataSource.OpenConnectionAsync();
+        var tx = releaseSql == null ? await conn.BeginTransactionAsync() : null;
+
+        await using (var cmd = conn.CreateCommand())
+        {
+            cmd.CommandText = lockSql;
+            cmd.Transaction = tx;
+            await cmd.ExecuteNonQueryAsync();
+        }
+
+        return new HeldLock(conn, tx, releaseSql);
+    }
+
+    private sealed class HeldLock(DbConnection conn, DbTransaction? tx, string? releaseSql) : IAsyncDisposable
+    {
+        public async ValueTask DisposeAsync()
+        {
+            if (tx != null)
+            {
+                await tx.CommitAsync();
+            }
+            else
+            {
+                await using var cmd = conn.CreateCommand();
+                cmd.CommandText = releaseSql;
+                await cmd.ExecuteNonQueryAsync();
+            }
+
+            await conn.DisposeAsync();
+        }
+    }
+
+    [Fact]
+    public async Task a_dead_letter_marked_replayable_while_the_move_is_running_is_not_lost()
+    {
+        var envelope = await deadLetter(Failures.CardDeclined("raced"));
+
+        var held = await blockDeadLetterHistoryWritesAsync();
+        if (held == null) Assert.Skip("This store has no way to hold the move between its statements");
+
+        Task moving;
+        Task marking;
+        await using (held)
+        {
+            // The move copies the replayable dead letters (none yet) to the inbox, then stops on the history table
+            moving = moveReplayableDeadLettersAsync();
+            await Task.Delay(1.Seconds());
+            moving.IsCompleted.ShouldBeFalse("The move should be waiting on the history table lock");
+
+            // Marked after the copy and before the delete. Some databases block this until the move commits,
+            // so give it a moment and carry on
+            marking = theDeadLetters.ReplayMatchingAsync(new DeadLetterEnvelopeQuery([envelope.Id]),
+                CancellationToken.None);
+            await Task.WhenAny(marking, Task.Delay(1.Seconds()));
+        }
+
+        await moving.WaitAsync(30.Seconds());
+        await marking.WaitAsync(30.Seconds());
+
+        // Wherever the move left it, it has to be somewhere
+        var inbox = await thePersistence.Admin.AllIncomingAsync();
+        var stillDeadLettered = await theDeadLetters.DeadLetterEnvelopeByIdAsync(envelope.Id);
+        (inbox.Any(x => x.Id == envelope.Id) || stillDeadLettered != null)
+            .ShouldBeTrue("The replayed message is in neither the inbox nor the dead letter queue");
+
+        // and the next pass delivers it, recording one replay
+        await moveReplayableDeadLettersAsync();
+
+        (await thePersistence.Admin.AllIncomingAsync()).ShouldContain(x => x.Id == envelope.Id);
+        (await theDeadLetters.DeadLetterEnvelopeByIdAsync(envelope.Id)).ShouldBeNull();
+
+        await thePersistence.Inbox.MoveToDeadLetterStorageAsync(envelope, Failures.CardDeclined("again"));
+        (await theDeadLetters.DeadLetterEnvelopeByIdAsync(envelope.Id))!.ReplayCount.ShouldBe(1);
+    }
 }