natsdotnet/tests/NATS.Server.Tests/Routes/RouteConnectionTests.cs

using System.Net;
using System.Net.Sockets;
using System.Text;
using Microsoft.Extensions.Logging.Abstractions;
using NATS.Client.Core;
using NATS.Server.Configuration;
using NATS.Server.Routes;
using NATS.Server.Subscriptions;

namespace NATS.Server.Tests.Routes;

/// <summary>
/// Tests for route connection establishment, handshake, reconnection, and lifecycle.
/// Ported from Go: server/routes_test.go.
/// </summary>
public class RouteConnectionTests
{
    // -- Helpers --

    private static async Task<(NatsServer Server, CancellationTokenSource Cts)> StartServerAsync(
        NatsOptions opts)
    {
        var server = new NatsServer(opts, NullLoggerFactory.Instance);
        var cts = new CancellationTokenSource();
        _ = server.StartAsync(cts.Token);
        await server.WaitForReadyAsync();
        return (server, cts);
    }

    private static NatsOptions MakeClusterOpts(string? clusterName = null, string? seed = null)
    {
        return new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Cluster = new ClusterOptions
            {
                Name = clusterName ?? Guid.NewGuid().ToString("N"),
                Host = "127.0.0.1",
                Port = 0,
                Routes = seed is null ? [] : [seed],
            },
        };
    }

    private static async Task WaitForRouteFormation(NatsServer a, NatsServer b, int timeoutSeconds = 5)
    {
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(timeoutSeconds));
        while (!timeout.IsCancellationRequested &&
               (Interlocked.Read(ref a.Stats.Routes) == 0 ||
                Interlocked.Read(ref b.Stats.Routes) == 0))
        {
            await Task.Delay(50, timeout.Token).ContinueWith(_ => { }, TaskScheduler.Default);
        }
    }

    private static async Task WaitForCondition(Func<bool> predicate, int timeoutMs = 5000)
    {
        using var cts = new CancellationTokenSource(timeoutMs);
        while (!cts.IsCancellationRequested)
        {
            if (predicate())
                return;
            await Task.Delay(20, cts.Token).ContinueWith(_ => { }, TaskScheduler.Default);
        }

        throw new TimeoutException("Condition not met.");
    }

    private static async Task DisposeServers(params (NatsServer Server, CancellationTokenSource Cts)[] servers)
    {
        foreach (var (server, cts) in servers)
        {
            await cts.CancelAsync();
            server.Dispose();
            cts.Dispose();
        }
    }

    // -- Tests --

    // Go: TestSeedSolicitWorks server/routes_test.go:365
    [Fact]
    public async Task Seed_solicit_establishes_route_connection()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var optsA = MakeClusterOpts(cluster);
        var a = await StartServerAsync(optsA);

        var optsB = MakeClusterOpts(cluster, a.Server.ClusterListen!);
        var b = await StartServerAsync(optsB);

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
            Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestSeedSolicitWorks server/routes_test.go:365 (message delivery)
    [Fact]
    public async Task Seed_solicit_delivers_messages_across_route()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);

            await using var subscriber = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{a.Server.Port}",
            });
            await subscriber.ConnectAsync();
            await using var sub = await subscriber.SubscribeCoreAsync<string>("foo");
            await subscriber.PingAsync();

            await WaitForCondition(() => b.Server.HasRemoteInterest("foo"));

            await using var publisher = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{b.Server.Port}",
            });
            await publisher.ConnectAsync();
            await publisher.PublishAsync("foo", "Hello");

            using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
            var msg = await sub.Msgs.ReadAsync(receiveTimeout.Token);
            msg.Data.ShouldBe("Hello");
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestChainedSolicitWorks server/routes_test.go:481
    [Fact]
    public async Task Three_servers_form_full_mesh_via_seed()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));
        var c = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            await WaitForRouteFormation(a.Server, c.Server);

            // Verify message delivery across the 3-node cluster
            await using var subscriber = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{a.Server.Port}",
            });
            await subscriber.ConnectAsync();
            await using var sub = await subscriber.SubscribeCoreAsync<string>("chain.test");
            await subscriber.PingAsync();

            await WaitForCondition(() => c.Server.HasRemoteInterest("chain.test"));

            await using var publisher = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{c.Server.Port}",
            });
            await publisher.ConnectAsync();
            await publisher.PublishAsync("chain.test", "chained");

            using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
            var msg = await sub.Msgs.ReadAsync(receiveTimeout.Token);
            msg.Data.ShouldBe("chained");
        }
        finally
        {
            await DisposeServers(a, b, c);
        }
    }

    // Go: TestRoutesToEachOther server/routes_test.go:759
    [Fact]
    public async Task Mutual_route_solicitation_resolves_to_single_route()
    {
        // Both servers point routes at each other, should still form a single cluster
        var cluster = Guid.NewGuid().ToString("N");
        var optsA = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Cluster = new ClusterOptions
            {
                Name = cluster,
                Host = "127.0.0.1",
                Port = 0,
            },
        };
        var a = await StartServerAsync(optsA);

        var optsB = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Cluster = new ClusterOptions
            {
                Name = cluster,
                Host = "127.0.0.1",
                Port = 0,
                Routes = [a.Server.ClusterListen!],
            },
        };
        var b = await StartServerAsync(optsB);

        // Also point A's routes at B (mutual solicitation)
        // We can't change routes dynamically, so we just verify that the route forms properly
        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
            Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteRTT server/routes_test.go:1203
    [Fact]
    public async Task Route_stats_tracked_after_formation()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
            Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteConfig server/routes_test.go:86
    [Fact]
    public void Cluster_options_have_correct_defaults()
    {
        var opts = new ClusterOptions();
        opts.Port.ShouldBe(6222);
        opts.Host.ShouldBe("0.0.0.0");
        opts.PoolSize.ShouldBe(3);
        opts.Routes.ShouldNotBeNull();
        opts.Routes.Count.ShouldBe(0);
    }

    // Go: TestRouteConfig server/routes_test.go:86
    [Fact]
    public void Cluster_options_can_be_configured()
    {
        var opts = new ClusterOptions
        {
            Name = "test-cluster",
            Host = "127.0.0.1",
            Port = 7244,
            PoolSize = 5,
            Routes = ["127.0.0.1:7245", "127.0.0.1:7246"],
        };

        opts.Name.ShouldBe("test-cluster");
        opts.Port.ShouldBe(7244);
        opts.PoolSize.ShouldBe(5);
        opts.Routes.Count.ShouldBe(2);
    }

    // Go: TestRouteReconnectExponentialBackoff server/routes_test.go:1758
    [Fact]
    public async Task Route_reconnects_after_peer_restart()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var clusterListenA = a.Server.ClusterListen!;

        var b = await StartServerAsync(MakeClusterOpts(cluster, clusterListenA));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);

            // Stop server B
            await b.Cts.CancelAsync();
            b.Server.Dispose();
            b.Cts.Dispose();

            // Wait for A to notice B is gone
            await WaitForCondition(() => Interlocked.Read(ref a.Server.Stats.Routes) == 0, 5000);

            // Restart B
            b = await StartServerAsync(MakeClusterOpts(cluster, clusterListenA));
            await WaitForRouteFormation(a.Server, b.Server);

            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
            Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteReconnectExponentialBackoff server/routes_test.go:1758
    [Fact]
    public async Task Route_reconnects_and_resumes_message_forwarding()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var clusterListenA = a.Server.ClusterListen!;
        var b = await StartServerAsync(MakeClusterOpts(cluster, clusterListenA));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);

            // Stop and restart B
            await b.Cts.CancelAsync();
            b.Server.Dispose();
            b.Cts.Dispose();

            b = await StartServerAsync(MakeClusterOpts(cluster, clusterListenA));
            await WaitForRouteFormation(a.Server, b.Server);

            // Verify forwarding works after reconnect
            await using var subscriber = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{a.Server.Port}",
            });
            await subscriber.ConnectAsync();
            await using var sub = await subscriber.SubscribeCoreAsync<string>("reconnect.test");
            await subscriber.PingAsync();

            await WaitForCondition(() => b.Server.HasRemoteInterest("reconnect.test"));

            await using var publisher = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{b.Server.Port}",
            });
            await publisher.ConnectAsync();
            await publisher.PublishAsync("reconnect.test", "after-restart");

            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
            var msg = await sub.Msgs.ReadAsync(timeout.Token);
            msg.Data.ShouldBe("after-restart");
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRoutePool server/routes_test.go:1966
    [Fact]
    public async Task Route_pool_establishes_configured_number_of_connections()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var optsA = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Cluster = new ClusterOptions
            {
                Name = cluster,
                Host = "127.0.0.1",
                Port = 0,
                PoolSize = 3,
            },
        };
        var a = await StartServerAsync(optsA);

        var optsB = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Cluster = new ClusterOptions
            {
                Name = cluster,
                Host = "127.0.0.1",
                Port = 0,
                PoolSize = 3,
                Routes = [a.Server.ClusterListen!],
            },
        };
        var b = await StartServerAsync(optsB);

        try
        {
            await WaitForCondition(() => Interlocked.Read(ref a.Server.Stats.Routes) >= 3, 5000);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThanOrEqualTo(3);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRoutePoolSizeDifferentOnEachServer server/routes_test.go:2254
    [Fact]
    public async Task Route_pool_size_of_one_still_forwards_messages()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var optsA = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Cluster = new ClusterOptions
            {
                Name = cluster,
                Host = "127.0.0.1",
                Port = 0,
                PoolSize = 1,
            },
        };
        var a = await StartServerAsync(optsA);

        var optsB = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
            Cluster = new ClusterOptions
            {
                Name = cluster,
                Host = "127.0.0.1",
                Port = 0,
                PoolSize = 1,
                Routes = [a.Server.ClusterListen!],
            },
        };
        var b = await StartServerAsync(optsB);

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);

            await using var subscriber = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{a.Server.Port}",
            });
            await subscriber.ConnectAsync();
            await using var sub = await subscriber.SubscribeCoreAsync<string>("pool.one");
            await subscriber.PingAsync();

            await WaitForCondition(() => b.Server.HasRemoteInterest("pool.one"));

            await using var publisher = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{b.Server.Port}",
            });
            await publisher.ConnectAsync();
            await publisher.PublishAsync("pool.one", "single-pool");

            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
            var msg = await sub.Msgs.ReadAsync(timeout.Token);
            msg.Data.ShouldBe("single-pool");
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteHandshake (low-level handshake)
    [Fact]
    public async Task Route_connection_outbound_handshake_exchanges_server_ids()
    {
        using var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;

        using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
        using var routeSocket = await listener.AcceptSocketAsync();
        await using var route = new RouteConnection(routeSocket);
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));

        var handshakeTask = route.PerformOutboundHandshakeAsync("LOCAL_SERVER", timeout.Token);

        var received = await ReadLineAsync(remoteSocket, timeout.Token);
        received.ShouldBe("ROUTE LOCAL_SERVER");

        await WriteLineAsync(remoteSocket, "ROUTE REMOTE_SERVER", timeout.Token);
        await handshakeTask;

        route.RemoteServerId.ShouldBe("REMOTE_SERVER");
    }

    // Go: TestRouteHandshake inbound direction
    [Fact]
    public async Task Route_connection_inbound_handshake_exchanges_server_ids()
    {
        using var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;

        using var remoteSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        await remoteSocket.ConnectAsync(IPAddress.Loopback, port);
        using var routeSocket = await listener.AcceptSocketAsync();
        await using var route = new RouteConnection(routeSocket);
        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));

        var handshakeTask = route.PerformInboundHandshakeAsync("LOCAL_SERVER", timeout.Token);

        await WriteLineAsync(remoteSocket, "ROUTE REMOTE_SERVER", timeout.Token);
        await handshakeTask;

        var received = await ReadLineAsync(remoteSocket, timeout.Token);
        received.ShouldBe("ROUTE LOCAL_SERVER");
        route.RemoteServerId.ShouldBe("REMOTE_SERVER");
    }

    // Go: TestRouteNoCrashOnAddingSubToRoute server/routes_test.go:1131
    [Fact]
    public async Task Many_subscriptions_propagate_across_route()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);

            await using var nc = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{a.Server.Port}",
            });
            await nc.ConnectAsync();

            var subs = new List<IAsyncDisposable>();
            for (var i = 0; i < 50; i++)
            {
                var sub = await nc.SubscribeCoreAsync<string>($"many.subs.{i}");
                subs.Add(sub);
            }

            await nc.PingAsync();

            // Verify at least some interest propagated
            await WaitForCondition(() => b.Server.HasRemoteInterest("many.subs.0"));
            await WaitForCondition(() => b.Server.HasRemoteInterest("many.subs.49"));

            b.Server.HasRemoteInterest("many.subs.0").ShouldBeTrue();
            b.Server.HasRemoteInterest("many.subs.49").ShouldBeTrue();

            foreach (var sub in subs)
                await sub.DisposeAsync();
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteSendLocalSubsWithLowMaxPending server/routes_test.go:1098
    [Fact]
    public async Task Subscriptions_propagate_with_many_subscribers()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);

            await using var nc = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{a.Server.Port}",
            });
            await nc.ConnectAsync();

            var subs = new List<IAsyncDisposable>();
            for (var i = 0; i < 20; i++)
            {
                var sub = await nc.SubscribeCoreAsync<string>($"local.sub.{i}");
                subs.Add(sub);
            }

            await nc.PingAsync();

            await WaitForCondition(() => b.Server.HasRemoteInterest("local.sub.0"), 10000);
            await WaitForCondition(() => b.Server.HasRemoteInterest("local.sub.19"), 10000);

            b.Server.HasRemoteInterest("local.sub.0").ShouldBeTrue();
            b.Server.HasRemoteInterest("local.sub.19").ShouldBeTrue();

            foreach (var sub in subs)
                await sub.DisposeAsync();
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteCloseTLSConnection server/routes_test.go:1290 (basic close test, no TLS)
    [Fact]
    public async Task Route_connection_close_decrements_stats()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);

            // Stop B - A's route count should drop
            await b.Cts.CancelAsync();
            b.Server.Dispose();
            b.Cts.Dispose();

            await WaitForCondition(() => Interlocked.Read(ref a.Server.Stats.Routes) == 0, 5000);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBe(0);
        }
        finally
        {
            await a.Cts.CancelAsync();
            a.Server.Dispose();
            a.Cts.Dispose();
        }
    }

    // Go: TestRouteDuplicateServerName server/routes_test.go:1444
    [Fact]
    public async Task Cluster_with_different_server_ids_form_routes()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var optsA = MakeClusterOpts(cluster);
        optsA.ServerName = "server-alpha";
        var a = await StartServerAsync(optsA);

        var optsB = MakeClusterOpts(cluster, a.Server.ClusterListen!);
        optsB.ServerName = "server-beta";
        var b = await StartServerAsync(optsB);

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            a.Server.ServerName.ShouldBe("server-alpha");
            b.Server.ServerName.ShouldBe("server-beta");
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteIPResolutionAndRouteToSelf server/routes_test.go:1415
    [Fact]
    public void Server_without_cluster_has_null_cluster_listen()
    {
        var opts = new NatsOptions
        {
            Host = "127.0.0.1",
            Port = 0,
        };
        var server = new NatsServer(opts, NullLoggerFactory.Instance);
        server.ClusterListen.ShouldBeNull();
        server.Dispose();
    }

    // Go: TestBlockedShutdownOnRouteAcceptLoopFailure server/routes_test.go:634
    [Fact]
    public async Task Server_with_cluster_can_be_shut_down_cleanly()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));

        await a.Cts.CancelAsync();
        a.Server.Dispose();
        a.Cts.Dispose();
        // If we get here without timeout, shutdown worked properly
    }

    // Go: TestRoutePings server/routes_test.go:4376
    [Fact]
    public async Task Route_stays_alive_with_periodic_activity()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);

            // Route stays alive after some time
            await Task.Delay(500);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
            Interlocked.Read(ref b.Server.Stats.Routes).ShouldBeGreaterThan(0);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestServerRoutesWithClients server/routes_test.go:216
    [Fact]
    public async Task Multiple_messages_flow_across_route()
    {
        var cluster = Guid.NewGuid().ToString("N");
        var optsA = MakeClusterOpts(cluster);
        optsA.Cluster!.PoolSize = 1;
        var a = await StartServerAsync(optsA);
        var optsB = MakeClusterOpts(cluster, a.Server.ClusterListen!);
        optsB.Cluster!.PoolSize = 1;
        var b = await StartServerAsync(optsB);

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);

            await using var subscriber = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{a.Server.Port}",
            });
            await subscriber.ConnectAsync();
            await using var sub = await subscriber.SubscribeCoreAsync<string>("multi.msg");
            await subscriber.PingAsync();
            await WaitForCondition(() => b.Server.HasRemoteInterest("multi.msg"));

            await using var publisher = new NatsConnection(new NatsOpts
            {
                Url = $"nats://127.0.0.1:{b.Server.Port}",
            });
            await publisher.ConnectAsync();

            for (var i = 0; i < 10; i++)
            {
                await publisher.PublishAsync("multi.msg", $"msg-{i}");
            }

            var received = new HashSet<string>();
            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
            for (var i = 0; i < 10; i++)
            {
                var msg = await sub.Msgs.ReadAsync(timeout.Token);
                received.Add(msg.Data!);
            }

            received.Count.ShouldBe(10);
            for (var i = 0; i < 10; i++)
                received.ShouldContain($"msg-{i}");
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // Go: TestRouteClusterNameConflictBetweenStaticAndDynamic server/routes_test.go:1374
    [Fact]
    public async Task Route_with_named_cluster_forms_correctly()
    {
        var cluster = "named-cluster-test";
        var a = await StartServerAsync(MakeClusterOpts(cluster));
        var b = await StartServerAsync(MakeClusterOpts(cluster, a.Server.ClusterListen!));

        try
        {
            await WaitForRouteFormation(a.Server, b.Server);
            Interlocked.Read(ref a.Server.Stats.Routes).ShouldBeGreaterThan(0);
        }
        finally
        {
            await DisposeServers(a, b);
        }
    }

    // -- Wire-level helpers --

    private static async Task<string> ReadLineAsync(Socket socket, CancellationToken ct)
    {
        var bytes = new List<byte>(64);
        var single = new byte[1];
        while (true)
        {
            var read = await socket.ReceiveAsync(single, SocketFlags.None, ct);
            if (read == 0) break;
            if (single[0] == (byte)'\n') break;
            if (single[0] != (byte)'\r')
                bytes.Add(single[0]);
        }

        return Encoding.ASCII.GetString([.. bytes]);
    }

    private static Task WriteLineAsync(Socket socket, string line, CancellationToken ct)
        => socket.SendAsync(Encoding.ASCII.GetBytes($"{line}\r\n"), SocketFlags.None, ct).AsTask();
}