From 5885b8c38c9b9d8afc105f6ae34a24ce0403d7cc Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Fri, 23 Oct 2026 14:23:59 -0500 Subject: [PATCH] GH-5055 Dead letter outgoing envelopes a transport cannot map instead of reporting them sent - Azure Service Bus: an envelope whose mapping throws is reported through MarkSerializationFailureAsync and left out of what is reported sent (the success path used to report the whole original batch) - SNS: mapping failures are reported as serialization failures, not successes; a chunk with nothing mappable is not published. Under ServerlessWithOutbox the durability pass now counts them as OutgoingDeadLettered, not OutgoingSent - Kafka: only the unmappable envelope is a serialization failure; it used to take the whole batch down with it - SendingAgent.MarkSerializationFailureAsync moves the envelopes to dead letter storage when the application has a message store (every mode). The durable agent deletes the outbox row once the dead letter is written, and leaves it for a later sweep when that write fails - The ServerlessWithOutbox single-envelope inline send dead letters an unsendable envelope right away, consistent with the grouped inline send and the durability pass (GH-5047) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DCggXgxnG3EBXoR9rrq37U --- .../transports/sqs/large-messages.md | 8 +- ...tgoing_envelopes_are_dead_lettered_5055.cs | 182 ++++++++++++++++++ .../serverless_with_outbox_sns_5044.cs | 41 +++- ...ppable_envelopes_are_dead_lettered_5055.cs | 115 +++++++++++ .../Internal/SnsSenderProtocol.cs | 30 ++- .../AmazonSqsSubscriberConfiguration.cs | 2 +- .../Internal/SqsSenderProtocol.cs | 2 +- ...less_with_outbox_azure_service_bus_5044.cs | 10 +- ...reported_as_serialization_failures_5055.cs | 106 ++++++++++ .../Internal/AzureServiceBusSenderProtocol.cs | 15 +- .../Wolverine.Kafka/KafkaSenderProtocol.cs | 37 +++- .../Kafka/Wolverine.Kafka/KafkaTopic.cs | 5 +- .../Durability/DurableSendingAgent.cs | 48 ++++- .../Durability/UnsendableOutgoingEnvelopes.cs | 16 +- .../Transports/Sending/SendingAgent.cs | 43 ++++- 15 files changed, 611 insertions(+), 49 deletions(-) create mode 100644 src/Testing/CoreTests/Persistence/Durability/unmappable_outgoing_envelopes_are_dead_lettered_5055.cs create mode 100644 src/Transports/AWS/Wolverine.AmazonSns.Tests/unmappable_envelopes_are_dead_lettered_5055.cs create mode 100644 src/Transports/Azure/Wolverine.AzureServiceBus.Tests/unmappable_envelopes_are_reported_as_serialization_failures_5055.cs diff --git a/docs/guide/messaging/transports/sqs/large-messages.md b/docs/guide/messaging/transports/sqs/large-messages.md index 3d449c0080..8763393fe5 100644 --- a/docs/guide/messaging/transports/sqs/large-messages.md +++ b/docs/guide/messaging/transports/sqs/large-messages.md @@ -7,9 +7,9 @@ InvalidParameterValue - Message must be shorter than 262144 bytes. (SenderFault: ``` `SenderFault: true` means SQS is telling you the request itself is wrong, so the identical request will -fail identically forever. Wolverine treats an oversized message accordingly: it is **logged and -discarded** rather than retried, so a message that outgrew the limit produces one error rather than an -endless flood of them. +fail identically forever. Wolverine treats an oversized message accordingly: it is **logged and moved to +dead letter storage** (when the application has a message store, otherwise it is only logged) rather than +retried, so a message that outgrew the limit produces one error rather than an endless flood of them. If you want the message to actually get through, there are two answers. @@ -106,7 +106,7 @@ opts.MessagePartitioning.GlobalPartitioned(topology => where it was always going to land. Only a message with no group id at all gets a synthesized one, purely to hold the set together. -* **There is a ceiling.** A message needing more than 10 fragments is discarded with an error pointing at +* **There is a ceiling.** A message needing more than 10 fragments is dead lettered with an error pointing at claim checks. At that size this is a storage problem, not a framing one. * **On a FIFO queue** each fragment gets a distinct `MessageDeduplicationId` (the envelope's, suffixed with diff --git a/src/Testing/CoreTests/Persistence/Durability/unmappable_outgoing_envelopes_are_dead_lettered_5055.cs b/src/Testing/CoreTests/Persistence/Durability/unmappable_outgoing_envelopes_are_dead_lettered_5055.cs new file mode 100644 index 0000000000..40fba0d166 --- /dev/null +++ b/src/Testing/CoreTests/Persistence/Durability/unmappable_outgoing_envelopes_are_dead_lettered_5055.cs @@ -0,0 +1,182 @@ +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using CoreTests.Runtime; +using Wolverine; +using Wolverine.ComplianceTests; +using Wolverine.Logging; +using Wolverine.Persistence.Durability; +using Wolverine.Runtime; +using Wolverine.Transports; +using Wolverine.Transports.Sending; +using Wolverine.Transports.Tcp; +using Xunit; + +namespace CoreTests.Persistence.Durability; + +/// +/// GH-5055. When a transport cannot map an outgoing envelope to its broker message, it reports the envelope through +/// ISenderCallback.MarkSerializationFailureAsync, since mapping it again can never succeed. The sending agent then +/// moves it to dead letter storage, in every durability mode, rather than only logging it. A durable agent deletes +/// its outbox row once the dead letter is written +/// +public class unmappable_outgoing_envelopes_are_dead_lettered_5055 : IAsyncDisposable +{ + private readonly MockWolverineRuntime theRuntime = new(); + private readonly IMessageOutbox theOutbox = Substitute.For(); + private readonly UnmappableProtocol theProtocol = new(); + private readonly TcpEndpoint theEndpoint = new(2346); + private readonly BatchedSender theSender; + private readonly List _disposables = new(); + + public unmappable_outgoing_envelopes_are_dead_lettered_5055() + { + theSender = new BatchedSender(theEndpoint, theProtocol, CancellationToken.None, NullLogger.Instance); + } + + public async ValueTask DisposeAsync() + { + foreach (var disposable in _disposables) await disposable.DisposeAsync(); + } + + private DurableSendingAgent durableAgent() + { + var agent = new DurableSendingAgent(theSender, theRuntime.Options.Durability, NullLogger.Instance, + Substitute.For(), theOutbox, theEndpoint, theRuntime, null); + _disposables.Add(agent); + return agent; + } + + private Envelope envelope() + { + var envelope = ObjectMother.Envelope(); + envelope.Destination = theEndpoint.Uri; + envelope.DeliverBy = null; + return envelope; + } + + // What the transport protocols do on a mapping failure, one envelope of three + private Task sendBatchOfThreeAsync(ISenderCallback agent, out Envelope[] mapped, out Envelope unmappable) + { + var one = envelope(); + unmappable = envelope(); + var three = envelope(); + theProtocol.Unmappable.Add(unmappable.Id); + mapped = [one, three]; + + return theProtocol.SendBatchAsync(agent, new OutgoingMessageBatch(theEndpoint.Uri, [one, unmappable, three])); + } + + [Fact] + public async Task a_durable_agent_dead_letters_the_unmappable_envelope_and_deletes_only_its_rows_once_sent() + { + var agent = durableAgent(); + + await sendBatchOfThreeAsync(agent, out var mapped, out var unmappable); + await agent.DisposeAsync(); // flushes the background deletes + + await theRuntime.Storage.Inbox.Received(1).MoveToDeadLetterStorageAsync(unmappable, + Arg.Is(e => e is UnsendableEnvelopeException)); + foreach (var envelope in mapped) + { + await theRuntime.Storage.Inbox.DidNotReceive().MoveToDeadLetterStorageAsync(envelope, Arg.Any()); + } + + theRuntime.MessageTracking.Received(1).MovedToErrorQueue(unmappable, Arg.Any()); + + // The two sent and the one dead lettered all leave the outbox, each exactly once + await theOutbox.Received(1).DeleteOutgoingAsync(Arg.Is(x => + x.Length == 2 && x.Contains(mapped[0]) && x.Contains(mapped[1]))); + await theOutbox.Received(1).DeleteOutgoingAsync(Arg.Is(x => + x.Length == 1 && x[0] == unmappable)); + } + + [Fact] + public async Task a_durable_agent_leaves_the_row_in_the_outbox_when_the_dead_letter_cannot_be_written() + { + var agent = durableAgent(); + theRuntime.Storage.Inbox.MoveToDeadLetterStorageAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromException(new InvalidOperationException("database is down"))); + + await sendBatchOfThreeAsync(agent, out _, out var unmappable); + await agent.DisposeAsync(); + + // Nothing is lost: a later recovery sweep tries it again + await theOutbox.DidNotReceive().DeleteOutgoingAsync(Arg.Is(x => x.Contains(unmappable))); + } + + [Fact] + public async Task a_buffered_agent_dead_letters_the_unmappable_envelope() + { + var agent = new BufferedSendingAgent(NullLogger.Instance, Substitute.For(), theSender, + theRuntime.Options.Durability, theEndpoint, theRuntime, null); + _disposables.Add(agent); + + await sendBatchOfThreeAsync(agent, out var mapped, out var unmappable); + + await theRuntime.Storage.Inbox.Received(1).MoveToDeadLetterStorageAsync(unmappable, Arg.Any()); + await theRuntime.Storage.Inbox.DidNotReceive().MoveToDeadLetterStorageAsync(mapped[0], Arg.Any()); + } + + [Fact] + public async Task the_serverless_inline_send_dead_letters_an_unmappable_envelope_right_away() + { + theRuntime.Options.Durability.Mode = DurabilityMode.ServerlessWithOutbox; + var agent = durableAgent(); + + var unmappable = envelope(); + theProtocol.Unmappable.Add(unmappable.Id); + + await agent.StoreAndForwardAsync(unmappable); + + await theRuntime.Storage.Inbox.Received(1).MoveToDeadLetterStorageAsync(unmappable, Arg.Any()); + await theOutbox.Received(1).DeleteOutgoingAsync(unmappable); + } + + [Fact] + public async Task the_serverless_inline_send_leaves_a_failed_send_in_the_outbox() + { + theRuntime.Options.Durability.Mode = DurabilityMode.ServerlessWithOutbox; + var agent = durableAgent(); + + var throttled = envelope(); + theProtocol.Rejected.Add(throttled.Id); + + await agent.StoreAndForwardAsync(throttled); + + await theRuntime.Storage.Inbox.DidNotReceiveWithAnyArgs().MoveToDeadLetterStorageAsync(default!, default); + await theOutbox.DidNotReceive().DeleteOutgoingAsync(throttled); + } + + /// + /// Reports a batch the way the fixed Azure Service Bus, SNS and Kafka protocols do when their mapper throws for + /// some of it: those as a serialization failure, and only what was really sent as successful + /// + private class UnmappableProtocol : ISenderProtocol + { + public HashSet Unmappable { get; } = new(); + public HashSet Rejected { get; } = new(); + + public async Task SendBatchAsync(ISenderCallback callback, OutgoingMessageBatch batch) + { + var unmappable = batch.Messages.Where(x => Unmappable.Contains(x.Id)).ToList(); + var rejected = batch.Messages.Where(x => Rejected.Contains(x.Id)).ToList(); + var sent = batch.Messages.Except(unmappable).Except(rejected).ToList(); + + if (unmappable.Count > 0) + { + await callback.MarkSerializationFailureAsync(new OutgoingMessageBatch(batch.Destination, unmappable)); + } + + if (sent.Count > 0) + { + await callback.MarkSuccessfulAsync(new OutgoingMessageBatch(batch.Destination, sent)); + } + + if (rejected.Count > 0) + { + await callback.MarkProcessingFailureAsync(new OutgoingMessageBatch(batch.Destination, rejected), + new InvalidOperationException("Throttled")); + } + } + } +} diff --git a/src/Transports/AWS/Wolverine.AmazonSns.Tests/serverless_with_outbox_sns_5044.cs b/src/Transports/AWS/Wolverine.AmazonSns.Tests/serverless_with_outbox_sns_5044.cs index f868bc5bad..35895208a0 100644 --- a/src/Transports/AWS/Wolverine.AmazonSns.Tests/serverless_with_outbox_sns_5044.cs +++ b/src/Transports/AWS/Wolverine.AmazonSns.Tests/serverless_with_outbox_sns_5044.cs @@ -166,14 +166,27 @@ public async Task message_batch_size_over_the_publish_batch_limit() $"inline: {delivered} delivered, {rows.Count} outbox rows left; after a pass: {deliveredByPass} delivered, {rowsAfterPass.Count} rows left; protocol calls: {string.Join(",", counting.Batches.Select(x => x.Length))}"); } + private async Task> deadLetterIdsAsync() + { + await using var conn = new NpgsqlConnection(Servers.PostgresConnectionString); + await conn.OpenAsync(Token); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"select id from {Schema}.wolverine_dead_letters"; + var ids = new List(); + await using var reader = await cmd.ExecuteReaderAsync(Token); + while (await reader.ReadAsync(Token)) ids.Add(reader.GetGuid(0)); + return ids; + } + [Fact] - public async Task an_envelope_the_mapper_cannot_map_is_reported_successful() + public async Task an_envelope_the_mapper_cannot_map_is_dead_lettered_by_the_durability_pass() { var counting = wrap(Poison); // Outbox rows as a failed inline send leaves them: 2 good, 1 the mapper throws on var runtime = _host.GetRuntime(); var uri = _transport.Topics[Poison].Uri; + var envelopes = new List(); foreach (var number in new[] { 1, Serverless5044PoisonMapper.PoisonNumber, 2 }) { var envelope = new Envelope(new Serverless5044SnsPoisonMessage(number)) @@ -183,6 +196,7 @@ public async Task an_envelope_the_mapper_cannot_map_is_reported_successful() }; envelope.Data = runtime.Options.DefaultSerializer.Write(envelope); await runtime.Storage.Outbox.StoreOutgoingAsync(envelope, 0); + envelopes.Add(envelope); } await makeOutboxStaleAsync(); @@ -190,13 +204,25 @@ public async Task an_envelope_the_mapper_cannot_map_is_reported_successful() var delivered = await drainAsync(Poison, 2, 10.Seconds()); - // Observed facts, recorded: 2 delivered, and all 3 rows deleted -- the unmappable one is counted sent + // GH-5055: 2 delivered, and the unmappable one moved to dead letter storage instead of counted as sent delivered.ShouldBe(2); + report.OutgoingSent.ShouldBe(2); + report.OutgoingDeadLettered.ShouldBe(1); + (await deadLetterIdsAsync()).ShouldBe([envelopes[1].Id]); (await outboxRowsAsync()).ShouldBeEmpty(); - report.OutgoingSent.ShouldBe(3); counting.Batches.Select(x => x.Length).ToArray().ShouldBe(new[] { 3 }); } + [Fact] + public async Task an_envelope_the_mapper_cannot_map_is_dead_lettered_by_the_inline_send() + { + await _host.MessageBus().InvokeAsync(new Serverless5044SnsPoisonTrigger([1, Serverless5044PoisonMapper.PoisonNumber, 2]), Token); + + (await drainAsync(Poison, 2, 10.Seconds())).ShouldBe(2); + (await deadLetterIdsAsync()).Count.ShouldBe(1); + (await outboxRowsAsync()).ShouldBeEmpty(); + } + [Fact] public async Task a_topic_that_no_longer_exists_leaves_every_row() { @@ -219,6 +245,8 @@ public record Serverless5044SnsWideMessage(int Number); public record Serverless5044SnsPoisonMessage(int Number); +public record Serverless5044SnsPoisonTrigger(int[] Numbers); + public static class Serverless5044SnsHandler { public static OutgoingMessages Handle(Serverless5044SnsTrigger trigger) @@ -228,6 +256,13 @@ public static OutgoingMessages Handle(Serverless5044SnsTrigger trigger) for (var i = 0; i < trigger.Wide; i++) messages.Add(new Serverless5044SnsWideMessage(i)); return messages; } + + public static OutgoingMessages Handle(Serverless5044SnsPoisonTrigger trigger) + { + var messages = new OutgoingMessages(); + foreach (var number in trigger.Numbers) messages.Add(new Serverless5044SnsPoisonMessage(number)); + return messages; + } } internal class Serverless5044PoisonMapper : ISnsEnvelopeMapper diff --git a/src/Transports/AWS/Wolverine.AmazonSns.Tests/unmappable_envelopes_are_dead_lettered_5055.cs b/src/Transports/AWS/Wolverine.AmazonSns.Tests/unmappable_envelopes_are_dead_lettered_5055.cs new file mode 100644 index 0000000000..dd2b1dcf0c --- /dev/null +++ b/src/Transports/AWS/Wolverine.AmazonSns.Tests/unmappable_envelopes_are_dead_lettered_5055.cs @@ -0,0 +1,115 @@ +using Amazon.SQS.Model; +using IntegrationTests; +using JasperFx.Core; +using JasperFx.Resources; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Npgsql; +using Shouldly; +using Wolverine.AmazonSqs; +using Wolverine.Postgresql; +using Wolverine.Runtime; +using Wolverine.Tracking; + +namespace Wolverine.AmazonSns.Tests; + +/// +/// GH-5055, against LocalStack. An envelope the topic's mapper cannot map to an SNS message can never be sent, so +/// the durable sending agent moves it to dead letter storage and deletes its outbox row. The other envelopes of +/// the same batch are delivered. It used to be reported as sent with the rest, with only a log line left +/// +public class unmappable_envelopes_are_dead_lettered_5055 : IAsyncLifetime +{ + private const string Schema = "gh5055_sns"; + private readonly string _name = $"gh5055-sns-{Guid.NewGuid():N}"; + private IHost _host = null!; + + private static CancellationToken Token => TestContext.Current.CancellationToken; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(Serverless5044SnsHandler)); + opts.Policies.DisableConventionalLocalRouting(); + opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, Schema); + + opts.UseAmazonSqsTransportLocally().AutoProvision(); + opts.UseAmazonSnsTransportLocally().AutoProvision(); + + // A mapper that throws on the poison message, standing in for any mapping bug + opts.PublishMessage().ToSnsTopic(_name).SubscribeSqsQueue(_name) + .InteropWith(new Serverless5044PoisonMapper()).UseDurableOutbox(); + + opts.Services.AddResourceSetupOnStartup(); + }).StartAsync(Token); + + await _host.GetRuntime().Storage.Admin.ClearAllAsync(); + } + + public async ValueTask DisposeAsync() + { + var options = _host.GetRuntime().Options; + foreach (var queue in options.AmazonSqsTransport().Queues.Where(x => x.QueueName.StartsWith("gh5055"))) + { + try { await queue.TeardownAsync(NullLogger.Instance); } catch { } + } + + foreach (var topic in options.AmazonSnsTransport().Topics.Where(x => x.TopicName.StartsWith("gh5055"))) + { + try { await topic.TeardownAsync(NullLogger.Instance); } catch { } + } + + await _host.StopAsync(); + _host.Dispose(); + } + + private async Task countAsync(string table) + { + await using var conn = new NpgsqlConnection(Servers.PostgresConnectionString); + await conn.OpenAsync(Token); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"select count(*) from {Schema}.{table}"; + return (long)(await cmd.ExecuteScalarAsync(Token))!; + } + + private async Task receiveAsync(int expected) + { + var sqs = _host.GetRuntime().Options.AmazonSnsTransport().SqsClient!; + var url = (await sqs.GetQueueUrlAsync(_name, Token)).QueueUrl; + var count = 0; + var deadline = DateTimeOffset.UtcNow.Add(15.Seconds()); + while (DateTimeOffset.UtcNow < deadline && count < expected) + { + var response = await sqs.ReceiveMessageAsync( + new ReceiveMessageRequest(url) { MaxNumberOfMessages = 10, WaitTimeSeconds = 1 }, Token); + var messages = response.Messages ?? []; + count += messages.Count; + foreach (var message in messages) await sqs.DeleteMessageAsync(url, message.ReceiptHandle, Token); + } + + return count; + } + + [Fact] + public async Task two_of_three_are_delivered_and_the_unmappable_one_is_dead_lettered() + { + await _host.MessageBus().InvokeAsync( + new Serverless5044SnsPoisonTrigger([1, Serverless5044PoisonMapper.PoisonNumber, 2]), Token); + + (await receiveAsync(2)).ShouldBe(2); + + // The dead letter is written and the outbox rows are deleted in the background + var deadline = DateTimeOffset.UtcNow.Add(10.Seconds()); + while (DateTimeOffset.UtcNow < deadline && + (await countAsync("wolverine_dead_letters") == 0 || await countAsync("wolverine_outgoing_envelopes") > 0)) + { + await Task.Delay(100, Token); + } + + (await countAsync("wolverine_dead_letters")).ShouldBe(1); + (await countAsync("wolverine_outgoing_envelopes")).ShouldBe(0); + } +} diff --git a/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs b/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs index b2e95ea556..8f608db77d 100644 --- a/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs +++ b/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs @@ -47,6 +47,20 @@ private async Task sendChunkAsync(ISenderCallback callback, OutgoingMessageBatch { var snsBatch = new OutgoingSnsBatch(_topic, _logger, batch.Messages); + // GH-5055. Mapping it again can never succeed, so it is dead lettered rather than retried, and it is not + // reported as sent with the rest, which it used to be + if (snsBatch.MappingFailures.Count > 0) + { + await callback.MarkSerializationFailureAsync( + new OutgoingMessageBatch(batch.Destination, snsBatch.MappingFailures)); + + // PublishBatch refuses a request without entries + if (snsBatch.Request.PublishBatchRequestEntries.Count == 0) return; + + batch = new OutgoingMessageBatch(batch.Destination, + batch.Messages.Where(x => !snsBatch.MappingFailures.Contains(x)).ToList()); + } + try { var response = await _sns.PublishBatchAsync(snsBatch.Request); @@ -64,9 +78,14 @@ internal class OutgoingSnsBatch { private readonly Dictionary _envelopes = new(); private readonly List _mappingFailures = new(); - + public PublishBatchRequest Request { get; } + /// + /// Envelopes the topic's mapper threw on. They are not in the request, and can never be sent + /// + public IReadOnlyList MappingFailures => _mappingFailures; + public OutgoingSnsBatch(AmazonSnsTopic topic, ILogger logger, IEnumerable envelopes) { var entries = new List(); @@ -104,7 +123,8 @@ public OutgoingSnsBatch(AmazonSnsTopic topic, ILogger logger, IEnumerable 0) + { + await callback.MarkSuccessfulAsync(new OutgoingMessageBatch(batch.Destination, successes)); + } await callback.MarkProcessingFailureAsync(new OutgoingMessageBatch(batch.Destination, fails)); } diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsSubscriberConfiguration.cs b/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsSubscriberConfiguration.cs index 017c7028f9..6f0339b4a6 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsSubscriberConfiguration.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/AmazonSqsSubscriberConfiguration.cs @@ -44,7 +44,7 @@ public AmazonSqsSubscriberConfiguration EnableFairQueueMessageGroups() /// /// Split a message whose body would exceed SQS's hard 256KB limit into several SQS messages, and /// reassemble it on the receiving side. Without this an oversized message is rejected by SQS with - /// a permanent SenderFault, and Wolverine discards it rather than retrying forever. + /// a permanent SenderFault, and Wolverine moves it to dead letter storage rather than retrying forever. /// /// /// diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs index f6a2411fc5..dfb7b895b4 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs @@ -71,7 +71,7 @@ public async Task SendBatchAsync(ISenderCallback callback, OutgoingMessageBatch // Permanently unsendable, not a transient failure: SQS answers an oversized message with a // SenderFault, and retrying it produces the same SenderFault forever. That infinite retry is // how this presents in production - a flood of identical errors rather than one. Route it to - // the serialization-failure path, which logs and drops rather than re-queueing. + // the serialization-failure path, which dead letters rather than re-queueing (GH-5055). await callback.MarkSerializationFailureAsync( new OutgoingMessageBatch(batch.Destination, unsendable)); } diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/serverless_with_outbox_azure_service_bus_5044.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/serverless_with_outbox_azure_service_bus_5044.cs index b66517a64d..b6d118e326 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/serverless_with_outbox_azure_service_bus_5044.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/serverless_with_outbox_azure_service_bus_5044.cs @@ -83,7 +83,7 @@ public async Task happy_path_one_broker_send_for_the_batch() } [Fact] - public async Task an_unmappable_envelope_is_reported_accepted_although_it_was_never_sent() + public async Task an_unmappable_envelope_is_reported_as_a_serialization_failure() { var envelopes = new[] { envelope("a"), envelope(Unmappable), envelope("c") }; var result = new InlineSendResult(); @@ -92,9 +92,11 @@ public async Task an_unmappable_envelope_is_reported_accepted_although_it_was_ne _sentBatches.SelectMany(x => x).ShouldBe(["a", "c"]); - // Observed and recorded: the unmappable one counts as accepted, so its outbox row would be deleted. The - // protocol logs it as discarded, the same in every durability mode. A design question, not asserted as a bug - result.WasAccepted(envelopes[1]).ShouldBeTrue(); + // GH-5055: it is not accepted, and the inline send and the durability pass dead letter it + result.WasAccepted(envelopes[0]).ShouldBeTrue(); + result.WasAccepted(envelopes[2]).ShouldBeTrue(); + result.WasAccepted(envelopes[1]).ShouldBeFalse(); + result.IsSerializationFailure(envelopes[1], out _).ShouldBeTrue(); } [Fact] diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/unmappable_envelopes_are_reported_as_serialization_failures_5055.cs b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/unmappable_envelopes_are_reported_as_serialization_failures_5055.cs new file mode 100644 index 0000000000..46fcd9f8d8 --- /dev/null +++ b/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/unmappable_envelopes_are_reported_as_serialization_failures_5055.cs @@ -0,0 +1,106 @@ +using Azure.Messaging.ServiceBus; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Shouldly; +using Wolverine.AzureServiceBus.Internal; +using Wolverine.Runtime; +using Wolverine.Transports; +using Wolverine.Transports.Sending; +using Xunit; + +namespace Wolverine.AzureServiceBus.Tests; + +/// +/// GH-5055, unit level with a faked ServiceBusSender (no emulator). An envelope the outgoing mapper throws on can +/// never be sent, so the protocol reports it through MarkSerializationFailureAsync, which dead letters it, and +/// only reports the envelopes it really sent as successful. It used to log the envelope as discarded and then +/// report the whole original batch as sent, so its outbox row was deleted with nothing but a log line left +/// +public class unmappable_envelopes_are_reported_as_serialization_failures_5055 +{ + private const string Unmappable = "gh5055-unmappable"; + + private readonly ISenderCallback theCallback = Substitute.For(); + private readonly List _sent = new(); + private readonly AzureServiceBusQueue theQueue = new(new AzureServiceBusTransport(), "gh5055-queue"); + private readonly AzureServiceBusSenderProtocol theProtocol; + + public unmappable_envelopes_are_reported_as_serialization_failures_5055() + { + var runtime = Substitute.For(); + runtime.LoggerFactory.Returns(NullLoggerFactory.Instance); + runtime.Cancellation.Returns(CancellationToken.None); + + // A user's mapper that cannot handle one of the messages + var mapper = Substitute.For>(); + mapper.When(x => x.MapEnvelopeToOutgoing(Arg.Any(), Arg.Any())) + .Do(call => + { + var envelope = call.Arg(); + if (envelope.MessageType == Unmappable) throw new InvalidOperationException("cannot map"); + call.Arg().Subject = envelope.MessageType; + call.Arg().SessionId = "one"; + }); + + var stores = new Dictionary>(); + var sender = Substitute.For(); + sender.CreateMessageBatchAsync(Arg.Any()).Returns(_ => + { + var store = new List(); + var batch = ServiceBusModelFactory.ServiceBusMessageBatch(262144, store); + stores[batch] = store; + return new ValueTask(batch); + }); + sender.SendMessagesAsync(Arg.Any(), Arg.Any()) + .Returns(call => + { + _sent.AddRange(stores[call.Arg()].Select(m => m.Subject)); + return Task.CompletedTask; + }); + + theProtocol = new AzureServiceBusSenderProtocol(runtime, theQueue, mapper, sender); + } + + private static Envelope envelope(string messageType) + { + return new Envelope { Id = Guid.NewGuid(), MessageType = messageType, Data = [1, 2, 3] }; + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task two_of_three_are_sent_and_the_unmappable_one_is_a_serialization_failure(bool partitioned) + { + theQueue.Options.EnablePartitioning = partitioned; + + var one = envelope("one"); + var unmappable = envelope(Unmappable); + var three = envelope("three"); + + await theProtocol.SendBatchAsync(theCallback, + new OutgoingMessageBatch(theQueue.Uri, [one, unmappable, three])); + + _sent.ShouldBe(["one", "three"]); + + await theCallback.Received(1).MarkSerializationFailureAsync(Arg.Is(b => + b.Messages.Count == 1 && b.Messages[0] == unmappable)); + + await theCallback.Received(1).MarkSuccessfulAsync(Arg.Is(b => + b.Messages.Count == 2 && b.Messages.Contains(one) && b.Messages.Contains(three))); + await theCallback.DidNotReceive().MarkSuccessfulAsync(Arg.Is(b => + b.Messages.Contains(unmappable))); + await theCallback.DidNotReceiveWithAnyArgs().MarkProcessingFailureAsync(default!, default); + } + + [Fact] + public async Task a_batch_with_nothing_mappable_sends_nothing_and_reports_nothing_successful() + { + var unmappable = envelope(Unmappable); + + await theProtocol.SendBatchAsync(theCallback, new OutgoingMessageBatch(theQueue.Uri, [unmappable])); + + _sent.ShouldBeEmpty(); + await theCallback.Received(1).MarkSerializationFailureAsync(Arg.Any()); + await theCallback.DidNotReceiveWithAnyArgs().MarkSuccessfulAsync(default(OutgoingMessageBatch)!); + } +} diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusSenderProtocol.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusSenderProtocol.cs index db21a08f44..c0c7cb1920 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusSenderProtocol.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusSenderProtocol.cs @@ -29,6 +29,7 @@ public async Task SendBatchAsync(ISenderCallback callback, OutgoingMessageBatch await _endpoint.InitializeAsync(_logger); var messages = new List<(Envelope Envelope, ServiceBusMessage Message)>(batch.Messages.Count); + List? unmappable = null; foreach (var envelope in batch.Messages) { @@ -41,11 +42,23 @@ public async Task SendBatchAsync(ISenderCallback callback, OutgoingMessageBatch catch (Exception e) { _logger.LogError(e, - "Error trying to translate envelope {Envelope} to a ServiceBusMessage object. Message will be discarded.", + "Error trying to translate envelope {Envelope} to a ServiceBusMessage object. It can never be sent, so it is reported as a serialization failure", envelope); + (unmappable ??= new()).Add(envelope); } } + // GH-5055. Mapping it again can never succeed, so it is dead lettered rather than retried. It is also left + // out of what is reported sent below, which used to be the whole original batch + if (unmappable != null) + { + await callback.MarkSerializationFailureAsync(new OutgoingMessageBatch(batch.Destination, unmappable)); + + if (messages.Count == 0) return; + + batch = new OutgoingMessageBatch(batch.Destination, messages.Select(x => x.Envelope).ToList()); + } + if (_endpoint.IsPartitioned) await sendPartitionedBatches(callback, messages, batch); else diff --git a/src/Transports/Kafka/Wolverine.Kafka/KafkaSenderProtocol.cs b/src/Transports/Kafka/Wolverine.Kafka/KafkaSenderProtocol.cs index efd47b6566..3afdcf73ec 100644 --- a/src/Transports/Kafka/Wolverine.Kafka/KafkaSenderProtocol.cs +++ b/src/Transports/Kafka/Wolverine.Kafka/KafkaSenderProtocol.cs @@ -1,4 +1,6 @@ using Confluent.Kafka; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using Wolverine.Kafka.Internals; using Wolverine.Transports; using Wolverine.Transports.Sending; @@ -10,6 +12,11 @@ public class KafkaSenderProtocol : ISenderProtocol, IDisposable private readonly KafkaTopic _topic; private readonly IProducer _producer; + /// + /// Where an envelope the mapper throws on is logged with the original exception + /// + internal ILogger Logger { get; set; } = NullLogger.Instance; + public KafkaSenderProtocol(KafkaTopic topic) : this(topic, topic.Parent) { @@ -41,23 +48,35 @@ public async Task SendBatchAsync(ISenderCallback callback, OutgoingMessageBatch // the acked records are reported successful and only the rest failed, // each as its own OutgoingMessageBatch, the way the SQS sender does. - // Phase 1 — map every envelope to a Confluent Message<>. A mapping / - // serialization failure on any envelope before we've started enqueuing - // means we haven't put anything on the wire yet — treat as a clean - // serialization failure on the batch. + // Phase 1 — map every envelope to a Confluent Message<>. GH-5055: an + // envelope the mapper throws on can never be sent, so only that one is + // reported as a serialization failure, which dead letters it. Reporting + // the whole batch took every envelope that mapped fine down with it var prepared = new List<(Envelope envelope, Message message)>(batch.Messages.Count); - try + List? unmappable = null; + foreach (var envelope in batch.Messages) { - foreach (var envelope in batch.Messages) + try { var message = await _topic.EnvelopeMapper!.CreateMessage(envelope); prepared.Add((envelope, message)); } + catch (Exception e) + { + Logger.LogError(e, + "Error while mapping envelope {Envelope} to a Kafka message for {Uri}. It can never be sent, so it is reported as a serialization failure", + envelope, _topic.Uri); + (unmappable ??= new()).Add(envelope); + } } - catch (Exception) + + if (unmappable != null) { - await callback.MarkSerializationFailureAsync(batch); - return; + await callback.MarkSerializationFailureAsync(new OutgoingMessageBatch(batch.Destination, unmappable)); + + if (prepared.Count == 0) return; + + batch = new OutgoingMessageBatch(batch.Destination, prepared.Select(x => x.envelope).ToList()); } // Phase 2 — enqueue every produce without awaiting so the Confluent diff --git a/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs b/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs index 81794e5b3c..9b704b6c4e 100644 --- a/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs +++ b/src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs @@ -333,8 +333,9 @@ protected override ISender CreateSender(IWolverineRuntime runtime) return SendsInline ? new InlineKafkaSender(this) - : new BatchedSender(this, new KafkaSenderProtocol(this), runtime.Cancellation, - runtime.LoggerFactory.CreateLogger()); + : new BatchedSender(this, + new KafkaSenderProtocol(this) { Logger = runtime.LoggerFactory.CreateLogger() }, + runtime.Cancellation, runtime.LoggerFactory.CreateLogger()); } public override bool TryBuildDeadLetterSender(IWolverineRuntime runtime, out ISender? deadLetterSender) diff --git a/src/Wolverine/Persistence/Durability/DurableSendingAgent.cs b/src/Wolverine/Persistence/Durability/DurableSendingAgent.cs index c67e41fff0..0a05f81e4b 100644 --- a/src/Wolverine/Persistence/Durability/DurableSendingAgent.cs +++ b/src/Wolverine/Persistence/Durability/DurableSendingAgent.cs @@ -221,12 +221,22 @@ public override Task MarkSuccessfulAsync(OutgoingMessageBatch outgoing) /// GH-3926: an envelope the transport has judged permanently unsendable has to leave the outgoing /// table as well as the in-memory queue. Logging it and stopping there leaves the row behind, and /// the durability agent re-reads and re-sends it on every recovery sweep -- which is precisely the - /// endless retry this path exists to break. + /// endless retry this path exists to break. GH-5055: it is moved to dead letter storage first, and + /// a row whose dead letter could not be written stays in the outbox for a later recovery sweep, so + /// nothing is lost /// public override async Task MarkSerializationFailureAsync(OutgoingMessageBatch outgoing) { - await base.MarkSerializationFailureAsync(outgoing); - await _deleteOutgoingMany.PostAsync(outgoing.Messages.ToArray()); + var notMoved = await MoveUnsendableToDeadLetterStorageAsync(outgoing); + + var toDelete = notMoved.Count == 0 + ? outgoing.Messages.ToArray() + : outgoing.Messages.Where(x => !notMoved.Contains(x)).ToArray(); + + if (toDelete.Length > 0) + { + await _deleteOutgoingMany.PostAsync(toDelete); + } } public override Task MarkSuccessfulAsync(Envelope outgoing) @@ -265,25 +275,43 @@ protected override async Task storeAndForwardAsync(Envelope envelope) // next durability pass to recover private async Task sendInlineAsync(Envelope envelope) { - try - { - await _sender.SendToBrokerAsync(envelope); - } - catch (Exception e) + // The grouped send never throws, and tells a send that failed apart from one that never can succeed + var result = await _sender.SendToBrokerAsync([envelope]); + + if (!result.WasAccepted(envelope)) { - _logger.LogError(e, + // GH-5055. The transport can never send it, like an envelope its mapper throws on, so it goes to dead + // letter storage now rather than failing again in every later durability pass + if (result.IsSerializationFailure(envelope, out _)) + { + var notMoved = await MoveUnsendableToDeadLetterStorageAsync( + new OutgoingMessageBatch(envelope.Destination ?? Destination, [envelope])); + if (notMoved.Count == 0) + { + await deleteInlineSentAsync(envelope); + } + + return; + } + + _logger.LogError(result.Exception, "Unable to send outgoing message {Envelope} inline to {Destination}. It is left in the outbox to be recovered by a later durability pass", envelope, envelope.Destination); return; } + await deleteInlineSentAsync(envelope); + } + + private async Task deleteInlineSentAsync(Envelope envelope) + { try { await _outbox.DeleteOutgoingAsync(envelope); } catch (Exception e) { - // The message was sent, so the worst case is a duplicate send by a later durability pass + // The message was sent (or dead lettered), so the worst case is a duplicate send by a later durability pass _logger.LogError(e, "Unable to delete the outbox row of message {Envelope} that was already sent inline. A later durability pass may send it again", envelope); diff --git a/src/Wolverine/Persistence/Durability/UnsendableOutgoingEnvelopes.cs b/src/Wolverine/Persistence/Durability/UnsendableOutgoingEnvelopes.cs index d73dd4b07e..e3ad8654d2 100644 --- a/src/Wolverine/Persistence/Durability/UnsendableOutgoingEnvelopes.cs +++ b/src/Wolverine/Persistence/Durability/UnsendableOutgoingEnvelopes.cs @@ -7,8 +7,8 @@ namespace Wolverine.Persistence.Durability; /// GH-5047. What the inline send and durability pass do with an /// outgoing envelope the transport reported as a serialization failure, such as an SQS message over the size limit /// without fragmenting. Sending it again can never succeed, so it is moved to dead letter storage, where it can be -/// seen, rather than retried by every pass. In every other mode a durable sending agent deletes such a row -/// () +/// seen, rather than retried by every pass. GH-5055: a sending agent does the same in every other mode +/// () /// internal static class UnsendableOutgoingEnvelopes { @@ -39,3 +39,15 @@ public static async Task TryMoveToDeadLetterStorageAsync(IWolverineRuntime return true; } } + +/// +/// GH-5055. Recorded with the dead letter of an outgoing envelope its transport reported as one it can never send, +/// like an envelope the transport's mapper threw on. The transport logged the original error when it happened +/// +internal class UnsendableEnvelopeException : Exception +{ + public UnsendableEnvelopeException(Envelope envelope, Uri destination) : base( + $"The transport for {destination} can never send envelope {envelope}, because it could not be mapped or serialized to a broker message. The transport's error log has the original exception") + { + } +} diff --git a/src/Wolverine/Transports/Sending/SendingAgent.cs b/src/Wolverine/Transports/Sending/SendingAgent.cs index 9f69b5ef7c..5d03b2ba0d 100644 --- a/src/Wolverine/Transports/Sending/SendingAgent.cs +++ b/src/Wolverine/Transports/Sending/SendingAgent.cs @@ -78,18 +78,45 @@ Task ISenderCallback.MarkTimedOutAsync(OutgoingMessageBatch outgoing) /// /// The envelopes in this batch can never be sent to this destination -- the transport has already - /// decided that retrying would fail identically forever. Log them and let them go. A buffered - /// agent is done at that point; a durable one also has to clear them out of the outgoing table, - /// which is why this is virtual. + /// decided that retrying would fail identically forever, like an envelope its mapper cannot turn into + /// a broker message. GH-5055: they are moved to dead letter storage when the application has a message + /// store, so they can be seen and replayed, rather than only logged. A durable agent also has to clear + /// them out of the outgoing table, which is why this is virtual. /// - public virtual Task MarkSerializationFailureAsync(OutgoingMessageBatch outgoing) + public virtual async Task MarkSerializationFailureAsync(OutgoingMessageBatch outgoing) + { + await MoveUnsendableToDeadLetterStorageAsync(outgoing); + } + + /// + /// GH-5055. Move what the transport can never send to dead letter storage. Answers the envelopes whose dead + /// letter could not be written, which a durable agent leaves in its outbox. Without a message store there is + /// nowhere to move them, so they are only logged, as before + /// + internal async Task> MoveUnsendableToDeadLetterStorageAsync(OutgoingMessageBatch outgoing) { _logger.OutgoingBatchFailed(outgoing); - var exception = new Exception("Serialization failure with outgoing envelopes " + - outgoing.Messages.Select(x => x.ToString()).Join(", ")); - _logger.LogError(exception, "Serialization failure"); - return Task.CompletedTask; + if (_runtime == null || _runtime.Storage is NullMessageStore) + { + var exception = new Exception("Serialization failure with outgoing envelopes " + + outgoing.Messages.Select(x => x.ToString()).Join(", ")); + _logger.LogError(exception, "Serialization failure"); + return []; + } + + List? notMoved = null; + foreach (var envelope in outgoing.Messages) + { + var exception = new UnsendableEnvelopeException(envelope, outgoing.Destination); + if (!await UnsendableOutgoingEnvelopes.TryMoveToDeadLetterStorageAsync(_runtime, + envelope.Store ?? _runtime.Storage, envelope, exception, _logger)) + { + (notMoved ??= new()).Add(envelope); + } + } + + return notMoved ?? (IReadOnlyList)[]; } Task ISenderCallback.MarkQueueDoesNotExistAsync(OutgoingMessageBatch outgoing)