From 09b4630953caf06a0b411b25bb3d479ed232ca19 Mon Sep 17 00:00:00 2001
From: JustMe <mr.toing@outlook.com>
Date: Mon, 19 Oct 2026 09:28:29 +0800
Subject: [PATCH] fix(probe): rebuild the Agent session when the datagram path
 goes silent [CI]

---
 .../src/MoShiro.Probe/Link/AgentConnection.cs | 25 +++++++++++++++++--
 1 file changed, 23 insertions(+), 2 deletions(-)

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
 
     /// <summary>收尾的每一步最多等这么久：QUIC 连接或控制连接的关闭卡在原生层时，不能让整个重拨循环跟着卡住。</summary>
     private static readonly TimeSpan s_teardownLimit = TimeSpan.FromSeconds(5);
+
+    /// <summary>一直在发测量报文、却这么久没有收到 Agent 的任何一个：数据面已经不通，重建会话。要比一次线路中断长（中断恢复后接着测），又不能拖到几个小时。</summary>
+    private static readonly TimeSpan s_dataIdleLimit = TimeSpan.FromSeconds(90);
     private static readonly RpcRouter<ConnectionSession> s_router = BuildRouter();
 
     /// <summary>对等面入站方法的路由器（只读，测试用）。</summary>
@@ -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
     }
 
     /// <summary>切块（每 10 秒）和时钟同步（每 30 秒）。</summary>
-    private async Task MaintainAsync(RpcSession<ConnectionSession> rpc, ConnectionSession session, ClockEstimator clock, CancellationToken ct)
+    private async Task MaintainAsync(RpcSession<ConnectionSession> 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)
             {