From ca8506f15c535fe80d0891508a895592a8a7c680 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Mon, 15 Dec 2025 04:25:40 +0000 Subject: [PATCH 01/13] Add validation for subjects --- sandbox/MicroBenchmark/MicroBenchmark.csproj | 4 +- sandbox/MicroBenchmark/RequestReplyBench.cs | 2 + .../MicroBenchmark/SubjectValidationBench.cs | 84 +++++++++++++ sandbox/MicroBenchmark/Subscribe.cs | 2 + .../Commands/ProtocolWriter.cs | 30 +++++ .../ProtocolWriterTests.cs | 119 ++++++++++++++++++ 6 files changed, 240 insertions(+), 1 deletion(-) create mode 100644 sandbox/MicroBenchmark/SubjectValidationBench.cs create mode 100644 tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs diff --git a/sandbox/MicroBenchmark/MicroBenchmark.csproj b/sandbox/MicroBenchmark/MicroBenchmark.csproj index 63a2702bd..97246f8d4 100644 --- a/sandbox/MicroBenchmark/MicroBenchmark.csproj +++ b/sandbox/MicroBenchmark/MicroBenchmark.csproj @@ -2,11 +2,13 @@ Exe - net8.0 + net8.0 + $(TargetFrameworks);net481 enable enable false true + 12 diff --git a/sandbox/MicroBenchmark/RequestReplyBench.cs b/sandbox/MicroBenchmark/RequestReplyBench.cs index 4c7ed5973..c1e7839b2 100644 --- a/sandbox/MicroBenchmark/RequestReplyBench.cs +++ b/sandbox/MicroBenchmark/RequestReplyBench.cs @@ -1,3 +1,4 @@ +#if !NET481 using BenchmarkDotNet.Attributes; using NATS.Client.Core; @@ -43,3 +44,4 @@ private static async Task GetResultAsync(NatsConnection nats) return result; } } +#endif diff --git a/sandbox/MicroBenchmark/SubjectValidationBench.cs b/sandbox/MicroBenchmark/SubjectValidationBench.cs new file mode 100644 index 000000000..1a26b01ec --- /dev/null +++ b/sandbox/MicroBenchmark/SubjectValidationBench.cs @@ -0,0 +1,84 @@ +using System.Text; +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Jobs; +using NATS.Client.Core; +using NATS.Client.Core.Commands; + +namespace MicroBenchmark; + +[MemoryDiagnoser] +[SimpleJob(RuntimeMoniker.Net481)] +[SimpleJob(RuntimeMoniker.Net80)] +public class SubjectValidationBench +{ + private const int MessageCount = 1024; + private const string Subject = "foo.bar.baz"; + private const string ReplyTo = "reply.to.subject"; + private static readonly string Data = new('0', 128); + + private ProtocolWriter _protocolWriter = null!; + private NatsBufferWriter _bufferWriter = null!; + private ReadOnlyMemory _payload; + private NatsConnection _nats = null!; + + [GlobalSetup] + public async Task Setup() + { + _protocolWriter = new ProtocolWriter(Encoding.UTF8); + _bufferWriter = new NatsBufferWriter(1024 * 1024); + _payload = new byte[128]; + _nats = new NatsConnection(); + await _nats.ConnectAsync(); + } + + [GlobalCleanup] + public async Task Cleanup() + { + _bufferWriter.Dispose(); + await _nats.DisposeAsync(); + } + + [Benchmark] + public void WritePublish_NoReplyTo() + { + for (var i = 0; i < MessageCount; i++) + { + _protocolWriter.WritePublish(_bufferWriter, Subject, replyTo: null, headers: null, _payload); + } + + _bufferWriter.Clear(); + } + + [Benchmark] + public void WritePublish_WithReplyTo() + { + for (var i = 0; i < MessageCount; i++) + { + _protocolWriter.WritePublish(_bufferWriter, Subject, ReplyTo, headers: null, _payload); + } + + _bufferWriter.Clear(); + } + + [Benchmark] + public async Task PublishAsync_NoReplyTo() + { + for (var i = 0; i < MessageCount; i++) + { + await _nats.PublishAsync(Subject, Data); + } + + await _nats.PingAsync(); + } + + [Benchmark] + public async Task PublishAsync_WithReplyTo() + { + for (var i = 0; i < MessageCount; i++) + { + await _nats.PublishAsync(Subject, Data, replyTo: ReplyTo); + } + + await _nats.PingAsync(); + } +} diff --git a/sandbox/MicroBenchmark/Subscribe.cs b/sandbox/MicroBenchmark/Subscribe.cs index 7b1eed671..f1d951f5a 100644 --- a/sandbox/MicroBenchmark/Subscribe.cs +++ b/sandbox/MicroBenchmark/Subscribe.cs @@ -87,6 +87,7 @@ public async Task CoreRead() } } +#if !NET481 [Benchmark] public async Task CoreReadAll() { @@ -102,6 +103,7 @@ public async Task CoreReadAll() } } } +#endif // limit pub to the same rate across benchmarks // pub in batches so that groups of messages are available diff --git a/src/NATS.Client.Core/Commands/ProtocolWriter.cs b/src/NATS.Client.Core/Commands/ProtocolWriter.cs index d85b368c2..88f3e6fdc 100644 --- a/src/NATS.Client.Core/Commands/ProtocolWriter.cs +++ b/src/NATS.Client.Core/Commands/ProtocolWriter.cs @@ -36,6 +36,12 @@ internal sealed class ProtocolWriter private static readonly ulong PongNewLine = BinaryPrimitives.ReadUInt64LittleEndian("PONG\r\n "u8); private static readonly ulong UnsubSpace = BinaryPrimitives.ReadUInt64LittleEndian("UNSUB "u8); +#if NET8_0_OR_GREATER + private static readonly SearchValues WhitespaceChars = SearchValues.Create([' ', '\r', '\n', '\t']); +#else + private static readonly char[] WhitespaceChars = [' ', '\r', '\n', '\t']; +#endif + private readonly Encoding _subjectEncoding; public ProtocolWriter(Encoding subjectEncoding) => _subjectEncoding = subjectEncoding; @@ -79,6 +85,12 @@ public void WritePong(IBufferWriter writer) // HPUB [reply-to] <#header bytes> <#total bytes>\r\n[headers]\r\n\r\n[payload]\r\n public void WritePublish(IBufferWriter writer, string subject, string? replyTo, ReadOnlyMemory? headers, ReadOnlyMemory payload) { + CheckSubjectForWhitespace(subject); + if (replyTo != null) + { + CheckSubjectForWhitespace(replyTo); + } + if (headers == null) { WritePub(writer, subject, replyTo, payload); @@ -93,6 +105,12 @@ public void WritePublish(IBufferWriter writer, string subject, string? rep // SUB [queue group] public void WriteSubscribe(IBufferWriter writer, int sid, string subject, string? queueGroup, int? maxMsgs) { + CheckSubjectForWhitespace(subject); + if (queueGroup != null) + { + CheckSubjectForWhitespace(queueGroup); + } + // 'SUB ' + subject +' '+ sid +'\r\n' var ctrlLen = SubSpaceLength + _subjectEncoding.GetByteCount(subject) + 1 + MaxIntStringLength + NewLineLength; @@ -188,6 +206,18 @@ public void WriteUnsubscribe(IBufferWriter writer, int sid, int? maxMessag [MethodImpl(MethodImplOptions.NoInlining)] private static void ThrowOnUtf8FormatFail() => throw new NatsException("Can not format integer."); + [MethodImpl(MethodImplOptions.NoInlining)] + private static void ThrowOnSubjectFormatFail() => throw new NatsException("Subject cannot be empty or contain whitespace."); + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static void CheckSubjectForWhitespace(ReadOnlySpan subject) + { + if (subject.IsEmpty || subject.IndexOfAny(WhitespaceChars) >= 0) + { + ThrowOnSubjectFormatFail(); + } + } + // PUB [reply-to] <#bytes>\r\n[payload]\r\n private void WritePub(IBufferWriter writer, string subject, string? replyTo, ReadOnlyMemory payload) { diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs new file mode 100644 index 000000000..f05231507 --- /dev/null +++ b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs @@ -0,0 +1,119 @@ +using System.Text; + +// ReSharper disable AccessToDisposedClosure +namespace NATS.Client.CoreUnit.Tests; + +public class ProtocolWriterTests +{ + private readonly ProtocolWriter _protocolWriter = new(Encoding.UTF8); + + [Fact] + public void WritePublish_ValidSubject_DoesNotThrow() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, "foo.bar", null, null, ReadOnlyMemory.Empty); + action.Should().NotThrow(); + } + + [Fact] + public void WritePublish_ValidSubjectWithReplyTo_DoesNotThrow() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, "foo.bar", "reply.to", null, ReadOnlyMemory.Empty); + action.Should().NotThrow(); + } + + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + [InlineData("foo\rbar")] + [InlineData("foo\nbar")] + public void WritePublish_SubjectWithWhitespace_Throws(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + } + + [Theory] + [InlineData("reply to")] + [InlineData("reply\tto")] + [InlineData("reply\rto")] + [InlineData("reply\nto")] + public void WritePublish_ReplyToWithWhitespace_Throws(string replyTo) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, "foo.bar", replyTo, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + } + + [Fact] + public void WriteSubscribe_ValidSubject_DoesNotThrow() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", null, null); + action.Should().NotThrow(); + } + + [Fact] + public void WriteSubscribe_ValidSubjectWithQueueGroup_DoesNotThrow() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", "queue-group", null); + action.Should().NotThrow(); + } + + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + [InlineData("foo\rbar")] + [InlineData("foo\nbar")] + public void WriteSubscribe_SubjectWithWhitespace_Throws(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, subject, null, null); + action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + } + + [Theory] + [InlineData("queue group")] + [InlineData("queue\tgroup")] + [InlineData("queue\rgroup")] + [InlineData("queue\ngroup")] + public void WriteSubscribe_QueueGroupWithWhitespace_Throws(string queueGroup) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", queueGroup, null); + action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + } + + // Test for whitespace at the START of strings (index 0) + [Theory] + [InlineData(" foo")] + [InlineData("\tfoo")] + [InlineData("\rfoo")] + [InlineData("\nfoo")] + public void WritePublish_SubjectWithLeadingWhitespace_Throws(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + } + + // Test for empty strings + [Fact] + public void WritePublish_EmptySubject_Throws() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, string.Empty, null, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + } + + [Fact] + public void WriteSubscribe_EmptySubject_Throws() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, string.Empty, null, null); + action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + } +} From 8d4a31592f1f45cbb94216a8b43d35d7d1224d20 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Mon, 15 Dec 2025 07:27:57 +0000 Subject: [PATCH 02/13] Add protocol-breaking whitespace validation for subject/replyTo/queueGroup --- src/NATS.Client.Core/Commands/ProtocolWriter.cs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/NATS.Client.Core/Commands/ProtocolWriter.cs b/src/NATS.Client.Core/Commands/ProtocolWriter.cs index 88f3e6fdc..5a35e0cad 100644 --- a/src/NATS.Client.Core/Commands/ProtocolWriter.cs +++ b/src/NATS.Client.Core/Commands/ProtocolWriter.cs @@ -36,6 +36,10 @@ internal sealed class ProtocolWriter private static readonly ulong PongNewLine = BinaryPrimitives.ReadUInt64LittleEndian("PONG\r\n "u8); private static readonly ulong UnsubSpace = BinaryPrimitives.ReadUInt64LittleEndian("UNSUB "u8); + // Used for subject/replyTo/queueGroup validation to prevent protocol-breaking whitespace. + // Static field ensures zero allocations per call. SearchValues (NET8+) uses SIMD vectorization; + // char[] (older TFMs) uses optimized IndexOfAny for <=5 chars. Adds ~5% overhead to full publish + // operations with zero GC pressure - acceptable trade-off for input safety. #if NET8_0_OR_GREATER private static readonly SearchValues WhitespaceChars = SearchValues.Create([' ', '\r', '\n', '\t']); #else From 17c0a694a4833b86ecc65a8fc41e5ee285a225fd Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Mon, 15 Dec 2025 10:51:47 +0000 Subject: [PATCH 03/13] Enhance subject and queue group validation to match Go client standards and add comprehensive test coverage --- .../Commands/ProtocolWriter.cs | 36 ++++++-- .../ProtocolWriterTests.cs | 84 ++++++++++++++----- 2 files changed, 91 insertions(+), 29 deletions(-) diff --git a/src/NATS.Client.Core/Commands/ProtocolWriter.cs b/src/NATS.Client.Core/Commands/ProtocolWriter.cs index 5a35e0cad..49e1fb7c4 100644 --- a/src/NATS.Client.Core/Commands/ProtocolWriter.cs +++ b/src/NATS.Client.Core/Commands/ProtocolWriter.cs @@ -89,10 +89,10 @@ public void WritePong(IBufferWriter writer) // HPUB [reply-to] <#header bytes> <#total bytes>\r\n[headers]\r\n\r\n[payload]\r\n public void WritePublish(IBufferWriter writer, string subject, string? replyTo, ReadOnlyMemory? headers, ReadOnlyMemory payload) { - CheckSubjectForWhitespace(subject); + ValidateSubject(subject); if (replyTo != null) { - CheckSubjectForWhitespace(replyTo); + ValidateSubject(replyTo); } if (headers == null) @@ -109,10 +109,10 @@ public void WritePublish(IBufferWriter writer, string subject, string? rep // SUB [queue group] public void WriteSubscribe(IBufferWriter writer, int sid, string subject, string? queueGroup, int? maxMsgs) { - CheckSubjectForWhitespace(subject); + ValidateSubject(subject); if (queueGroup != null) { - CheckSubjectForWhitespace(queueGroup); + ValidateQueueGroup(queueGroup); } // 'SUB ' + subject +' '+ sid +'\r\n' @@ -211,14 +211,34 @@ public void WriteUnsubscribe(IBufferWriter writer, int sid, int? maxMessag private static void ThrowOnUtf8FormatFail() => throw new NatsException("Can not format integer."); [MethodImpl(MethodImplOptions.NoInlining)] - private static void ThrowOnSubjectFormatFail() => throw new NatsException("Subject cannot be empty or contain whitespace."); + private static void ThrowOnBadSubject() => throw new NatsException("Subject is invalid."); + [MethodImpl(MethodImplOptions.NoInlining)] + private static void ThrowOnBadQueueGroup() => throw new NatsException("Queue group is invalid."); + + // Validates subjects: no whitespace, no empty tokens (.foo, foo., foo..bar) + // Matches Go client's badSubject() validation + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static void ValidateSubject(ReadOnlySpan subject) + { + if (subject.IsEmpty || + subject.IndexOfAny(WhitespaceChars) >= 0 || + subject[0] == '.' || + subject[^1] == '.' || + subject.IndexOf("..") >= 0) + { + ThrowOnBadSubject(); + } + } + + // Validates queue groups: no whitespace only (dots are allowed) + // Matches Go client's badQueue() validation [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static void CheckSubjectForWhitespace(ReadOnlySpan subject) + private static void ValidateQueueGroup(ReadOnlySpan queueGroup) { - if (subject.IsEmpty || subject.IndexOfAny(WhitespaceChars) >= 0) + if (queueGroup.IsEmpty || queueGroup.IndexOfAny(WhitespaceChars) >= 0) { - ThrowOnSubjectFormatFail(); + ThrowOnBadQueueGroup(); } } diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs index f05231507..18f240194 100644 --- a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs @@ -23,16 +23,21 @@ public void WritePublish_ValidSubjectWithReplyTo_DoesNotThrow() action.Should().NotThrow(); } + // Whitespace validation for subjects [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] [InlineData("foo\rbar")] [InlineData("foo\nbar")] + [InlineData(" foo")] + [InlineData("\tfoo")] + [InlineData("\rfoo")] + [InlineData("\nfoo")] public void WritePublish_SubjectWithWhitespace_Throws(string subject) { using var writer = new NatsBufferWriter(); var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); - action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + action.Should().Throw().WithMessage("Subject is invalid."); } [Theory] @@ -44,9 +49,36 @@ public void WritePublish_ReplyToWithWhitespace_Throws(string replyTo) { using var writer = new NatsBufferWriter(); var action = () => _protocolWriter.WritePublish(writer, "foo.bar", replyTo, null, ReadOnlyMemory.Empty); - action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + // Empty subject validation + [Fact] + public void WritePublish_EmptySubject_Throws() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, string.Empty, null, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + // Dot token validation (matches Go client's badSubject) + [Theory] + [InlineData(".foo")] + [InlineData("foo.")] + [InlineData(".")] + [InlineData("..")] + [InlineData("foo..bar")] + [InlineData("foo...bar")] + [InlineData(".foo.bar")] + [InlineData("foo.bar.")] + public void WritePublish_SubjectWithInvalidDots_Throws(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject is invalid."); } + // Subscribe tests [Fact] public void WriteSubscribe_ValidSubject_DoesNotThrow() { @@ -63,6 +95,19 @@ public void WriteSubscribe_ValidSubjectWithQueueGroup_DoesNotThrow() action.Should().NotThrow(); } + // Queue group can have dots (unlike subjects, no token validation) + [Theory] + [InlineData("queue.group")] + [InlineData(".queue")] + [InlineData("queue.")] + [InlineData("queue..group")] + public void WriteSubscribe_QueueGroupWithDots_DoesNotThrow(string queueGroup) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", queueGroup, null); + action.Should().NotThrow(); + } + [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] @@ -72,7 +117,7 @@ public void WriteSubscribe_SubjectWithWhitespace_Throws(string subject) { using var writer = new NatsBufferWriter(); var action = () => _protocolWriter.WriteSubscribe(writer, 1, subject, null, null); - action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + action.Should().Throw().WithMessage("Subject is invalid."); } [Theory] @@ -84,36 +129,33 @@ public void WriteSubscribe_QueueGroupWithWhitespace_Throws(string queueGroup) { using var writer = new NatsBufferWriter(); var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", queueGroup, null); - action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + action.Should().Throw().WithMessage("Queue group is invalid."); } - // Test for whitespace at the START of strings (index 0) - [Theory] - [InlineData(" foo")] - [InlineData("\tfoo")] - [InlineData("\rfoo")] - [InlineData("\nfoo")] - public void WritePublish_SubjectWithLeadingWhitespace_Throws(string subject) + [Fact] + public void WriteSubscribe_EmptySubject_Throws() { using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); - action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, string.Empty, null, null); + action.Should().Throw().WithMessage("Subject is invalid."); } - // Test for empty strings [Fact] - public void WritePublish_EmptySubject_Throws() + public void WriteSubscribe_EmptyQueueGroup_Throws() { using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, string.Empty, null, null, ReadOnlyMemory.Empty); - action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", string.Empty, null); + action.Should().Throw().WithMessage("Queue group is invalid."); } - [Fact] - public void WriteSubscribe_EmptySubject_Throws() + [Theory] + [InlineData(".foo")] + [InlineData("foo.")] + [InlineData("foo..bar")] + public void WriteSubscribe_SubjectWithInvalidDots_Throws(string subject) { using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, string.Empty, null, null); - action.Should().Throw().WithMessage("Subject cannot be empty or contain whitespace."); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, subject, null, null); + action.Should().Throw().WithMessage("Subject is invalid."); } } From 45023938abc5a5c8b6bb2528bdf66030fafe3657 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Mon, 15 Dec 2025 11:28:09 +0000 Subject: [PATCH 04/13] Add skipSubjectValidation option to improve performance and update tests accordingly --- .../MicroBenchmark/SubjectValidationBench.cs | 54 +++++++++---------- .../Commands/CommandWriter.cs | 2 +- .../Commands/ProtocolWriter.cs | 25 ++++++--- src/NATS.Client.Core/NatsOpts.cs | 16 ++++++ .../ProtocolWriterTests.cs | 38 +++++++++++++ 5 files changed, 99 insertions(+), 36 deletions(-) diff --git a/sandbox/MicroBenchmark/SubjectValidationBench.cs b/sandbox/MicroBenchmark/SubjectValidationBench.cs index 1a26b01ec..11bc44838 100644 --- a/sandbox/MicroBenchmark/SubjectValidationBench.cs +++ b/sandbox/MicroBenchmark/SubjectValidationBench.cs @@ -1,8 +1,6 @@ -using System.Text; using BenchmarkDotNet.Attributes; using BenchmarkDotNet.Jobs; using NATS.Client.Core; -using NATS.Client.Core.Commands; namespace MicroBenchmark; @@ -12,73 +10,73 @@ namespace MicroBenchmark; public class SubjectValidationBench { private const int MessageCount = 1024; - private const string Subject = "foo.bar.baz"; - private const string ReplyTo = "reply.to.subject"; + private const string ShortSubject = "foo.bar.baz"; + private const string LongSubject = "org.company.division.team.project.service.module.component.action.event.type.version.region"; + private const string ShortReplyTo = "reply.to.subject"; + private const string LongReplyTo = "org.company.division.team.project.service.module.component.reply.inbox.unique.identifier"; private static readonly string Data = new('0', 128); - private ProtocolWriter _protocolWriter = null!; - private NatsBufferWriter _bufferWriter = null!; - private ReadOnlyMemory _payload; - private NatsConnection _nats = null!; + private NatsConnection _natsValidation = null!; + private NatsConnection _natsNoValidation = null!; [GlobalSetup] public async Task Setup() { - _protocolWriter = new ProtocolWriter(Encoding.UTF8); - _bufferWriter = new NatsBufferWriter(1024 * 1024); - _payload = new byte[128]; - _nats = new NatsConnection(); - await _nats.ConnectAsync(); + _natsValidation = new NatsConnection(NatsOpts.Default with { SkipSubjectValidation = false }); + await _natsValidation.ConnectAsync(); + + _natsNoValidation = new NatsConnection(NatsOpts.Default with { SkipSubjectValidation = true }); + await _natsNoValidation.ConnectAsync(); } [GlobalCleanup] public async Task Cleanup() { - _bufferWriter.Dispose(); - await _nats.DisposeAsync(); + await _natsValidation.DisposeAsync(); + await _natsNoValidation.DisposeAsync(); } - [Benchmark] - public void WritePublish_NoReplyTo() + [Benchmark(Baseline = true)] + public async Task PublishAsync_Short_NoValidation() { for (var i = 0; i < MessageCount; i++) { - _protocolWriter.WritePublish(_bufferWriter, Subject, replyTo: null, headers: null, _payload); + await _natsNoValidation.PublishAsync(ShortSubject, Data, replyTo: ShortReplyTo); } - _bufferWriter.Clear(); + await _natsNoValidation.PingAsync(); } [Benchmark] - public void WritePublish_WithReplyTo() + public async Task PublishAsync_Short_WithValidation() { for (var i = 0; i < MessageCount; i++) { - _protocolWriter.WritePublish(_bufferWriter, Subject, ReplyTo, headers: null, _payload); + await _natsValidation.PublishAsync(ShortSubject, Data, replyTo: ShortReplyTo); } - _bufferWriter.Clear(); + await _natsValidation.PingAsync(); } [Benchmark] - public async Task PublishAsync_NoReplyTo() + public async Task PublishAsync_Long_NoValidation() { for (var i = 0; i < MessageCount; i++) { - await _nats.PublishAsync(Subject, Data); + await _natsNoValidation.PublishAsync(LongSubject, Data, replyTo: LongReplyTo); } - await _nats.PingAsync(); + await _natsNoValidation.PingAsync(); } [Benchmark] - public async Task PublishAsync_WithReplyTo() + public async Task PublishAsync_Long_WithValidation() { for (var i = 0; i < MessageCount; i++) { - await _nats.PublishAsync(Subject, Data, replyTo: ReplyTo); + await _natsValidation.PublishAsync(LongSubject, Data, replyTo: LongReplyTo); } - await _nats.PingAsync(); + await _natsValidation.PingAsync(); } } diff --git a/src/NATS.Client.Core/Commands/CommandWriter.cs b/src/NATS.Client.Core/Commands/CommandWriter.cs index ea43a4e10..9ba1768b2 100644 --- a/src/NATS.Client.Core/Commands/CommandWriter.cs +++ b/src/NATS.Client.Core/Commands/CommandWriter.cs @@ -71,7 +71,7 @@ public CommandWriter(string name, NatsConnection connection, ObjectPool pool, Na _counter = counter; _defaultCommandTimeout = overrideCommandTimeout ?? opts.CommandTimeout; _enqueuePing = enqueuePing; - _protocolWriter = new ProtocolWriter(opts.SubjectEncoding); + _protocolWriter = new ProtocolWriter(opts.SubjectEncoding, opts.SkipSubjectValidation); _channelLock = Channel.CreateBounded(1); _channelSize = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleWriter = true, SingleReader = true }); _headerWriter = new HeaderWriter(opts.HeaderEncoding); diff --git a/src/NATS.Client.Core/Commands/ProtocolWriter.cs b/src/NATS.Client.Core/Commands/ProtocolWriter.cs index 49e1fb7c4..ec56a0374 100644 --- a/src/NATS.Client.Core/Commands/ProtocolWriter.cs +++ b/src/NATS.Client.Core/Commands/ProtocolWriter.cs @@ -47,8 +47,13 @@ internal sealed class ProtocolWriter #endif private readonly Encoding _subjectEncoding; + private readonly bool _skipSubjectValidation; - public ProtocolWriter(Encoding subjectEncoding) => _subjectEncoding = subjectEncoding; + public ProtocolWriter(Encoding subjectEncoding, bool skipSubjectValidation = false) + { + _subjectEncoding = subjectEncoding; + _skipSubjectValidation = skipSubjectValidation; + } // https://docs.nats.io/reference/reference-protocols/nats-protocol#connect // CONNECT {["option_name":option_value],...} @@ -89,10 +94,13 @@ public void WritePong(IBufferWriter writer) // HPUB [reply-to] <#header bytes> <#total bytes>\r\n[headers]\r\n\r\n[payload]\r\n public void WritePublish(IBufferWriter writer, string subject, string? replyTo, ReadOnlyMemory? headers, ReadOnlyMemory payload) { - ValidateSubject(subject); - if (replyTo != null) + if (!_skipSubjectValidation) { - ValidateSubject(replyTo); + ValidateSubject(subject); + if (replyTo != null) + { + ValidateSubject(replyTo); + } } if (headers == null) @@ -109,10 +117,13 @@ public void WritePublish(IBufferWriter writer, string subject, string? rep // SUB [queue group] public void WriteSubscribe(IBufferWriter writer, int sid, string subject, string? queueGroup, int? maxMsgs) { - ValidateSubject(subject); - if (queueGroup != null) + if (!_skipSubjectValidation) { - ValidateQueueGroup(queueGroup); + ValidateSubject(subject); + if (queueGroup != null) + { + ValidateQueueGroup(queueGroup); + } } // 'SUB ' + subject +' '+ sid +'\r\n' diff --git a/src/NATS.Client.Core/NatsOpts.cs b/src/NATS.Client.Core/NatsOpts.cs index 910ed4afc..575e0883d 100644 --- a/src/NATS.Client.Core/NatsOpts.cs +++ b/src/NATS.Client.Core/NatsOpts.cs @@ -214,6 +214,22 @@ public sealed record NatsOpts /// public bool PublishTimeoutOnDisconnected { get; init; } = false; + /// + /// Gets or sets a value indicating whether to skip full subject validation. + /// The default is false, meaning subjects are fully validated for whitespace and empty tokens. + /// + /// + /// + /// When set to false (default), subjects are validated to ensure they don't contain + /// whitespace characters (space, tab, CR, LF) and don't have empty tokens (e.g., ".foo", "foo.", "foo..bar"). + /// + /// + /// When set to true, only empty subject validation is performed, which can improve + /// performance in high-throughput scenarios where subjects are known to be valid. + /// + /// + public bool SkipSubjectValidation { get; init; } = false; + internal NatsUri[] GetSeedUris(bool suppressRandomization = false) { var urls = Url.Split(','); diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs index 18f240194..5e6a1bb1b 100644 --- a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs @@ -6,6 +6,7 @@ namespace NATS.Client.CoreUnit.Tests; public class ProtocolWriterTests { private readonly ProtocolWriter _protocolWriter = new(Encoding.UTF8); + private readonly ProtocolWriter _protocolWriterNoValidation = new(Encoding.UTF8, skipSubjectValidation: true); [Fact] public void WritePublish_ValidSubject_DoesNotThrow() @@ -158,4 +159,41 @@ public void WriteSubscribe_SubjectWithInvalidDots_Throws(string subject) var action = () => _protocolWriter.WriteSubscribe(writer, 1, subject, null, null); action.Should().Throw().WithMessage("Subject is invalid."); } + + // SkipSubjectValidation tests + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + [InlineData(".foo")] + [InlineData("foo.")] + [InlineData("foo..bar")] + public void WritePublish_WithSkipValidation_DoesNotThrow(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriterNoValidation.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + action.Should().NotThrow(); + } + + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + [InlineData(".foo")] + [InlineData("foo.")] + [InlineData("foo..bar")] + public void WriteSubscribe_WithSkipValidation_DoesNotThrow(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriterNoValidation.WriteSubscribe(writer, 1, subject, null, null); + action.Should().NotThrow(); + } + + [Theory] + [InlineData("queue group")] + [InlineData("queue\tgroup")] + public void WriteSubscribe_QueueGroupWithSkipValidation_DoesNotThrow(string queueGroup) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriterNoValidation.WriteSubscribe(writer, 1, "foo.bar", queueGroup, null); + action.Should().NotThrow(); + } } From 41d8d64a11026661cdb4b0f126865f0f672958d3 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Mon, 15 Dec 2025 12:28:13 +0000 Subject: [PATCH 05/13] Improve subject validation logic with adaptive short/long path optimization and remove dot-based validation tests --- .../MicroBenchmark/SubjectValidationBench.cs | 2 +- .../Commands/ProtocolWriter.cs | 31 ++++++++++++---- .../ProtocolWriterTests.cs | 36 +------------------ 3 files changed, 26 insertions(+), 43 deletions(-) diff --git a/sandbox/MicroBenchmark/SubjectValidationBench.cs b/sandbox/MicroBenchmark/SubjectValidationBench.cs index 11bc44838..44feb1534 100644 --- a/sandbox/MicroBenchmark/SubjectValidationBench.cs +++ b/sandbox/MicroBenchmark/SubjectValidationBench.cs @@ -9,7 +9,7 @@ namespace MicroBenchmark; [SimpleJob(RuntimeMoniker.Net80)] public class SubjectValidationBench { - private const int MessageCount = 1024; + private const int MessageCount = 100_000; private const string ShortSubject = "foo.bar.baz"; private const string LongSubject = "org.company.division.team.project.service.module.component.action.event.type.version.region"; private const string ShortReplyTo = "reply.to.subject"; diff --git a/src/NATS.Client.Core/Commands/ProtocolWriter.cs b/src/NATS.Client.Core/Commands/ProtocolWriter.cs index ec56a0374..d211e8db6 100644 --- a/src/NATS.Client.Core/Commands/ProtocolWriter.cs +++ b/src/NATS.Client.Core/Commands/ProtocolWriter.cs @@ -227,19 +227,36 @@ public void WriteUnsubscribe(IBufferWriter writer, int sid, int? maxMessag [MethodImpl(MethodImplOptions.NoInlining)] private static void ThrowOnBadQueueGroup() => throw new NatsException("Queue group is invalid."); - // Validates subjects: no whitespace, no empty tokens (.foo, foo., foo..bar) - // Matches Go client's badSubject() validation + // Validates subjects: no whitespace (protocol-breaking characters) + // Uses adaptive algorithm: manual loop for short subjects (< 16 chars) with + // branch-predicted early exit, SIMD-optimized IndexOfAny for longer subjects. [MethodImpl(MethodImplOptions.AggressiveInlining)] private static void ValidateSubject(ReadOnlySpan subject) { - if (subject.IsEmpty || - subject.IndexOfAny(WhitespaceChars) >= 0 || - subject[0] == '.' || - subject[^1] == '.' || - subject.IndexOf("..") >= 0) + if (subject.IsEmpty) { ThrowOnBadSubject(); } + + if (subject.Length < 16) + { + // Fast path for short subjects - branch predictor learns most chars are > ' ' + foreach (var c in subject) + { + if (c <= ' ' && (c == ' ' || c == '\t' || c == '\r' || c == '\n')) + { + ThrowOnBadSubject(); + } + } + } + else + { + // SIMD path for long subjects + if (subject.IndexOfAny(WhitespaceChars) >= 0) + { + ThrowOnBadSubject(); + } + } } // Validates queue groups: no whitespace only (dots are allowed) diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs index 5e6a1bb1b..0106cefe0 100644 --- a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs @@ -62,23 +62,6 @@ public void WritePublish_EmptySubject_Throws() action.Should().Throw().WithMessage("Subject is invalid."); } - // Dot token validation (matches Go client's badSubject) - [Theory] - [InlineData(".foo")] - [InlineData("foo.")] - [InlineData(".")] - [InlineData("..")] - [InlineData("foo..bar")] - [InlineData("foo...bar")] - [InlineData(".foo.bar")] - [InlineData("foo.bar.")] - public void WritePublish_SubjectWithInvalidDots_Throws(string subject) - { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); - action.Should().Throw().WithMessage("Subject is invalid."); - } - // Subscribe tests [Fact] public void WriteSubscribe_ValidSubject_DoesNotThrow() @@ -96,7 +79,7 @@ public void WriteSubscribe_ValidSubjectWithQueueGroup_DoesNotThrow() action.Should().NotThrow(); } - // Queue group can have dots (unlike subjects, no token validation) + // Queue group can have dots [Theory] [InlineData("queue.group")] [InlineData(".queue")] @@ -149,24 +132,10 @@ public void WriteSubscribe_EmptyQueueGroup_Throws() action.Should().Throw().WithMessage("Queue group is invalid."); } - [Theory] - [InlineData(".foo")] - [InlineData("foo.")] - [InlineData("foo..bar")] - public void WriteSubscribe_SubjectWithInvalidDots_Throws(string subject) - { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, subject, null, null); - action.Should().Throw().WithMessage("Subject is invalid."); - } - // SkipSubjectValidation tests [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] - [InlineData(".foo")] - [InlineData("foo.")] - [InlineData("foo..bar")] public void WritePublish_WithSkipValidation_DoesNotThrow(string subject) { using var writer = new NatsBufferWriter(); @@ -177,9 +146,6 @@ public void WritePublish_WithSkipValidation_DoesNotThrow(string subject) [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] - [InlineData(".foo")] - [InlineData("foo.")] - [InlineData("foo..bar")] public void WriteSubscribe_WithSkipValidation_DoesNotThrow(string subject) { using var writer = new NatsBufferWriter(); From 9ba3b22073782075497e77b91e1799371da423c5 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Mon, 15 Dec 2025 12:52:32 +0000 Subject: [PATCH 06/13] Enhance wildcard subject validation with new test cases and refine subject validation details in comments and documentation --- .../Commands/ProtocolWriter.cs | 4 ++-- src/NATS.Client.Core/NatsOpts.cs | 10 ++++---- .../ProtocolWriterTests.cs | 23 +++++++++++++++++++ 3 files changed, 30 insertions(+), 7 deletions(-) diff --git a/src/NATS.Client.Core/Commands/ProtocolWriter.cs b/src/NATS.Client.Core/Commands/ProtocolWriter.cs index d211e8db6..fc12b955b 100644 --- a/src/NATS.Client.Core/Commands/ProtocolWriter.cs +++ b/src/NATS.Client.Core/Commands/ProtocolWriter.cs @@ -38,8 +38,8 @@ internal sealed class ProtocolWriter // Used for subject/replyTo/queueGroup validation to prevent protocol-breaking whitespace. // Static field ensures zero allocations per call. SearchValues (NET8+) uses SIMD vectorization; - // char[] (older TFMs) uses optimized IndexOfAny for <=5 chars. Adds ~5% overhead to full publish - // operations with zero GC pressure - acceptable trade-off for input safety. + // char[] (older TFMs) uses optimized IndexOfAny for <=5 chars. Adds ~3% overhead on .NET 8 + // and ~5% on .NET Framework to full publish operations, with zero GC pressure. #if NET8_0_OR_GREATER private static readonly SearchValues WhitespaceChars = SearchValues.Create([' ', '\r', '\n', '\t']); #else diff --git a/src/NATS.Client.Core/NatsOpts.cs b/src/NATS.Client.Core/NatsOpts.cs index 575e0883d..021595445 100644 --- a/src/NATS.Client.Core/NatsOpts.cs +++ b/src/NATS.Client.Core/NatsOpts.cs @@ -215,16 +215,16 @@ public sealed record NatsOpts public bool PublishTimeoutOnDisconnected { get; init; } = false; /// - /// Gets or sets a value indicating whether to skip full subject validation. - /// The default is false, meaning subjects are fully validated for whitespace and empty tokens. + /// Gets or sets a value indicating whether to skip subject validation. + /// The default is false, meaning subjects are validated for whitespace and empty strings. /// /// /// - /// When set to false (default), subjects are validated to ensure they don't contain - /// whitespace characters (space, tab, CR, LF) and don't have empty tokens (e.g., ".foo", "foo.", "foo..bar"). + /// When set to false (default), subjects are validated to ensure they are not empty + /// and don't contain whitespace characters (space, tab, CR, LF). /// /// - /// When set to true, only empty subject validation is performed, which can improve + /// When set to true, all subject validation is bypassed. This can improve /// performance in high-throughput scenarios where subjects are known to be valid. /// /// diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs index 0106cefe0..4189109c7 100644 --- a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs @@ -24,6 +24,21 @@ public void WritePublish_ValidSubjectWithReplyTo_DoesNotThrow() action.Should().NotThrow(); } + // Wildcard subjects should be valid + [Theory] + [InlineData("foo.*")] + [InlineData("foo.>")] + [InlineData("*")] + [InlineData(">")] + [InlineData("foo.*.bar")] + [InlineData("foo.bar.>")] + public void WritePublish_WildcardSubject_DoesNotThrow(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + action.Should().NotThrow(); + } + // Whitespace validation for subjects [Theory] [InlineData("foo bar")] @@ -62,6 +77,14 @@ public void WritePublish_EmptySubject_Throws() action.Should().Throw().WithMessage("Subject is invalid."); } + [Fact] + public void WritePublish_EmptyReplyTo_Throws() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, "foo.bar", string.Empty, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject is invalid."); + } + // Subscribe tests [Fact] public void WriteSubscribe_ValidSubject_DoesNotThrow() From 4aaf9a60908ba2a512c496c96b39a8a0f095e009 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Mon, 15 Dec 2025 13:18:36 +0000 Subject: [PATCH 07/13] Add test cases for short and long subject whitespace validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expanded tests to include separate validations for short (<16 chars) and long (≥16 chars) subject paths, with specific checks for whitespace handling. --- .../ProtocolWriterTests.cs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs index 4189109c7..720b2fc84 100644 --- a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs @@ -39,7 +39,7 @@ public void WritePublish_WildcardSubject_DoesNotThrow(string subject) action.Should().NotThrow(); } - // Whitespace validation for subjects + // Whitespace validation for subjects (short path < 16 chars) [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] @@ -49,7 +49,21 @@ public void WritePublish_WildcardSubject_DoesNotThrow(string subject) [InlineData("\tfoo")] [InlineData("\rfoo")] [InlineData("\nfoo")] - public void WritePublish_SubjectWithWhitespace_Throws(string subject) + public void WritePublish_ShortSubjectWithWhitespace_Throws(string subject) + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + // Whitespace validation for subjects (long path >= 16 chars, SIMD) + [Theory] + [InlineData("foo.bar.baz.qux witespace")] + [InlineData("foo.bar.baz.qux\twithtab")] + [InlineData("foo.bar.baz.qux\rwithcr")] + [InlineData("foo.bar.baz.qux\nwithlf")] + [InlineData(" foo.bar.baz.qux.start")] + public void WritePublish_LongSubjectWithWhitespace_Throws(string subject) { using var writer = new NatsBufferWriter(); var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); From 26731258e7d48a1eb291cbca4c0a8803f6b226a1 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Tue, 16 Dec 2025 12:38:38 +0000 Subject: [PATCH 08/13] Refactor subject validation logic: implement centralized validation in `SubjectValidator`, remove inline checks, and streamline public APIs. Update tests for comprehensive validation coverage. --- .../Commands/CommandWriter.cs | 2 +- .../Commands/ProtocolWriter.cs | 81 +---- .../Internal/SubjectValidator.cs | 106 ++++++ .../NatsConnection.Publish.cs | 12 + .../NatsConnection.RequestReply.cs | 10 + .../NatsConnection.RequestSub.cs | 7 + .../NatsConnection.Subscribe.cs | 13 + src/NATS.Client.Core/NatsOpts.cs | 12 +- .../ProtocolWriterTests.cs | 336 ++++++++++++++---- 9 files changed, 428 insertions(+), 151 deletions(-) create mode 100644 src/NATS.Client.Core/Internal/SubjectValidator.cs diff --git a/src/NATS.Client.Core/Commands/CommandWriter.cs b/src/NATS.Client.Core/Commands/CommandWriter.cs index 9ba1768b2..ea43a4e10 100644 --- a/src/NATS.Client.Core/Commands/CommandWriter.cs +++ b/src/NATS.Client.Core/Commands/CommandWriter.cs @@ -71,7 +71,7 @@ public CommandWriter(string name, NatsConnection connection, ObjectPool pool, Na _counter = counter; _defaultCommandTimeout = overrideCommandTimeout ?? opts.CommandTimeout; _enqueuePing = enqueuePing; - _protocolWriter = new ProtocolWriter(opts.SubjectEncoding, opts.SkipSubjectValidation); + _protocolWriter = new ProtocolWriter(opts.SubjectEncoding); _channelLock = Channel.CreateBounded(1); _channelSize = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleWriter = true, SingleReader = true }); _headerWriter = new HeaderWriter(opts.HeaderEncoding); diff --git a/src/NATS.Client.Core/Commands/ProtocolWriter.cs b/src/NATS.Client.Core/Commands/ProtocolWriter.cs index fc12b955b..d7c8f13ca 100644 --- a/src/NATS.Client.Core/Commands/ProtocolWriter.cs +++ b/src/NATS.Client.Core/Commands/ProtocolWriter.cs @@ -36,23 +36,11 @@ internal sealed class ProtocolWriter private static readonly ulong PongNewLine = BinaryPrimitives.ReadUInt64LittleEndian("PONG\r\n "u8); private static readonly ulong UnsubSpace = BinaryPrimitives.ReadUInt64LittleEndian("UNSUB "u8); - // Used for subject/replyTo/queueGroup validation to prevent protocol-breaking whitespace. - // Static field ensures zero allocations per call. SearchValues (NET8+) uses SIMD vectorization; - // char[] (older TFMs) uses optimized IndexOfAny for <=5 chars. Adds ~3% overhead on .NET 8 - // and ~5% on .NET Framework to full publish operations, with zero GC pressure. -#if NET8_0_OR_GREATER - private static readonly SearchValues WhitespaceChars = SearchValues.Create([' ', '\r', '\n', '\t']); -#else - private static readonly char[] WhitespaceChars = [' ', '\r', '\n', '\t']; -#endif - private readonly Encoding _subjectEncoding; - private readonly bool _skipSubjectValidation; - public ProtocolWriter(Encoding subjectEncoding, bool skipSubjectValidation = false) + public ProtocolWriter(Encoding subjectEncoding) { _subjectEncoding = subjectEncoding; - _skipSubjectValidation = skipSubjectValidation; } // https://docs.nats.io/reference/reference-protocols/nats-protocol#connect @@ -94,15 +82,6 @@ public void WritePong(IBufferWriter writer) // HPUB [reply-to] <#header bytes> <#total bytes>\r\n[headers]\r\n\r\n[payload]\r\n public void WritePublish(IBufferWriter writer, string subject, string? replyTo, ReadOnlyMemory? headers, ReadOnlyMemory payload) { - if (!_skipSubjectValidation) - { - ValidateSubject(subject); - if (replyTo != null) - { - ValidateSubject(replyTo); - } - } - if (headers == null) { WritePub(writer, subject, replyTo, payload); @@ -117,15 +96,6 @@ public void WritePublish(IBufferWriter writer, string subject, string? rep // SUB [queue group] public void WriteSubscribe(IBufferWriter writer, int sid, string subject, string? queueGroup, int? maxMsgs) { - if (!_skipSubjectValidation) - { - ValidateSubject(subject); - if (queueGroup != null) - { - ValidateQueueGroup(queueGroup); - } - } - // 'SUB ' + subject +' '+ sid +'\r\n' var ctrlLen = SubSpaceLength + _subjectEncoding.GetByteCount(subject) + 1 + MaxIntStringLength + NewLineLength; @@ -221,55 +191,6 @@ public void WriteUnsubscribe(IBufferWriter writer, int sid, int? maxMessag [MethodImpl(MethodImplOptions.NoInlining)] private static void ThrowOnUtf8FormatFail() => throw new NatsException("Can not format integer."); - [MethodImpl(MethodImplOptions.NoInlining)] - private static void ThrowOnBadSubject() => throw new NatsException("Subject is invalid."); - - [MethodImpl(MethodImplOptions.NoInlining)] - private static void ThrowOnBadQueueGroup() => throw new NatsException("Queue group is invalid."); - - // Validates subjects: no whitespace (protocol-breaking characters) - // Uses adaptive algorithm: manual loop for short subjects (< 16 chars) with - // branch-predicted early exit, SIMD-optimized IndexOfAny for longer subjects. - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static void ValidateSubject(ReadOnlySpan subject) - { - if (subject.IsEmpty) - { - ThrowOnBadSubject(); - } - - if (subject.Length < 16) - { - // Fast path for short subjects - branch predictor learns most chars are > ' ' - foreach (var c in subject) - { - if (c <= ' ' && (c == ' ' || c == '\t' || c == '\r' || c == '\n')) - { - ThrowOnBadSubject(); - } - } - } - else - { - // SIMD path for long subjects - if (subject.IndexOfAny(WhitespaceChars) >= 0) - { - ThrowOnBadSubject(); - } - } - } - - // Validates queue groups: no whitespace only (dots are allowed) - // Matches Go client's badQueue() validation - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static void ValidateQueueGroup(ReadOnlySpan queueGroup) - { - if (queueGroup.IsEmpty || queueGroup.IndexOfAny(WhitespaceChars) >= 0) - { - ThrowOnBadQueueGroup(); - } - } - // PUB [reply-to] <#bytes>\r\n[payload]\r\n private void WritePub(IBufferWriter writer, string subject, string? replyTo, ReadOnlyMemory payload) { diff --git a/src/NATS.Client.Core/Internal/SubjectValidator.cs b/src/NATS.Client.Core/Internal/SubjectValidator.cs new file mode 100644 index 000000000..896ed312a --- /dev/null +++ b/src/NATS.Client.Core/Internal/SubjectValidator.cs @@ -0,0 +1,106 @@ +using System.Runtime.CompilerServices; + +namespace NATS.Client.Core.Internal; + +/// +/// Validates NATS subjects and queue groups at the API level. +/// +/// +/// Subject validation prevents protocol-breaking whitespace characters from being sent to the server. +/// This validation is performed at the top-level API (Publish/Subscribe) for immediate error feedback. +/// +internal static class SubjectValidator +{ + // Used for subject/replyTo/queueGroup validation to prevent protocol-breaking whitespace. + // Static field ensures zero allocations per call. SearchValues (NET8+) uses SIMD vectorization; + // char[] (older TFMs) uses optimized IndexOfAny for <=5 chars. +#if NET8_0_OR_GREATER + private static readonly System.Buffers.SearchValues WhitespaceChars = System.Buffers.SearchValues.Create([' ', '\r', '\n', '\t']); +#else + private static readonly char[] WhitespaceChars = [' ', '\r', '\n', '\t']; +#endif + + /// + /// Validates a subject string for whitespace characters. + /// + /// The subject to validate. + /// Thrown when the subject is empty or contains whitespace. + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static void ValidateSubject(string? subject) + { + if (string.IsNullOrEmpty(subject)) + { + ThrowOnBadSubject(); + } + + ValidateSubjectSpan(subject.AsSpan()); + } + + /// + /// Validates a reply-to string for whitespace characters (if not null). + /// + /// The reply-to to validate, or null. + /// Thrown when the reply-to contains whitespace. + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static void ValidateReplyTo(string? replyTo) + { + if (replyTo != null) + { + if (replyTo.Length == 0) + { + ThrowOnBadSubject(); + } + + ValidateSubjectSpan(replyTo.AsSpan()); + } + } + + /// + /// Validates a queue group string for whitespace characters (if not null). + /// + /// The queue group to validate, or null. + /// Thrown when the queue group is empty or contains whitespace. + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static void ValidateQueueGroup(string? queueGroup) + { + if (queueGroup != null) + { + if (queueGroup.Length == 0 || queueGroup.AsSpan().IndexOfAny(WhitespaceChars) >= 0) + { + ThrowOnBadQueueGroup(); + } + } + } + + // Uses adaptive algorithm: manual loop for short subjects (< 16 chars) with + // branch-predicted early exit, SIMD-optimized IndexOfAny for longer subjects. + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static void ValidateSubjectSpan(ReadOnlySpan subject) + { + if (subject.Length < 16) + { + // Fast path for short subjects - branch predictor learns most chars are > ' ' + foreach (var c in subject) + { + if (c <= ' ' && (c == ' ' || c == '\t' || c == '\r' || c == '\n')) + { + ThrowOnBadSubject(); + } + } + } + else + { + // SIMD path for long subjects + if (subject.IndexOfAny(WhitespaceChars) >= 0) + { + ThrowOnBadSubject(); + } + } + } + + [MethodImpl(MethodImplOptions.NoInlining)] + private static void ThrowOnBadSubject() => throw new NatsException("Subject is invalid."); + + [MethodImpl(MethodImplOptions.NoInlining)] + private static void ThrowOnBadQueueGroup() => throw new NatsException("Queue group is invalid."); +} diff --git a/src/NATS.Client.Core/NatsConnection.Publish.cs b/src/NATS.Client.Core/NatsConnection.Publish.cs index 0b6a91161..54a92dd35 100644 --- a/src/NATS.Client.Core/NatsConnection.Publish.cs +++ b/src/NATS.Client.Core/NatsConnection.Publish.cs @@ -8,6 +8,12 @@ public partial class NatsConnection /// public ValueTask PublishAsync(string subject, NatsHeaders? headers = default, string? replyTo = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default) { + if (!Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + SubjectValidator.ValidateReplyTo(replyTo); + } + if (Telemetry.HasListeners()) { using var activity = Telemetry.StartSendActivity($"{SpanDestinationName(subject)} {Telemetry.Constants.PublishActivityName}", this, subject, replyTo); @@ -35,6 +41,12 @@ public ValueTask PublishAsync(string subject, NatsHeaders? headers = default, st /// public ValueTask PublishAsync(string subject, T? data, NatsHeaders? headers = default, string? replyTo = default, INatsSerialize? serializer = default, NatsPubOpts? opts = default, CancellationToken cancellationToken = default) { + if (!Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + SubjectValidator.ValidateReplyTo(replyTo); + } + if (Telemetry.HasListeners()) { using var activity = Telemetry.StartSendActivity($"{SpanDestinationName(subject)} {Telemetry.Constants.PublishActivityName}", this, subject, replyTo); diff --git a/src/NATS.Client.Core/NatsConnection.RequestReply.cs b/src/NATS.Client.Core/NatsConnection.RequestReply.cs index f0dec997f..951ef8936 100644 --- a/src/NATS.Client.Core/NatsConnection.RequestReply.cs +++ b/src/NATS.Client.Core/NatsConnection.RequestReply.cs @@ -33,6 +33,11 @@ public async ValueTask> RequestAsync( NatsSubOpts? replyOpts = default, CancellationToken cancellationToken = default) { + if (!Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + } + if (Telemetry.HasListeners()) { using var activity = Telemetry.StartSendActivity($"{SpanDestinationName(subject)} {Telemetry.Constants.RequestReplyActivityName}", this, subject, null); @@ -113,6 +118,11 @@ public async IAsyncEnumerable> RequestManyAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) .ConfigureAwait(false); diff --git a/src/NATS.Client.Core/NatsConnection.RequestSub.cs b/src/NATS.Client.Core/NatsConnection.RequestSub.cs index 8763d8967..fbc2c9cac 100644 --- a/src/NATS.Client.Core/NatsConnection.RequestSub.cs +++ b/src/NATS.Client.Core/NatsConnection.RequestSub.cs @@ -1,3 +1,5 @@ +using NATS.Client.Core.Internal; + namespace NATS.Client.Core; public partial class NatsConnection @@ -13,6 +15,11 @@ public async ValueTask> CreateRequestSubAsync( NatsSubOpts? replyOpts = default, CancellationToken cancellationToken = default) { + if (!Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + } + var replyTo = NewInbox(); replySerializer ??= Opts.SerializerRegistry.GetDeserializer(); diff --git a/src/NATS.Client.Core/NatsConnection.Subscribe.cs b/src/NATS.Client.Core/NatsConnection.Subscribe.cs index 770bd6421..f53926929 100644 --- a/src/NATS.Client.Core/NatsConnection.Subscribe.cs +++ b/src/NATS.Client.Core/NatsConnection.Subscribe.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using System.Runtime.CompilerServices; +using NATS.Client.Core.Internal; namespace NATS.Client.Core; @@ -8,6 +9,12 @@ public partial class NatsConnection /// public async IAsyncEnumerable> SubscribeAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, [EnumeratorCancellation] CancellationToken cancellationToken = default) { + if (!Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + SubjectValidator.ValidateQueueGroup(queueGroup); + } + serializer ??= Opts.SerializerRegistry.GetDeserializer(); await using var sub = new NatsSub(this, _subscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); @@ -24,6 +31,12 @@ public async IAsyncEnumerable> SubscribeAsync(string subject, stri /// public async ValueTask> SubscribeCoreAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default) { + if (!Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + SubjectValidator.ValidateQueueGroup(queueGroup); + } + serializer ??= Opts.SerializerRegistry.GetDeserializer(); var sub = new NatsSub(this, _subscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); await AddSubAsync(sub, cancellationToken).ConfigureAwait(false); diff --git a/src/NATS.Client.Core/NatsOpts.cs b/src/NATS.Client.Core/NatsOpts.cs index 021595445..ffcc20d89 100644 --- a/src/NATS.Client.Core/NatsOpts.cs +++ b/src/NATS.Client.Core/NatsOpts.cs @@ -216,19 +216,19 @@ public sealed record NatsOpts /// /// Gets or sets a value indicating whether to skip subject validation. - /// The default is false, meaning subjects are validated for whitespace and empty strings. + /// The default is true, meaning subject validation is disabled. /// /// /// - /// When set to false (default), subjects are validated to ensure they are not empty - /// and don't contain whitespace characters (space, tab, CR, LF). + /// When set to true (default), all subject validation is bypassed. /// /// - /// When set to true, all subject validation is bypassed. This can improve - /// performance in high-throughput scenarios where subjects are known to be valid. + /// When set to false, subjects are validated to ensure they are not empty + /// and don't contain whitespace characters (space, tab, CR, LF). This can help + /// catch invalid subjects early but adds minor overhead. /// /// - public bool SkipSubjectValidation { get; init; } = false; + public bool SkipSubjectValidation { get; init; } = true; internal NatsUri[] GetSeedUris(bool suppressRandomization = false) { diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs index 720b2fc84..a9ddd5205 100644 --- a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs @@ -1,4 +1,5 @@ using System.Text; +using NATS.Client.Core.Internal; // ReSharper disable AccessToDisposedClosure namespace NATS.Client.CoreUnit.Tests; @@ -6,7 +7,6 @@ namespace NATS.Client.CoreUnit.Tests; public class ProtocolWriterTests { private readonly ProtocolWriter _protocolWriter = new(Encoding.UTF8); - private readonly ProtocolWriter _protocolWriterNoValidation = new(Encoding.UTF8, skipSubjectValidation: true); [Fact] public void WritePublish_ValidSubject_DoesNotThrow() @@ -24,6 +24,30 @@ public void WritePublish_ValidSubjectWithReplyTo_DoesNotThrow() action.Should().NotThrow(); } + // Subscribe tests + [Fact] + public void WriteSubscribe_ValidSubject_DoesNotThrow() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", null, null); + action.Should().NotThrow(); + } + + [Fact] + public void WriteSubscribe_ValidSubjectWithQueueGroup_DoesNotThrow() + { + using var writer = new NatsBufferWriter(); + var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", "queue-group", null); + action.Should().NotThrow(); + } +} + +/// +/// Tests for subject validation at the API level. +/// Validation is performed in SubjectValidator and called from NatsConnection.PublishAsync/SubscribeAsync. +/// +public class SubjectValidatorTests +{ // Wildcard subjects should be valid [Theory] [InlineData("foo.*")] @@ -32,10 +56,9 @@ public void WritePublish_ValidSubjectWithReplyTo_DoesNotThrow() [InlineData(">")] [InlineData("foo.*.bar")] [InlineData("foo.bar.>")] - public void WritePublish_WildcardSubject_DoesNotThrow(string subject) + public void ValidateSubject_WildcardSubject_DoesNotThrow(string subject) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + var action = () => SubjectValidator.ValidateSubject(subject); action.Should().NotThrow(); } @@ -49,10 +72,9 @@ public void WritePublish_WildcardSubject_DoesNotThrow(string subject) [InlineData("\tfoo")] [InlineData("\rfoo")] [InlineData("\nfoo")] - public void WritePublish_ShortSubjectWithWhitespace_Throws(string subject) + public void ValidateSubject_ShortSubjectWithWhitespace_Throws(string subject) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + var action = () => SubjectValidator.ValidateSubject(subject); action.Should().Throw().WithMessage("Subject is invalid."); } @@ -63,10 +85,9 @@ public void WritePublish_ShortSubjectWithWhitespace_Throws(string subject) [InlineData("foo.bar.baz.qux\rwithcr")] [InlineData("foo.bar.baz.qux\nwithlf")] [InlineData(" foo.bar.baz.qux.start")] - public void WritePublish_LongSubjectWithWhitespace_Throws(string subject) + public void ValidateSubject_LongSubjectWithWhitespace_Throws(string subject) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); + var action = () => SubjectValidator.ValidateSubject(subject); action.Should().Throw().WithMessage("Subject is invalid."); } @@ -75,44 +96,38 @@ public void WritePublish_LongSubjectWithWhitespace_Throws(string subject) [InlineData("reply\tto")] [InlineData("reply\rto")] [InlineData("reply\nto")] - public void WritePublish_ReplyToWithWhitespace_Throws(string replyTo) + public void ValidateReplyTo_WithWhitespace_Throws(string replyTo) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, "foo.bar", replyTo, null, ReadOnlyMemory.Empty); + var action = () => SubjectValidator.ValidateReplyTo(replyTo); action.Should().Throw().WithMessage("Subject is invalid."); } // Empty subject validation [Fact] - public void WritePublish_EmptySubject_Throws() + public void ValidateSubject_EmptySubject_Throws() { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, string.Empty, null, null, ReadOnlyMemory.Empty); + var action = () => SubjectValidator.ValidateSubject(string.Empty); action.Should().Throw().WithMessage("Subject is invalid."); } [Fact] - public void WritePublish_EmptyReplyTo_Throws() + public void ValidateSubject_NullSubject_Throws() { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, "foo.bar", string.Empty, null, ReadOnlyMemory.Empty); + var action = () => SubjectValidator.ValidateSubject(null); action.Should().Throw().WithMessage("Subject is invalid."); } - // Subscribe tests [Fact] - public void WriteSubscribe_ValidSubject_DoesNotThrow() + public void ValidateReplyTo_EmptyReplyTo_Throws() { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", null, null); - action.Should().NotThrow(); + var action = () => SubjectValidator.ValidateReplyTo(string.Empty); + action.Should().Throw().WithMessage("Subject is invalid."); } [Fact] - public void WriteSubscribe_ValidSubjectWithQueueGroup_DoesNotThrow() + public void ValidateReplyTo_NullReplyTo_DoesNotThrow() { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", "queue-group", null); + var action = () => SubjectValidator.ValidateReplyTo(null); action.Should().NotThrow(); } @@ -122,81 +137,274 @@ public void WriteSubscribe_ValidSubjectWithQueueGroup_DoesNotThrow() [InlineData(".queue")] [InlineData("queue.")] [InlineData("queue..group")] - public void WriteSubscribe_QueueGroupWithDots_DoesNotThrow(string queueGroup) + public void ValidateQueueGroup_WithDots_DoesNotThrow(string queueGroup) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", queueGroup, null); + var action = () => SubjectValidator.ValidateQueueGroup(queueGroup); action.Should().NotThrow(); } + [Theory] + [InlineData("queue group")] + [InlineData("queue\tgroup")] + [InlineData("queue\rgroup")] + [InlineData("queue\ngroup")] + public void ValidateQueueGroup_WithWhitespace_Throws(string queueGroup) + { + var action = () => SubjectValidator.ValidateQueueGroup(queueGroup); + action.Should().Throw().WithMessage("Queue group is invalid."); + } + + [Fact] + public void ValidateQueueGroup_EmptyQueueGroup_Throws() + { + var action = () => SubjectValidator.ValidateQueueGroup(string.Empty); + action.Should().Throw().WithMessage("Queue group is invalid."); + } + + [Fact] + public void ValidateQueueGroup_NullQueueGroup_DoesNotThrow() + { + var action = () => SubjectValidator.ValidateQueueGroup(null); + action.Should().NotThrow(); + } +} + +/// +/// Tests for subject validation at the NatsConnection API level. +/// These tests verify that validation happens immediately when calling PublishAsync/SubscribeAsync, +/// before any connection attempt is made. +/// Note: SkipSubjectValidation defaults to true, so tests must explicitly enable validation. +/// +public class NatsConnectionSubjectValidationTests +{ + private static readonly NatsOpts OptsWithValidation = new() { SkipSubjectValidation = false }; + + // PublishAsync tests [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] [InlineData("foo\rbar")] [InlineData("foo\nbar")] - public void WriteSubscribe_SubjectWithWhitespace_Throws(string subject) + public async Task PublishAsync_SubjectWithWhitespace_ThrowsImmediately(string subject) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, subject, null, null); - action.Should().Throw().WithMessage("Subject is invalid."); + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.PublishAsync(subject); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); + } + + [Fact] + public async Task PublishAsync_EmptySubject_ThrowsImmediately() + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.PublishAsync(string.Empty); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } [Theory] - [InlineData("queue group")] - [InlineData("queue\tgroup")] - [InlineData("queue\rgroup")] - [InlineData("queue\ngroup")] - public void WriteSubscribe_QueueGroupWithWhitespace_Throws(string queueGroup) + [InlineData("reply to")] + [InlineData("reply\tto")] + public async Task PublishAsync_ReplyToWithWhitespace_ThrowsImmediately(string replyTo) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", queueGroup, null); - action.Should().Throw().WithMessage("Queue group is invalid."); + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.PublishAsync("foo.bar", replyTo: replyTo); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } [Fact] - public void WriteSubscribe_EmptySubject_Throws() + public async Task PublishAsync_EmptyReplyTo_ThrowsImmediately() { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, string.Empty, null, null); - action.Should().Throw().WithMessage("Subject is invalid."); + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.PublishAsync("foo.bar", replyTo: string.Empty); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } - [Fact] - public void WriteSubscribe_EmptyQueueGroup_Throws() + // PublishAsync tests + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + public async Task PublishAsyncT_SubjectWithWhitespace_ThrowsImmediately(string subject) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", string.Empty, null); - action.Should().Throw().WithMessage("Queue group is invalid."); + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.PublishAsync(subject, "data"); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } - // SkipSubjectValidation tests + // SubscribeAsync tests [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] - public void WritePublish_WithSkipValidation_DoesNotThrow(string subject) + [InlineData("foo\rbar")] + [InlineData("foo\nbar")] + public async Task SubscribeAsync_SubjectWithWhitespace_ThrowsImmediately(string subject) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriterNoValidation.WritePublish(writer, subject, null, null, ReadOnlyMemory.Empty); - action.Should().NotThrow(); + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => + { + await foreach (var _ in nats.SubscribeAsync(subject)) + { + // Should not reach here + } + }; + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); + } + + [Fact] + public async Task SubscribeAsync_EmptySubject_ThrowsImmediately() + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => + { + await foreach (var _ in nats.SubscribeAsync(string.Empty)) + { + // Should not reach here + } + }; + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } + [Theory] + [InlineData("queue group")] + [InlineData("queue\tgroup")] + public async Task SubscribeAsync_QueueGroupWithWhitespace_ThrowsImmediately(string queueGroup) + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => + { + await foreach (var _ in nats.SubscribeAsync("foo.bar", queueGroup: queueGroup)) + { + // Should not reach here + } + }; + await action.Should().ThrowAsync().WithMessage("Queue group is invalid."); + } + + [Fact] + public async Task SubscribeAsync_EmptyQueueGroup_ThrowsImmediately() + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => + { + await foreach (var _ in nats.SubscribeAsync("foo.bar", queueGroup: string.Empty)) + { + // Should not reach here + } + }; + await action.Should().ThrowAsync().WithMessage("Queue group is invalid."); + } + + // SubscribeCoreAsync tests [Theory] [InlineData("foo bar")] [InlineData("foo\tbar")] - public void WriteSubscribe_WithSkipValidation_DoesNotThrow(string subject) + public async Task SubscribeCoreAsync_SubjectWithWhitespace_ThrowsImmediately(string subject) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriterNoValidation.WriteSubscribe(writer, 1, subject, null, null); - action.Should().NotThrow(); + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.SubscribeCoreAsync(subject); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } [Theory] [InlineData("queue group")] [InlineData("queue\tgroup")] - public void WriteSubscribe_QueueGroupWithSkipValidation_DoesNotThrow(string queueGroup) + public async Task SubscribeCoreAsync_QueueGroupWithWhitespace_ThrowsImmediately(string queueGroup) { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriterNoValidation.WriteSubscribe(writer, 1, "foo.bar", queueGroup, null); - action.Should().NotThrow(); + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.SubscribeCoreAsync("foo.bar", queueGroup: queueGroup); + await action.Should().ThrowAsync().WithMessage("Queue group is invalid."); + } + + // SkipSubjectValidation option tests (default is true, so validation is skipped by default) + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + public async Task PublishAsync_WithSkipValidation_DoesNotThrowOnInvalidSubject(string subject) + { + // Default SkipSubjectValidation=true, validation is bypassed at the API level. + // The call will fail later (e.g., connection timeout) but not due to validation. + await using var nats = new NatsConnection(); + + // This should not throw NatsException for invalid subject + // It will throw something else (timeout, connection error) since we're not connected + Func action = async () => + { + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); + await nats.PublishAsync(subject, cancellationToken: cts.Token); + }; + + // Should NOT throw "Subject is invalid." + var ex = await action.Should().ThrowAsync(); + ex.Which.Message.Should().NotBe("Subject is invalid."); + } + + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + public async Task SubscribeAsync_WithSkipValidation_DoesNotThrowOnInvalidSubject(string subject) + { + // Default SkipSubjectValidation=true + await using var nats = new NatsConnection(); + + // This should not throw NatsException for invalid subject + Func action = async () => + { + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); + await foreach (var _ in nats.SubscribeAsync(subject, cancellationToken: cts.Token)) + { + // Should not reach here + } + }; + + // Should NOT throw "Subject is invalid." - will throw timeout or connection error instead + var ex = await action.Should().ThrowAsync(); + ex.Which.Message.Should().NotBe("Subject is invalid."); + } + + // RequestAsync tests + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + [InlineData("foo\rbar")] + [InlineData("foo\nbar")] + public async Task RequestAsync_SubjectWithWhitespace_ThrowsImmediately(string subject) + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.RequestAsync(subject, "data"); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); + } + + [Fact] + public async Task RequestAsync_EmptySubject_ThrowsImmediately() + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.RequestAsync(string.Empty, "data"); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); + } + + // RequestManyAsync tests + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + public async Task RequestManyAsync_SubjectWithWhitespace_ThrowsImmediately(string subject) + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => + { + await foreach (var msg in nats.RequestManyAsync(subject, "data")) + { + // Should not reach here + } + }; + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); + } + + // CreateRequestSubAsync tests + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + public async Task CreateRequestSubAsync_SubjectWithWhitespace_ThrowsImmediately(string subject) + { + await using var nats = new NatsConnection(OptsWithValidation); + var action = async () => await nats.CreateRequestSubAsync(subject, "data"); + await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } } From 918e38de915e26c20cf223abc8f74547ba6a2945 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Tue, 16 Dec 2025 12:47:05 +0000 Subject: [PATCH 09/13] Refactor and expand subject, replyTo, and queue group validation tests. Split `ProtocolWriterTests` into focused test classes and improve test coverage for whitespace handling and edge cases. --- ...> NatsConnectionSubjectValidationTests.cs} | 171 ------------------ .../SubjectValidatorTests.cs | 124 +++++++++++++ 2 files changed, 124 insertions(+), 171 deletions(-) rename tests/NATS.Client.CoreUnit.Tests/{ProtocolWriterTests.cs => NatsConnectionSubjectValidationTests.cs} (61%) create mode 100644 tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs diff --git a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs b/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs similarity index 61% rename from tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs rename to tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs index a9ddd5205..eeec0b6c9 100644 --- a/tests/NATS.Client.CoreUnit.Tests/ProtocolWriterTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs @@ -4,177 +4,6 @@ // ReSharper disable AccessToDisposedClosure namespace NATS.Client.CoreUnit.Tests; -public class ProtocolWriterTests -{ - private readonly ProtocolWriter _protocolWriter = new(Encoding.UTF8); - - [Fact] - public void WritePublish_ValidSubject_DoesNotThrow() - { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, "foo.bar", null, null, ReadOnlyMemory.Empty); - action.Should().NotThrow(); - } - - [Fact] - public void WritePublish_ValidSubjectWithReplyTo_DoesNotThrow() - { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WritePublish(writer, "foo.bar", "reply.to", null, ReadOnlyMemory.Empty); - action.Should().NotThrow(); - } - - // Subscribe tests - [Fact] - public void WriteSubscribe_ValidSubject_DoesNotThrow() - { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", null, null); - action.Should().NotThrow(); - } - - [Fact] - public void WriteSubscribe_ValidSubjectWithQueueGroup_DoesNotThrow() - { - using var writer = new NatsBufferWriter(); - var action = () => _protocolWriter.WriteSubscribe(writer, 1, "foo.bar", "queue-group", null); - action.Should().NotThrow(); - } -} - -/// -/// Tests for subject validation at the API level. -/// Validation is performed in SubjectValidator and called from NatsConnection.PublishAsync/SubscribeAsync. -/// -public class SubjectValidatorTests -{ - // Wildcard subjects should be valid - [Theory] - [InlineData("foo.*")] - [InlineData("foo.>")] - [InlineData("*")] - [InlineData(">")] - [InlineData("foo.*.bar")] - [InlineData("foo.bar.>")] - public void ValidateSubject_WildcardSubject_DoesNotThrow(string subject) - { - var action = () => SubjectValidator.ValidateSubject(subject); - action.Should().NotThrow(); - } - - // Whitespace validation for subjects (short path < 16 chars) - [Theory] - [InlineData("foo bar")] - [InlineData("foo\tbar")] - [InlineData("foo\rbar")] - [InlineData("foo\nbar")] - [InlineData(" foo")] - [InlineData("\tfoo")] - [InlineData("\rfoo")] - [InlineData("\nfoo")] - public void ValidateSubject_ShortSubjectWithWhitespace_Throws(string subject) - { - var action = () => SubjectValidator.ValidateSubject(subject); - action.Should().Throw().WithMessage("Subject is invalid."); - } - - // Whitespace validation for subjects (long path >= 16 chars, SIMD) - [Theory] - [InlineData("foo.bar.baz.qux witespace")] - [InlineData("foo.bar.baz.qux\twithtab")] - [InlineData("foo.bar.baz.qux\rwithcr")] - [InlineData("foo.bar.baz.qux\nwithlf")] - [InlineData(" foo.bar.baz.qux.start")] - public void ValidateSubject_LongSubjectWithWhitespace_Throws(string subject) - { - var action = () => SubjectValidator.ValidateSubject(subject); - action.Should().Throw().WithMessage("Subject is invalid."); - } - - [Theory] - [InlineData("reply to")] - [InlineData("reply\tto")] - [InlineData("reply\rto")] - [InlineData("reply\nto")] - public void ValidateReplyTo_WithWhitespace_Throws(string replyTo) - { - var action = () => SubjectValidator.ValidateReplyTo(replyTo); - action.Should().Throw().WithMessage("Subject is invalid."); - } - - // Empty subject validation - [Fact] - public void ValidateSubject_EmptySubject_Throws() - { - var action = () => SubjectValidator.ValidateSubject(string.Empty); - action.Should().Throw().WithMessage("Subject is invalid."); - } - - [Fact] - public void ValidateSubject_NullSubject_Throws() - { - var action = () => SubjectValidator.ValidateSubject(null); - action.Should().Throw().WithMessage("Subject is invalid."); - } - - [Fact] - public void ValidateReplyTo_EmptyReplyTo_Throws() - { - var action = () => SubjectValidator.ValidateReplyTo(string.Empty); - action.Should().Throw().WithMessage("Subject is invalid."); - } - - [Fact] - public void ValidateReplyTo_NullReplyTo_DoesNotThrow() - { - var action = () => SubjectValidator.ValidateReplyTo(null); - action.Should().NotThrow(); - } - - // Queue group can have dots - [Theory] - [InlineData("queue.group")] - [InlineData(".queue")] - [InlineData("queue.")] - [InlineData("queue..group")] - public void ValidateQueueGroup_WithDots_DoesNotThrow(string queueGroup) - { - var action = () => SubjectValidator.ValidateQueueGroup(queueGroup); - action.Should().NotThrow(); - } - - [Theory] - [InlineData("queue group")] - [InlineData("queue\tgroup")] - [InlineData("queue\rgroup")] - [InlineData("queue\ngroup")] - public void ValidateQueueGroup_WithWhitespace_Throws(string queueGroup) - { - var action = () => SubjectValidator.ValidateQueueGroup(queueGroup); - action.Should().Throw().WithMessage("Queue group is invalid."); - } - - [Fact] - public void ValidateQueueGroup_EmptyQueueGroup_Throws() - { - var action = () => SubjectValidator.ValidateQueueGroup(string.Empty); - action.Should().Throw().WithMessage("Queue group is invalid."); - } - - [Fact] - public void ValidateQueueGroup_NullQueueGroup_DoesNotThrow() - { - var action = () => SubjectValidator.ValidateQueueGroup(null); - action.Should().NotThrow(); - } -} - -/// -/// Tests for subject validation at the NatsConnection API level. -/// These tests verify that validation happens immediately when calling PublishAsync/SubscribeAsync, -/// before any connection attempt is made. -/// Note: SkipSubjectValidation defaults to true, so tests must explicitly enable validation. -/// public class NatsConnectionSubjectValidationTests { private static readonly NatsOpts OptsWithValidation = new() { SkipSubjectValidation = false }; diff --git a/tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs b/tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs new file mode 100644 index 000000000..91f18769c --- /dev/null +++ b/tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs @@ -0,0 +1,124 @@ +namespace NATS.Client.CoreUnit.Tests; + +public class SubjectValidatorTests +{ + // Wildcard subjects should be valid + [Theory] + [InlineData("foo.*")] + [InlineData("foo.>")] + [InlineData("*")] + [InlineData(">")] + [InlineData("foo.*.bar")] + [InlineData("foo.bar.>")] + public void ValidateSubject_WildcardSubject_DoesNotThrow(string subject) + { + var action = () => SubjectValidator.ValidateSubject(subject); + action.Should().NotThrow(); + } + + // Whitespace validation for subjects (short path < 16 chars) + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + [InlineData("foo\rbar")] + [InlineData("foo\nbar")] + [InlineData(" foo")] + [InlineData("\tfoo")] + [InlineData("\rfoo")] + [InlineData("\nfoo")] + public void ValidateSubject_ShortSubjectWithWhitespace_Throws(string subject) + { + var action = () => SubjectValidator.ValidateSubject(subject); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + // Whitespace validation for subjects (long path >= 16 chars, SIMD) + [Theory] + [InlineData("foo.bar.baz.qux witespace")] + [InlineData("foo.bar.baz.qux\twithtab")] + [InlineData("foo.bar.baz.qux\rwithcr")] + [InlineData("foo.bar.baz.qux\nwithlf")] + [InlineData(" foo.bar.baz.qux.start")] + public void ValidateSubject_LongSubjectWithWhitespace_Throws(string subject) + { + var action = () => SubjectValidator.ValidateSubject(subject); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + [Theory] + [InlineData("reply to")] + [InlineData("reply\tto")] + [InlineData("reply\rto")] + [InlineData("reply\nto")] + public void ValidateReplyTo_WithWhitespace_Throws(string replyTo) + { + var action = () => SubjectValidator.ValidateReplyTo(replyTo); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + // Empty subject validation + [Fact] + public void ValidateSubject_EmptySubject_Throws() + { + var action = () => SubjectValidator.ValidateSubject(string.Empty); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + [Fact] + public void ValidateSubject_NullSubject_Throws() + { + var action = () => SubjectValidator.ValidateSubject(null); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + [Fact] + public void ValidateReplyTo_EmptyReplyTo_Throws() + { + var action = () => SubjectValidator.ValidateReplyTo(string.Empty); + action.Should().Throw().WithMessage("Subject is invalid."); + } + + [Fact] + public void ValidateReplyTo_NullReplyTo_DoesNotThrow() + { + var action = () => SubjectValidator.ValidateReplyTo(null); + action.Should().NotThrow(); + } + + // Queue group can have dots + [Theory] + [InlineData("queue.group")] + [InlineData(".queue")] + [InlineData("queue.")] + [InlineData("queue..group")] + public void ValidateQueueGroup_WithDots_DoesNotThrow(string queueGroup) + { + var action = () => SubjectValidator.ValidateQueueGroup(queueGroup); + action.Should().NotThrow(); + } + + [Theory] + [InlineData("queue group")] + [InlineData("queue\tgroup")] + [InlineData("queue\rgroup")] + [InlineData("queue\ngroup")] + public void ValidateQueueGroup_WithWhitespace_Throws(string queueGroup) + { + var action = () => SubjectValidator.ValidateQueueGroup(queueGroup); + action.Should().Throw().WithMessage("Queue group is invalid."); + } + + [Fact] + public void ValidateQueueGroup_EmptyQueueGroup_Throws() + { + var action = () => SubjectValidator.ValidateQueueGroup(string.Empty); + action.Should().Throw().WithMessage("Queue group is invalid."); + } + + [Fact] + public void ValidateQueueGroup_NullQueueGroup_DoesNotThrow() + { + var action = () => SubjectValidator.ValidateQueueGroup(null); + action.Should().NotThrow(); + } +} From de6c47182862aa1aa0b42ce14a2a7d95980695da Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Tue, 16 Dec 2025 15:08:56 +0000 Subject: [PATCH 10/13] Add synchronous subject and queue group validation to public NATS APIs Ensure subject and queue group validation occurs before async operations in `SubscribeAsync`, `SubscribeCoreAsync`, and `RequestManyAsync`. Update tests to verify synchronous exceptions and add comprehensive validation coverage. --- .../NatsConnection.RequestReply.cs | 19 +++++- .../NatsConnection.Subscribe.cs | 18 +++++- src/NATS.Client.Core/NatsSubBase.cs | 6 ++ .../NatsConnectionSubjectValidationTests.cs | 60 +++++++++++++++++++ .../ServicesTests.cs | 53 ++++++++++++++++ 5 files changed, 151 insertions(+), 5 deletions(-) diff --git a/src/NATS.Client.Core/NatsConnection.RequestReply.cs b/src/NATS.Client.Core/NatsConnection.RequestReply.cs index 951ef8936..4b5a60c49 100644 --- a/src/NATS.Client.Core/NatsConnection.RequestReply.cs +++ b/src/NATS.Client.Core/NatsConnection.RequestReply.cs @@ -108,7 +108,7 @@ public ValueTask> RequestAsync( cancellationToken: cancellationToken); /// - public async IAsyncEnumerable> RequestManyAsync( + public IAsyncEnumerable> RequestManyAsync( string subject, TRequest? data, NatsHeaders? headers = default, @@ -116,13 +116,28 @@ public async IAsyncEnumerable> RequestManyAsync? replySerializer = default, NatsPubOpts? requestOpts = default, NatsSubOpts? replyOpts = default, - [EnumeratorCancellation] CancellationToken cancellationToken = default) + CancellationToken cancellationToken = default) { + // Validate synchronously before returning the async enumerable + // so that invalid subjects throw immediately when RequestManyAsync is called if (!Opts.SkipSubjectValidation) { SubjectValidator.ValidateSubject(subject); } + return RequestManyInternalAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken); + } + + private async IAsyncEnumerable> RequestManyInternalAsync( + string subject, + TRequest? data, + NatsHeaders? headers, + INatsSerialize? requestSerializer, + INatsDeserialize? replySerializer, + NatsPubOpts? requestOpts, + NatsSubOpts? replyOpts, + [EnumeratorCancellation] CancellationToken cancellationToken) + { replyOpts = SetReplyManyOptsDefaults(replyOpts); await using var sub = await CreateRequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) .ConfigureAwait(false); diff --git a/src/NATS.Client.Core/NatsConnection.Subscribe.cs b/src/NATS.Client.Core/NatsConnection.Subscribe.cs index f53926929..11965dfdc 100644 --- a/src/NATS.Client.Core/NatsConnection.Subscribe.cs +++ b/src/NATS.Client.Core/NatsConnection.Subscribe.cs @@ -1,4 +1,3 @@ -using System.Collections.Concurrent; using System.Runtime.CompilerServices; using NATS.Client.Core.Internal; @@ -7,14 +6,21 @@ namespace NATS.Client.Core; public partial class NatsConnection { /// - public async IAsyncEnumerable> SubscribeAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, [EnumeratorCancellation] CancellationToken cancellationToken = default) + public IAsyncEnumerable> SubscribeAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default) { + // Validate synchronously before returning the async enumerable + // so that invalid subjects throw immediately when SubscribeAsync is called if (!Opts.SkipSubjectValidation) { SubjectValidator.ValidateSubject(subject); SubjectValidator.ValidateQueueGroup(queueGroup); } + return SubscribeInternalAsync(subject, queueGroup, serializer, opts, cancellationToken); + } + + private async IAsyncEnumerable> SubscribeInternalAsync(string subject, string? queueGroup, INatsDeserialize? serializer, NatsSubOpts? opts, [EnumeratorCancellation] CancellationToken cancellationToken) + { serializer ??= Opts.SerializerRegistry.GetDeserializer(); await using var sub = new NatsSub(this, _subscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); @@ -29,14 +35,20 @@ public async IAsyncEnumerable> SubscribeAsync(string subject, stri } /// - public async ValueTask> SubscribeCoreAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default) + public ValueTask> SubscribeCoreAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default) { + // Validate synchronously so invalid subjects throw immediately if (!Opts.SkipSubjectValidation) { SubjectValidator.ValidateSubject(subject); SubjectValidator.ValidateQueueGroup(queueGroup); } + return SubscribeCoreInternalAsync(subject, queueGroup, serializer, opts, cancellationToken); + } + + private async ValueTask> SubscribeCoreInternalAsync(string subject, string? queueGroup, INatsDeserialize? serializer, NatsSubOpts? opts, CancellationToken cancellationToken) + { serializer ??= Opts.SerializerRegistry.GetDeserializer(); var sub = new NatsSub(this, _subscriptionManager.GetManagerFor(subject), subject, queueGroup, opts, serializer, cancellationToken); await AddSubAsync(sub, cancellationToken).ConfigureAwait(false); diff --git a/src/NATS.Client.Core/NatsSubBase.cs b/src/NATS.Client.Core/NatsSubBase.cs index 175cd7646..a5f7b2e45 100644 --- a/src/NATS.Client.Core/NatsSubBase.cs +++ b/src/NATS.Client.Core/NatsSubBase.cs @@ -67,6 +67,12 @@ protected NatsSubBase( NatsSubOpts? opts, CancellationToken cancellationToken = default) { + if (!connection.Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + SubjectValidator.ValidateQueueGroup(queueGroup); + } + _logger = connection.Opts.LoggerFactory.CreateLogger(); _debug = _logger.IsEnabled(LogLevel.Debug); _manager = manager; diff --git a/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs b/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs index eeec0b6c9..3abc70fb4 100644 --- a/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs @@ -236,4 +236,64 @@ public async Task CreateRequestSubAsync_SubjectWithWhitespace_ThrowsImmediately( var action = async () => await nats.CreateRequestSubAsync(subject, "data"); await action.Should().ThrowAsync().WithMessage("Subject is invalid."); } + + // Tests that verify exceptions are thrown SYNCHRONOUSLY (before any await/iteration) + // This is critical for good user experience - users should get immediate feedback + [Fact] + public void SubscribeAsync_ThrowsSynchronously_BeforeIteration() + { + var nats = new NatsConnection(OptsWithValidation); + + // Exception should be thrown HERE, when calling SubscribeAsync, + // NOT when iterating the result + var exception = Assert.Throws(() => + { + _ = nats.SubscribeAsync("foo bar"); // No await, no iteration + }); + + Assert.Equal("Subject is invalid.", exception.Message); + } + + [Fact] + public void SubscribeAsync_QueueGroup_ThrowsSynchronously_BeforeIteration() + { + var nats = new NatsConnection(OptsWithValidation); + + var exception = Assert.Throws(() => + { + _ = nats.SubscribeAsync("valid.subject", queueGroup: "queue group"); + }); + + Assert.Equal("Queue group is invalid.", exception.Message); + } + + [Fact] + public void SubscribeCoreAsync_ThrowsSynchronously_BeforeAwait() + { + var nats = new NatsConnection(OptsWithValidation); + + // Exception should be thrown HERE, when calling SubscribeCoreAsync, + // NOT when awaiting the result + var exception = Assert.Throws(() => + { + _ = nats.SubscribeCoreAsync("foo bar"); // No await + }); + + Assert.Equal("Subject is invalid.", exception.Message); + } + + [Fact] + public void RequestManyAsync_ThrowsSynchronously_BeforeIteration() + { + var nats = new NatsConnection(OptsWithValidation); + + // Exception should be thrown HERE, when calling RequestManyAsync, + // NOT when iterating the result + var exception = Assert.Throws(() => + { + _ = nats.RequestManyAsync("foo bar", "data"); + }); + + Assert.Equal("Subject is invalid.", exception.Message); + } } diff --git a/tests/NATS.Client.Services.Tests/ServicesTests.cs b/tests/NATS.Client.Services.Tests/ServicesTests.cs index 560cffe05..89215eca5 100644 --- a/tests/NATS.Client.Services.Tests/ServicesTests.cs +++ b/tests/NATS.Client.Services.Tests/ServicesTests.cs @@ -480,4 +480,57 @@ await s2.AddEndpointAsync( Assert.Equal(2, count); } } + + [Theory] + [InlineData("foo bar")] + [InlineData("foo\tbar")] + public async Task AddEndpoint_SubjectWithWhitespace_ThrowsWhenValidationEnabled(string subject) + { + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cancellationToken = cts.Token; + + await using var server = await NatsServerProcess.StartAsync(); + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, SkipSubjectValidation = false }); + var svc = new NatsSvcContext(nats); + + await using var s1 = await svc.AddServiceAsync("s1", "1.0.0", cancellationToken: cancellationToken); + + var exception = await Assert.ThrowsAsync(async () => + { + await s1.AddEndpointAsync( + subject: subject, + handler: _ => ValueTask.CompletedTask, + cancellationToken: cancellationToken); + }); + + Assert.Equal("Subject is invalid.", exception.Message); + } + + [Theory] + [InlineData("queue group")] + [InlineData("queue\tgroup")] + public async Task AddEndpoint_QueueGroupWithWhitespace_ThrowsWhenValidationEnabled(string queueGroup) + { + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var cancellationToken = cts.Token; + + await using var server = await NatsServerProcess.StartAsync(); + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, SkipSubjectValidation = false }); + var svc = new NatsSvcContext(nats); + + await using var s1 = await svc.AddServiceAsync( + new NatsSvcConfig("s1", "1.0.0") { UseQueueGroup = false }, + cancellationToken: cancellationToken); + + var exception = await Assert.ThrowsAsync(async () => + { + await s1.AddEndpointAsync( + subject: "valid.subject", + queueGroup: queueGroup, + handler: _ => ValueTask.CompletedTask, + cancellationToken: cancellationToken); + }); + + Assert.Equal("Queue group is invalid.", exception.Message); + } } From d90b9b6db14d3d9fcf422eadf95ac432ce7a64fc Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Tue, 16 Dec 2025 15:31:11 +0000 Subject: [PATCH 11/13] Refactor and reorder `SubscribeCoreAsync` and `RequestManyInternalAsync` implementations. Update tests to improve clarity and replace unused variable names. Resolve BenchmarkDotNet dependency conflict for `net481`. --- sandbox/MicroBenchmark/MicroBenchmark.csproj | 2 + .../NatsConnection.RequestReply.cs | 40 +++++++++---------- .../NatsConnection.Subscribe.cs | 26 ++++++------ .../NatsConnectionSubjectValidationTests.cs | 10 ++--- 4 files changed, 40 insertions(+), 38 deletions(-) diff --git a/sandbox/MicroBenchmark/MicroBenchmark.csproj b/sandbox/MicroBenchmark/MicroBenchmark.csproj index 97246f8d4..414afa07d 100644 --- a/sandbox/MicroBenchmark/MicroBenchmark.csproj +++ b/sandbox/MicroBenchmark/MicroBenchmark.csproj @@ -13,6 +13,8 @@ + + diff --git a/src/NATS.Client.Core/NatsConnection.RequestReply.cs b/src/NATS.Client.Core/NatsConnection.RequestReply.cs index 4b5a60c49..a3ddc8c86 100644 --- a/src/NATS.Client.Core/NatsConnection.RequestReply.cs +++ b/src/NATS.Client.Core/NatsConnection.RequestReply.cs @@ -128,26 +128,6 @@ public IAsyncEnumerable> RequestManyAsync( return RequestManyInternalAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken); } - private async IAsyncEnumerable> RequestManyInternalAsync( - string subject, - TRequest? data, - NatsHeaders? headers, - INatsSerialize? requestSerializer, - INatsDeserialize? replySerializer, - NatsPubOpts? requestOpts, - NatsSubOpts? replyOpts, - [EnumeratorCancellation] CancellationToken cancellationToken) - { - replyOpts = SetReplyManyOptsDefaults(replyOpts); - await using var sub = await CreateRequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) - .ConfigureAwait(false); - - await foreach (var msg in sub.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false)) - { - yield return msg; - } - } - [SkipLocalsInit] internal static string NewInbox(string prefix) { @@ -173,6 +153,26 @@ static void WriteBuffer(Span buffer, (string prefix, int pfxLen) state) #endif } + private async IAsyncEnumerable> RequestManyInternalAsync( + string subject, + TRequest? data, + NatsHeaders? headers, + INatsSerialize? requestSerializer, + INatsDeserialize? replySerializer, + NatsPubOpts? requestOpts, + NatsSubOpts? replyOpts, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + replyOpts = SetReplyManyOptsDefaults(replyOpts); + await using var sub = await CreateRequestSubAsync(subject, data, headers, requestSerializer, replySerializer, requestOpts, replyOpts, cancellationToken) + .ConfigureAwait(false); + + await foreach (var msg in sub.Msgs.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return msg; + } + } + private NatsSubOpts SetReplyOptsDefaults(NatsSubOpts? replyOpts) { var opts = replyOpts ?? ReplyOptsDefault; diff --git a/src/NATS.Client.Core/NatsConnection.Subscribe.cs b/src/NATS.Client.Core/NatsConnection.Subscribe.cs index 11965dfdc..ede74b4e4 100644 --- a/src/NATS.Client.Core/NatsConnection.Subscribe.cs +++ b/src/NATS.Client.Core/NatsConnection.Subscribe.cs @@ -19,6 +19,19 @@ public IAsyncEnumerable> SubscribeAsync(string subject, string? qu return SubscribeInternalAsync(subject, queueGroup, serializer, opts, cancellationToken); } + /// + public ValueTask> SubscribeCoreAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default) + { + // Validate synchronously so invalid subjects throw immediately + if (!Opts.SkipSubjectValidation) + { + SubjectValidator.ValidateSubject(subject); + SubjectValidator.ValidateQueueGroup(queueGroup); + } + + return SubscribeCoreInternalAsync(subject, queueGroup, serializer, opts, cancellationToken); + } + private async IAsyncEnumerable> SubscribeInternalAsync(string subject, string? queueGroup, INatsDeserialize? serializer, NatsSubOpts? opts, [EnumeratorCancellation] CancellationToken cancellationToken) { serializer ??= Opts.SerializerRegistry.GetDeserializer(); @@ -34,19 +47,6 @@ private async IAsyncEnumerable> SubscribeInternalAsync(string subj } } - /// - public ValueTask> SubscribeCoreAsync(string subject, string? queueGroup = default, INatsDeserialize? serializer = default, NatsSubOpts? opts = default, CancellationToken cancellationToken = default) - { - // Validate synchronously so invalid subjects throw immediately - if (!Opts.SkipSubjectValidation) - { - SubjectValidator.ValidateSubject(subject); - SubjectValidator.ValidateQueueGroup(queueGroup); - } - - return SubscribeCoreInternalAsync(subject, queueGroup, serializer, opts, cancellationToken); - } - private async ValueTask> SubscribeCoreInternalAsync(string subject, string? queueGroup, INatsDeserialize? serializer, NatsSubOpts? opts, CancellationToken cancellationToken) { serializer ??= Opts.SerializerRegistry.GetDeserializer(); diff --git a/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs b/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs index 3abc70fb4..0c38cd6b6 100644 --- a/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/NatsConnectionSubjectValidationTests.cs @@ -69,7 +69,7 @@ public async Task SubscribeAsync_SubjectWithWhitespace_ThrowsImmediately(string await using var nats = new NatsConnection(OptsWithValidation); var action = async () => { - await foreach (var _ in nats.SubscribeAsync(subject)) + await foreach (var unused in nats.SubscribeAsync(subject)) { // Should not reach here } @@ -83,7 +83,7 @@ public async Task SubscribeAsync_EmptySubject_ThrowsImmediately() await using var nats = new NatsConnection(OptsWithValidation); var action = async () => { - await foreach (var _ in nats.SubscribeAsync(string.Empty)) + await foreach (var unused in nats.SubscribeAsync(string.Empty)) { // Should not reach here } @@ -99,7 +99,7 @@ public async Task SubscribeAsync_QueueGroupWithWhitespace_ThrowsImmediately(stri await using var nats = new NatsConnection(OptsWithValidation); var action = async () => { - await foreach (var _ in nats.SubscribeAsync("foo.bar", queueGroup: queueGroup)) + await foreach (var unused in nats.SubscribeAsync("foo.bar", queueGroup: queueGroup)) { // Should not reach here } @@ -113,7 +113,7 @@ public async Task SubscribeAsync_EmptyQueueGroup_ThrowsImmediately() await using var nats = new NatsConnection(OptsWithValidation); var action = async () => { - await foreach (var _ in nats.SubscribeAsync("foo.bar", queueGroup: string.Empty)) + await foreach (var unused in nats.SubscribeAsync("foo.bar", queueGroup: string.Empty)) { // Should not reach here } @@ -177,7 +177,7 @@ public async Task SubscribeAsync_WithSkipValidation_DoesNotThrowOnInvalidSubject Func action = async () => { using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); - await foreach (var _ in nats.SubscribeAsync(subject, cancellationToken: cts.Token)) + await foreach (var unused in nats.SubscribeAsync(subject, cancellationToken: cts.Token)) { // Should not reach here } From 96d1dabe2336d7dc65a1bec25cd4485306c4fbeb Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Tue, 16 Dec 2025 20:15:10 +0000 Subject: [PATCH 12/13] Fix format --- tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs b/tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs index 91f18769c..5270b4991 100644 --- a/tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs +++ b/tests/NATS.Client.CoreUnit.Tests/SubjectValidatorTests.cs @@ -1,4 +1,4 @@ -namespace NATS.Client.CoreUnit.Tests; +namespace NATS.Client.CoreUnit.Tests; public class SubjectValidatorTests { From 69b2118be5df093023967b07c7e1bcb8a20ad4e8 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Wed, 17 Dec 2025 10:00:58 +0000 Subject: [PATCH 13/13] Simplify `SubjectValidator` by removing unnecessary `AsSpan` calls in subject validation methods. --- src/NATS.Client.Core/Internal/SubjectValidator.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/NATS.Client.Core/Internal/SubjectValidator.cs b/src/NATS.Client.Core/Internal/SubjectValidator.cs index 896ed312a..08995f539 100644 --- a/src/NATS.Client.Core/Internal/SubjectValidator.cs +++ b/src/NATS.Client.Core/Internal/SubjectValidator.cs @@ -33,7 +33,7 @@ public static void ValidateSubject(string? subject) ThrowOnBadSubject(); } - ValidateSubjectSpan(subject.AsSpan()); + ValidateSubjectSpan(subject); } /// @@ -51,7 +51,7 @@ public static void ValidateReplyTo(string? replyTo) ThrowOnBadSubject(); } - ValidateSubjectSpan(replyTo.AsSpan()); + ValidateSubjectSpan(replyTo); } }