diff --git a/backend/fleet/src/Fleet.Core/Bounded.cs b/backend/fleet/src/Fleet.Core/Bounded.cs new file mode 100644 index 0000000..277e98a --- /dev/null +++ b/backend/fleet/src/Fleet.Core/Bounded.cs @@ -0,0 +1,32 @@ +using System.Net.Sockets; +using System.Net.WebSockets; + +namespace Fleet.Core; + +/// +/// 收尾类的等待（关闭、释放、等任务结束）一律要有上限：对端没了、原生层不回调、某个处理器卡住时， +/// 没有上限的等待会让状态机永远停在「已连接」，既不重连也不报错。超时就放弃等待并交给调用方记日志；会话结束时的常见异常按正常噪声吞掉。 +/// +public static class Bounded +{ + /// 收尾的一步最多等这么久。 + public static readonly TimeSpan Teardown = TimeSpan.FromSeconds(5); + + /// 等 最多 ；超时返回 false（任务继续在后台收尾，不再等它），否则返回 true。 + public static async Task WaitAsync(Task work, TimeSpan limit, TimeProvider time) + { + try + { + await work.WaitAsync(limit, time); + return true; + } + catch (TimeoutException) + { + return false; + } + catch (Exception ex) when (ex is OperationCanceledException or IOException or ObjectDisposedException or InvalidOperationException or SocketException or WebSocketException) + { + return true; // 会话结束时的正常噪声：这一步已经结束了 + } + } +} diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs index e920a57..8b61b19 100644 --- a/backend/fleet/src/Fleet.Link/FleetLink.cs +++ b/backend/fleet/src/Fleet.Link/FleetLink.cs @@ -279,6 +279,7 @@ public sealed class FleetLink Task cancellation = Task.Delay(Timeout.Infinite, ct); Task finished = await Task.WhenAny(run, pump, heartbeat, cancellation); string finishedReason = DescribeFinishedTask(finished, run, pump, heartbeat, session.TransportFailure); + _session = null; // 链路已经结束：上报方不能再往这条会话上写，收尾卡住时也一样 if (ct.IsCancellationRequested) { try @@ -295,15 +296,15 @@ public sealed class FleetLink } await sessionCts.CancelAsync(); - await Task.WhenAll(Quiet(run), Quiet(pump), Quiet(heartbeat)); + await Bounded.WaitAsync(Task.WhenAll(Quiet(run), Quiet(pump), Quiet(heartbeat)), Bounded.Teardown, _time); return finishedReason; } finally { _session = null; await sessionCts.CancelAsync(); - await Quiet(run); - await session.DisposeAsync(); + await Bounded.WaitAsync(Quiet(run), Bounded.Teardown, _time); + await Bounded.WaitAsync(session.DisposeAsync().AsTask(), Bounded.Teardown, _time); } } diff --git a/backend/fleet/src/Fleet.Link/Transports.cs b/backend/fleet/src/Fleet.Link/Transports.cs index 14d864f..3426fb0 100644 --- a/backend/fleet/src/Fleet.Link/Transports.cs +++ b/backend/fleet/src/Fleet.Link/Transports.cs @@ -118,6 +118,7 @@ public sealed class WsTransport : IRpcTransport public sealed class TlsFrameTransport : IRpcTransport { private const int HeaderBytes = 5; + private static readonly TimeSpan s_sendTimeout = TimeSpan.FromSeconds(20); private readonly Stream _stream; private readonly int _maxFrameBytes; @@ -139,16 +140,29 @@ public sealed class TlsFrameTransport : IRpcTransport byte[] header = new byte[HeaderBytes]; BinaryPrimitives.WriteUInt32LittleEndian(header, (uint)frame.Data.Length); header[4] = (byte)frame.Kind; - await _sendLock.WaitAsync(ct); + // 对端黑洞时写会一直挂着并占着发送锁，心跳、时钟同步和重连都跟着卡住：超时就掐掉这条连接 + using var timeout = new CancellationTokenSource(s_sendTimeout); + using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token); + bool lockTaken = false; try { - await _stream.WriteAsync(header, ct); - await _stream.WriteAsync(frame.Data, ct); - await _stream.FlushAsync(ct); + await _sendLock.WaitAsync(linked.Token); + lockTaken = true; + await _stream.WriteAsync(header, linked.Token); + await _stream.WriteAsync(frame.Data, linked.Token); + await _stream.FlushAsync(linked.Token); + } + catch (OperationCanceledException) when (timeout.IsCancellationRequested && !ct.IsCancellationRequested) + { + await _stream.DisposeAsync(); + throw new TimeoutException($"peer send timed out after {s_sendTimeout.TotalSeconds:0} seconds"); } finally { - _sendLock.Release(); + if (lockTaken) + { + _sendLock.Release(); + } } } diff --git a/backend/fleet/src/Fleet.Rpc/RpcSession.cs b/backend/fleet/src/Fleet.Rpc/RpcSession.cs index 918807a..c9c0e78 100644 --- a/backend/fleet/src/Fleet.Rpc/RpcSession.cs +++ b/backend/fleet/src/Fleet.Rpc/RpcSession.cs @@ -50,6 +50,7 @@ public sealed class RpcSession : IAsyncDisposable private readonly IRpcTransport _transport; private readonly RpcRouter _router; private readonly TContext _context; + private static readonly TimeSpan s_dispatcherGrace = TimeSpan.FromSeconds(5); private readonly TimeProvider _time; private readonly RpcSessionOptions _options; private readonly ConcurrentDictionary> _pending = new(); @@ -135,14 +136,8 @@ public sealed class RpcSession : IAsyncDisposable } await linked.CancelAsync(); - try - { - await dispatcher; - } - catch (OperationCanceledException) - { - // 正常结束 - } + // 某个入站处理器卡住了也不能拖住整个会话的结束（链路据此重连）：它随取消令牌自己收尾 + await Bounded.WaitAsync(dispatcher, s_dispatcherGrace, _time); } } diff --git a/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs b/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs index 4239a9e..949566a 100644 --- a/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs +++ b/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs @@ -16,6 +16,9 @@ public sealed class FleetServerOptions public TimeSpan RequestTimeout { get; set; } = TimeSpan.FromSeconds(15); + /// 处理节点的一条消息（心跳、hello、收件箱）最多用这么久：钩子卡在数据库或外部查询里时，不能让这条连接停在「在线」而收不到任何心跳超时。超过就断开，节点按退避重连。 + public TimeSpan HandlerTimeout { get; set; } = TimeSpan.FromSeconds(30); + /// 节点违反协议（方法没有声明、方向、类型或角色不对）的次数上限；超过就断开这条连接。 public int MaxViolations { get; set; } = 20; } diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs index 9be6688..333d9a8 100644 --- a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs @@ -34,12 +34,20 @@ internal sealed partial class NodeConnection { if (TryParse(frame.Data, out RpcMessage? message)) { - await HandleMessageAsync(message, ct); + if (!await Bounded.WaitAsync(HandleMessageAsync(message, ct), _host.Options.HandlerTimeout, _host.Time)) + { + _closeReason ??= "handler timeout"; + return; + } } } else { - await HandleBinaryAsync(frame.Data, ct); + if (!await Bounded.WaitAsync(HandleBinaryAsync(frame.Data, ct), _host.Options.HandlerTimeout, _host.Time)) + { + _closeReason ??= "handler timeout"; + return; + } } } } diff --git a/backend/fleet/tests/Fleet.Tests/BoundedTests.cs b/backend/fleet/tests/Fleet.Tests/BoundedTests.cs new file mode 100644 index 0000000..ced8933 --- /dev/null +++ b/backend/fleet/tests/Fleet.Tests/BoundedTests.cs @@ -0,0 +1,28 @@ +using Fleet.Core; + +namespace Fleet.Tests; + +public sealed class BoundedTests +{ + [Fact] + public async Task WaitAsync_ReturnsFalse_WhenTheWorkNeverFinishes() + { + var never = new TaskCompletionSource(); + Assert.False(await Bounded.WaitAsync(never.Task, TimeSpan.FromMilliseconds(100), TimeProvider.System)); + } + + [Fact] + public async Task WaitAsync_ReturnsTrue_WhenTheWorkFinishes_OrFailsWithSessionNoise() + { + Assert.True(await Bounded.WaitAsync(Task.CompletedTask, TimeSpan.FromSeconds(5), TimeProvider.System)); + Assert.True(await Bounded.WaitAsync(Task.FromException(new IOException("closed")), TimeSpan.FromSeconds(5), TimeProvider.System)); + Assert.True(await Bounded.WaitAsync(Task.FromException(new ObjectDisposedException("x")), TimeSpan.FromSeconds(5), TimeProvider.System)); + Assert.True(await Bounded.WaitAsync(Task.FromCanceled(new CancellationToken(true)), TimeSpan.FromSeconds(5), TimeProvider.System)); + } + + [Fact] + public async Task WaitAsync_DoesNotHideRealFailures() + { + await Assert.ThrowsAsync(() => Bounded.WaitAsync(Task.FromException(new FormatException()), TimeSpan.FromSeconds(5), TimeProvider.System)); + } +} diff --git a/backend/fleet/tests/Fleet.Tests/LinkTests.cs b/backend/fleet/tests/Fleet.Tests/LinkTests.cs index 284ea3b..e8d45f9 100644 --- a/backend/fleet/tests/Fleet.Tests/LinkTests.cs +++ b/backend/fleet/tests/Fleet.Tests/LinkTests.cs @@ -275,6 +275,27 @@ public sealed class LinkTests await run.WaitAsync(TimeSpan.FromSeconds(5)); } + [Fact] + public async Task Session_RunEnds_EvenWhenAnInboundHandlerIsStuck() + { + (TlsFrameTransport left, TlsFrameTransport right) = await PairAsync(); + MethodTable table = TestTables.Of(TestTables.Req("right.stuck")); + var stuck = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); // 不理会取消令牌：模拟卡在原生层的处理器 + var rightRouter = new RpcRouter(RpcSide.Listener, table).On("right.stuck", LinkJson.Default.Number, LinkJson.Default.Number, (_, _, _) => new ValueTask>(stuck.Task)); + await using var leftSession = new RpcSession(left, new RpcRouter(RpcSide.Dialer, table), "left", TimeProvider.System); + await using var rightSession = new RpcSession(right, rightRouter, "right", TimeProvider.System); + Task leftRun = leftSession.RunAsync(default); + Task rightRun = rightSession.RunAsync(default); + + _ = leftSession.RequestAsync("right.stuck", new Number { Value = 1 }, LinkJson.Default.Number, LinkJson.Default.Number, default, TimeSpan.FromSeconds(30)).AsTask(); + await Task.Delay(300); // 请求到达并进入处理器 + await leftSession.CloseAsync("bye", default); + + await rightRun.WaitAsync(TimeSpan.FromSeconds(10)); // 以前这里会一直等到处理器返回 + stuck.TrySetResult(Result.Ok(new Number { Value = 0 })); + await leftRun.WaitAsync(TimeSpan.FromSeconds(5)); + } + // ───────────────────────── 点对点握手 ───────────────────────── private sealed class PeerFixture : IAsyncDisposable diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs index a7c4b28..1ae62e1 100644 --- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs +++ b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs @@ -27,9 +27,6 @@ public sealed class AgentConnection : IDisposable /// Agent 的任何帧（时钟回复、任务集）都算存活；比时钟同步间隔长得多，容得下差线路上的黑洞，但不会让一个 Agent 已经丢掉的会话永远挂着。 private static readonly TimeSpan s_idleLimit = TimeSpan.FromSeconds(120); - /// 收尾的每一步最多等这么久：QUIC 连接或控制连接的关闭卡在原生层时，不能让整个重拨循环跟着卡住。 - private static readonly TimeSpan s_teardownLimit = TimeSpan.FromSeconds(5); - /// 一直在发测量报文、却这么久没有收到 Agent 的任何一个：数据面已经不通，重建会话。要比一次线路中断长（中断恢复后接着测），又不能拖到几个小时。 private static readonly TimeSpan s_dataIdleLimit = TimeSpan.FromSeconds(90); private static readonly RpcRouter s_router = BuildRouter(); @@ -347,18 +344,10 @@ public sealed class AgentConnection : IDisposable private async Task BoundedAsync(string step, Func work) { - try - { - await work().WaitAsync(s_teardownLimit, _env.Time); - } - catch (TimeoutException) + if (!await Bounded.WaitAsync(work(), Bounded.Teardown, _env.Time)) { _env.Log.Warn("agent session teardown step timed out, moving on", ("agent", Endpoint.AgentId), ("step", step)); } - catch (Exception ex) when (ex is OperationCanceledException or IOException or ObjectDisposedException or System.Net.Sockets.SocketException or InvalidOperationException) - { - // 会话结束时的正常噪声 - } } private static async Task Quiet(Task task) diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs index 9e03202..e4df037 100644 --- a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs +++ b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs @@ -271,7 +271,15 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection if (Volatile.Read(ref _handle) != 0) { - await _shutdown.Task.ConfigureAwait(false); + // 原生层偶尔不回报关闭完成：等一会儿就放手，调用方（重拨循环、有序退出）不能跟着卡住 + try + { + await _shutdown.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); + } + catch (TimeoutException) + { + _log.Warn("QUIC connection did not report shutdown completion in time; moving on"); + } } }