From 090eeb32cd69634fac83bc9e4d30eb20d16382f7 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Fri, 6 Nov 2026 06:26:22 -0600 Subject: [PATCH] GH-5395 A dead letter marked replayable mid-move is no longer deleted without reaching the inbox MoveReplayableErrorMessagesToIncomingOperation's four statements each selected replayable = true on their own. At READ COMMITTED (PostgreSQL, SQL Server, Oracle) a dead letter marked after the inbox insert ran was deleted by the final statement without ever being copied. The history and delete statements now act only on dead letters with a matching (uncommitted) inbox row from the insert. New compliance test holds the move on a history table lock to open the window deterministically; red before the fix on PostgreSQL, SQL Server and Oracle. Co-Authored-By: Claude Opus 5.5 --- .../dead_letter_failure_group_compliance.cs | 7 ++ .../dead_letter_failure_group_compliance.cs | 5 ++ .../dead_letter_failure_group_compliance.cs | 5 ++ .../dead_letter_failure_group_compliance.cs | 5 ++ ...layableErrorMessagesToIncomingOperation.cs | 20 +++-- .../DeadLetterFailureGroupCompliance.cs | 90 +++++++++++++++++++ 6 files changed, 127 insertions(+), 5 deletions(-) diff --git a/src/Persistence/MySql/MySqlTests/dead_letter_failure_group_compliance.cs b/src/Persistence/MySql/MySqlTests/dead_letter_failure_group_compliance.cs index c7bd5de8a..c58b21167 100644 --- a/src/Persistence/MySql/MySqlTests/dead_letter_failure_group_compliance.cs +++ b/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 BuildCleanHost() opts.Durability.Mode = DurabilityMode.Solo; }).StartAsync(); } + + // LOCK TABLES is not transactional, so it needs its own release + protected override Task blockDeadLetterHistoryWritesAsync() => + holdLockAsync( + $"lock tables {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} read", + "unlock tables"); } diff --git a/src/Persistence/Oracle/OracleTests/dead_letter_failure_group_compliance.cs b/src/Persistence/Oracle/OracleTests/dead_letter_failure_group_compliance.cs index 8b51cc942..30a92170d 100644 --- a/src/Persistence/Oracle/OracleTests/dead_letter_failure_group_compliance.cs +++ b/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 BuildCleanHost() opts.Durability.Mode = DurabilityMode.Solo; }).StartAsync(); } + + protected override Task blockDeadLetterHistoryWritesAsync() => + holdLockAsync( + $"lock table {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} in share mode"); } diff --git a/src/Persistence/PostgresqlTests/dead_letter_failure_group_compliance.cs b/src/Persistence/PostgresqlTests/dead_letter_failure_group_compliance.cs index e22648b7f..f0d721797 100644 --- a/src/Persistence/PostgresqlTests/dead_letter_failure_group_compliance.cs +++ b/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 BuildCleanHost() opts.Durability.Mode = DurabilityMode.Solo; }).StartAsync(); } + + protected override Task blockDeadLetterHistoryWritesAsync() => + holdLockAsync( + $"lock table {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} in share mode"); } diff --git a/src/Persistence/SqlServerTests/Persistence/dead_letter_failure_group_compliance.cs b/src/Persistence/SqlServerTests/Persistence/dead_letter_failure_group_compliance.cs index a818e82f2..20b1f0a41 100644 --- a/src/Persistence/SqlServerTests/Persistence/dead_letter_failure_group_compliance.cs +++ b/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 BuildCleanHost() opts.Durability.DeadLetterQueueExpirationEnabled = true; }).StartAsync(); } + + protected override Task blockDeadLetterHistoryWritesAsync() => + holdLockAsync( + $"select count(*) from {((IMessageDatabase)thePersistence).TableNameFor(DatabaseConstants.DeadLetterHistoryTable)} with (tablock, holdlock)"); } diff --git a/src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs b/src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs index 780c1bf4b..d2177fb6f 100644 --- a/src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs +++ b/src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs @@ -31,14 +31,25 @@ 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(); @@ -46,13 +57,12 @@ public void ConfigureCommand(DbCommandBuilder builder) $"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 exceptions, CancellationToken token) diff --git a/src/Testing/Wolverine.ComplianceTests/DeadLetterFailureGroupCompliance.cs b/src/Testing/Wolverine.ComplianceTests/DeadLetterFailureGroupCompliance.cs index 9ec84316a..fdf314136 100644 --- a/src/Testing/Wolverine.ComplianceTests/DeadLetterFailureGroupCompliance.cs +++ b/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); } + + /// + /// 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 + /// + protected virtual Task blockDeadLetterHistoryWritesAsync() => + Task.FromResult(null); + + /// + /// Run on a connection of its own and keep it open until disposed. A + /// transactional lock is released by the commit, any other by + /// + protected async Task 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); + } }