diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs index 607f93f..e920a57 100644 --- a/backend/fleet/src/Fleet.Link/FleetLink.cs +++ b/backend/fleet/src/Fleet.Link/FleetLink.cs @@ -64,6 +64,9 @@ public sealed class FleetLinkOptions /// Optional reason for an orderly client shutdown, so the server can distinguish updates from service stops. public Func? ShutdownReason { get; init; } + /// 连上服务器的整个过程(TCP、TLS、升级)最多等多久;卡住就放弃,由外层退避重试。 + public TimeSpan ConnectTimeout { get; init; } = TimeSpan.FromSeconds(20); + public TimeSpan HeartbeatInterval { get; init; } = TimeSpan.FromSeconds(10); public int MaxFrameBytes { get; init; } = 16 << 20; @@ -96,7 +99,7 @@ public sealed class FleetLink private readonly INodeLog _log; private volatile RpcSession? _session; private volatile bool _stopped; - private static readonly TimeSpan s_connectTimeout = TimeSpan.FromSeconds(20); + private static readonly TimeSpan s_tcpTimeout = TimeSpan.FromSeconds(4); private int _failedConnects; @@ -179,7 +182,7 @@ public sealed class FleetLink using var handler = new SocketsHttpHandler { ConnectCallback = (context, token) => ConnectTcpAsync(context, attempt, token) }; using var invoker = new HttpMessageInvoker(handler, disposeHandler: false); using CancellationTokenSource timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); - timeout.CancelAfter(s_connectTimeout); + timeout.CancelAfter(_options.ConnectTimeout); try { await socket.ConnectAsync(uri, invoker, timeout.Token); @@ -233,7 +236,7 @@ public sealed class FleetLink socket.Options.KeepAliveInterval = TimeSpan.FromSeconds(15); socket.Options.KeepAliveTimeout = TimeSpan.FromSeconds(20); // 对端 20 秒没有回 Pong 就判定连接已死 var builder = new UriBuilder(_options.Server) { Scheme = _options.Server.Scheme == "https" ? "wss" : "ws", Path = RpcPath }; - await socket.ConnectAsync(builder.Uri, ct); + await ConnectAsync(socket, builder.Uri, ct); await using var transport = new SecureTransport(new WsTransport(socket, _options.MaxFrameBytes)); string? failure = await AuthenticateAsync(transport, ct); diff --git a/backend/fleet/tests/Fleet.Tests/LinkConnectTimeoutTests.cs b/backend/fleet/tests/Fleet.Tests/LinkConnectTimeoutTests.cs new file mode 100644 index 0000000..8108095 --- /dev/null +++ b/backend/fleet/tests/Fleet.Tests/LinkConnectTimeoutTests.cs @@ -0,0 +1,98 @@ +using System.Net; +using System.Net.Sockets; +using Fleet.Core; +using Fleet.Identity; +using Fleet.Link; +using Fleet.Rpc; + +namespace Fleet.Tests; + +/// 面板那头只接住连接、什么都不回(线路把握手吞了、代理卡住):节点要在限定时间内放弃这次连接重试,而不是一直等着。 +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(); + 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("prb_test"), + ConnectTimeout = TimeSpan.FromMilliseconds(500), + Backoff = new Backoff(TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(50)), + }; + var router = new RpcRouter(RpcSide.Node, FleetMethodTable.Table); + var link = new FleetLink(options, router, [], new Outbox(1 << 16), TimeProvider.System, new SilentLog()); + var reasons = new List(); + 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(); + await running; + 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) + { + } + } +}