From d5a3d1eb04ccd8ca65cb2a289cc4488af3fce04b Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Thu, 30 Jul 2026 13:06:46 -0500 Subject: [PATCH] Fix InvalidCastException starting a NATS listener with a dead-letter subject (GH-3739) NatsEndpoint.BuildListenerAsync resolved the dead-letter sender by casting the result of IEndpointCollection.GetOrBuildSendingAgent(...) straight to ISender. That method returns an ISendingAgent -- a BufferedSendingAgent, DurableSendingAgent or InlineSendingAgent -- and none of those implement ISender, so the cast threw InvalidCastException during WolverineRuntime.StartAsync for every listener configured with a dead-letter subject, regardless of the destination endpoint's sending mode. The whole host failed to start; the only workaround was to drop the dead-letter subject and let poison messages be silently discarded. Reach through to the sender the agent wraps instead, the same way EndpointCollection already does when it resolves connection state. The underlying sender is deliberately what we want here rather than the agent: NatsListener.MoveToErrorsAsync publishes the poison message and only then terminates JetStream delivery, so the forward has to reach the broker before the terminate. Enqueuing on a buffered agent would return before the message was on the wire, and a crash in between would lose it. The NATS suite had no dead-letter coverage at all, which is how this survived two releases. NatsDeadLetterSubjectTests adds both the startup regression and an end-to-end check that a poison message really lands on the configured dead-letter subject with its failure metadata. Both fail with the reported InvalidCastException without the fix. Co-Authored-By: Claude Opus 5 (1M context) --- .../NatsDeadLetterSubjectTests.cs | 162 ++++++++++++++++++ .../Wolverine.Nats/Internal/NatsEndpoint.cs | 34 +++- 2 files changed, 195 insertions(+), 1 deletion(-) create mode 100644 src/Transports/NATS/Wolverine.Nats.Tests/NatsDeadLetterSubjectTests.cs diff --git a/src/Transports/NATS/Wolverine.Nats.Tests/NatsDeadLetterSubjectTests.cs b/src/Transports/NATS/Wolverine.Nats.Tests/NatsDeadLetterSubjectTests.cs new file mode 100644 index 000000000..e99f85a5e --- /dev/null +++ b/src/Transports/NATS/Wolverine.Nats.Tests/NatsDeadLetterSubjectTests.cs @@ -0,0 +1,162 @@ +using System.Text; +using IntegrationTests; +using JasperFx.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.Attributes; +using Wolverine.ErrorHandling; +using Wolverine.Nats.Internal; +using Wolverine.Runtime; +using Wolverine.Transports.Sending; +using Xunit; + +namespace Wolverine.Nats.Tests; + +/// +/// Regression coverage for GH-3739: NatsEndpoint.BuildListenerAsync resolved the dead-letter sender by +/// casting the result of IEndpointCollection.GetOrBuildSendingAgent(...) straight to ISender. That +/// method returns an ISendingAgent — a BufferedSendingAgent, DurableSendingAgent or +/// InlineSendingAgent — and none of those implement ISender, so the cast threw +/// InvalidCastException during WolverineRuntime.StartAsync for *every* listener configured with a +/// dead-letter subject, regardless of the destination endpoint's sending mode. The whole host failed to start; +/// the only workaround was to drop the dead-letter subject and let poison messages be discarded. +/// +/// The NATS suite had no dead-letter coverage at all, which is how this survived two releases. +/// +[Collection("NATS Integration")] +[Trait("Category", "Integration")] +public class NatsDeadLetterSubjectTests +{ + private readonly ITestOutputHelper _output; + + public NatsDeadLetterSubjectTests(ITestOutputHelper output) => _output = output; + + [Fact] + public async Task listener_with_a_dead_letter_subject_starts_up() + { + var natsUrl = NatsTestHelpers.ResolveUrl(); + if (!await NatsTestHelpers.IsNatsAvailable(natsUrl)) return; + + var id = Guid.NewGuid().ToString("N"); + var stream = $"DLQSTART_{id}"; + var subject = $"dlqstart.{id}.incoming"; + var deadLetterSubject = $"dlqstart-errors.{id}"; + + // Before the fix this threw InvalidCastException out of StartAsync + using var host = await Host.CreateDefaultBuilder() + .ConfigureLogging(l => l.AddXunitLogging(_output)) + .UseWolverine(opts => + { + opts.UseNats(natsUrl) + .AutoProvision() + .DefineStream(stream, s => s.WithSubjects($"dlqstart.{id}.>")); + + opts.Policies.DisableConventionalLocalRouting(); + + opts.ListenToNatsSubject(subject) + .UseJetStream(stream, $"dlqstart-consumer-{id}") + .ConfigureDeadLetterQueue(1, deadLetterSubject); + }) + .StartAsync(cancellationToken: TestContext.Current.CancellationToken); + + var runtime = host.Services.GetRequiredService(); + + // ...and the sender it resolves has to be the transport-level ISender the listener needs, not the + // ISendingAgent wrapper. NatsListener.MoveToErrorsAsync publishes the poison message and only then + // terminates JetStream delivery, so the forward must reach the broker before the terminate — enqueuing + // on a buffered agent would return before the message was on the wire. + var transport = runtime.Options.Transports.GetOrCreate(); + var dlqEndpoint = transport.EndpointForSubject(deadLetterSubject); + var agent = runtime.Endpoints.GetOrBuildSendingAgent(dlqEndpoint.Uri); + + agent.ShouldNotBeAssignableTo(); + agent.ShouldBeAssignableTo() + .Sender.ShouldBeOfType(); + } + + [Fact] + public async Task poison_message_is_forwarded_to_the_configured_dead_letter_subject() + { + var natsUrl = NatsTestHelpers.ResolveUrl(); + if (!await NatsTestHelpers.IsNatsAvailable(natsUrl)) return; + + var id = Guid.NewGuid().ToString("N"); + var stream = $"DLQFORWARD_{id}"; + var subject = $"dlqforward.{id}.incoming"; + + // Deliberately outside the stream's subject space so the dead-letter copy is a plain core-NATS + // publish that a raw subscriber can observe without JetStream getting in the way + var deadLetterSubject = $"dlqforward-errors.{id}"; + + PoisonMessageHandler.Reset(); + + // Subscribe before the host starts so the forward can't be missed + await using var dlqSubscription = await NatsTestHelpers.SubscribeRawAsync(natsUrl, deadLetterSubject); + + using var host = await Host.CreateDefaultBuilder() + .ConfigureLogging(l => l.AddXunitLogging(_output)) + .UseWolverine(opts => + { + opts.UseNats(natsUrl) + .AutoProvision() + .DefineStream(stream, s => s.WithSubjects($"dlqforward.{id}.>")); + + opts.Policies.DisableConventionalLocalRouting(); + + // MaxDeliveryAttempts of 1 means the very first failure is already terminal, so the test + // doesn't have to wait out a JetStream redelivery cycle + opts.ListenToNatsSubject(subject) + .UseJetStream(stream, $"dlqforward-consumer-{id}") + .ConfigureDeadLetterQueue(1, deadLetterSubject); + + opts.PublishMessage().ToNatsSubject(subject).UseJetStream(stream).SendInline(); + + opts.Policies.OnException().MoveToErrorQueue(); + }) + .StartAsync(cancellationToken: TestContext.Current.CancellationToken); + + await host.MessageBus().SendAsync(new PoisonMessage(id)); + + await PoisonMessageHandler.WaitForAttemptAsync(); + + var deadLettered = await dlqSubscription.ReadAsync(30.Seconds()); + deadLettered.ShouldNotBeNull(); + + // The forwarded copy carries the original body plus the failure metadata NatsListener stamps on it + Encoding.UTF8.GetString(deadLettered.Value.Data!).ShouldContain(id); + deadLettered.Value.Headers.ShouldNotBeNull(); + deadLettered.Value.Headers!["x-dlq-original-subject"].ToString().ShouldBe(subject); + } +} + +public record PoisonMessage(string Id); + +public class PoisonPillException : Exception +{ + public PoisonPillException(string message) : base(message) + { + } +} + +[WolverineHandler] +public static class PoisonMessageHandler +{ + private static TaskCompletionSource _attempted = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public static void Reset() + { + _attempted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + public static void Handle(PoisonMessage message) + { + _attempted.TrySetResult(); + throw new PoisonPillException($"This message is poison: {message.Id}"); + } + + public static Task WaitForAttemptAsync() + { + return _attempted.Task.WaitAsync(30.Seconds()); + } +} diff --git a/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs index 4554ac7db..b84cb2214 100644 --- a/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs +++ b/src/Transports/NATS/Wolverine.Nats/Internal/NatsEndpoint.cs @@ -249,7 +249,7 @@ IReceiver receiver if (!string.IsNullOrEmpty(DeadLetterSubject)) { var dlqEndpoint = _transport.EndpointForSubject(DeadLetterSubject); - deadLetterSender = (ISender)runtime.Endpoints.GetOrBuildSendingAgent(dlqEndpoint.Uri); + deadLetterSender = resolveDeadLetterSender(runtime, dlqEndpoint); } var useJetStream = UseJetStream && _transport.Configuration.EnableJetStream; @@ -297,6 +297,38 @@ IReceiver receiver return compound; } + /// + /// Resolve the transport-level for a dead-letter subject. + /// + /// + /// + /// hands back an — + /// a BufferedSendingAgent, DurableSendingAgent or — and none + /// of those implement . Casting the agent straight to therefore + /// threw an during listener startup for every listener with a configured + /// dead-letter subject (GH-3739). Reach through to the sender the agent wraps instead, the same way + /// EndpointCollection does when it resolves connection state. + /// + /// + /// The underlying sender is deliberately what we want here rather than the agent: + /// publishes the poison message and only then terminates JetStream delivery, so the forward has to reach the + /// broker before the terminate. Enqueuing on a buffered agent would return before the message was on the wire + /// and a crash in between would lose it. + /// + /// + private ISender resolveDeadLetterSender(IWolverineRuntime runtime, NatsEndpoint dlqEndpoint) + { + var agent = runtime.Endpoints.GetOrBuildSendingAgent(dlqEndpoint.Uri); + + return agent switch + { + SendingAgent sendingAgent => sendingAgent.Sender, + InlineSendingAgent inlineAgent => inlineAgent.Sender, + _ => throw new InvalidOperationException( + $"Unable to resolve an {nameof(ISender)} for the dead letter subject '{DeadLetterSubject}' of the NATS listener at '{Uri}'. The sending agent at '{dlqEndpoint.Uri}' was a {agent.GetType().FullName}.") + }; + } + private async ValueTask startListenerAsync( IWolverineRuntime runtime, IReceiver receiver,