fix(quic): exclude locally rejected datagrams [CI]

backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs

@@ -51,7 +51,7 @@ public sealed class AgentConnection : IDisposable
 
     public KnownAgent Endpoint { get; }
 
-    public bool IsConnected => _session is not null;
+    public bool IsConnected => _session is not null && Volatile.Read(ref _activeDatagrams)?.IsConnected == true;
 
     public int RoleCount => _session?.RoleCount ?? 0;
 
@@ -91,16 +91,26 @@ public sealed class AgentConnection : IDisposable
                 p95 = sorted[(int)Math.Ceiling(sorted.Length * 0.95) - 1];
             }
 
-            QuicTransportSnapshot? transport = Volatile.Read(ref _activeDatagrams)?.GetTransportSnapshot();
+            QuicDatagramTransport? datagrams = Volatile.Read(ref _activeDatagrams);
+            bool connected = _session is not null && datagrams?.IsConnected == true;
+            ProbeLinkAttemptState state = _attemptState;
+            string? lastError = _lastError;
+            if (!connected && state == ProbeLinkAttemptState.Connected)
+            {
+                state = ProbeLinkAttemptState.RetryWait;
+                lastError ??= "QUIC datagram transport is closed; reconnecting";
+            }
+
+            QuicTransportSnapshot? transport = datagrams?.GetTransportSnapshot();
             return new ProbeQuicLinkHealth
             {
                 AgentId = Endpoint.AgentId,
                 IpVersion = Endpoint.IpVersion,
-                Connected = IsConnected,
+                Connected = connected,
-                State = _attemptState,
+                State = state,
-                LastError = _lastError,
+                LastError = lastError,
                 RetryAttempt = _lastRetryAttempt,
-                RetryInMs = _attemptState == ProbeLinkAttemptState.RetryWait && _retryAt is { } retryAt
+                RetryInMs = state == ProbeLinkAttemptState.RetryWait && _retryAt is { } retryAt
                     ? Math.Max(0, (long)(retryAt - _env.Time.GetUtcNow()).TotalMilliseconds)
                     : null,
                 HandshakeAttempts = _handshakeAttempts,
@@ -282,9 +292,21 @@ public sealed class AgentConnection : IDisposable
             _env.Log.Info("agent session up", ("agent", Endpoint.AgentId), ("offset_us", clock.OffsetUs));
             Task maintenance = MaintainAsync(rpc, session, clock, cts.Token);
             Task finished = await Task.WhenAny(run, receive, pump, maintenance);
+            string reason = finished == run
+                ? "closed by the Agent control connection"
+                : finished == receive
+                    ? "QUIC datagram transport closed"
+                    : finished == pump
+                        ? "Probe outbox pump stopped"
+                        : "connection maintenance stopped";
+            if (finished.IsFaulted && finished.Exception?.GetBaseException() is { } error)
+            {
+                reason = $"{error.GetType().Name}: {error.Message}";
+            }
+
             await cts.CancelAsync();
             await Task.WhenAll(Quiet(run), Quiet(receive), Quiet(pump), Quiet(maintenance));
-            return finished == run ? "closed by the agent" : "session task failed";
+            return reason;
         }
         finally
         {

backend/shared/MoShiro.Measurement/Blocks/Echo.cs

@@ -1,5 +1,6 @@
 using System.Net;
 using System.Net.Sockets;
+using Fleet.Core;
 using MoShiro.Protocol;
 
 namespace MoShiro.Measurement;
@@ -26,6 +27,7 @@ internal sealed class EchoRows
     private readonly int _t4;
     private readonly long _graceUs;
     private readonly bool _settleOnReply;
+    private long _discarded;
 
     public EchoRows(KindSpec kind, int capacity, long graceUs, bool settleOnReply)
     {
@@ -38,7 +40,7 @@ internal sealed class EchoRows
         _t4 = _log.Store.ColumnIndex("t4");
     }
 
-    public long Count => _log.Count;
+    public long Count => _log.Count - Interlocked.Read(ref _discarded);
 
     public uint Add(int size, long t1Us)
     {
@@ -56,6 +58,17 @@ internal sealed class EchoRows
         }
     }
 
+    public bool Discard(uint seq)
+    {
+        if (!_log.Store.Remove(seq))
+        {
+            return false;
+        }
+
+        Interlocked.Increment(ref _discarded);
+        return true;
+    }
+
     public RawStream? Take(long nowUs, bool all) =>
         _log.Take(
             key => (_settleOnReply && _log.Store.IsSet(key, _t4)) || (_log.Store.TryGet(key, _t1, out long t1) && t1 <= nowUs - _graceUs),
@@ -68,6 +81,8 @@ internal sealed class EchoRows
 public sealed class DatagramEchoDialer : BlockBase
 {
     private readonly EchoRows _rows;
+    private readonly object _gate = new();
+    private long _nextSendFailureWarningUs;
 
     public DatagramEchoDialer(BlockEnv env, BlockSpec spec)
         : base(env, spec)
@@ -92,9 +107,21 @@ public sealed class DatagramEchoDialer : BlockBase
             while (await timer.WaitForNextTickAsync(ct))
             {
                 long t1 = Env.Clock.NowUs();
-                uint seq = _rows.Add(size, t1);
+                lock (_gate)
-                MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, session.SessionId, Spec.StreamId, seq, t1), session.Key);
+                {
-                datagrams.Send(session, target, packet);
+                    uint seq = _rows.Add(size, t1);
+                    MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, session.SessionId, Spec.StreamId, seq, t1), session.Key);
+                    if (!datagrams.Send(session, target, packet))
+                    {
+                        _rows.Discard(seq);
+                        if (t1 >= _nextSendFailureWarningUs)
+                        {
+                            Env.Log.Warn("QUIC echo datagram was not queued; omitted from loss samples",
+                                ("stream", Spec.StreamId), ("seq", seq), ("bytes", size));
+                            _nextSendFailureWarningUs = t1 + 5_000_000;
+                        }
+                    }
+                }
             }
         }
         catch (OperationCanceledException)
@@ -109,7 +136,12 @@ public sealed class DatagramEchoDialer : BlockBase
 
     public override int Flush(FlushMode mode, long peerProduced = 0)
     {
-        RawStream? stream = _rows.Take(Env.Clock.NowUs(), mode == FlushMode.All);
+        RawStream? stream;
+        lock (_gate)
+        {
+            stream = _rows.Take(Env.Clock.NowUs(), mode == FlushMode.All);
+        }
+
         if (stream is null)
         {
             return 0;
@@ -123,7 +155,10 @@ public sealed class DatagramEchoDialer : BlockBase
     {
         if (delivery.Header.Type == MeasurementDatagramType.EchoReply)
         {
-            _rows.Reply(delivery.Header.Seq, delivery.ReceiveUs);
+            lock (_gate)
+            {
+                _rows.Reply(delivery.Header.Seq, delivery.ReceiveUs);
+            }
         }
     }
 }

backend/shared/MoShiro.Measurement/Blocks/Train.cs

@@ -1,5 +1,6 @@
 using System.Diagnostics;
 using System.Net;
+using Fleet.Core;
 using MoShiro.Protocol;
 
 namespace MoShiro.Measurement;
@@ -121,7 +122,13 @@ public sealed class TrainListener : BlockBase
             {
                 long sendUs = Env.Clock.NowUs();
                 MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.Data, session.SessionId, Spec.StreamId, seq, sendUs), session.Key);
-                Env.Datagrams!.Send(session, target, packet);
+                if (!Env.Datagrams!.Send(session, target, packet))
+                {
+                    Env.Log.Warn("QUIC train datagram was not queued; stopping the train without counting it as loss",
+                        ("stream", Spec.StreamId), ("seq", seq), ("bytes", size));
+                    return;
+                }
+
                 Span<long> row = [size, sendUs];
                 _records.SetRow(seq, row);
                 Interlocked.Exchange(ref _sent, seq + 1L);
@@ -171,7 +178,10 @@ public sealed class TrainDialer : BlockBase
             for (uint seq = 0; seq < 3; seq++)
             {
                 MeasurementDatagram.Write(hello, new MeasurementDatagramHeader(MeasurementDatagramType.Data, session.SessionId, Spec.StreamId, seq, Env.Clock.NowUs()), session.Key);
-                datagrams.Send(session, target, hello);
+                if (!datagrams.Send(session, target, hello))
+                {
+                    throw new BlockFaultException($"QUIC datagram send failed during path discovery (seq {seq}).");
+                }
             }
 
             await WaitUntilCancelledAsync(ct);

backend/shared/MoShiro.Measurement/Core/RowStore.cs

@@ -138,6 +138,36 @@ public sealed class RowStore
         }
     }
 
+    /// <summary>Remove a row that was reserved locally but never accepted by the transport.</summary>
+    internal bool Remove(long key)
+    {
+        lock (_gate)
+        {
+            if (key < 0)
+            {
+                return false;
+            }
+
+            int slot = (int)(key % _capacity);
+            if (_keys[slot] != key)
+            {
+                return false;
+            }
+
+            _keys[slot] = -1;
+            for (int i = 0; i < _columns.Count; i++)
+            {
+                _values[i][slot] = 0;
+                if (_valid[i] is { } valid)
+                {
+                    valid[slot] = false;
+                }
+            }
+
+            return true;
+        }
+    }
+
     /// <summary>读一个值；行不存在或该列没有设置（可空列）返回 false。</summary>
     public bool TryGet(long key, int column, out long value)
     {

backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs

@@ -99,7 +99,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
         return RemoteEndPoint;
     }
 
-    public bool IsReady => _connected.Task.IsCompletedSuccessfully && Volatile.Read(ref _sendEnabled) != 0;
+    public bool IsReady => Volatile.Read(ref _handle) != 0 && _connected.Task.IsCompletedSuccessfully && Volatile.Read(ref _sendEnabled) != 0;
 
     public int MaxDatagramBytes => Volatile.Read(ref _maxDatagramBytes);
 

backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs

@@ -128,13 +128,16 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
     public void Unregister(uint streamId) => _router.Unregister(streamId);
 
     /// <summary>Sends one measurement datagram on the connection for the supplied session.</summary>
-    public void Send(DataSession session, EndPoint target, ReadOnlySpan<byte> packet)
+    public bool Send(DataSession session, EndPoint target, ReadOnlySpan<byte> packet)
     {
         _ = target; // A QUIC connection owns its path; the Agent learns it from authenticated datagrams.
         if (session.DatagramConnection?.Send(packet) == true)
         {
             _router.Counters.PacketOut();
+            return true;
         }
+
+        return false;
     }
 
     public async Task RunAsync(CancellationToken ct)

backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs

@@ -307,6 +307,20 @@ public sealed class MeasurementTests
         Assert.Equal([1L], rows.Take(0, all: true)!.Find("seq")!.Values);
     }
 
+    [Fact]
+    public void EchoRows_DiscardedLocalSendIsNotReportedAsNetworkLoss()
+    {
+        var rows = new EchoRows(Kinds.UdpEcho, 64, EchoRows.DatagramGraceUs, settleOnReply: false);
+        uint unsent = rows.Add(64, 0);
+        Assert.True(rows.Discard(unsent));
+
+        uint sent = rows.Add(64, 1_000_000);
+        Assert.Equal(1, rows.Count);
+        RawStream cut = rows.Take(3_500_000, all: false)!;
+        Assert.Equal([(long)sent], cut.Find("seq")!.Values);
+        Assert.Equal([false], cut.Find("t4")!.Valid!);
+    }
+
     // ---- 积木目录 ----
 
     [Fact]

scripts/netdiag/udp_path.py

@@ -0,0 +1,235 @@
+#!/usr/bin/env python3
+"""Small dependency-free UDP path test with sequence, loss, and RTT reporting.
+
+Run a bounded echo receiver on the remote host:
+    python3 scripts/netdiag/udp_path.py server --bind 0.0.0.0 --port 443 --duration 45
+
+Then run the client from the source host:
+    python3 scripts/netdiag/udp_path.py client --target 203.0.113.10 --port 443 \
+        --count 251 --rate 31 --size 64 --timeout 3 --socket-mode sendto
+
+The server echoes each valid datagram unchanged. The client uses its sequence and
+monotonic send timestamp to report lost replies and round-trip latency. It uses
+no TCP control channel, so TCP service occupancy on the same port is irrelevant.
+"""
+
+from __future__ import annotations
+
+import argparse
+import math
+import secrets
+import select
+import socket
+import struct
+import sys
+import time
+from collections import Counter
+
+
+HEADER = struct.Struct("!8sIQ")  # random run id, sequence, monotonic ns
+MAX_DATAGRAM = 65_507
+
+
+def percentile(values: list[float], p: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    index = max(0, math.ceil(p * len(ordered)) - 1)
+    return ordered[index]
+
+
+def format_ms(value: float | None) -> str:
+    return "n/a" if value is None else f"{value:.3f} ms"
+
+
+def is_probe_packet(data: bytes) -> bool:
+    return len(data) >= HEADER.size
+
+
+def run_server(args: argparse.Namespace) -> int:
+    sock = socket.socket(socket.AF_INET6 if ":" in args.bind else socket.AF_INET, socket.SOCK_DGRAM)
+    if sock.family == socket.AF_INET6:
+        sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
+    try:
+        sock.bind((args.bind, args.port))
+    except OSError as exc:
+        print(f"bind failed on {args.bind}:{args.port}: {exc}", file=sys.stderr)
+        return 2
+
+    print(f"listening udp://{args.bind}:{args.port} for {args.duration:g}s", flush=True)
+    deadline = time.monotonic() + args.duration if args.duration > 0 else None
+    received = echoed = malformed = duplicates = 0
+    sequences: set[tuple[str, bytes, int]] = set()
+    sources: Counter[str] = Counter()
+    next_report = time.monotonic() + 1
+    sock.settimeout(0.25)
+    try:
+        while deadline is None or time.monotonic() < deadline:
+            try:
+                data, address = sock.recvfrom(MAX_DATAGRAM)
+            except socket.timeout:
+                continue
+            received += 1
+            source = f"{address[0]}:{address[1]}"
+            sources[source] += 1
+            if not is_probe_packet(data):
+                malformed += 1
+                continue
+            run_id, sequence, _ = HEADER.unpack_from(data)
+            key = (source, run_id, sequence)
+            if key in sequences:
+                duplicates += 1
+            else:
+                sequences.add(key)
+            try:
+                sock.sendto(data, address)
+                echoed += 1
+            except OSError as exc:
+                print(f"reply failed to {source}: {exc}", file=sys.stderr, flush=True)
+            if time.monotonic() >= next_report:
+                print(f"server progress received={received} unique={len(sequences)} echoed={echoed}", flush=True)
+                next_report += 1
+    except KeyboardInterrupt:
+        pass
+    finally:
+        sock.close()
+
+    source_summary = ", ".join(f"{source}={count}" for source, count in sources.most_common()) or "none"
+    print(
+        f"server received={received} unique={len(sequences)} echoed={echoed} "
+        f"malformed={malformed} duplicate_seq={duplicates} sources=[{source_summary}]",
+        flush=True,
+    )
+    return 0
+
+
+def receive_replies(
+    sock: socket.socket,
+    pending: dict[int, int],
+    rtts_ms: list[float],
+    duplicates: list[int],
+    wrong_source: list[int],
+    receive_errors: list[str],
+    run_id: bytes,
+    target: tuple[str, int],
+    wait_seconds: float,
+) -> None:
+    deadline = time.monotonic() + wait_seconds
+    while time.monotonic() < deadline:
+        remaining = deadline - time.monotonic()
+        readable, _, _ = select.select([sock], [], [], max(0.0, min(remaining, 0.05)))
+        if not readable:
+            continue
+        try:
+            data, address = sock.recvfrom(MAX_DATAGRAM)
+        except OSError as exc:
+            if not receive_errors:
+                receive_errors.append(f"{type(exc).__name__}: {exc}")
+            continue
+        if address[1] != target[1] or address[0] not in (target[0], "::ffff:" + target[0]):
+            wrong_source[0] += 1
+            continue
+        if not is_probe_packet(data):
+            continue
+        reply_run_id, sequence, sent_ns = HEADER.unpack_from(data)
+        if reply_run_id != run_id:
+            wrong_source[0] += 1
+            continue
+        if sequence not in pending:
+            duplicates[0] += 1
+            continue
+        pending.pop(sequence)
+        rtts_ms.append((time.monotonic_ns() - sent_ns) / 1_000_000)
+
+
+def run_client(args: argparse.Namespace) -> int:
+    if args.size < HEADER.size or args.size > MAX_DATAGRAM:
+        print(f"--size must be between {HEADER.size} and {MAX_DATAGRAM}", file=sys.stderr)
+        return 2
+    try:
+        addresses = socket.getaddrinfo(args.target, args.port, type=socket.SOCK_DGRAM)
+    except OSError as exc:
+        print(f"resolve failed for {args.target}: {exc}", file=sys.stderr)
+        return 2
+    family, socktype, proto, _, target = addresses[0]
+    sock = socket.socket(family, socktype, proto)
+    pending: dict[int, int] = {}
+    rtts_ms: list[float] = []
+    duplicates = [0]
+    wrong_source = [0]
+    receive_errors: list[str] = []
+    send_errors = 0
+    run_id = secrets.token_bytes(8)
+    local_endpoint = None
+    payload_padding = bytes(args.size - HEADER.size)
+
+    try:
+        if args.socket_mode == "connected":
+            sock.connect(target)
+        start = time.monotonic()
+        for sequence in range(args.count):
+            send_at = start + sequence / args.rate
+            while True:
+                remaining = send_at - time.monotonic()
+                if remaining <= 0:
+                    break
+                receive_replies(sock, pending, rtts_ms, duplicates, wrong_source, receive_errors, run_id, target, max(0.0, min(remaining, 0.05)))
+
+            sent_ns = time.monotonic_ns()
+            packet = HEADER.pack(run_id, sequence, sent_ns) + payload_padding
+            try:
+                if args.socket_mode == "connected":
+                    sock.send(packet)
+                else:
+                    sock.sendto(packet, target)
+                if local_endpoint is None:
+                    local_endpoint = sock.getsockname()
+                pending[sequence] = sent_ns
+            except OSError as exc:
+                send_errors += 1
+                print(f"send failed seq={sequence}: {exc}", file=sys.stderr, flush=True)
+
+        receive_replies(sock, pending, rtts_ms, duplicates, wrong_source, receive_errors, run_id, target, args.timeout)
+    finally:
+        sock.close()
+
+    sent = args.count - send_errors
+    received = len(rtts_ms)
+    loss = 0.0 if sent == 0 else 100.0 * (sent - received) / sent
+    elapsed = args.count / args.rate
+    print(f"target={args.target}:{args.port} local={local_endpoint} socket_mode={args.socket_mode} size={args.size} rate={args.rate:g} pps duration≈{elapsed:.2f}s run_id={run_id.hex()}")
+    print(f"sent={sent} replies={received} timeout_loss={loss:.2f}% send_errors={send_errors} duplicates={duplicates[0]} wrong_source={wrong_source[0]} receive_errors={receive_errors}")
+    print(f"rtt p50={format_ms(percentile(rtts_ms, 0.50))} p95={format_ms(percentile(rtts_ms, 0.95))} max={format_ms(max(rtts_ms) if rtts_ms else None)}")
+    return 0 if received > 0 else 1
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    modes = parser.add_subparsers(dest="mode", required=True)
+    server = modes.add_parser("server", help="run a bounded or continuous UDP echo receiver")
+    server.add_argument("--bind", default="0.0.0.0")
+    server.add_argument("--port", type=int, required=True)
+    server.add_argument("--duration", type=float, default=45, help="seconds; 0 runs until Ctrl-C")
+    server.set_defaults(run=run_server)
+    client = modes.add_parser("client", help="send sequenced UDP probes and measure replies")
+    client.add_argument("--target", required=True)
+    client.add_argument("--port", type=int, required=True)
+    client.add_argument("--count", type=int, default=251)
+    client.add_argument("--rate", type=float, default=31, help="datagrams per second")
+    client.add_argument("--size", type=int, default=64, help="UDP payload size in bytes")
+    client.add_argument("--timeout", type=float, default=3, help="reply drain timeout in seconds")
+    client.add_argument(
+        "--socket-mode",
+        choices=("sendto", "connected"),
+        default="sendto",
+        help="send with sendto() or a connected UDP socket; compare these when diagnosing VM networking",
+    )
+    client.set_defaults(run=run_client)
+    return parser
+
+
+if __name__ == "__main__":
+    arguments = build_parser().parse_args()
+    if getattr(arguments, "count", 1) < 1 or getattr(arguments, "rate", 1) <= 0:
+        raise SystemExit("--count must be positive and --rate must be greater than zero")
+    raise SystemExit(arguments.run(arguments))