feat(quic): replace probe UDP data plane with QUIC datagrams

backend/tests/MoShiro.TestSupport/ProbeSimulator.cs

@@ -8,13 +8,14 @@ using Fleet.Core;
 using Fleet.Identity;
 using Fleet.Link;
 using Fleet.Rpc;
+using MoShiro.Measurement;
 using MoShiro.Protocol;
 
 namespace MoShiro.TestSupport;
 
 /// <summary>
 /// 假 Probe：通过真实的点对点握手（TLS 固定证书 + 签名 + ECDH）连接真实的 Agent，接收 <c>line_task_set</c>，
-/// 在自己的 UDP 套接字上发回显包、接收单向数据流、估计时钟偏移，并把 Probe 一侧的列块经发件箱通道 17 交给 Agent。
+/// 通过 QUIC Datagram 发回显包、接收单向数据流、估计时钟偏移，并把 Probe 一侧的列块经发件箱通道 17 交给 Agent。
 /// 它实现 Probe 的协议职责，是将来真实 Probe 的参照。
 /// </summary>
 public sealed class ProbeSimulator : IAsyncDisposable
@@ -30,8 +31,9 @@ public sealed class ProbeSimulator : IAsyncDisposable
     private readonly CancellationTokenSource _stop = new();
     private NodeId _agentId;
     private RpcSession<ProbeSimulator>? _rpc;
-    private UdpClient? _udp;
-    private IPEndPoint? _agentUdp;
+    private QuicDatagramTransport? _quic;
+    private DataSession? _dataSession;
+    private IPEndPoint? _agentQuic;
     private PeerSessionInfo? _info;
     private Task? _receiver;
     private Task? _run;
@@ -72,13 +74,22 @@ public sealed class ProbeSimulator : IAsyncDisposable
         (PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(host, controlPort, agentPublicKey, _identity, ProbeId, "0.0.0-sim", 8 << 20, time ?? TimeProvider.System, ct);
         _info = info;
         _agentKey = agentPublicKey;
-        _udp = new UdpClient(new IPEndPoint(IPAddress.Loopback, 0));
-        _agentUdp = new IPEndPoint(IPAddress.Parse(host), udpPort);
+        _dataSession = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key);
+        _agentQuic = new IPEndPoint(IPAddress.Parse(host), udpPort);
+        _quic = new QuicDatagramTransport(
+            _dataSession,
+            new MonoClock(time ?? TimeProvider.System),
+            new DataPlaneCounters(),
+            new TestNodeLog(),
+            _agentQuic,
+            agentPublicKey,
+            0);
         _rpc = new RpcSession<ProbeSimulator>(transport, _router, this, time ?? TimeProvider.System);
         _rpc.OnBinary = (_, _) => ValueTask.CompletedTask;
         _run = _rpc.RunAsync(_stop.Token);
-        _receiver = Task.Run(ReceiveLoopAsync, _stop.Token);
+        _receiver = _quic.RunAsync(_stop.Token);
         await SyncClockAsync();
+        await _quic.ConnectAsync(ct);
     }
 
     public async Task<LineTaskSet> WaitForTasksAsync(Func<LineTaskSet, bool> predicate, TimeSpan? timeout = null)
@@ -123,19 +134,21 @@ public sealed class ProbeSimulator : IAsyncDisposable
         int size = (int)task.Params["size"].GetInt64();
         var collector = new EchoCollector();
         _echoes[task.StreamId] = collector;
+        _quic!.Register(task.StreamId, OnDatagram);
         long startUtc = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds();
         byte[] packet = new byte[size];
         for (uint seq = 0; seq < count; seq++)
         {
             long t1 = NowUs();
             collector.Sent[seq] = t1;
-            UdpPacket.Write(packet, new UdpHeader(UdpPacketType.EchoRequest, _info!.Credentials.SessionIdValue, task.StreamId, seq, t1), _info.Credentials.Key);
-            await _udp!.SendAsync(packet, packet.Length, _agentUdp!);
+            MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, _info!.Credentials.SessionIdValue, task.StreamId, seq, t1), _info.Credentials.Key);
+            _quic.Send(_dataSession!, _agentQuic!, packet);
             await Task.Delay(interval, ct);
         }
 
         await Task.Delay(300, ct); // 等最后的回复
         _echoes.TryRemove(task.StreamId, out _);
+        _quic.Unregister(task.StreamId);
 
         var msg = new StreamBuilder(new StreamSpec("msg", [Col.Seq("seq"), Col.U16("size"), Col.TsProbe("t1"), Col.TsProbe("t4", nullable: true)]));
         foreach ((uint seq, long t1) in collector.Sent.OrderBy(p => p.Key))
@@ -191,22 +204,28 @@ public sealed class ProbeSimulator : IAsyncDisposable
     }
 
     /// <summary>发送任意字节到 Agent 的 UDP 端口（用来测试 Agent 对垃圾包的处理）。</summary>
-    public Task SendRawUdpAsync(byte[] datagram) => _udp!.SendAsync(datagram, datagram.Length, _agentUdp!);
+    public Task SendRawUdpAsync(byte[] datagram)
+    {
+        _quic!.Send(_dataSession!, _agentQuic!, datagram);
+        return Task.CompletedTask;
+    }
 
     /// <summary>用伪造的密钥发一个回显包。</summary>
     public Task SendForgedEchoAsync(LineTask task, uint seq, byte[]? key = null)
     {
         byte[] packet = new byte[64];
-        UdpPacket.Write(packet, new UdpHeader(UdpPacketType.EchoRequest, _info!.Credentials.SessionIdValue, task.StreamId, seq, NowUs()), key ?? new byte[32]);
-        return _udp!.SendAsync(packet, packet.Length, _agentUdp!);
+        MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, _info!.Credentials.SessionIdValue, task.StreamId, seq, NowUs()), key ?? new byte[32]);
+        _quic!.Send(_dataSession!, _agentQuic!, packet);
+        return Task.CompletedTask;
     }
 
     /// <summary>重放：用同一个序号、同样的密钥再发一次。</summary>
     public Task SendEchoAsync(LineTask task, uint seq)
     {
         byte[] packet = new byte[64];
-        UdpPacket.Write(packet, new UdpHeader(UdpPacketType.EchoRequest, _info!.Credentials.SessionIdValue, task.StreamId, seq, NowUs()), _info.Credentials.Key);
-        return _udp!.SendAsync(packet, packet.Length, _agentUdp!);
+        MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, _info!.Credentials.SessionIdValue, task.StreamId, seq, NowUs()), _info.Credentials.Key);
+        _quic!.Send(_dataSession!, _agentQuic!, packet);
+        return Task.CompletedTask;
     }
 
     public int EchoRepliesSeen(uint streamId) => _echoes.TryGetValue(streamId, out EchoCollector? c) ? c.Received.Count : 0;
@@ -227,7 +246,7 @@ public sealed class ProbeSimulator : IAsyncDisposable
     public async ValueTask DisposeAsync()
     {
         await _stop.CancelAsync();
-        _udp?.Dispose();
+        _quic?.Dispose();
         if (_rpc is not null)
         {
             await _rpc.DisposeAsync();
@@ -277,41 +296,18 @@ public sealed class ProbeSimulator : IAsyncDisposable
         return header;
     }
 
-    private async Task ReceiveLoopAsync()
+    private void OnDatagram(in QuicDatagramDelivery delivery)
     {
-        while (!_stop.IsCancellationRequested)
+        long now = delivery.ReceiveUs;
+        MeasurementDatagramHeader header = delivery.Header;
+        if (header.Type == MeasurementDatagramType.EchoReply && _echoes.TryGetValue(header.StreamId, out EchoCollector? echo))
         {
-            UdpReceiveResult result;
-            try
-            {
-                result = await _udp!.ReceiveAsync(_stop.Token);
-            }
-            catch (Exception ex) when (ex is OperationCanceledException or ObjectDisposedException or SocketException)
-            {
-                if (_stop.IsCancellationRequested || ex is ObjectDisposedException)
-                {
-                    return;
-                }
-
-                continue;
-            }
-
-            long now = NowUs();
-            byte[] data = result.Buffer;
-            if (!UdpPacket.TryParseHeader(data, out UdpHeader header) || !UdpPacket.VerifyMac(data, _info!.Credentials.Key))
-            {
-                continue;
-            }
-
-            if (header.Type == UdpPacketType.EchoReply && _echoes.TryGetValue(header.StreamId, out EchoCollector? echo))
-            {
-                echo.Received.TryAdd(header.Seq, now);
-            }
-            else if (header.Type == UdpPacketType.Data && _trains.TryGetValue(header.StreamId, out TrainCollector? train))
-            {
-                train.Received.TryAdd(header.Seq, now);
-                train.Size = data.Length;
-            }
+            echo.Received.TryAdd(header.Seq, now);
+        }
+        else if (header.Type == MeasurementDatagramType.Data && _trains.TryGetValue(header.StreamId, out TrainCollector? train))
+        {
+            train.Received.TryAdd(header.Seq, now);
+            train.Size = delivery.Payload.Length;
         }
     }
 
@@ -366,33 +362,39 @@ public sealed class ProbeSimulator : IAsyncDisposable
 
     private readonly ConcurrentDictionary<uint, TestPhaseStartParams> _activePhases = new();
 
-    private async Task<bool> StartPhaseAsync(TestPhaseStartParams start, CancellationToken ct)
+    private Task<bool> StartPhaseAsync(TestPhaseStartParams start, CancellationToken ct)
     {
         Phases.Enqueue(start);
         if (RefusePhases)
         {
-            return false;
+            return Task.FromResult(false);
         }
 
         _activePhases[start.StreamId] = start;
         if (start.Kind == "udp_train")
         {
             _trains[start.StreamId] = new TrainCollector { StartUtcMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds() };
+            _quic!.Register(start.StreamId, OnDatagram);
             // 先发几个探活包，让 Agent 学到（最新的）源地址，再回复 accepted
             byte[] hello = new byte[64];
             for (uint seq = 0; seq < 3; seq++)
             {
-                UdpPacket.Write(hello, new UdpHeader(UdpPacketType.Data, _info!.Credentials.SessionIdValue, start.StreamId, seq, NowUs()), _info.Credentials.Key);
-                await _udp!.SendAsync(hello.AsMemory(), _agentUdp!, ct);
+                MeasurementDatagram.Write(hello, new MeasurementDatagramHeader(MeasurementDatagramType.Data, _info!.Credentials.SessionIdValue, start.StreamId, seq, NowUs()), _info.Credentials.Key);
+                _quic.Send(_dataSession!, _agentQuic!, hello);
             }
         }
 
-        return true;
+        _ = ct;
+        return Task.FromResult(true);
     }
 
     private async Task<int> EndPhaseAsync(TestPhaseEndParams end, CancellationToken ct)
     {
         _activePhases.TryRemove(end.StreamId, out TestPhaseStartParams? start);
+        if (start is not null)
+        {
+            _quic!.Unregister(start.StreamId);
+        }
         if (start is null || DropFragments)
         {
             _trains.TryRemove(end.StreamId, out _);
@@ -456,4 +458,9 @@ public sealed class ProbeSimulator : IAsyncDisposable
 
         public long StartUtcMs { get; init; }
     }
+
+    private sealed class TestNodeLog : INodeLog
+    {
+        public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields) { }
+    }
 }