MoShiro/backend/fleet/tests/Fleet.Tests/LinkConnectTimeoutTests.cs

using System.Net;
using System.Net.Sockets;
using Fleet.Core;
using Fleet.Identity;
using Fleet.Link;
using Fleet.Rpc;

namespace Fleet.Tests;

/// <summary>面板那头只接住连接、什么都不回（线路把握手吞了、代理卡住）：节点要在限定时间内放弃这次连接重试，而不是一直等着。</summary>
public sealed class LinkConnectTimeoutTests
{
    [Fact]
    public async Task ASilentServer_IsGivenUp_AfterTheConnectTimeout_AndRetried()
    {
        using var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var accepted = new List<TcpClient>();
        using var acceptLoop = new CancellationTokenSource();
        Task accepting = Task.Run(async () =>
        {
            try
            {
                while (true)
                {
                    accepted.Add(await listener.AcceptTcpClientAsync(acceptLoop.Token));
                }
            }
            catch (OperationCanceledException)
            {
                // 测试结束
            }
        });

        var options = new FleetLinkOptions
        {
            Server = new Uri($"http://127.0.0.1:{((IPEndPoint)listener.LocalEndpoint).Port}"),
            Identity = NodeIdentity.Generate(),
            NodeId = new NodeId("dlr_test"),
            ConnectTimeout = TimeSpan.FromMilliseconds(500),
            Backoff = new Backoff(TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50)),
        };
        var router = new RpcRouter<LinkContext>(RpcSide.Node, FleetMethodTable.Table);
        var link = new FleetLink(options, router, [], new Outbox(1 << 16), TimeProvider.System, new SilentLog());
        var reasons = new List<string>();
        link.OnDisconnected = reason =>
        {
            lock (reasons)
            {
                reasons.Add(reason);
            }
        };

        using var stop = new CancellationTokenSource();
        Task running = link.RunAsync(stop.Token);
        try
        {
            using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(10));
            while (true)
            {
                lock (reasons)
                {
                    if (reasons.Count >= 2)
                    {
                        break;
                    }
                }

                await Task.Delay(50, deadline.Token);
            }

            lock (reasons)
            {
                Assert.All(reasons, reason => Assert.Contains("timed out", reason, StringComparison.OrdinalIgnoreCase));
            }

            Assert.True(accepted.Count >= 2, "the second attempt connects again");
        }
        finally
        {
            await stop.CancelAsync();
            try
            {
                await running;
            }
            catch (OperationCanceledException) when (stop.IsCancellationRequested)
            {
                // Canceling the retry delay ends RunAsync with the requested cancellation.
            }

            await acceptLoop.CancelAsync();
            await accepting;
            foreach (TcpClient client in accepted)
            {
                client.Dispose();
            }
        }
    }

    private sealed class SilentLog : INodeLog
    {
        public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields)
        {
        }
    }
}