fix(probe): bound every step of tearing down an Agent session [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was canceled

This commit is contained in:
justme committed 2026-10-18 19:42:43 +08:00
1 parent f9be98e801
commit 57ae0752aa
1 file changed
+27 -7
@@ -26,6 +26,9 @@ public sealed class AgentConnection : IDisposable
/// <summary>Agent 的任何帧(时钟回复、任务集)都算存活;比时钟同步间隔长得多,容得下差线路上的黑洞,但不会让一个 Agent 已经丢掉的会话永远挂着。</summary>
private static readonly TimeSpan s_idleLimit = TimeSpan.FromSeconds(120);
/// <summary>收尾的每一步最多等这么久:QUIC 连接或控制连接的关闭卡在原生层时,不能让整个重拨循环跟着卡住。</summary>
private static readonly TimeSpan s_teardownLimit = TimeSpan.FromSeconds(5);
private static readonly RpcRouter<ConnectionSession> s_router = BuildRouter();
/// <summary>对等面入站方法的路由器(只读,测试用)。</summary>
@@ -209,7 +212,7 @@ public sealed class AgentConnection : IDisposable
var remote = (IPEndPoint)info.Remote!;
IReadOnlyList<int> 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<Task> 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