Fix consume 503 handling

src/NATS.Client.JetStream/INatsJSNotification.cs

@@ -7,7 +7,30 @@ public interface INatsJSNotification
 
 public record NatsJSTimeoutNotification : INatsJSNotification
 {
+    public static readonly NatsJSTimeoutNotification Default = new();
+
     public string Name => "Timeout";
 }
 
+public record NatsJSNoRespondersNotification : INatsJSNotification
+{
+    public static readonly NatsJSNoRespondersNotification Default = new();
+
+    public string Name => "No Responders";
+}
+
+public record NatsJSLeadershipChangeNotification : INatsJSNotification
+{
+    public static readonly NatsJSLeadershipChangeNotification Default = new();
+
+    public string Name => "Leadership Change";
+}
+
+public record NatsJSMessageSizeExceedsMaxBytesNotification : INatsJSNotification
+{
+    public static readonly NatsJSMessageSizeExceedsMaxBytesNotification Default = new();
+
+    public string Name => "Message Size Exceeds MaxBytes";
+}
+
 public record NatsJSProtocolNotification(string Name, int HeaderCode, string HeaderMessageText) : INatsJSNotification;

src/NATS.Client.JetStream/Internal/NatsJSConsume.cs

@@ -104,7 +104,7 @@ public NatsJSConsume(
             static state =>
             {
                 var self = (NatsJSConsume<TMsg>)state!;
-                self._notificationChannel?.Notify(new NatsJSTimeoutNotification());
+                self._notificationChannel?.Notify(NatsJSTimeoutNotification.Default);
 
                 if (self._cancellationToken.IsCancellationRequested)
                 {
@@ -338,15 +338,30 @@ protected override async ValueTask ReceiveInternalAsync(
                     if (headers is { Code: 408, Message: NatsHeaders.Messages.RequestTimeout })
                     {
                     }
-                    else if (headers is { Code: 409, Message: NatsHeaders.Messages.MessageSizeExceedsMaxBytes })
+                    else if (headers is { Code: 100, Message: NatsHeaders.Messages.IdleHeartbeat })
                     {
+                        // No action is required for idle heartbeat notifications.
+                        // This branch is intentionally left empty.
                     }
-                    else if (headers is { Code: 100, Message: NatsHeaders.Messages.IdleHeartbeat })
+                    else if (headers is { Code: 409, Message: NatsHeaders.Messages.MessageSizeExceedsMaxBytes })
                     {
+                        _logger.LogWarning(NatsJSLogEvents.MessageSizeExceedsMaxBytes, "Message Size Exceeds MaxBytes");
+                        _notificationChannel?.Notify(NatsJSMessageSizeExceedsMaxBytesNotification.Default);
                     }
                     else if (headers.Code == 409 && string.Equals(headers.MessageText, "Leadership Change", StringComparison.OrdinalIgnoreCase))
                     {
                         _logger.LogDebug(NatsJSLogEvents.LeadershipChange, "Leadership Change");
+                        _notificationChannel?.Notify(NatsJSLeadershipChangeNotification.Default);
+                        lock (_pendingGate)
+                        {
+                            _pendingBytes = 0;
+                            _pendingMsgs = 0;
+                        }
+                    }
+                    else if (headers.Code == 503)
+                    {
+                        _logger.LogDebug(NatsJSLogEvents.NoResponders, "503 no responders");
+                        _notificationChannel?.Notify(NatsJSNoRespondersNotification.Default);
                         lock (_pendingGate)
                         {
                             _pendingBytes = 0;
@@ -360,8 +375,8 @@ protected override async ValueTask ReceiveInternalAsync(
                     }
                     else
                     {
-                        _notificationChannel?.Notify(new NatsJSProtocolNotification("Unhandled protocol message", headers.Code, headers.MessageText));
                         _logger.LogWarning(NatsJSLogEvents.ProtocolMessage, "Unhandled protocol message: {Code} {Description}", headers.Code, headers.MessageText);
+                        _notificationChannel?.Notify(new NatsJSProtocolNotification("Unhandled protocol message", headers.Code, headers.MessageText));
                     }
                 }
                 else

src/NATS.Client.JetStream/Internal/NatsJSExtensionsInternal.cs

@@ -21,6 +21,7 @@ public static bool HasTerminalJSError(this NatsHeaders headers)
             if (headers.MessageText.StartsWith("Exceeded MaxRequestBatch", StringComparison.OrdinalIgnoreCase)
                 || headers.MessageText.StartsWith("Exceeded MaxRequestExpires", StringComparison.OrdinalIgnoreCase)
                 || headers.MessageText.StartsWith("Exceeded MaxRequestMaxBytes", StringComparison.OrdinalIgnoreCase)
+                || headers.MessageText.StartsWith("Consumer is push based", StringComparison.OrdinalIgnoreCase)
                 || headers.MessageText.StartsWith("Exceeded MaxWaiting", StringComparison.OrdinalIgnoreCase))
                 return true;
         }

src/NATS.Client.JetStream/Internal/NatsJSFetch.cs

@@ -94,7 +94,7 @@ public NatsJSFetch(
             static state =>
             {
                 var self = (NatsJSFetch<TMsg>)state!;
-                self._notificationChannel?.Notify(new NatsJSTimeoutNotification());
+                self._notificationChannel?.Notify(NatsJSTimeoutNotification.Default);
 
                 self.EndSubscription(NatsSubEndReason.IdleHeartbeatTimeout);
                 if (self._debug)

src/NATS.Client.JetStream/NatsJSLogEvents.cs

@@ -23,4 +23,6 @@ public static class NatsJSLogEvents
     public static readonly EventId RecreateConsumer = new(2017, nameof(RecreateConsumer));
     public static readonly EventId Internal = new(2018, nameof(Internal));
     public static readonly EventId Retry = new(2019, nameof(Retry));
+    public static readonly EventId NoResponders = new(2020, nameof(NoResponders));
+    public static readonly EventId MessageSizeExceedsMaxBytes = new(2021, nameof(MessageSizeExceedsMaxBytes));
 }

src/NATS.Client.JetStream/NatsJSOpts.cs

@@ -115,6 +115,31 @@ public record NatsJSConsumeOpts
     /// </summary>
     public int? ThresholdBytes { get; init; }
 
+    /// <summary>
+    /// A handler function invoked for notifications related to consumption, allowing custom handling of
+    /// operational updates or state changes during the consumer's lifecycle.
+    /// </summary>
+    /// <remarks>
+    /// Possible notifications include:
+    /// <para>
+    /// <see cref="NatsJSProtocolNotification"/> is sent when the server sends a header not handled by the client.
+    /// </para>
+    /// <para>
+    /// <see cref="NatsJSMessageSizeExceedsMaxBytesNotification"/> is sent when a message exceeds the maximum bytes limit.
+    /// </para>
+    /// <para>
+    /// <see cref="NatsJSLeadershipChangeNotification"/> is sent when the JetStream leadership changes.
+    /// </para>
+    /// <para>
+    /// <see cref="NatsJSNoRespondersNotification"/> is sent when there are no responders for a request.
+    /// This is a generic NATS server 503 error which may be caused by various reasons, however, in most cases
+    /// it will indicate that the cluster's state is changing and a server is restarting.
+    /// </para>
+    /// <para>
+    /// <see cref="NatsJSTimeoutNotification"/> is sent when the client times out waiting for a response.
+    /// It is triggered when the server does not respond within twice the idle heartbeat duration.
+    /// </para>
+    /// </remarks>
     public Func<INatsJSNotification, CancellationToken, Task>? NotificationHandler { get; init; }
 
     /// <summary>

src/NATS.Client.JetStream/NatsJSOrderedConsumer.cs

@@ -110,7 +110,7 @@ public async IAsyncEnumerable<NatsJSMsg<T>> ConsumeAsync<T>(
                         }
                         catch (NatsJSTimeoutException e)
                         {
-                            notificationHandler?.Invoke(new NatsJSTimeoutNotification(), cancellationToken);
+                            notificationHandler?.Invoke(NatsJSTimeoutNotification.Default, cancellationToken);
                             _logger.LogWarning(NatsJSLogEvents.Retry, "{Error}. Retrying...", e.Message);
                             goto CONSUME_LOOP;
                         }

tests/NATS.Client.Core2.Tests/SendBufferTest.cs

@@ -100,13 +100,13 @@ void Log(string m)
         await using var server = new MockServer(
             handler: (client, cmd) =>
             {
-                if (cmd.Name == "PUB")
+                if (cmd.Name == "(PRE)PUB")
                 {
                     lock (pubs)
                         pubs.Add($"PUB {cmd.Subject}");
                 }
 
-                if (cmd is { Name: "PUB", Subject: "close" })
+                if (cmd is { Name: "(PRE)PUB", Subject: "close" })
                 {
                     client.Close();
                 }

tests/NATS.Client.JetStream.Tests/ConsumeResponseTest.cs

@@ -0,0 +1,71 @@
+using System.Text;
+using NATS.Client.TestUtilities;
+using NATS.Net;
+
+namespace NATS.Client.JetStream.Tests;
+
+public class ConsumeResponseTest
+{
+    [Fact]
+    public async Task Consume_response()
+    {
+        var headers = new Stack<string>();
+        headers.Push("NATS/1.0 400 Bad Test Request");
+        headers.Push("NATS/1.0 999 Unknown Error");
+        headers.Push("NATS/1.0 503");
+        headers.Push("NATS/1.0 409 Message Size Exceeds MaxBytes");
+        headers.Push("NATS/1.0 409 Leadership Change");
+
+        await using var ms = new MockServer((_, cmd) =>
+        {
+            if (cmd.Name == "PUB" && cmd.Subject.Contains("CONSUMER.INFO"))
+            {
+                cmd.Reply(payload: """{"stream_name":"x","name":"x"}""");
+            }
+            else if (cmd.Name == "PUB" && cmd.Subject.Contains("CONSUMER.MSG.NEXT"))
+            {
+                if (headers.Peek() != null)
+                {
+                    var header = headers.Pop();
+                    cmd.Reply(headers: header);
+                }
+            }
+
+            return Task.CompletedTask;
+        });
+
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+        await using var nats = new NatsConnection(new NatsOpts { Url = ms.Url });
+        var js = nats.CreateJetStreamContext();
+        var consumer = await js.GetConsumerAsync("x", "x", cts.Token);
+
+        var notifications = new List<INatsJSNotification>();
+        var exception = await Assert.ThrowsAsync<NatsJSProtocolException>(async () =>
+        {
+            await foreach (var msg in consumer.ConsumeAsync<string>(
+                               opts: new NatsJSConsumeOpts
+                               {
+                                   MaxMsgs = 1000,
+                                   IdleHeartbeat = TimeSpan.FromSeconds(1),
+                                   NotificationHandler = (n, ct) =>
+                                   {
+                                       notifications.Add(n);
+                                       return Task.CompletedTask;
+                                   },
+                               },
+                               cancellationToken: cts.Token))
+            {
+            }
+        });
+
+        Assert.Equal(400, exception.HeaderCode);
+        Assert.Equal("Bad Test Request", exception.HeaderMessageText);
+
+        var types = notifications.Select(n => n.GetType()).ToList();
+        Assert.Contains(typeof(NatsJSTimeoutNotification), types);
+        Assert.Contains(typeof(NatsJSNoRespondersNotification), types);
+        Assert.Contains(typeof(NatsJSLeadershipChangeNotification), types);
+        Assert.Contains(typeof(NatsJSMessageSizeExceedsMaxBytesNotification), types);
+        Assert.Contains(typeof(NatsJSProtocolNotification), types);
+    }
+}

tests/NATS.Client.JetStream.Tests/DirectGetTest.cs

@@ -0,0 +1,60 @@
+using NATS.Client.Core2.Tests;
+using NATS.Client.JetStream.Models;
+using NATS.Client.TestUtilities;
+
+namespace NATS.Client.JetStream.Tests;
+
+[Collection("nats-server")]
+public class DirectGetTest
+{
+    private readonly ITestOutputHelper _output;
+    private readonly NatsServerFixture _server;
+
+    public DirectGetTest(ITestOutputHelper output, NatsServerFixture server)
+    {
+        _output = output;
+        _server = server;
+    }
+
+    [SkipIfNatsServer(versionEarlierThan: "2.10.28")]
+    public async Task Direct_get_returns_503_no_responders()
+    {
+        // https://github.com/nats-io/nats-server/commit/ce309b79d99552996e18dce47dc04bdc730c0d84
+        // When we fail to deliver a message through a service import respond with no responders.
+        await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
+        var prefix = _server.GetNextId();
+        var js = new NatsJSContext(nats);
+
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+
+        var s1 = await js.CreateStreamAsync(
+            new StreamConfig($"{prefix}S1", [$"{prefix}s1"]) { AllowDirect = false },
+            cancellationToken: cts.Token);
+
+        await js.PublishAsync($"{prefix}s1", "x", cancellationToken: cts.Token);
+
+        await Assert.ThrowsAsync<NatsNoRespondersException>(async () => await s1.GetDirectAsync<string>(
+            new StreamMsgGetRequest { Seq = 1 },
+            cancellationToken: cts.Token));
+    }
+
+    [SkipIfNatsServer(versionLaterThan: "2.10.28")]
+    public async Task Direct_get_returns_no_reply()
+    {
+        await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
+        var prefix = _server.GetNextId();
+        var js = new NatsJSContext(nats);
+
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+
+        var s1 = await js.CreateStreamAsync(
+            new StreamConfig($"{prefix}S1", [$"{prefix}s1"]) { AllowDirect = false },
+            cancellationToken: cts.Token);
+
+        await js.PublishAsync($"{prefix}s1", "x", cancellationToken: cts.Token);
+
+        await Assert.ThrowsAsync<NatsNoReplyException>(async () => await s1.GetDirectAsync<string>(
+            new StreamMsgGetRequest { Seq = 1 },
+            cancellationToken: cts.Token));
+    }
+}