Add direct reply subject checks

src/NATS.Client.Core/Internal/ReplyTask.cs

@@ -124,11 +124,14 @@ public ReplyTask<TReply> CreateReplyTask<TReply>(INatsDeserialize<TReply>? deser
 
     public void Return(long id) => _replies.TryRemove(id, out _);
 
-    public void SetResult(long id, string? replyTo, in ReadOnlySequence<byte> payloadBuffer, in ReadOnlySequence<byte>? headersBuffer)
+    public bool TrySetResult(long id, string? replyTo, in ReadOnlySequence<byte> payloadBuffer, in ReadOnlySequence<byte>? headersBuffer)
     {
         if (_replies.TryGetValue(id, out var rt))
         {
             rt.SetResult(replyTo, payloadBuffer, headersBuffer);
+            return true;
         }
+
+        return false;
     }
 }

src/NATS.Client.Core/NatsConnection.cs

@@ -297,8 +297,13 @@ internal ValueTask PublishToClientHandlersAsync(string subject, string? replyTo,
 
                 if (long.TryParse(idString, out var id))
                 {
-                    _replyTaskFactory.SetResult(id, replyTo, payloadBuffer, headersBuffer);
-                    return default;
+                    if (_replyTaskFactory.TrySetResult(id, replyTo, payloadBuffer, headersBuffer))
+                    {
+                        return default;
+                    }
+
+                    // if we can't set the result, either the task is already timed out or
+                    // it's not a reply to a request.
                 }
 
                 // if we can't parse the id, it's not a reply.