Round-trip Envelope.DeduplicationId, and let a ping reach a FIFO destination (GH-3793)

src/Testing/CoreTests/Serialization/serialization_and_deserialization.cs

@@ -231,6 +231,18 @@ public void group_id()
         incoming.GroupId.ShouldBe(outgoing.GroupId);
     }
 
+    /// <summary>
+    ///     GH-3793. Without this the durable outbox round-trip silently dropped the
+    ///     MessageDeduplicationId, and every recovered envelope was rejected outright by
+    ///     an SNS/SQS FIFO destination that doesn't have ContentBasedDeduplication turned on.
+    /// </summary>
+    [Fact]
+    public void deduplication_id()
+    {
+        outgoing.DeduplicationId = Guid.NewGuid().ToString();
+        incoming.DeduplicationId.ShouldBe(outgoing.DeduplicationId);
+    }
+
     [Fact]
     public void partition_key()
     {

src/Transports/AWS/Wolverine.AmazonSns.Tests/Internal/sns_fifo_deduplication_id_3793.cs

@@ -0,0 +1,93 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Wolverine.AmazonSns.Internal;
+
+namespace Wolverine.AmazonSns.Tests.Internal;
+
+// GH-3793. Two separate SNS defects, both verified against LocalStack:
+//
+//  1. A FIFO topic without ContentBasedDeduplication rejects any publish that carries no
+//     MessageDeduplicationId ("The topic should either have ContentBasedDeduplication enabled or
+//     MessageDeduplicationId provided explicitly"). Wolverine's circuit-resume ping never had one,
+//     so a latched sender could not probe its way back on such a topic.
+//  2. A *standard* topic rejects a MessageDeduplicationId outright ("the request includes
+//     MessageDeduplicationId parameter that is not valid for this topic type"), so the mapping has
+//     to be gated on the topic type the way AmazonSqsQueue already gates it.
+public class sns_fifo_deduplication_id_3793
+{
+    private static AmazonSnsTopic TopicFor(string name)
+    {
+        return new AmazonSnsTopic(name, new AmazonSnsTransport())
+        {
+            Mapper = new DefaultSnsEnvelopeMapper()
+        };
+    }
+
+    private static string? DeduplicationIdFor(AmazonSnsTopic topic, Envelope envelope)
+    {
+        var batch = new OutgoingSnsBatch(topic, NullLogger.Instance, [envelope]);
+        return batch.Request.PublishBatchRequestEntries.ShouldHaveSingleItem().MessageDeduplicationId;
+    }
+
+    private static Envelope theEnvelope()
+    {
+        return new Envelope
+        {
+            Data = [1, 2, 3],
+            MessageType = "probe-message",
+            Destination = new Uri("sns://probe-topic")
+        };
+    }
+
+    [Fact]
+    public void fifo_detection_follows_the_required_suffix()
+    {
+        TopicFor("orders.fifo").IsFifoTopic.ShouldBeTrue();
+        TopicFor("orders").IsFifoTopic.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void an_explicit_deduplication_id_still_wins_on_a_fifo_topic()
+    {
+        var envelope = theEnvelope();
+        envelope.DeduplicationId = "dedup-1";
+
+        DeduplicationIdFor(TopicFor("orders.fifo"), envelope).ShouldBe("dedup-1");
+    }
+
+    [Fact]
+    public void a_standard_topic_never_gets_a_deduplication_id()
+    {
+        var envelope = theEnvelope();
+        envelope.DeduplicationId = "dedup-1";
+
+        DeduplicationIdFor(TopicFor("orders"), envelope).ShouldBeNull();
+    }
+
+    [Fact]
+    public void a_ping_falls_back_to_the_envelope_id_on_a_fifo_topic()
+    {
+        var ping = Envelope.ForPing(new Uri("sns://orders.fifo"));
+
+        DeduplicationIdFor(TopicFor("orders.fifo"), ping).ShouldBe(ping.Id.ToString());
+    }
+
+    [Fact]
+    public void two_pings_never_share_a_deduplication_id()
+    {
+        // Content-based deduplication would collapse these into one -- every ping body is the
+        // same four bytes -- which is exactly why the ping needs an explicit, unique id.
+        var topic = TopicFor("orders.fifo");
+
+        DeduplicationIdFor(topic, Envelope.ForPing(new Uri("sns://orders.fifo")))
+            .ShouldNotBe(DeduplicationIdFor(topic, Envelope.ForPing(new Uri("sns://orders.fifo"))));
+    }
+
+    [Fact]
+    public void an_ordinary_envelope_with_no_deduplication_id_still_gets_none()
+    {
+        // Only the ping gets the fallback. A user publishing to a FIFO topic with
+        // ContentBasedDeduplication enabled must keep getting content-based behavior.
+        DeduplicationIdFor(TopicFor("orders.fifo"), theEnvelope()).ShouldBeNull();
+    }
+}

src/Transports/AWS/Wolverine.AmazonSns.Tests/fifo_topic_publishing_3793.cs

@@ -0,0 +1,86 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Wolverine.AmazonSns.Internal;
+using Wolverine.Runtime.Serialization;
+
+namespace Wolverine.AmazonSns.Tests;
+
+/// <summary>
+///     GH-3793, end to end against a real FIFO topic with <c>ContentBasedDeduplication</c> turned off.
+///     Both cases below used to fail deterministically with
+///     "Invalid parameter: The topic should either have ContentBasedDeduplication enabled or
+///     MessageDeduplicationId provided explicitly", which is the failure the reporter hit on every
+///     envelope recovered out of the durable outbox after an outage.
+/// </summary>
+public class fifo_topic_publishing_3793 : IAsyncLifetime
+{
+    private const string TopicName = "gh3793_fifo.fifo";
+
+    private IHost _host = null!;
+    private AmazonSnsTopic _topic = null!;
+
+    public async ValueTask InitializeAsync()
+    {
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.UseAmazonSnsTransportLocally().AutoProvision();
+
+                opts.PublishMessage<FifoMessage>()
+                    .ToSnsTopic(TopicName)
+                    .ConfigureTopicCreation(request =>
+                    {
+                        request.Attributes ??= new Dictionary<string, string>();
+                        request.Attributes["FifoTopic"] = "true";
+                        request.Attributes["ContentBasedDeduplication"] = "false";
+                    });
+            }).StartAsync();
+
+        _topic = _host.Services.GetRequiredService<WolverineOptions>()
+            .AmazonSnsTransport().Topics.Single(x => x.TopicName == TopicName);
+
+        await _topic.InitializeAsync(NullLogger.Instance);
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _topic.TeardownAsync(NullLogger.Instance);
+        await _host.StopAsync();
+        _host.Dispose();
+    }
+
+    [Fact]
+    public async Task an_envelope_recovered_from_durable_storage_can_still_be_published()
+    {
+        var envelope = new Envelope
+        {
+            Data = [1, 2, 3],
+            MessageType = "probe-message",
+            GroupId = "group-under-test",
+            DeduplicationId = Guid.NewGuid().ToString(),
+            Destination = _topic.Uri
+        };
+
+        // This is the outbox round trip: recovery after a restart, or reassignment while the
+        // sending agent is latched, both hand the sender an envelope rebuilt from these bytes.
+        var recovered = EnvelopeSerializer.Deserialize(EnvelopeSerializer.Serialize(envelope));
+
+        recovered.DeduplicationId.ShouldBe(envelope.DeduplicationId);
+
+        await Should.NotThrowAsync(() => _topic.SendMessageAsync(recovered, NullLogger.Instance));
+    }
+
+    [Fact]
+    public async Task the_circuit_resume_ping_can_be_published()
+    {
+        // The latched sender probes with this before it will unlatch, so if the ping can't be
+        // published the endpoint stays latched forever no matter how healthy SNS is.
+        var ping = Envelope.ForPing(_topic.Uri);
+
+        await Should.NotThrowAsync(() => _topic.SendMessageAsync(ping, NullLogger.Instance));
+    }
+}
+
+public record FifoMessage(string Name);

src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs

@@ -67,6 +67,29 @@ internal ISnsEnvelopeMapper BuildMapper(IWolverineRuntime runtime)
 
     public string TopicName { get; }
     public string TopicArn { get; set; }
+
+    // AWS requires FIFO topics to carry the ".fifo" suffix, so the name is the only thing
+    // we need to tell the two topic types apart. Mirrors AmazonSqsQueue.IsFifoQueue.
+    internal bool IsFifoTopic => TopicName.EndsWith(".fifo", StringComparison.OrdinalIgnoreCase);
+
+    /// <summary>
+    ///     The <c>MessageDeduplicationId</c> to publish for this envelope, or null for none. A FIFO topic
+    ///     without <c>ContentBasedDeduplication</c> rejects any publish that carries no
+    ///     <c>MessageDeduplicationId</c> at all, and Wolverine's own circuit-resume ping never has one --
+    ///     so a latched sender could never probe its way back on such a topic. Fall back to the envelope
+    ///     id for pings, which is unique per probe and is exactly the semantic we want (two pings must
+    ///     never dedupe against each other, which content-based deduplication would happily do since every
+    ///     ping body is identical). See GH-3793.
+    /// </summary>
+    internal static string? DetermineDeduplicationId(Envelope envelope)
+    {
+        if (envelope.DeduplicationId.IsNotEmpty())
+        {
+            return envelope.DeduplicationId;
+        }
+
+        return envelope.IsPing() ? envelope.Id.ToString() : null;
+    }
 
     [ChildDescription]
     public CreateTopicRequest Configuration { get; }
@@ -136,9 +159,16 @@ internal async Task SendMessageAsync(Envelope envelope, ILogger logger)
             request.MessageGroupId = envelope.GroupId;
         }
 
-        if (envelope.DeduplicationId.IsNotEmpty())
+        // SNS rejects MessageDeduplicationId outright on a standard topic ("the request includes
+        // MessageDeduplicationId parameter that is not valid for this topic type"), so this has to be
+        // gated the same way AmazonSqsQueue gates it. See GH-3793.
+        if (IsFifoTopic)
         {
-            request.MessageDeduplicationId = envelope.DeduplicationId;
+            var deduplicationId = DetermineDeduplicationId(envelope);
+            if (deduplicationId.IsNotEmpty())
+            {
+                request.MessageDeduplicationId = deduplicationId;
+            }
         }
 
         foreach (var attribute in Mapper!.ToAttributes(envelope))

src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs

@@ -64,9 +64,13 @@ public OutgoingSnsBatch(AmazonSnsTopic topic, ILogger logger, IEnumerable<Envelo
                 {
                     entry.MessageGroupId = envelope.GroupId;
                 }
-                if (envelope.DeduplicationId.IsNotEmpty())
+                if (topic.IsFifoTopic)
                 {
-                    entry.MessageDeduplicationId = envelope.DeduplicationId;
+                    var deduplicationId = AmazonSnsTopic.DetermineDeduplicationId(envelope);
+                    if (deduplicationId.IsNotEmpty())
+                    {
+                        entry.MessageDeduplicationId = deduplicationId;
+                    }
                 }
 
                 foreach (var attribute in topic.Mapper.ToAttributes(envelope))

src/Transports/AWS/Wolverine.AmazonSqs.Tests/Internal/sqs_fifo_deduplication_id_3793.cs

@@ -0,0 +1,75 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Wolverine.AmazonSqs.Internal;
+using Wolverine.ComplianceTests;
+
+namespace Wolverine.AmazonSqs.Tests.Internal;
+
+// GH-3793: a FIFO queue that does not have ContentBasedDeduplication turned on rejects any send
+// with no MessageDeduplicationId at all ("The queue should either have ContentBasedDeduplication
+// enabled or MessageDeduplicationId provided explicitly"). Wolverine's circuit-resume ping carries
+// no deduplication id of its own, so a latched sender could never probe its way back on such a
+// queue -- the probe itself failed deterministically.
+public class sqs_fifo_deduplication_id_3793
+{
+    private static AmazonSqsQueue QueueFor(string name)
+    {
+        return new AmazonSqsQueue(name, new AmazonSqsTransport())
+        {
+            Mapper = new DefaultSqsEnvelopeMapper()
+        };
+    }
+
+    private static string? DeduplicationIdFor(AmazonSqsQueue queue, Envelope envelope)
+    {
+        var batch = new OutgoingSqsBatch(queue, NullLogger.Instance, [envelope]);
+        return batch.Request.Entries.ShouldHaveSingleItem().MessageDeduplicationId;
+    }
+
+    [Fact]
+    public void an_explicit_deduplication_id_still_wins()
+    {
+        var envelope = ObjectMother.Envelope();
+        envelope.DeduplicationId = "dedup-1";
+
+        DeduplicationIdFor(QueueFor("orders.fifo"), envelope).ShouldBe("dedup-1");
+    }
+
+    [Fact]
+    public void a_ping_falls_back_to_the_envelope_id_on_a_fifo_queue()
+    {
+        var ping = Envelope.ForPing(new Uri("sqs://orders.fifo"));
+
+        DeduplicationIdFor(QueueFor("orders.fifo"), ping).ShouldBe(ping.Id.ToString());
+    }
+
+    [Fact]
+    public void two_pings_never_share_a_deduplication_id()
+    {
+        // Content-based deduplication would collapse these into one -- every ping body is the
+        // same four bytes -- which is exactly why the ping needs an explicit, unique id.
+        var queue = QueueFor("orders.fifo");
+
+        DeduplicationIdFor(queue, Envelope.ForPing(new Uri("sqs://orders.fifo")))
+            .ShouldNotBe(DeduplicationIdFor(queue, Envelope.ForPing(new Uri("sqs://orders.fifo"))));
+    }
+
+    [Fact]
+    public void an_ordinary_envelope_with_no_deduplication_id_still_gets_none()
+    {
+        // Only the ping gets the fallback. A user publishing to a FIFO queue with
+        // ContentBasedDeduplication enabled must keep getting content-based behavior.
+        var envelope = ObjectMother.Envelope();
+        envelope.DeduplicationId = null;
+
+        DeduplicationIdFor(QueueFor("orders.fifo"), envelope).ShouldBeNull();
+    }
+
+    [Fact]
+    public void a_ping_to_a_standard_queue_gets_no_deduplication_id()
+    {
+        var ping = Envelope.ForPing(new Uri("sqs://orders"));
+
+        DeduplicationIdFor(QueueFor("orders"), ping).ShouldBeNull();
+    }
+}

src/Transports/AWS/Wolverine.AmazonSqs/Internal/AmazonSqsQueue.cs

@@ -61,6 +61,25 @@ internal AmazonSqsQueue(string queueName, AmazonSqsTransport parent) : base(
 
     internal bool IsFifoQueue => QueueName.EndsWith(".fifo", StringComparison.OrdinalIgnoreCase);
 
+    /// <summary>
+    ///     The <c>MessageDeduplicationId</c> to send for this envelope, or null for none. A FIFO queue
+    ///     without <c>ContentBasedDeduplication</c> rejects any send that carries no
+    ///     <c>MessageDeduplicationId</c> at all, and Wolverine's own circuit-resume ping never has one --
+    ///     so a latched sender could never probe its way back on such a queue. Fall back to the envelope
+    ///     id for pings, which is unique per probe and is exactly the semantic we want (two pings must
+    ///     never dedupe against each other, which content-based deduplication would happily do since every
+    ///     ping body is identical). See GH-3793.
+    /// </summary>
+    internal static string? DetermineDeduplicationId(Envelope envelope)
+    {
+        if (envelope.DeduplicationId.IsNotEmpty())
+        {
+            return envelope.DeduplicationId;
+        }
+
+        return envelope.IsPing() ? envelope.Id.ToString() : null;
+    }
+
     /// <summary>
     ///     Opt this standard (non-FIFO) queue into Amazon SQS fair queues by mapping
     ///     <see cref="Envelope.GroupId"/> to the SQS <c>MessageGroupId</c> on outgoing messages.
@@ -359,9 +378,10 @@ internal async Task SendMessageAsync(Envelope envelope, ILogger logger)
                 request.MessageGroupId = groupId;
             }
 
-            if (envelope.DeduplicationId.IsNotEmpty())
+            var deduplicationId = DetermineDeduplicationId(envelope);
+            if (deduplicationId.IsNotEmpty())
             {
-                request.MessageDeduplicationId = envelope.DeduplicationId;
+                request.MessageDeduplicationId = deduplicationId;
             }
         }
         else if (EnableFairQueueMessageGroups)

src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs

@@ -204,9 +204,10 @@ public OutgoingSqsBatch(AmazonSqsQueue queue, ILogger logger, IEnumerable<Envelo
                     {
                         entry.MessageGroupId = groupId;
                     }
-                    if (envelope.DeduplicationId.IsNotEmpty())
+                    var deduplicationId = AmazonSqsQueue.DetermineDeduplicationId(envelope);
+                    if (deduplicationId.IsNotEmpty())
                     {
-                        entry.MessageDeduplicationId = envelope.DeduplicationId;
+                        entry.MessageDeduplicationId = deduplicationId;
                     }
                 }
                 else if (queue.EnableFairQueueMessageGroups)