fix(link): actually bound the server connect with a timeout [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful

This commit is contained in:
justme committed 2026-10-18 03:30:43 +08:00
1 parent 68c7113adf
commit 1a3feddc9d
2 files changed
+104 -3

No files matched your search

+6 -3
View File
@@ -64,6 +64,9 @@ public sealed class FleetLinkOptions
/// <summary>Optional reason for an orderly client shutdown, so the server can distinguish updates from service stops.</summary>
public Func<string>? ShutdownReason { get; init; }
/// <summary>连上服务器的整个过程(TCP、TLS、升级)最多等多久;卡住就放弃,由外层退避重试。</summary>
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<LinkContext>? _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);
@@ -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;
/// <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("prb_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();
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)
{
}
}
}