diff --git a/src/Testing/CoreTests/Serialization/serialization_and_deserialization.cs b/src/Testing/CoreTests/Serialization/serialization_and_deserialization.cs index fdddb9227..2a38d23f3 100644 --- a/src/Testing/CoreTests/Serialization/serialization_and_deserialization.cs +++ b/src/Testing/CoreTests/Serialization/serialization_and_deserialization.cs @@ -231,6 +231,18 @@ public void group_id() incoming.GroupId.ShouldBe(outgoing.GroupId); } + /// + /// 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. + /// + [Fact] + public void deduplication_id() + { + outgoing.DeduplicationId = Guid.NewGuid().ToString(); + incoming.DeduplicationId.ShouldBe(outgoing.DeduplicationId); + } + [Fact] public void partition_key() { diff --git a/src/Transports/AWS/Wolverine.AmazonSns.Tests/Internal/sns_fifo_deduplication_id_3793.cs b/src/Transports/AWS/Wolverine.AmazonSns.Tests/Internal/sns_fifo_deduplication_id_3793.cs new file mode 100644 index 000000000..2f1ae090b --- /dev/null +++ b/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(); + } +} diff --git a/src/Transports/AWS/Wolverine.AmazonSns.Tests/fifo_topic_publishing_3793.cs b/src/Transports/AWS/Wolverine.AmazonSns.Tests/fifo_topic_publishing_3793.cs new file mode 100644 index 000000000..99b4d3922 --- /dev/null +++ b/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; + +/// +/// GH-3793, end to end against a real FIFO topic with ContentBasedDeduplication 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. +/// +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() + .ToSnsTopic(TopicName) + .ConfigureTopicCreation(request => + { + request.Attributes ??= new Dictionary(); + request.Attributes["FifoTopic"] = "true"; + request.Attributes["ContentBasedDeduplication"] = "false"; + }); + }).StartAsync(); + + _topic = _host.Services.GetRequiredService() + .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); diff --git a/src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs b/src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs index dba0639b7..3455d292c 100644 --- a/src/Transports/AWS/Wolverine.AmazonSns/Internal/AmazonSnsTopic.cs +++ b/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); + + /// + /// The MessageDeduplicationId to publish for this envelope, or null for none. A FIFO topic + /// without ContentBasedDeduplication rejects any publish that carries no + /// MessageDeduplicationId 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. + /// + 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)) diff --git a/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs b/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs index a0491dd55..0ef80f1c7 100644 --- a/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs +++ b/src/Transports/AWS/Wolverine.AmazonSns/Internal/SnsSenderProtocol.cs @@ -64,9 +64,13 @@ public OutgoingSnsBatch(AmazonSnsTopic topic, ILogger logger, IEnumerable QueueName.EndsWith(".fifo", StringComparison.OrdinalIgnoreCase); + /// + /// The MessageDeduplicationId to send for this envelope, or null for none. A FIFO queue + /// without ContentBasedDeduplication rejects any send that carries no + /// MessageDeduplicationId 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. + /// + internal static string? DetermineDeduplicationId(Envelope envelope) + { + if (envelope.DeduplicationId.IsNotEmpty()) + { + return envelope.DeduplicationId; + } + + return envelope.IsPing() ? envelope.Id.ToString() : null; + } + /// /// Opt this standard (non-FIFO) queue into Amazon SQS fair queues by mapping /// to the SQS MessageGroupId 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) diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs index 480d6c7c7..674485d35 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsSenderProtocol.cs @@ -204,9 +204,10 @@ public OutgoingSqsBatch(AmazonSqsQueue queue, ILogger logger, IEnumerable