GH-5055 Dead letter outgoing envelopes a transport cannot map

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

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;
+
+/// <summary>
+/// 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
+/// </summary>
+public class unmappable_outgoing_envelopes_are_dead_lettered_5055 : IAsyncDisposable
+{
+    private readonly MockWolverineRuntime theRuntime = new();
+    private readonly IMessageOutbox theOutbox = Substitute.For<IMessageOutbox>();
+    private readonly UnmappableProtocol theProtocol = new();
+    private readonly TcpEndpoint theEndpoint = new(2346);
+    private readonly BatchedSender theSender;
+    private readonly List<IAsyncDisposable> _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<IMessageTracker>(), 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<Exception>(e => e is UnsendableEnvelopeException));
+        foreach (var envelope in mapped)
+        {
+            await theRuntime.Storage.Inbox.DidNotReceive().MoveToDeadLetterStorageAsync(envelope, Arg.Any<Exception>());
+        }
+
+        theRuntime.MessageTracking.Received(1).MovedToErrorQueue(unmappable, Arg.Any<Exception>());
+
+        // The two sent and the one dead lettered all leave the outbox, each exactly once
+        await theOutbox.Received(1).DeleteOutgoingAsync(Arg.Is<Envelope[]>(x =>
+            x.Length == 2 && x.Contains(mapped[0]) && x.Contains(mapped[1])));
+        await theOutbox.Received(1).DeleteOutgoingAsync(Arg.Is<Envelope[]>(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<Envelope>(), Arg.Any<Exception>())
+            .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<Envelope[]>(x => x.Contains(unmappable)));
+    }
+
+    [Fact]
+    public async Task a_buffered_agent_dead_letters_the_unmappable_envelope()
+    {
+        var agent = new BufferedSendingAgent(NullLogger.Instance, Substitute.For<IMessageTracker>(), 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<Exception>());
+        await theRuntime.Storage.Inbox.DidNotReceive().MoveToDeadLetterStorageAsync(mapped[0], Arg.Any<Exception>());
+    }
+
+    [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<Exception>());
+        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);
+    }
+
+    /// <summary>
+    /// 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
+    /// </summary>
+    private class UnmappableProtocol : ISenderProtocol
+    {
+        public HashSet<Guid> Unmappable { get; } = new();
+        public HashSet<Guid> 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"));
+            }
+        }
+    }
+}

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<List<Guid>> 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<Guid>();
+        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<Envelope>();
         foreach (var number in new[] { 1, Serverless5044PoisonMapper.PoisonNumber, 2 })
         {
             var envelope = new Envelope(new Serverless5044SnsPoisonMessage(number))
@@ -183,20 +196,33 @@ 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();
         var report = await _host.RunDurabilityPassAsync(maxDuration: 30.Seconds(), token: Token);
 
         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