diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs
index b7b9be4..a7c4b28 100644
--- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs
+++ b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs
@@ -29,6 +29,9 @@ public sealed class AgentConnection : IDisposable
/// 收尾的每一步最多等这么久: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();
/// 对等面入站方法的路由器(只读,测试用)。
@@ -296,7 +299,7 @@ public sealed class AgentConnection : IDisposable
_session = session;
SetAttempt(ProbeLinkAttemptState.Connected, null, null, 0);
_env.Log.Info("agent session up", ("agent", Endpoint.AgentId), ("offset_us", clock.OffsetUs));
- Task maintenance = MaintainAsync(rpc, session, clock, cts.Token);
+ Task maintenance = MaintainAsync(rpc, session, clock, datagrams, cts.Token);
Task finished = await Task.WhenAny(run, receive, pump, maintenance);
string reason = finished == run
? "closed by the Agent control connection"
@@ -393,10 +396,13 @@ public sealed class AgentConnection : IDisposable
}
/// 切块(每 10 秒)和时钟同步(每 30 秒)。
- private async Task MaintainAsync(RpcSession rpc, ConnectionSession session, ClockEstimator clock, CancellationToken ct)
+ private async Task MaintainAsync(RpcSession rpc, ConnectionSession session, ClockEstimator clock, QuicDatagramTransport datagrams, CancellationToken ct)
{
using var flush = new PeriodicTimer(s_flushInterval, _env.Time);
long lastClockUs = _env.Clock.NowUs();
+ long seenIn = datagrams.Counters.PacketsIn;
+ long seenOut = datagrams.Counters.PacketsOut;
+ long inChangedUs = lastClockUs;
while (await flush.WaitForNextTickAsync(ct))
{
if (rpc.IdleFor > s_idleLimit)
@@ -406,6 +412,21 @@ public sealed class AgentConnection : IDisposable
throw new IOException("the Agent stopped answering on this session");
}
+ // 数据面也要有回应:控制连接(TCP)还活着,QUIC 这条路却可能早已黑洞(一直在发、什么也收不到),表现为去程 100% 丢包、没有回程
+ long packetsIn = datagrams.Counters.PacketsIn;
+ long packetsOut = datagrams.Counters.PacketsOut;
+ if (packetsIn != seenIn)
+ {
+ seenIn = packetsIn;
+ inChangedUs = _env.Clock.NowUs();
+ }
+ else if (packetsOut != seenOut && _env.Clock.NowUs() - inChangedUs > s_dataIdleLimit.TotalMicroseconds)
+ {
+ _env.Log.Warn("agent datagram path is silent, reconnecting", ("agent", Endpoint.AgentId), ("idle_s", (_env.Clock.NowUs() - inChangedUs) / 1_000_000));
+ throw new IOException("the Agent has not answered any measurement datagram");
+ }
+
+ seenOut = packetsOut;
session.FlushLineBlocks();
if (_env.Clock.NowUs() - lastClockUs >= s_clockInterval.TotalMicroseconds)
{