Default request-reply to Direct mode

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

@@ -17,19 +17,21 @@ internal sealed class ReplyTask<T> : ReplyTaskBase, IDisposable
     private readonly NatsConnection _connection;
     private readonly INatsDeserialize<T> _deserializer;
     private readonly TimeSpan _requestTimeout;
+    private readonly bool _throwIfNoResponders;
     private readonly TaskCompletionSource _tcs;
     private NatsMsg<T> _msg;
     private long _replyBytes;
     private bool _isNoResponders;
 
-    public ReplyTask(ReplyTaskFactory factory, long id, string subject, NatsConnection connection, INatsDeserialize<T> deserializer, TimeSpan requestTimeout)
+    public ReplyTask(ReplyTaskFactory factory, long id, string subject, NatsConnection connection, INatsDeserialize<T> deserializer, TimeSpan requestTimeout, bool throwIfNoResponders)
     {
         _factory = factory;
         _id = id;
         Subject = subject;
         _connection = connection;
         _deserializer = deserializer;
         _requestTimeout = TimeoutValidation.Validate(requestTimeout, nameof(requestTimeout), Timeout.InfiniteTimeSpan);
+        _throwIfNoResponders = throwIfNoResponders;
         _tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
 #if NET9_0_OR_GREATER
         _gate = new System.Threading.Lock();
@@ -63,6 +65,13 @@ await _tcs.Task
             isNoResponders = _isNoResponders;
         }
 
+        // Match the SharedInbox path: when ThrowIfNoResponders is set, a 503 sentinel
+        // surfaces as NatsNoRespondersException rather than an empty message.
+        if (isNoResponders && _throwIfNoResponders)
+        {
+            throw new NatsNoRespondersException();
+        }
+
         // Count only messages actually delivered to the caller. Late replies that arrive
         // after a timeout still hit SetResult, but the user never sees them, so the
         // counters belong here on the success path, not in SetResult. 503 NoResponders
@@ -123,7 +132,7 @@ public ReplyTaskFactory(NatsConnection connection)
         _replies = new ConcurrentDictionary<long, ReplyTaskBase>();
     }
 
-    public ReplyTask<TReply> CreateReplyTask<TReply>(INatsDeserialize<TReply>? deserializer, TimeSpan? requestTimeout)
+    public ReplyTask<TReply> CreateReplyTask<TReply>(INatsDeserialize<TReply>? deserializer, TimeSpan? requestTimeout, bool throwIfNoResponders)
     {
         deserializer ??= _serializerRegistry.GetDeserializer<TReply>();
         var id = Interlocked.Increment(ref _nextId);
@@ -149,7 +158,7 @@ public ReplyTask<TReply> CreateReplyTask<TReply>(INatsDeserialize<TReply>? deser
             subject = _inboxPrefixString + id;
         }
 
-        var rt = new ReplyTask<TReply>(this, id, subject, _connection, deserializer, requestTimeout ?? _requestTimeout);
+        var rt = new ReplyTask<TReply>(this, id, subject, _connection, deserializer, requestTimeout ?? _requestTimeout, throwIfNoResponders);
         _replies.TryAdd(id, rt);
         return rt;
     }

src/NATS.Client.Core/NatsConnection.RequestReply.cs

@@ -7,16 +7,16 @@ namespace NATS.Client.Core;
 
 public partial class NatsConnection
 {
+    // ThrowIfNoResponders is intentionally left unset here so it falls through to the
+    // default derived in SetBaseReplyOptsDefaults (throw unless Direct was set explicitly).
     private static readonly NatsSubOpts ReplyOptsDefault = new NatsSubOpts
     {
         MaxMsgs = 1,
-        ThrowIfNoResponders = true,
     };
 
     private static readonly NatsSubOpts ReplyManyOptsDefault = new NatsSubOpts
     {
         StopOnEmptyMsg = true,
-        ThrowIfNoResponders = true,
     };
 
     /// <inheritdoc />
@@ -53,7 +53,7 @@ public async ValueTask<NatsMsg<TReply>> RequestAsync<TRequest, TReply>(
 
                     if (Opts.RequestReplyMode == NatsRequestReplyMode.Direct)
                     {
-                        using var rt = _replyTaskFactory.CreateReplyTask(replySerializer, replyOpts.Timeout);
+                        using var rt = _replyTaskFactory.CreateReplyTask(replySerializer, replyOpts.Timeout, replyOpts.ThrowIfNoResponders ?? true);
                         requestSerializer ??= Opts.SerializerRegistry.GetSerializer<TRequest>();
                         await PublishAsync(subject, data, headers, rt.Subject, requestSerializer, requestOpts, cancellationToken).ConfigureAwait(false);
                         var msg = await rt.GetResultAsync(cancellationToken).ConfigureAwait(false);
@@ -85,7 +85,7 @@ public async ValueTask<NatsMsg<TReply>> RequestAsync<TRequest, TReply>(
 
             if (Opts.RequestReplyMode == NatsRequestReplyMode.Direct)
             {
-                using var rt = _replyTaskFactory.CreateReplyTask(replySerializer, replyOpts.Timeout);
+                using var rt = _replyTaskFactory.CreateReplyTask(replySerializer, replyOpts.Timeout, replyOpts.ThrowIfNoResponders ?? true);
                 requestSerializer ??= Opts.SerializerRegistry.GetSerializer<TRequest>();
                 await PublishAsync(subject, data, headers, rt.Subject, requestSerializer, requestOpts, cancellationToken).ConfigureAwait(false);
                 return await rt.GetResultAsync(cancellationToken).ConfigureAwait(false);
@@ -214,6 +214,13 @@ private NatsSubOpts SetReplyManyOptsDefaults(NatsSubOpts? replyOpts)
             opts = opts with { StopOnEmptyMsg = true };
         }
 
+        // RequestManyAsync always uses the shared inbox path, so it is unaffected by the
+        // Direct intentional-selection opt-out and keeps throwing on no-responders by default.
+        if (!opts.ThrowIfNoResponders.HasValue)
+        {
+            opts = opts with { ThrowIfNoResponders = true };
+        }
+
         return SetBaseReplyOptsDefaults(opts);
     }
 
@@ -226,7 +233,9 @@ private NatsSubOpts SetBaseReplyOptsDefaults(NatsSubOpts opts)
 
         if (!opts.ThrowIfNoResponders.HasValue)
         {
-            opts = opts with { ThrowIfNoResponders = true };
+            // Throw on no-responders by default, except when Direct was selected explicitly, which
+            // preserves Direct's pre-3.x behavior of returning the sentinel as a message.
+            opts = opts with { ThrowIfNoResponders = !Opts.DirectSetIntentionally };
         }
 
         return opts;

src/NATS.Client.Core/NatsOpts.cs

@@ -38,6 +38,12 @@ public sealed record NatsOpts
         AuthOpts = NatsAuthOpts.Default,
     };
 
+    // Backing fields for RequestReplyMode. _directSetIntentionally records whether Direct was set
+    // explicitly (vs inherited as the default) so no-responders behavior can be preserved across
+    // the 3.x default flip; see the RequestReplyMode init accessor.
+    private readonly NatsRequestReplyMode _requestReplyMode = NatsRequestReplyMode.Direct;
+    private readonly bool _directSetIntentionally;
+
     /// <summary>
     /// NATS server URL to connect to. (default: nats://localhost:4222)
     /// </summary>
@@ -260,8 +266,30 @@ public sealed record NatsOpts
     /// </para>
     /// The <see cref="RequestReplyMode"/> setting determines which mode is used during message exchanges
     /// initiated by <see cref="NatsConnection.RequestAsync{TRequest, TReply}"/> or other related methods.
+    /// Defaults to <see cref="NatsRequestReplyMode.Direct"/>. <see cref="NatsConnection.RequestManyAsync{TRequest, TReply}"/>
+    /// always uses the shared inbox path regardless of this setting.
+    /// <para>
+    /// No-responders handling depends on whether this is set explicitly. Left at its default, request-reply
+    /// throws <see cref="NatsNoRespondersException"/> on a 503 no-responders sentinel. Setting it explicitly to
+    /// <see cref="NatsRequestReplyMode.Direct"/> instead returns the sentinel as a message with
+    /// <see cref="NatsMsg{T}.HasNoResponders"/> set, preserving the behavior of Direct mode before 3.x. Use the
+    /// per-call <see cref="NatsSubOpts.ThrowIfNoResponders"/> to override either way.
+    /// </para>
     /// </remarks>
-    public NatsRequestReplyMode RequestReplyMode { get; init; } = NatsRequestReplyMode.SharedInbox;
+    public NatsRequestReplyMode RequestReplyMode
+    {
+        get => _requestReplyMode;
+        init
+        {
+            _requestReplyMode = value;
+
+            // Explicitly selecting Direct (as opposed to inheriting it as the default) opts out of
+            // throwing on no-responders, matching Direct's pre-3.x behavior. The _requestReplyMode
+            // field initializer bypasses this accessor, so an unset RequestReplyMode keeps the
+            // throwing default.
+            _directSetIntentionally = value == NatsRequestReplyMode.Direct;
+        }
+    }
 
     /// <summary>
     /// Factory for creating socket connections to the NATS server.
@@ -337,6 +365,12 @@ public sealed record NatsOpts
     /// </remarks>
     public bool SuppressSlowConsumerWarnings { get; init; } = false;
 
+    // True when RequestReplyMode was set to Direct explicitly (vs inherited as the default).
+    // NatsConnection.SetBaseReplyOptsDefaults derives the default no-responders behavior from this:
+    // explicit Direct returns the 503 sentinel as a message, everything else throws. A per-call
+    // NatsSubOpts.ThrowIfNoResponders still takes precedence.
+    internal bool DirectSetIntentionally => _directSetIntentionally;
+
     internal NatsUri[] GetSeedUris(bool suppressRandomization = false)
     {
         var urls = Url.Split(',');

src/NATS.Client.JetStream/NatsJSContext.cs

@@ -181,14 +181,16 @@ public async ValueTask<NatsResult<PubAckResponse>> TryPublishAsync<T>(
                 NatsMsg<PubAckResponse> msg;
                 try
                 {
+                    // ThrowIfNoResponders=false: the 503 sentinel comes back as a message so the
+                    // retry loop below can handle it, matching the shared-inbox path.
                     msg = await Connection.RequestAsync<T, PubAckResponse>(
                         subject: subject,
                         data: data,
                         headers: headers,
                         requestSerializer: serializer,
                         replySerializer: NatsJSJsonSerializer<PubAckResponse>.Default,
                         requestOpts: opts,
-                        replyOpts: new NatsSubOpts { Timeout = Opts.RequestTimeout },
+                        replyOpts: new NatsSubOpts { Timeout = Opts.RequestTimeout, ThrowIfNoResponders = false },
                         cancellationToken).ConfigureAwait(false);
                 }
                 catch (NatsNoReplyException)
@@ -432,11 +434,13 @@ internal async ValueTask<NatsResult<NatsJSResponse<TResponse>>> TryJSRequestAsyn
             NatsMsg<NatsJSApiResult<TResponse>> msg;
             try
             {
+                // ThrowIfNoResponders=false: inspect the 503 no-responders sentinel below
+                // rather than letting RequestAsync throw, matching the shared-inbox path.
                 msg = await Connection.RequestAsync<TRequest, NatsJSApiResult<TResponse>>(
                     subject: subject,
                     data: request,
                     headers: null,
-                    replyOpts: new NatsSubOpts { Timeout = Opts.RequestTimeout },
+                    replyOpts: new NatsSubOpts { Timeout = Opts.RequestTimeout, ThrowIfNoResponders = false },
                     requestSerializer: NatsJSJsonSerializer<TRequest>.Default,
                     replySerializer: NatsJSJsonDocumentSerializer<TResponse>.Default,
                     cancellationToken: cancellationToken).ConfigureAwait(false);

tests/NATS.Client.Core.Tests/SubscriptionTest.cs

@@ -31,7 +31,10 @@ public async Task Subscription_periodic_cleanup_test()
     {
         await using var server = await NatsServerProcess.StartAsync();
         var proxy = new NatsProxy(server.Port);
-        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", SubscriptionCleanUpInterval = TimeSpan.FromSeconds(1) });
+
+        // SharedInbox so the connection doesn't open an inbox subscription at connect,
+        // which would make the SUB frame count below never settle at 1.
+        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", SubscriptionCleanUpInterval = TimeSpan.FromSeconds(1), RequestReplyMode = NatsRequestReplyMode.SharedInbox });
 
         async Task Isolator()
         {
@@ -68,7 +71,10 @@ public async Task Subscription_cleanup_on_message_receive_test()
     {
         await using var server = await NatsServerProcess.StartAsync();
         var proxy = new NatsProxy(server.Port);
-        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", SubscriptionCleanUpInterval = TimeSpan.MaxValue });
+
+        // SharedInbox so the connection doesn't open an inbox subscription at connect,
+        // which would make the SUB frame count below never settle at 1.
+        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", SubscriptionCleanUpInterval = TimeSpan.MaxValue, RequestReplyMode = NatsRequestReplyMode.SharedInbox });
 
         async Task Isolator()
         {

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

@@ -204,7 +204,10 @@ public async Task Valid_hmsg_still_works()
         await using var server = new FakeServer(output);
 
         await server.Ready;
-        await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, ConnectTimeout = TimeSpan.FromSeconds(10) });
+
+        // SharedInbox so the connection doesn't open an inbox subscription at connect;
+        // the test injects an HMSG with sid 1, which must map to the "foo" subscription.
+        await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, ConnectTimeout = TimeSpan.FromSeconds(10), RequestReplyMode = NatsRequestReplyMode.SharedInbox });
         await nats.ConnectAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

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

@@ -22,7 +22,10 @@ public async Task Subscription_with_same_subject()
     {
         var nats1 = new NatsConnection(new NatsOpts { Url = _server.Url });
         var proxy = new NatsProxy(_server.Port);
-        var nats2 = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10) });
+
+        // SharedInbox so the connection doesn't open an inbox subscription at connect,
+        // which would add an extra SUB frame to the proxy capture asserted below.
+        var nats2 = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10), RequestReplyMode = NatsRequestReplyMode.SharedInbox });
 
         var sub1 = await nats2.SubscribeCoreAsync<int>("foo.bar");
         var sub2 = await nats2.SubscribeCoreAsync<int>("foo.bar");
@@ -112,7 +115,10 @@ await Retry.Until(
     public async Task Subscription_queue_group()
     {
         var proxy = new NatsProxy(_server.Port);
-        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10) });
+
+        // SharedInbox so the connection doesn't open an inbox subscription at connect,
+        // which would shift the SUB frames asserted by index below.
+        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10), RequestReplyMode = NatsRequestReplyMode.SharedInbox });
         var subject = $"{_server.GetNextId()}.foo";
 
         await using var sub1 = await nats.SubscribeCoreAsync<int>(subject, queueGroup: "group1");
@@ -213,7 +219,10 @@ void Log(string text)
 
         // Use a single server to test multiple scenarios to make test runs more efficient
         var proxy = new NatsProxy(_server.Port);
-        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10) });
+
+        // SharedInbox so the connection doesn't open an inbox subscription at connect,
+        // which would consume sid 1 and break the sid sequence asserted below.
+        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10), RequestReplyMode = NatsRequestReplyMode.SharedInbox });
         var sid = 0;
 
         Log("### Auto-unsubscribe after consuming max-msgs");
@@ -337,7 +346,10 @@ await Retry.Until(
     public async Task Reconnect_with_sub_and_additional_commands()
     {
         var proxy = new NatsProxy(_server.Port);
-        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10) });
+
+        // SharedInbox so the connection doesn't open an inbox subscription at connect,
+        // which would add an extra SUB frame to the proxy capture asserted below.
+        var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10), RequestReplyMode = NatsRequestReplyMode.SharedInbox });
 
         var subject = $"{_server.GetNextId()}.foo";
         var cmdSubject = $"{_server.GetNextId()}.bar";