Exclude NATS status frames from consumed metrics

src/NATS.Client.Core/NatsSubBase.cs

@@ -374,7 +374,12 @@ public virtual async ValueTask ReceiveAsync(string subject, string? replyTo, Rea
             // the hand-off to the subscription channel. received.bytes follows the same
             // convention. Messages dropped by cancellation, channel closure, or serializer
             // exceptions are not counted.
-            if (Telemetry.ConsumedMessages.Enabled || Telemetry.ReceivedBytes.Enabled)
+            //
+            // NATS status/control frames (no-responder 503s and JetStream heartbeats,
+            // flow-control, and protocol notifications) are consumed internally and never
+            // reach the application, so they are excluded from both counters. They are
+            // identifiable by the inline status code on the "NATS/1.0 <code>" header line.
+            if ((Telemetry.ConsumedMessages.Enabled || Telemetry.ReceivedBytes.Enabled) && !IsStatusMsg(headersBuffer))
             {
                 var tags = Telemetry.BuildMetricTags(Connection, Telemetry.Constants.OpRec);
                 if (Telemetry.ConsumedMessages.Enabled)
@@ -425,6 +430,21 @@ internal static bool IsHeader503(ReadOnlySequence<byte>? headersBuffer) =>
         headersBuffer is { Length: >= 12 }
         && headersBuffer.Value.Slice(8, 4).ToSpan().SequenceEqual(NoRespondersHeaderSequence);
 
+    // A NATS status/control frame carries an inline status code on the version line
+    // ("NATS/1.0 <code>"): a space then three ASCII digits at offset 8. Regular user
+    // headers have "\r\n" at that offset, so this never matches an application message.
+    internal static bool IsStatusMsg(ReadOnlySequence<byte>? headersBuffer)
+    {
+        if (headersBuffer is not { Length: >= 12 })
+            return false;
+
+        var code = headersBuffer.Value.Slice(8, 4).ToSpan();
+        return code[0] == (byte)' '
+            && code[1] is >= (byte)'0' and <= (byte)'9'
+            && code[2] is >= (byte)'0' and <= (byte)'9'
+            && code[3] is >= (byte)'0' and <= (byte)'9';
+    }
+
     internal void ClearException() => Interlocked.Exchange(ref _exception, null);
 
     /// <summary>

tests/NATS.Client.CoreUnit.Tests/StatusMsgTest.cs

@@ -0,0 +1,35 @@
+using System.Buffers;
+using System.Text;
+
+namespace NATS.Client.Core.Tests;
+
+public class StatusMsgTest
+{
+    [Theory]
+    [InlineData("NATS/1.0 100 Idle Heartbeat\r\n\r\n")] // JetStream idle heartbeat
+    [InlineData("NATS/1.0 503\r\n\r\n")] // no responders
+    [InlineData("NATS/1.0 408 Request Timeout\r\n\r\n")] // pull request timeout
+    [InlineData("NATS/1.0 409 Message Size Exceeds MaxBytes\r\n\r\n")] // flow control
+    [InlineData("NATS/1.0 100 FlowControl Request\r\n\r\n")]
+    public void Status_frames_are_detected(string header)
+    {
+        NatsSubBase.IsStatusMsg(Seq(header)).Should().BeTrue();
+    }
+
+    [Theory]
+    [InlineData("NATS/1.0\r\nFoo: Bar\r\n\r\n")] // regular user headers
+    [InlineData("NATS/1.0\r\n\r\n")] // headers present, no user keys, no status code
+    public void User_headers_are_not_status_frames(string header)
+    {
+        NatsSubBase.IsStatusMsg(Seq(header)).Should().BeFalse();
+    }
+
+    [Fact]
+    public void Null_or_short_buffer_is_not_a_status_frame()
+    {
+        NatsSubBase.IsStatusMsg(null).Should().BeFalse();
+        NatsSubBase.IsStatusMsg(Seq("NATS/1.0")).Should().BeFalse();
+    }
+
+    private static ReadOnlySequence<byte> Seq(string s) => new(Encoding.ASCII.GetBytes(s));
+}

tests/NATS.Net.OpenTelemetry.Tests/OpenTelemetryTest.cs

@@ -442,6 +442,48 @@ public async Task Direct_request_reply_receive_activity_is_disposed()
         await reg;
     }
 
+    [Fact]
+    public async Task Consumed_counter_excludes_jetstream_control_messages()
+    {
+        using var meter = new MeterTracker();
+        await using var server = await NatsServerProcess.StartAsync();
+        await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+
+        var js = new NatsJSContext(nats);
+
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+
+        await js.CreateStreamAsync(new StreamConfig { Name = "ctrl-stream", Subjects = ["ctrl.>"] }, cts.Token);
+        await js.CreateOrUpdateConsumerAsync("ctrl-stream", new ConsumerConfig("ctrl-consumer"), cts.Token);
+
+        for (var i = 0; i < 3; i++)
+            await js.PublishAsync("ctrl.subject", i, cancellationToken: cts.Token);
+
+        var consumer = await js.GetConsumerAsync("ctrl-stream", "ctrl-consumer", cts.Token);
+
+        // consumed.messages is connection-wide and already includes the PubAck and JS API
+        // replies from the setup above, so measure the delta across the fetch only.
+        long Consumed() => meter.LongMeasurements
+            .Where(m => m.Name == "messaging.client.consumed.messages")
+            .Sum(m => m.Value);
+
+        var before = Consumed();
+
+        // MaxMsgs greater than the 3 available, with a short expiry: the server delivers the
+        // 3 data messages on the fetch subscription and then a status frame to end the batch.
+        // Only the 3 data messages were delivered to the application, so the delta must be 3.
+        var received = 0;
+        await foreach (var msg in consumer.FetchAsync<int>(
+            new NatsJSFetchOpts { MaxMsgs = 10, Expires = TimeSpan.FromSeconds(2) },
+            cancellationToken: cts.Token))
+        {
+            received++;
+        }
+
+        received.Should().Be(3);
+        (Consumed() - before).Should().Be(3);
+    }
+
     private void AssertActivityData(string subject, IReadOnlyList<Activity> activityList, string expectedHost, string expectedClientId)
     {
         var activities = activityList.ToArray();

tools/site_src/documentation/advanced/opentelemetry.md

@@ -108,3 +108,9 @@ All instruments carry these tags:
 | `network.transport` | `tcp` | Transport protocol |
 
 `messaging.client.operation.duration` adds `error.type` (full exception type name) when the operation fails.
+
+`messaging.client.consumed.messages` and `nats.client.received.bytes` count only messages delivered to the
+application, per the OTel definition ("messages delivered to the application"). NATS status and control
+frames consumed internally by the client are excluded: no-responder `503` replies and JetStream heartbeats,
+flow-control, and protocol notifications. The two counters stay consistent, so `received.bytes / consumed.messages`
+reflects average delivered message size.