Fix InvalidCastException starting a NATS listener with a dead-letter subject (GH-3739)

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;
+
+/// <summary>
+/// Regression coverage for GH-3739: <c>NatsEndpoint.BuildListenerAsync</c> resolved the dead-letter sender by
+/// casting the result of <c>IEndpointCollection.GetOrBuildSendingAgent(...)</c> straight to <c>ISender</c>. That
+/// method returns an <c>ISendingAgent</c> — a <c>BufferedSendingAgent</c>, <c>DurableSendingAgent</c> or
+/// <c>InlineSendingAgent</c> — and none of those implement <c>ISender</c>, so the cast threw
+/// <c>InvalidCastException</c> during <c>WolverineRuntime.StartAsync</c> 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.
+/// </summary>
+[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<IWolverineRuntime>();
+
+        // ...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<NatsTransport>();
+        var dlqEndpoint = transport.EndpointForSubject(deadLetterSubject);
+        var agent = runtime.Endpoints.GetOrBuildSendingAgent(dlqEndpoint.Uri);
+
+        agent.ShouldNotBeAssignableTo<ISender>();
+        agent.ShouldBeAssignableTo<SendingAgent>()
+            .Sender.ShouldBeOfType<NatsSender>();
+    }
+
+    [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<PoisonMessage>().ToNatsSubject(subject).UseJetStream(stream).SendInline();
+
+                opts.Policies.OnException<PoisonPillException>().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());
+    }
+}

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;
     }
 
+    /// <summary>
+    /// Resolve the transport-level <see cref="ISender"/> for a dead-letter subject.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <see cref="IEndpointCollection.GetOrBuildSendingAgent"/> hands back an <see cref="ISendingAgent"/> —
+    /// a <c>BufferedSendingAgent</c>, <c>DurableSendingAgent</c> or <see cref="InlineSendingAgent"/> — and none
+    /// of those implement <see cref="ISender"/>. Casting the agent straight to <see cref="ISender"/> therefore
+    /// threw an <see cref="InvalidCastException"/> 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
+    /// <c>EndpointCollection</c> does when it resolves connection state.
+    /// </para>
+    /// <para>
+    /// The underlying sender is deliberately what we want here rather than the agent: <see cref="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.
+    /// </para>
+    /// </remarks>
+    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<NatsListener> startListenerAsync(
         IWolverineRuntime runtime,
         IReceiver receiver,