From 3b5eaf93fb5bd62ab1969b6e4fff459c0c64d76e Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Fri, 6 Nov 2026 06:43:44 -0600 Subject: [PATCH 1/2] MassTransit interop keeps the inbox id of a durable envelope when it reads the messageId MassTransitEnvelope.TransferData always replaced Envelope.Id with the MassTransit messageId when the handler pipeline deserialized the body. On SQS the MassTransit mapper leaves the id to the listener, which derives it from the SQS message, and the durable inbox stores the envelope under that id. Mark-handled, retry and dead-letter then matched no row, so the inbox row stayed Incoming and ran again on recovery. Same shape as the CloudEvents case in #5396. TransferData now leaves the id alone for an envelope that WasPersistedInInbox. Tests: CoreTests unit test, and an SQS (LocalStack) + PostgreSQL inbox test that fails without the fix with the row left Incoming. Co-Authored-By: Claude Opus 5.5 --- .../Interop/MassTransitEnvelopeTests.cs | 14 +++ ...terop_with_durable_inbox_keeps_inbox_id.cs | 95 +++++++++++++++++++ .../MassTransit/MassTransitEnvelope.cs | 4 +- 3 files changed, 112 insertions(+), 1 deletion(-) create mode 100644 src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_interop_with_durable_inbox_keeps_inbox_id.cs diff --git a/src/Testing/CoreTests/Runtime/Interop/MassTransitEnvelopeTests.cs b/src/Testing/CoreTests/Runtime/Interop/MassTransitEnvelopeTests.cs index 04742bf115..e6cda01724 100644 --- a/src/Testing/CoreTests/Runtime/Interop/MassTransitEnvelopeTests.cs +++ b/src/Testing/CoreTests/Runtime/Interop/MassTransitEnvelopeTests.cs @@ -68,6 +68,20 @@ public void map_the_message_id() theEnvelope.Id.ShouldBe(Guid.Parse(theMassTransitEnvelope.MessageId!)); } + [Fact] + public void an_envelope_held_in_the_inbox_keeps_the_id_it_was_stored_under() + { + var storedId = Guid.NewGuid(); + var envelope = new Envelope { Id = storedId, WasPersistedInInbox = true }; + + theMassTransitEnvelope.TransferData(envelope); + + envelope.Id.ShouldBe(storedId); + + // everything else still comes across + envelope.CorrelationId.ShouldBe(theMassTransitEnvelope.CorrelationId); + } + [Fact] public void map_the_correlation_id() { diff --git a/src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_interop_with_durable_inbox_keeps_inbox_id.cs b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_interop_with_durable_inbox_keeps_inbox_id.cs new file mode 100644 index 0000000000..1d5fe764bd --- /dev/null +++ b/src/Transports/AWS/Wolverine.AmazonSqs.Tests/masstransit_interop_with_durable_inbox_keeps_inbox_id.cs @@ -0,0 +1,95 @@ +using Amazon.Runtime; +using Amazon.SQS; +using Amazon.SQS.Model; +using IntegrationTests; +using JasperFx.Core; +using JasperFx.Resources; +using Microsoft.Extensions.Hosting; +using Npgsql; +using Shouldly; +using Wolverine.Postgresql; +using Xunit; + +namespace Wolverine.AmazonSqs.Tests; + +/// +/// The SQS MassTransit mapper leaves Envelope.Id to the listener, which derives it from the SQS message, and the +/// durable inbox stores the envelope under that id. The MassTransit messageId is only read later, when the handler +/// pipeline deserializes the body, and that read must not replace the id: marking the message handled finds its +/// inbox row by the id it was stored under. +/// +public class masstransit_interop_with_durable_inbox_keeps_inbox_id +{ + private const string Schema = "sqs_mt_durable_inbox"; + + [Fact] + public async Task a_handled_masstransit_message_is_marked_handled_in_the_inbox() + { + var queue = $"mt-inbox-{Guid.NewGuid():N}"; + + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.UseAmazonSqsTransportLocally().AutoProvision(); + opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, Schema); + + opts.ListenToSqsQueue(queue) + .UseMassTransitInterop() + .DefaultIncomingMessage() + .UseDurableInbox(); + + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(MassTransitInboxOrderHandler)); + opts.Services.AddResourceSetupOnStartup(); + }).StartAsync(TestContext.Current.CancellationToken); + + var client = new AmazonSQSClient(new BasicAWSCredentials("ignore", "ignore"), + new AmazonSQSConfig { ServiceURL = "http://localhost:4566" }); + var queueUrl = (await client.GetQueueUrlAsync(queue, TestContext.Current.CancellationToken)).QueueUrl; + + // What a MassTransit producer puts on the wire: its metadata, and the real message under "message" + var body = $$""" + { "messageId": "{{Guid.NewGuid()}}", "messageType": ["urn:message:Orders:MassTransitInboxOrder"], + "message": { "orderId": 5 } } + """; + await client.SendMessageAsync(new SendMessageRequest(queueUrl, body), TestContext.Current.CancellationToken); + + await MassTransitInboxOrderHandler.Handled.WaitAsync(60.Seconds(), TestContext.Current.CancellationToken); + + // The mark-handled write is not part of the handler's own commit, so give it a moment to land + var statuses = new List(); + var deadline = DateTimeOffset.UtcNow.AddSeconds(15); + await using var conn = new NpgsqlConnection(Servers.PostgresConnectionString); + await conn.OpenAsync(TestContext.Current.CancellationToken); + while (true) + { + statuses = await statusesAsync(conn, queue); + if (statuses.Count == 0 || statuses.All(x => x == "Handled") || DateTimeOffset.UtcNow > deadline) break; + await Task.Delay(100, TestContext.Current.CancellationToken); + } + + // Either retained as Handled or already deleted; never left Incoming + statuses.ShouldNotContain("Incoming", $"Inbox statuses: {string.Join(", ", statuses)}"); + } + + private static async Task> statusesAsync(NpgsqlConnection conn, string queue) + { + await using var cmd = new NpgsqlCommand( + $"select status from {Schema}.wolverine_incoming_envelopes where received_at like @uri", conn); + cmd.Parameters.AddWithValue("uri", $"%{queue}%"); + var result = new List(); + await using var reader = await cmd.ExecuteReaderAsync(TestContext.Current.CancellationToken); + while (await reader.ReadAsync(TestContext.Current.CancellationToken)) result.Add(reader.GetString(0)); + return result; + } +} + +public record MassTransitInboxOrder(int OrderId); + +public static class MassTransitInboxOrderHandler +{ + private static readonly TaskCompletionSource _handled = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public static Task Handled => _handled.Task; + + public static void Handle(MassTransitInboxOrder message) => _handled.TrySetResult(); +} diff --git a/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs b/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs index d48ed188d0..d6bfb8b17c 100644 --- a/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs +++ b/src/Wolverine/Runtime/Interop/MassTransit/MassTransitEnvelope.cs @@ -67,7 +67,9 @@ protected MassTransitEnvelope(Envelope envelope) public void TransferData(Envelope envelope) { - if (MessageId != null && Guid.TryParse(MessageId, out var id)) + // An envelope already stored in the durable inbox is found by the id it was stored under -- on SQS that is + // derived from the SQS message, not the MassTransit messageId -- so reading it must not replace that id + if (!envelope.WasPersistedInInbox && MessageId != null && Guid.TryParse(MessageId, out var id)) { envelope.Id = id; } From e69891d435907b665761c4f6f40634766c329de2 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Fri, 6 Nov 2026 07:34:18 -0600 Subject: [PATCH 2/2] Reset WithinCodegenCommand after the codegen-command test so later hosts are not MediatorOnly The GH-5391 test runs `codegen test` / `codegen write` through RunJasperFxCommands, which leaves the process-wide DynamicCodeBuilder.WithinCodegenCommand flag set. applyMetadataOnlyModeIfDetected then starts every later host in that test process in MediatorOnly mode, so whichever CoreTests ran after it in the same worker failed ("This operation is not allowed with Wolverine is bootstrapped in MediatorOnly mode") -- red on main at 6.62.1, green at a89e663dc, depending only on test order. Reset the flag after the commands and in finally, and run the class in a non-parallel collection so no host starts while the flag is set (the same arrangement Wolverine.Grpc.Tests uses for this flag). Co-Authored-By: Claude Opus 5.5 --- .../Configuration/reusing_compiled_assemblies.cs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/Testing/CoreTests/Configuration/reusing_compiled_assemblies.cs b/src/Testing/CoreTests/Configuration/reusing_compiled_assemblies.cs index 4e895de3be..da7775f4c5 100644 --- a/src/Testing/CoreTests/Configuration/reusing_compiled_assemblies.cs +++ b/src/Testing/CoreTests/Configuration/reusing_compiled_assemblies.cs @@ -19,6 +19,7 @@ public static void Handle(ReuseCompiledMessage message) } // GH-5378 / jasperfx#1114 +[Collection(CodegenCommandCollection.Name)] public class reusing_compiled_assemblies { private static Task startHost(bool reuse, TypeLoadMode mode = TypeLoadMode.Dynamic, @@ -103,13 +104,28 @@ public async Task codegen_test_and_write_commands_succeed_with_the_flag_on_and_h (await buildHost(true, TypeLoadMode.Auto, root).RunJasperFxCommands(["codegen", "write"])) .ShouldBe(0); + // The codegen command leaves this process-wide flag set, and while it is set every host starts in + // MediatorOnly mode -- these two, and every other test's host in this process + DynamicCodeBuilder.WithinCodegenCommand = false; + using var one = await startHost(true, TypeLoadMode.Auto, root); using var two = await startHost(true, TypeLoadMode.Auto, root); compiledHandlerType(one).Assembly.ShouldBeSameAs(compiledHandlerType(two).Assembly); } finally { + DynamicCodeBuilder.WithinCodegenCommand = false; Directory.Delete(root, true); } } } + +/// +/// Running a codegen command sets DynamicCodeBuilder.WithinCodegenCommand, a static flag that makes every +/// host started meanwhile come up in MediatorOnly mode, so nothing may run beside these tests +/// +[CollectionDefinition(Name, DisableParallelization = true)] +public class CodegenCommandCollection +{ + public const string Name = "CodegenCommand"; +}