Gate positive-path test connections with ConnectRetryAsync

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

@@ -1,5 +1,6 @@
 using System.Buffers;
 using System.Text;
+using NATS.Client.TestUtilities2;
 using Synadia.Orbit.Testing.NatsServerProcessManager;
 
 namespace NATS.Client.Core.Tests;
@@ -15,6 +16,7 @@ public async Task Sub_custom_builder_test()
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
 
         var subject = "foo.*";
         var builder = new NatsSubCustomTestBuilder(_output);

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

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Logging;
 using NATS.Client.TestUtilities;
+using NATS.Client.TestUtilities2;
 using Synadia.Orbit.Testing.NatsServerProcessManager;
 
 namespace NATS.Client.Core.Tests;
@@ -16,6 +17,7 @@ public async Task Protocol_parser_under_load(int size)
         var logger = new InMemoryTestLoggerFactory(LogLevel.Error);
         var opts = new NatsOpts { Url = server.Url, LoggerFactory = logger };
         var nats = new NatsConnection(opts);
+        await nats.ConnectRetryAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(120));
 

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

@@ -1,4 +1,5 @@
 using System.Net;
+using NATS.Client.TestUtilities2;
 using Synadia.Orbit.Testing.NatsServerProcessManager;
 
 namespace NATS.Client.Core.Tests;
@@ -35,6 +36,7 @@ public async Task Subscription_periodic_cleanup_test()
         // 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 });
+        await nats.ConnectRetryAsync();
 
         async Task Isolator()
         {
@@ -75,6 +77,7 @@ public async Task Subscription_cleanup_on_message_receive_test()
         // 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 });
+        await nats.ConnectRetryAsync();
 
         async Task Isolator()
         {
@@ -108,6 +111,7 @@ public async Task Auto_unsubscribe_on_max_messages_with_inbox_subscription_test(
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
         var subject = nats.NewInbox();
 
         await using var sub1 = await nats.SubscribeCoreAsync<int>(subject, opts: new NatsSubOpts { MaxMsgs = 1 });
@@ -147,6 +151,7 @@ public async Task Auto_unsubscribe_on_max_messages_test()
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
         const string subject = "foo1";
         const int maxMsgs = 99;
         var opts = new NatsSubOpts { MaxMsgs = maxMsgs };
@@ -177,6 +182,7 @@ public async Task Auto_unsubscribe_on_timeout_test()
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
 
         const string subject = "foo2";
         var opts = new NatsSubOpts { Timeout = TimeSpan.FromSeconds(1) };
@@ -200,6 +206,7 @@ public async Task Auto_unsubscribe_on_idle_timeout_test()
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
         const string subject = "foo3";
         var opts = new NatsSubOpts { IdleTimeout = TimeSpan.FromSeconds(3) };
 
@@ -231,6 +238,7 @@ public async Task Subscribe_with_no_timeout_sentinels_does_not_throw()
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
 
         const string subject = "foo-max";
         var opts = new NatsSubOpts
@@ -266,6 +274,7 @@ public async Task Manual_unsubscribe_test()
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
         const string subject = "foo4";
         await using var sub = await nats.SubscribeCoreAsync<int>(subject);
 
@@ -352,6 +361,7 @@ public async Task Serialization_exceptions()
     {
         await using var server = await NatsServerProcess.StartAsync();
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        await nats.ConnectRetryAsync();
 
         var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
 

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

@@ -1,6 +1,7 @@
 using Microsoft.Extensions.Logging;
 using NATS.Client.Core.Tests;
 using NATS.Client.TestUtilities;
+using NATS.Client.TestUtilities2;
 
 namespace NATS.Client.Core2.Tests;
 
@@ -29,6 +30,7 @@ public async Task Handle_permissions_violation()
             LoggerFactory = logger,
             AuthOpts = new NatsAuthOpts { Username = "u" },
         });
+        await nats.ConnectRetryAsync();
 
         var errors = new List<NatsServerErrorEventArgs>();
 

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

@@ -1,5 +1,6 @@
 using NATS.Client.Core2.Tests;
 using NATS.Client.Serializers.Json;
+using NATS.Client.TestUtilities2;
 
 namespace NATS.Client.Core.Tests;
 
@@ -24,6 +25,7 @@ public async Task Serialize_any_type()
             SerializerRegistry = NatsJsonSerializerRegistry.Default,
         };
         await using var nats = new NatsConnection(natsOpts);
+        await nats.ConnectRetryAsync();
 
         // in local runs server start is taking too long when running
         // the whole suite.
@@ -39,6 +41,7 @@ public async Task Serialize_any_type()
 
         // Default serializer won't work with random types
         await using var nats1 = new NatsConnection(new NatsOpts { Url = _server.Url });
+        await nats1.ConnectRetryAsync();
 
         var exception = await Assert.ThrowsAsync<NatsException>(() => nats1.PublishAsync(
             subject: "would.not.work",

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

@@ -1,4 +1,5 @@
 using NATS.Client.Core2.Tests;
+using NATS.Client.TestUtilities2;
 using Synadia.Orbit.Testing.NatsServerProcessManager;
 
 namespace NATS.Client.Core.Tests;
@@ -17,6 +18,7 @@ public MessageInterfaceTest(NatsServerFixture server)
     public async Task Sub_custom_builder_test()
     {
         await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
+        await nats.ConnectRetryAsync();
 
         var sync = 0;
         var sub = Task.Run(async () =>

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

@@ -25,7 +25,7 @@ public async Task Msg_with_payload_exceeding_max_payload_does_not_oom()
 
         await server.Ready;
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, LoggerFactory = logFactory, ConnectTimeout = TimeSpan.FromSeconds(30) });
-        await nats.ConnectAsync();
+        await nats.ConnectRetryAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
         await nats.SubscribeCoreAsync<int>("foo", cancellationToken: cts.Token);
@@ -53,7 +53,7 @@ public async Task Msg_with_negative_payload_length_does_not_oom()
 
         await server.Ready;
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, LoggerFactory = logFactory, ConnectTimeout = TimeSpan.FromSeconds(10) });
-        await nats.ConnectAsync();
+        await nats.ConnectRetryAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
         await nats.SubscribeCoreAsync<int>("foo", cancellationToken: cts.Token);
@@ -81,7 +81,7 @@ public async Task Hmsg_with_total_less_than_headers_does_not_oom()
 
         await server.Ready;
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, LoggerFactory = logFactory, ConnectTimeout = TimeSpan.FromSeconds(10) });
-        await nats.ConnectAsync();
+        await nats.ConnectRetryAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
         await nats.SubscribeCoreAsync<int>("foo", cancellationToken: cts.Token);
@@ -109,7 +109,7 @@ public async Task Hmsg_with_total_exceeding_max_payload_does_not_oom()
 
         await server.Ready;
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, LoggerFactory = logFactory, ConnectTimeout = TimeSpan.FromSeconds(10) });
-        await nats.ConnectAsync();
+        await nats.ConnectRetryAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
         await nats.SubscribeCoreAsync<int>("foo", cancellationToken: cts.Token);
@@ -138,7 +138,7 @@ public async Task Protocol_violation_exits_read_loop_cleanly()
 
         // MaxReconnectRetry=0 because FakeServer only accepts one connection
         await using var nats = new NatsConnection(new NatsOpts { Url = server.Url, LoggerFactory = logFactory, MaxReconnectRetry = 0 });
-        await nats.ConnectAsync();
+        await nats.ConnectRetryAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
         await nats.SubscribeCoreAsync<int>("foo", cancellationToken: cts.Token);
@@ -208,7 +208,7 @@ public async Task Valid_hmsg_still_works()
         // 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();
+        await nats.ConnectRetryAsync();
 
         using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
         var sub = await nats.SubscribeCoreAsync<byte[]>("foo", cancellationToken: cts.Token);

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

@@ -2,6 +2,7 @@
 using System.Text;
 using NATS.Client.Core2.Tests;
 using NATS.Client.Core2.Tests.ExtraUtils.FrameworkPolyfillExtensions;
+using NATS.Client.TestUtilities2;
 
 namespace NATS.Client.Core.Tests;
 
@@ -21,11 +22,13 @@ public ProtocolTest(ITestOutputHelper output, NatsServerFixture server)
     public async Task Subscription_with_same_subject()
     {
         var nats1 = new NatsConnection(new NatsOpts { Url = _server.Url });
+        await nats1.ConnectRetryAsync();
         var proxy = new NatsProxy(_server.Port);
 
         // 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 });
+        await nats2.ConnectRetryAsync();
 
         var sub1 = await nats2.SubscribeCoreAsync<int>("foo.bar");
         var sub2 = await nats2.SubscribeCoreAsync<int>("foo.bar");
@@ -119,6 +122,7 @@ public async Task Subscription_queue_group()
         // 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 });
+        await nats.ConnectRetryAsync();
         var subject = $"{_server.GetNextId()}.foo";
 
         await using var sub1 = await nats.SubscribeCoreAsync<int>(subject, queueGroup: "group1");
@@ -151,6 +155,7 @@ void Log(string text)
 
         var proxy = new NatsProxy(_server.Port);
         var nats = new NatsConnection(new NatsOpts { Url = $"nats://127.0.0.1:{proxy.Port}", ConnectTimeout = TimeSpan.FromSeconds(10) });
+        await nats.ConnectRetryAsync();
 
         var prefix = $"{_server.GetNextId()}.foo";
 
@@ -223,6 +228,7 @@ void Log(string text)
         // 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 });
+        await nats.ConnectRetryAsync();
         var sid = 0;
 
         Log("### Auto-unsubscribe after consuming max-msgs");
@@ -350,6 +356,7 @@ public async Task Reconnect_with_sub_and_additional_commands()
         // 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 });
+        await nats.ConnectRetryAsync();
 
         var subject = $"{_server.GetNextId()}.foo";
         var cmdSubject = $"{_server.GetNextId()}.bar";
@@ -406,6 +413,7 @@ await Retry.Until(
     public async Task Proactively_reject_payloads_over_the_threshold_set_by_server()
     {
         await using var nats = new NatsConnection(new NatsOpts { Url = _server.Url });
+        await nats.ConnectRetryAsync();
 
         var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));