From 76dc179a6a7837133b14cf5303ed8578ee697110 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Fri, 6 Nov 2026 16:18:18 -0600 Subject: [PATCH] Scheduled retries on the SQL Server, MySQL, Oracle and SQLite queues are atomic A scheduled retry of a message received on a database-backed queue deletes its inbox row and writes the envelope to the queue's scheduled table. On SQL Server, MySQL and SQLite the two statements ran autocommitted in one command, and on Oracle as two separate commands, so a scheduled write that failed left the inbox row deleted with nothing scheduled. The message then existed only in the receiver's in-memory retry block, and a node that died or shut down before that retry succeeded lost it. PostgreSQL was already atomic (one Npgsql command runs as one implicit transaction). Found by the out-of-process durability chaos rig: on SQL Server with a 5 second command timeout, ScheduleRetryAsync timed out and 3 of 18,030 messages, all ones that had gone through a scheduled retry, were in no table at the end. Each sender now runs the delete and the scheduled write in one ADO.NET transaction, like the GH-5107 outbox moves beside it. On SQL Server it is a client-side transaction rather than BEGIN TRANSACTION in the batch, so a command timeout's attention cannot leave it open on a pooled connection. Tests (one per store): a failed scheduled write -- a trigger, or on Oracle a message type too long for the column -- now leaves the inbox row in place, and the receiver's retry completes the move once the write can succeed. Each fails without its fix. Co-Authored-By: Claude Opus 5.5 --- .../Transport/scheduled_retry_is_atomic.cs | 124 ++++++++++++++++++ .../Transport/MySqlQueueSender.cs | 11 +- .../Transport/scheduled_retry_is_atomic.cs | 107 +++++++++++++++ .../Transport/OracleQueueSender.cs | 10 +- .../Transport/scheduled_retry_is_atomic.cs | 123 +++++++++++++++++ .../Transport/scheduled_retry_is_atomic.cs | 115 ++++++++++++++++ .../Transport/SqlServerQueueSender.cs | 8 ++ .../Transport/SqliteQueueSender.cs | 11 +- 8 files changed, 501 insertions(+), 8 deletions(-) create mode 100644 src/Persistence/MySql/MySqlTests/Transport/scheduled_retry_is_atomic.cs create mode 100644 src/Persistence/Oracle/OracleTests/Transport/scheduled_retry_is_atomic.cs create mode 100644 src/Persistence/SqlServerTests/Transport/scheduled_retry_is_atomic.cs create mode 100644 src/Persistence/SqliteTests/Transport/scheduled_retry_is_atomic.cs diff --git a/src/Persistence/MySql/MySqlTests/Transport/scheduled_retry_is_atomic.cs b/src/Persistence/MySql/MySqlTests/Transport/scheduled_retry_is_atomic.cs new file mode 100644 index 000000000..716fccbf1 --- /dev/null +++ b/src/Persistence/MySql/MySqlTests/Transport/scheduled_retry_is_atomic.cs @@ -0,0 +1,124 @@ +using IntegrationTests; +using MySqlConnector; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Wolverine; +using Wolverine.Configuration; +using Wolverine.MySql; +using Wolverine.MySql.Transport; +using Wolverine.Tracking; + +namespace MySqlTests.Transport; + +/// +/// A scheduled retry of a message received on a MySQL queue deletes its inbox row and writes it to the +/// queue's scheduled table. Those were two autocommitted statements in one command, so a scheduled write that failed -- a command +/// timeout or a deadlock, found by the durability chaos rig -- left the message in neither table, alive only in +/// the receiver's in-memory retry, and a node that died or shut down before that retry succeeded lost it. +/// +public class scheduled_retry_is_atomic : IAsyncLifetime +{ + private const string SchemaName = "wolverine_schedretry"; + private const string QueueName = "schedretry"; + + private IHost _host = null!; + private MySqlQueue _queue = null!; + + public async ValueTask InitializeAsync() + { + await using (var conn = new MySqlConnection(Servers.MySqlConnectionString)) + { + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"DROP DATABASE IF EXISTS {SchemaName}; CREATE DATABASE {SchemaName}"; + await cmd.ExecuteNonQueryAsync(); + } + + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery(); + opts.Durability.Mode = DurabilityMode.Solo; + opts.UseMySqlPersistenceAndTransport(Servers.MySqlConnectionString, SchemaName, + transportSchema: SchemaName); + + // Declared without a listener, so nothing but the test touches the queue + opts.PublishMessage().ToMySqlQueue(QueueName); + }).StartAsync(); + + _queue = _host.GetRuntime().Options.Transports.GetOrCreate().Queues[QueueName]; + _queue.Mode = EndpointMode.Durable; + await _queue.PurgeAsync(NullLogger.Instance); + await _host.GetRuntime().Storage.Admin.ClearAllAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + /// + /// What a durable listener has done by the time the handler fails: the envelope sits in the inbox + /// + private async Task receivedIntoTheInboxAsync() + { + var runtime = _host.GetRuntime(); + var envelope = runtime.RoutingFor(typeof(MySqlScheduledRetryMessage)) + .RouteForSend(new MySqlScheduledRetryMessage(Guid.NewGuid()), null).Single(); + envelope.Destination = _queue.Uri; + await runtime.Storage.Inbox.StoreIncomingAsync(envelope); + + envelope.ScheduledTime = DateTimeOffset.UtcNow.AddHours(1); + return envelope; + } + + private async Task isInTheInboxAsync(Envelope envelope) => + (await _host.GetRuntime().Storage.Admin.AllIncomingAsync()).Any(x => x.Id == envelope.Id); + + private async Task executeAsync(string sql) + { + await using var conn = new MySqlConnection(Servers.MySqlConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + await cmd.ExecuteNonQueryAsync(); + } + + [Fact] + public async Task a_scheduled_retry_moves_the_inbox_row_to_the_scheduled_table() + { + var envelope = await receivedIntoTheInboxAsync(); + + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } + + [Fact] + public async Task a_failed_scheduled_write_leaves_the_inbox_row_in_place() + { + var envelope = await receivedIntoTheInboxAsync(); + + await executeAsync( + $"CREATE TRIGGER {SchemaName}.fail_scheduled_writes BEFORE INSERT ON {_queue.ScheduledTable.Identifier} " + + "FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'the scheduled write fails'"); + + await Should.ThrowAsync(() => _queue.ScheduleRetryAsync(envelope, CancellationToken.None)); + + // Still in the inbox, so recovery runs it again if this node goes away + (await isInTheInboxAsync(envelope)).ShouldBeTrue(); + (await _queue.ScheduledCountAsync()).ShouldBe(0); + + // and the receiver's retry completes the move once the write can succeed + await executeAsync($"DROP TRIGGER {SchemaName}.fail_scheduled_writes"); + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } +} + +public record MySqlScheduledRetryMessage(Guid Id); diff --git a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs index e13158646..9cee31991 100644 --- a/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs +++ b/src/Persistence/MySql/Wolverine.MySql/Transport/MySqlQueueSender.cs @@ -114,13 +114,18 @@ public async Task ScheduleRetryAsync(Envelope envelope, CancellationToken cancel await using var conn = await _dataSource.OpenConnectionAsync(cancellationToken); try { - await conn.CreateCommand($"DELETE FROM {_queue.Parent.MessageStorageSchemaName}.{DatabaseConstants.IncomingTable} WHERE id = @id;" + _writeDirectlyToTheScheduledTable) + // The inbox delete and the scheduled write commit together or not at all: autocommitted one by one, a + // failed scheduled write left the inbox row deleted with nothing scheduled + await using var tx = await conn.BeginTransactionAsync(cancellationToken); + var cmd = conn.CreateCommand($"DELETE FROM {_queue.Parent.MessageStorageSchemaName}.{DatabaseConstants.IncomingTable} WHERE id = @id;" + _writeDirectlyToTheScheduledTable) .With("id", envelope.Id) .With("body", EnvelopeSerializer.Serialize(envelope)) .With("type", envelope.MessageType) .With("expires", envelope.DeliverBy) - .With("time", envelope.ScheduledTime) - .ExecuteNonQueryAsync(cancellationToken); + .With("time", envelope.ScheduledTime); + cmd.Transaction = tx; + await cmd.ExecuteNonQueryAsync(cancellationToken); + await tx.CommitAsync(cancellationToken); } finally { diff --git a/src/Persistence/Oracle/OracleTests/Transport/scheduled_retry_is_atomic.cs b/src/Persistence/Oracle/OracleTests/Transport/scheduled_retry_is_atomic.cs new file mode 100644 index 000000000..0a22ec650 --- /dev/null +++ b/src/Persistence/Oracle/OracleTests/Transport/scheduled_retry_is_atomic.cs @@ -0,0 +1,107 @@ +using IntegrationTests; +using Oracle.ManagedDataAccess.Client; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Wolverine; +using Wolverine.Configuration; +using Wolverine.Oracle; +using Wolverine.Oracle.Transport; +using Wolverine.Tracking; + +namespace OracleTests.Transport; + +/// +/// A scheduled retry of a message received on an Oracle queue deletes its inbox row and writes it to the +/// queue's scheduled table. Those were two separately committed commands, so a scheduled write that failed -- a command +/// timeout or a deadlock, found by the durability chaos rig -- left the message in neither table, alive only in +/// the receiver's in-memory retry, and a node that died or shut down before that retry succeeded lost it. +/// +public class scheduled_retry_is_atomic : IAsyncLifetime +{ + private const string QueueName = "schedretry"; + + private IHost _host = null!; + private OracleQueue _queue = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery(); + opts.Durability.Mode = DurabilityMode.Solo; + opts.PersistMessagesWithOracle(Servers.OracleConnectionString, "WOLVERINE") + .EnableMessageTransport(); + + // Declared without a listener, so nothing but the test touches the queue + opts.PublishMessage().ToOracleQueue(QueueName); + }).StartAsync(); + + _queue = _host.GetRuntime().Options.Transports.GetOrCreate().Queues[QueueName]; + _queue.Mode = EndpointMode.Durable; + await _queue.PurgeAsync(NullLogger.Instance); + await _host.GetRuntime().Storage.Admin.ClearAllAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + /// + /// What a durable listener has done by the time the handler fails: the envelope sits in the inbox + /// + private async Task receivedIntoTheInboxAsync() + { + var runtime = _host.GetRuntime(); + var envelope = runtime.RoutingFor(typeof(OracleScheduledRetryMessage)) + .RouteForSend(new OracleScheduledRetryMessage(Guid.NewGuid()), null).Single(); + envelope.Destination = _queue.Uri; + await runtime.Storage.Inbox.StoreIncomingAsync(envelope); + + envelope.ScheduledTime = DateTimeOffset.UtcNow.AddHours(1); + return envelope; + } + + private async Task isInTheInboxAsync(Envelope envelope) => + (await _host.GetRuntime().Storage.Admin.AllIncomingAsync()).Any(x => x.Id == envelope.Id); + + [Fact] + public async Task a_scheduled_retry_moves_the_inbox_row_to_the_scheduled_table() + { + var envelope = await receivedIntoTheInboxAsync(); + + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } + + [Fact] + public async Task a_failed_scheduled_write_leaves_the_inbox_row_in_place() + { + var envelope = await receivedIntoTheInboxAsync(); + + // Too long for the scheduled table's VARCHAR2(500) message_type, so the write fails with ORA-12899 after the + // delete -- the test user may not create a trigger in the shared WOLVERINE schema + var messageType = envelope.MessageType; + envelope.MessageType = new string('x', 600); + + await Should.ThrowAsync(() => _queue.ScheduleRetryAsync(envelope, CancellationToken.None)); + + // Still in the inbox, so recovery runs it again if this node goes away + (await isInTheInboxAsync(envelope)).ShouldBeTrue(); + (await _queue.ScheduledCountAsync()).ShouldBe(0); + + // and the receiver's retry completes the move once the write can succeed + envelope.MessageType = messageType; + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } +} + +public record OracleScheduledRetryMessage(Guid Id); diff --git a/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueueSender.cs b/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueueSender.cs index 2df07bfe5..d09c71c67 100644 --- a/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueueSender.cs +++ b/src/Persistence/Oracle/Wolverine.Oracle/Transport/OracleQueueSender.cs @@ -62,13 +62,19 @@ public async Task ScheduleRetryAsync(Envelope envelope, CancellationToken cancel await using var conn = await _dataSource.OpenConnectionAsync(cancellationToken); try { - // Delete from incoming, write to scheduled + // Delete from incoming, write to scheduled -- in one transaction. Committed one by one, a failed + // scheduled write left the inbox row deleted with nothing scheduled + await using var tx = (OracleTransaction)await conn.BeginTransactionAsync(cancellationToken); + await using var deleteCmd = conn.CreateCommand( $"DELETE FROM {_queue.Parent.MessageStorageSchemaName}.{DatabaseConstants.IncomingTable} WHERE id = :id"); + deleteCmd.Transaction = tx; deleteCmd.With("id", envelope.Id); await deleteCmd.ExecuteNonQueryAsync(cancellationToken); - await writeToScheduledTableAsync(envelope, cancellationToken, conn); + await writeToScheduledTableAsync(envelope, cancellationToken, conn, tx); + + await tx.CommitAsync(cancellationToken); } finally { diff --git a/src/Persistence/SqlServerTests/Transport/scheduled_retry_is_atomic.cs b/src/Persistence/SqlServerTests/Transport/scheduled_retry_is_atomic.cs new file mode 100644 index 000000000..a1996cb0e --- /dev/null +++ b/src/Persistence/SqlServerTests/Transport/scheduled_retry_is_atomic.cs @@ -0,0 +1,123 @@ +using IntegrationTests; +using Microsoft.Data.SqlClient; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Weasel.SqlServer; +using Wolverine; +using Wolverine.Configuration; +using Wolverine.SqlServer; +using Wolverine.SqlServer.Transport; +using Wolverine.Tracking; + +namespace SqlServerTests.Transport; + +/// +/// A scheduled retry of a message received on a SQL Server queue deletes its inbox row and writes it to the +/// queue's scheduled table. Those were two autocommitted statements, so a scheduled write that failed -- a command +/// timeout or a deadlock, found by the durability chaos rig -- left the message in neither table, alive only in +/// the receiver's in-memory retry, and a node that died or shut down before that retry succeeded lost it. +/// +public class scheduled_retry_is_atomic : IAsyncLifetime +{ + private const string SchemaName = "schedretry"; + private const string QueueName = "schedretry"; + + private IHost _host = null!; + private SqlServerQueue _queue = null!; + + public async ValueTask InitializeAsync() + { + await using (var conn = new SqlConnection(Servers.SqlServerConnectionString)) + { + await conn.OpenAsync(); + await conn.DropSchemaAsync(SchemaName); + } + + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery(); + opts.Durability.Mode = DurabilityMode.Solo; + opts.UseSqlServerPersistenceAndTransport(Servers.SqlServerConnectionString, SchemaName, + transportSchema: SchemaName).AutoProvision(); + + // Declared without a listener, so nothing but the test touches the queue + opts.PublishMessage().ToSqlServerQueue(QueueName); + }).StartAsync(); + + _queue = _host.GetRuntime().Options.Transports.OfType().Single().Queues[QueueName]; + _queue.Mode = EndpointMode.Durable; + await _queue.PurgeAsync(NullLogger.Instance); + await _host.GetRuntime().Storage.Admin.ClearAllAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + /// + /// What a durable listener has done by the time the handler fails: the envelope sits in the inbox + /// + private async Task receivedIntoTheInboxAsync() + { + var runtime = _host.GetRuntime(); + var envelope = runtime.RoutingFor(typeof(ScheduledRetryMessage)) + .RouteForSend(new ScheduledRetryMessage(Guid.NewGuid()), null).Single(); + envelope.Destination = _queue.Uri; + await runtime.Storage.Inbox.StoreIncomingAsync(envelope); + + envelope.ScheduledTime = DateTimeOffset.UtcNow.AddHours(1); + return envelope; + } + + private async Task isInTheInboxAsync(Envelope envelope) => + (await _host.GetRuntime().Storage.Admin.AllIncomingAsync()).Any(x => x.Id == envelope.Id); + + private async Task executeAsync(string sql) + { + await using var conn = new SqlConnection(Servers.SqlServerConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + await cmd.ExecuteNonQueryAsync(); + } + + [Fact] + public async Task a_scheduled_retry_moves_the_inbox_row_to_the_scheduled_table() + { + var envelope = await receivedIntoTheInboxAsync(); + + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } + + [Fact] + public async Task a_failed_scheduled_write_leaves_the_inbox_row_in_place() + { + var envelope = await receivedIntoTheInboxAsync(); + + await executeAsync( + $"create trigger {SchemaName}.fail_scheduled_writes on {_queue.ScheduledTable.Identifier} after insert, update " + + "as throw 50000, 'the scheduled write fails', 1;"); + + await Should.ThrowAsync(() => _queue.ScheduleRetryAsync(envelope, CancellationToken.None)); + + // Still in the inbox, so recovery runs it again if this node goes away + (await isInTheInboxAsync(envelope)).ShouldBeTrue(); + (await _queue.ScheduledCountAsync()).ShouldBe(0); + + // and the receiver's retry completes the move once the write can succeed + await executeAsync($"drop trigger {SchemaName}.fail_scheduled_writes"); + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } +} + +public record ScheduledRetryMessage(Guid Id); diff --git a/src/Persistence/SqliteTests/Transport/scheduled_retry_is_atomic.cs b/src/Persistence/SqliteTests/Transport/scheduled_retry_is_atomic.cs new file mode 100644 index 000000000..a38347aae --- /dev/null +++ b/src/Persistence/SqliteTests/Transport/scheduled_retry_is_atomic.cs @@ -0,0 +1,115 @@ +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Wolverine; +using Wolverine.Configuration; +using Wolverine.Sqlite; +using Wolverine.Sqlite.Transport; +using Wolverine.Tracking; + +namespace SqliteTests.Transport; + +/// +/// A scheduled retry of a message received on a SQLite queue deletes its inbox row and writes it to the +/// queue's scheduled table. Those were two autocommitted statements in one command, so a scheduled write that failed -- a command +/// timeout or a deadlock, found by the durability chaos rig -- left the message in neither table, alive only in +/// the receiver's in-memory retry, and a node that died or shut down before that retry succeeded lost it. +/// +public class scheduled_retry_is_atomic : IAsyncLifetime +{ + private const string QueueName = "schedretry"; + + private readonly SqliteTestDatabase _database = Servers.CreateDatabase(nameof(scheduled_retry_is_atomic)); + private IHost _host = null!; + private SqliteQueue _queue = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery(); + opts.Durability.Mode = DurabilityMode.Solo; + opts.UseSqlitePersistenceAndTransport(_database.ConnectionString).AutoProvision(); + + // Declared without a listener, so nothing but the test touches the queue + opts.PublishMessage().ToSqliteQueue(QueueName); + }).StartAsync(); + + _queue = _host.GetRuntime().Options.Transports.GetOrCreate().Queues[QueueName]; + _queue.Mode = EndpointMode.Durable; + await _queue.PurgeAsync(NullLogger.Instance); + await _host.GetRuntime().Storage.Admin.ClearAllAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + _database.Dispose(); + } + + /// + /// What a durable listener has done by the time the handler fails: the envelope sits in the inbox + /// + private async Task receivedIntoTheInboxAsync() + { + var runtime = _host.GetRuntime(); + var envelope = runtime.RoutingFor(typeof(SqliteScheduledRetryMessage)) + .RouteForSend(new SqliteScheduledRetryMessage(Guid.NewGuid()), null).Single(); + envelope.Destination = _queue.Uri; + await runtime.Storage.Inbox.StoreIncomingAsync(envelope); + + envelope.ScheduledTime = DateTimeOffset.UtcNow.AddHours(1); + return envelope; + } + + private async Task isInTheInboxAsync(Envelope envelope) => + (await _host.GetRuntime().Storage.Admin.AllIncomingAsync()).Any(x => x.Id == envelope.Id); + + private async Task executeAsync(string sql) + { + await using var conn = new SqliteConnection(_database.ConnectionString); + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = sql; + await cmd.ExecuteNonQueryAsync(); + } + + [Fact] + public async Task a_scheduled_retry_moves_the_inbox_row_to_the_scheduled_table() + { + var envelope = await receivedIntoTheInboxAsync(); + + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } + + [Fact] + public async Task a_failed_scheduled_write_leaves_the_inbox_row_in_place() + { + var envelope = await receivedIntoTheInboxAsync(); + + await executeAsync( + $"CREATE TRIGGER fail_scheduled_writes BEFORE INSERT ON {_queue.ScheduledTable.Identifier} " + + "BEGIN SELECT RAISE(ABORT, 'the scheduled write fails'); END;"); + + await Should.ThrowAsync(() => _queue.ScheduleRetryAsync(envelope, CancellationToken.None)); + + // Still in the inbox, so recovery runs it again if this node goes away + (await isInTheInboxAsync(envelope)).ShouldBeTrue(); + (await _queue.ScheduledCountAsync()).ShouldBe(0); + + // and the receiver's retry completes the move once the write can succeed + await executeAsync("DROP TRIGGER fail_scheduled_writes"); + await _queue.ScheduleRetryAsync(envelope, CancellationToken.None); + + (await isInTheInboxAsync(envelope)).ShouldBeFalse(); + (await _queue.ScheduledCountAsync()).ShouldBe(1); + } +} + +public record SqliteScheduledRetryMessage(Guid Id); diff --git a/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueueSender.cs b/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueueSender.cs index d0fe6f3e9..74a43ba32 100644 --- a/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueueSender.cs +++ b/src/Persistence/Wolverine.SqlServer/Transport/SqlServerQueueSender.cs @@ -120,13 +120,21 @@ public async Task ScheduleRetryAsync(Envelope envelope, CancellationToken cancel await conn.OpenAsync(cancellationToken); try { + // The inbox delete and the scheduled write commit together or not at all. Run as two autocommitted + // statements, a scheduled write that failed (a timeout, a deadlock) left the inbox row deleted with + // nothing scheduled, and the message lived on only in this node's memory until the retry block tried + // again -- a node that died or shut down first lost it. A client-side transaction, rather than + // BEGIN TRANSACTION in the batch, so a command timeout's attention rolls it back with the dispose. + await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(cancellationToken); await using var cmd = conn.CreateCommand(_deleteFromIncomingAndScheduleSql) .With("id", envelope.Id) .With("body", EnvelopeSerializer.Serialize(envelope)) .With("type", envelope.MessageType!) .With("expires", envelope.DeliverBy!) .With("time", envelope.ScheduledTime!); + cmd.Transaction = tx; await cmd.ExecuteNonQueryAsync(cancellationToken); + await tx.CommitAsync(cancellationToken); } finally { diff --git a/src/Persistence/Wolverine.Sqlite/Transport/SqliteQueueSender.cs b/src/Persistence/Wolverine.Sqlite/Transport/SqliteQueueSender.cs index 76bb2562e..e9a15d026 100644 --- a/src/Persistence/Wolverine.Sqlite/Transport/SqliteQueueSender.cs +++ b/src/Persistence/Wolverine.Sqlite/Transport/SqliteQueueSender.cs @@ -145,12 +145,17 @@ public async Task ScheduleRetryAsync(Envelope envelope, CancellationToken cancel { await using var conn = await _dataSource.OpenConnectionAsync(cancellationToken).ConfigureAwait(false); - await conn.CreateCommand(_deleteFromIncomingAndScheduleSql) + // The inbox delete and the scheduled write commit together or not at all: autocommitted one by one, a + // failed scheduled write left the inbox row deleted with nothing scheduled + await using var tx = await conn.BeginTransactionAsync(cancellationToken); + var cmd = conn.CreateCommand(_deleteFromIncomingAndScheduleSql) .With("id", $"{envelope.Id:D}") .With("body", EnvelopeSerializer.Serialize(envelope)) .With("type", envelope.MessageType!) - .With("time", (envelope.ScheduledTime ?? DateTimeOffset.UtcNow).UtcDateTime.ToString(SqliteDateTimeFormat)) - .ExecuteNonQueryAsync(cancellationToken); + .With("time", (envelope.ScheduledTime ?? DateTimeOffset.UtcNow).UtcDateTime.ToString(SqliteDateTimeFormat)); + cmd.Transaction = tx; + await cmd.ExecuteNonQueryAsync(cancellationToken); + await tx.CommitAsync(cancellationToken); } public async ValueTask SendAsync(Envelope envelope)