From 57ae0752aa944c62762e1e7541e3d9cd81348063 Mon Sep 17 00:00:00 2001 From: JustMe Date: Sun, 18 Oct 2026 19:42:43 +0800 Subject: [PATCH] fix(probe): bound every step of tearing down an Agent session [CI] --- .../src/MoShiro.Probe/Link/AgentConnection.cs | 34 +++++++++++++++---- 1 file changed, 27 insertions(+), 7 deletions(-) diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs index a585b4c..b7b9be4 100644 --- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs +++ b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs @@ -26,6 +26,9 @@ 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); private static readonly RpcRouter s_router = BuildRouter(); /// 对等面入站方法的路由器(只读,测试用)。 @@ -209,7 +212,7 @@ public sealed class AgentConnection : IDisposable var remote = (IPEndPoint)info.Remote!; IReadOnlyList quicPorts = Endpoint.EffectiveQuicPorts; int quicPort = quicPorts.Count == 0 ? 7443 : quicPorts[0]; - await using var datagrams = new QuicDatagramTransport( + var datagrams = new QuicDatagramTransport( data, _env.Clock, new DataPlaneCounters(), @@ -308,7 +311,7 @@ public sealed class AgentConnection : IDisposable } await cts.CancelAsync(); - await Task.WhenAll(Quiet(run), Quiet(receive), Quiet(pump), Quiet(maintenance)); + await BoundedAsync("stop tasks", () => Task.WhenAll(Quiet(run), Quiet(receive), Quiet(pump), Quiet(maintenance))); return reason; } finally @@ -316,11 +319,12 @@ public sealed class AgentConnection : IDisposable _session = null; Volatile.Write(ref _activeDatagrams, null); await cts.CancelAsync(); - await session.StopAllAsync(); - await Quiet(run); - await Quiet(receive); - await Quiet(pump); - await rpc.DisposeAsync(); + await BoundedAsync("stop blocks", session.StopAllAsync); + await BoundedAsync("wait control", () => Quiet(run)); + await BoundedAsync("wait datagrams", () => Quiet(receive)); + await BoundedAsync("wait pump", () => Quiet(pump)); + await BoundedAsync("dispose control", () => rpc.DisposeAsync().AsTask()); + await BoundedAsync("dispose quic", () => datagrams.DisposeAsync().AsTask()); } } @@ -338,6 +342,22 @@ public sealed class AgentConnection : IDisposable } } + private async Task BoundedAsync(string step, Func work) + { + try + { + await work().WaitAsync(s_teardownLimit, _env.Time); + } + catch (TimeoutException) + { + _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) { try