fix: prevent QUIC heartbeat stalls and isolate TCP retransmission counters [CI] [SKIP TESTS]

backend/probe/src/MoShiro.Probe/ProbeApp.cs

@@ -171,10 +171,13 @@ public sealed class ProbeApp : IDisposable
 
     private List<ProbeQuicLinkHealth> SnapshotQuicLinks()
     {
+        AgentConnection[] connections;
         lock (_connections)
         {
-            return [.. _connections.Values.Select(r => r.Connection.QuicHealth())];
+            connections = [.. _connections.Values.Select(r => r.Connection)];
         }
+
+        return [.. connections.Select(c => c.QuicHealth())];
     }
 
     public async Task RunAsync(CancellationToken ct)

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

@@ -33,6 +33,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
     private static readonly MsQuicApi.ConnectionCallback s_callback = OnEvent;
 
     private readonly object _sendGate = new();
+    private readonly object _nativeGate = new();
     private readonly TaskCompletionSource _connected = new(TaskCreationOptions.RunContinuationsAsynchronously);
     private readonly TaskCompletionSource _shutdown = new(TaskCreationOptions.RunContinuationsAsynchronously);
     private readonly GCHandle _self;
@@ -48,6 +49,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
     private int _sendEnabled;
     private int _disposeStarted;
     private double? _handshakeDurationMs;
+    private QuicTransportSnapshot? _transportSnapshot;
+    private long _statisticsAt;
 
     private MsQuicConnection(
         nint handle,
@@ -105,24 +108,33 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
 
     public double? HandshakeDurationMs => _handshakeDurationMs;
 
-    public bool TryGetStatistics(out MsQuicApi.QuicStatisticsV2 statistics)
+    public QuicTransportSnapshot? TransportSnapshot => Volatile.Read(ref _transportSnapshot);
+
+    // Called only from this connection's serialized native callbacks. GetParam is
+    // inline on the MsQuic worker; calling it from a heartbeat can wait forever
+    // for a stalled worker and take the entire Fleet heartbeat down with it.
+    private void UpdateTransportSnapshot()
     {
         nint handle = Volatile.Read(ref _handle);
-        if (handle == 0)
+        long now = Stopwatch.GetTimestamp();
+        if (handle == 0 || (_statisticsAt != 0 && Stopwatch.GetElapsedTime(_statisticsAt, now) < TimeSpan.FromSeconds(1)))
         {
-            statistics = default;
-            return false;
+            return;
         }
 
+        _statisticsAt = now;
         try
         {
-            statistics = MsQuicApi.GetStatisticsV2(handle);
-            return true;
+            MsQuicApi.QuicStatisticsV2 stats = MsQuicApi.GetStatisticsV2(handle);
+            ulong lost = stats.SendSuspectedLostPackets >= stats.SendSpuriousLostPackets
+                ? stats.SendSuspectedLostPackets - stats.SendSpuriousLostPackets : 0;
+            double? loss = stats.SendTotalPackets > 0 ? lost * 100.0 / stats.SendTotalPackets : null;
+            Volatile.Write(ref _transportSnapshot, new QuicTransportSnapshot(loss, stats.SendTotalPackets,
+                stats.SendCongestionCount, stats.SendEcnCongestionCount, stats.SendPathMtu));
         }
         catch (InvalidOperationException)
         {
-            statistics = default;
-            return false;
+            // Some event phases do not yet expose statistics. Retain the last snapshot.
         }
     }
 
@@ -214,12 +226,23 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
         var connection = new MsQuicConnection(handle, 0, 0, null, datagramHandler, log, closed, peerAddressChanged);
         MsQuicApi.SetContext(handle, GCHandle.ToIntPtr(connection._self));
         MsQuicApi.SetCallbackHandler(handle, s_callback, GCHandle.ToIntPtr(connection._self));
-        MsQuicApi.ConnectionSetConfiguration(handle, configuration);
         MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1);
+        // Transport parameters must be set before SetConfiguration starts TLS.
+        MsQuicApi.ConnectionSetConfiguration(handle, configuration);
         return connection;
     }
 
     public bool Send(ReadOnlySpan<byte> datagram)
+    {
+        // ConnectionClose must not free the handle or a pending send while
+        // its native buffer is still being prepared/submitted.
+        lock (_nativeGate)
+        {
+            return SendCore(datagram);
+        }
+    }
+
+    private bool SendCore(ReadOnlySpan<byte> datagram)
     {
         nint handle = Volatile.Read(ref _handle);
         if (handle == 0 || !IsReady || datagram.Length > MaxDatagramBytes)
@@ -264,9 +287,12 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
 
     public async ValueTask DisposeAsync()
     {
-        if (Interlocked.Exchange(ref _disposeStarted, 1) == 0 && Volatile.Read(ref _handle) is nint handle && handle != 0)
+        lock (_nativeGate)
         {
-            MsQuicApi.ConnectionShutdown(handle);
+            if (Interlocked.Exchange(ref _disposeStarted, 1) == 0 && Volatile.Read(ref _handle) is nint handle && handle != 0)
+            {
+                MsQuicApi.ConnectionShutdown(handle);
+            }
         }
 
         if (Volatile.Read(ref _handle) != 0)
@@ -396,6 +422,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
                     return connection.ValidatePeerCertificate(eventPointer);
             }
 
+            connection.UpdateTransportSnapshot();
             return 0;
         }
         catch (Exception ex)
@@ -418,10 +445,23 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
 
     private void OnDatagramStateChanged(nint eventPointer)
     {
-        byte sendEnabled = Marshal.ReadByte(eventPointer, IntPtr.Size);
-        int maxSendLength = (ushort)Marshal.ReadInt16(eventPointer, IntPtr.Size + 1);
+        (bool sendEnabled, int maxSendLength) = ReadDatagramState(eventPointer);
         Volatile.Write(ref _maxDatagramBytes, maxSendLength);
-        Volatile.Write(ref _sendEnabled, sendEnabled == 0 ? 0 : 1);
+        Volatile.Write(ref _sendEnabled, sendEnabled ? 1 : 0);
+    }
+
+    internal static (bool SendEnabled, int MaxSendLength) ReadDatagramState(nint eventPointer)
+    {
+        // BOOLEAN is followed by padding before the naturally aligned uint16_t.
+        var state = Marshal.PtrToStructure<DatagramState>(eventPointer + IntPtr.Size);
+        return (state.SendEnabled != 0, state.MaxSendLength);
+    }
+
+    [StructLayout(LayoutKind.Sequential)]
+    private struct DatagramState
+    {
+        public byte SendEnabled;
+        public ushort MaxSendLength;
     }
 
     private void OnDatagramReceived(nint eventPointer)
@@ -554,7 +594,17 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
             return;
         }
 
-        MsQuicApi.ConnectionClose(handle);
+        // RegistrationClose waits for native workers/children and is forbidden
+        // on a MsQuic callback. Return from the callback before waiting for cleanup.
+        ThreadPool.QueueUserWorkItem(_ => CloseHandles(handle));
+    }
+
+    private void CloseHandles(nint handle)
+    {
+        lock (_nativeGate)
+        {
+            MsQuicApi.ConnectionClose(handle);
+        }
         lock (_sendGate)
         {
             foreach ((nint context, SendState state) in _sends)

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

@@ -60,19 +60,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
 
     public double? HandshakeDurationMs => _client?.HandshakeDurationMs;
 
-    public QuicTransportSnapshot? GetTransportSnapshot()
-    {
-        if (_client is null || !_client.TryGetStatistics(out MsQuicApi.QuicStatisticsV2 stats))
-        {
-            return null;
-        }
-
-        ulong lost = stats.SendSuspectedLostPackets >= stats.SendSpuriousLostPackets
-            ? stats.SendSuspectedLostPackets - stats.SendSpuriousLostPackets
-            : 0;
-        double? packetLossRate = stats.SendTotalPackets > 0 ? lost * 100.0 / stats.SendTotalPackets : null;
-        return new QuicTransportSnapshot(packetLossRate, stats.SendTotalPackets, stats.SendCongestionCount, stats.SendEcnCongestionCount, stats.SendPathMtu);
-    }
+    public QuicTransportSnapshot? GetTransportSnapshot() => _client?.TransportSnapshot;
 
     public void Bind()
     {

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

@@ -70,14 +70,15 @@ public static class TcpInfo
 }
 
 /// <summary>
-/// 一条连接上周期性的内核采样（<c>tcp_info</c> 流）：每个采样一行 (ts, side, srtt, rttvar, cwnd, retrans_total, segs_out, bytes_acked)。
+/// 一条连接上周期性的内核采样（<c>tcp_info</c> 流）：每个采样一行 (ts, side, srtt, rttvar, cwnd, retrans_total, segs_out, bytes_acked, conn)。
 /// 两端各采样自己的套接字：Probe 的重传反映上行方向，Agent 的重传反映下行方向（代理场景里大流量走这个方向）。
-/// 累计计数器在 Backend 按 side 分别求差。
+/// 累计计数器在 Backend 按 (side, conn) 分别求差；conn 是当前记录器内分配的连接编号，不包含网络地址。
 /// </summary>
 public sealed class TcpInfoRecorder
 {
     private readonly MonoClock _clock;
     private readonly long _side;
+    private long _nextConnection;
 
     public TcpInfoRecorder(StreamSpec spec, Side side, MonoClock clock)
     {
@@ -96,6 +97,9 @@ public sealed class TcpInfoRecorder
             return;
         }
 
+        // A recorder survives reconnects, and listeners can briefly have more
+        // than one socket. Kernel counters belong to a socket, not just a side.
+        long connection = Interlocked.Increment(ref _nextConnection);
         try
         {
             using var timer = new PeriodicTimer(interval, _clock.Time);
@@ -107,7 +111,7 @@ public sealed class TcpInfoRecorder
                 }
 
                 long key = Rows.Append();
-                Rows.SetRow(key, [_clock.NowUs(), _side, s.SrttUs, s.RttVarUs, s.Cwnd, s.TotalRetrans, s.SegsOut, s.BytesAcked]);
+                Rows.SetRow(key, [_clock.NowUs(), _side, s.SrttUs, s.RttVarUs, s.Cwnd, s.TotalRetrans, s.SegsOut, s.BytesAcked, connection]);
             }
         }
         catch (OperationCanceledException)

backend/shared/MoShiro.Protocol/Catalog/Kinds/Kinds.cs

@@ -22,6 +22,7 @@ public static class Kinds
         [
             Col.TsLocal("ts"), Col.U8("side"), Col.U32("srtt_us"), Col.U32("rttvar_us"), Col.U32("cwnd"),
             Col.U32("retrans_total"), Col.U32("segs_out"), Col.U64("bytes_acked"),
+            new ColumnSpec("conn", ColumnType.U64, ColumnEnc.Plain, true),
         ],
         Union: true);
 

backend/shared/MoShiro.Protocol/Wire/TcpWire.cs

@@ -1,3 +1,4 @@
+using System.Buffers;
 using System.Buffers.Binary;
 using System.Security.Cryptography;
 using Fleet.Core;
@@ -124,9 +125,18 @@ public static class TcpWire
 
     public static async ValueTask WriteMessageAsync(Stream stream, ReadOnlyMemory<byte> body, CancellationToken ct)
     {
-        byte[] prefix = new byte[VarInt.MaxBytes];
-        int n = VarInt.Write(prefix, (ulong)body.Length);
-        await stream.WriteAsync(prefix.AsMemory(0, n), ct);
-        await stream.WriteAsync(body, ct);
+        byte[] frame = ArrayPool<byte>.Shared.Rent(checked(VarInt.MaxBytes + body.Length));
+        try
+        {
+            int n = VarInt.Write(frame, (ulong)body.Length);
+            body.CopyTo(frame.AsMemory(n));
+            // TCP_NODELAY is enabled: two writes otherwise create a tiny
+            // length-only segment before every echo body (and two TLS records).
+            await stream.WriteAsync(frame.AsMemory(0, n + body.Length), ct);
+        }
+        finally
+        {
+            ArrayPool<byte>.Shared.Return(frame);
+        }
     }
 }

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

@@ -203,7 +203,7 @@ public sealed class MeasurementTests
         var builder = new StreamBuilder(spec);
         foreach ((long ts, long side, long retrans) in rows)
         {
-            builder.Add(ts, side, 40_000, 5_000, 10, retrans, 100, 1_000);
+            builder.Add(ts, side, 40_000, 5_000, 10, retrans, 100, 1_000, null);
         }
 
         return builder.Build();
@@ -216,8 +216,8 @@ public sealed class MeasurementTests
         Assert.Equal(info.Columns.Count, info.ColumnsOf(Side.Agent).Count); // each side owns every column
         var agent = new RowStore(info, Side.Agent, 64);
         var recorder = new Source { Stores = { ["tcp_info"] = agent } };
-        agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000]);
-        agent.SetRow(agent.Append(), [600, 1, 40_000, 5_000, 10, 8, 120, 1_100]);
+        agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000, 1]);
+        agent.SetRow(agent.Append(), [600, 1, 40_000, 5_000, 10, 8, 120, 1_100, 1]);
 
         var msg = EchoFragment((0, 1000, 3000));
         var fragment = new RawChunk(msg.Header, [msg.Streams[0], InfoRows(info, (100, 0, 1))]);
@@ -227,6 +227,8 @@ public sealed class MeasurementTests
         Assert.Equal(3, merged.Count);
         Assert.Equal([0L, 1L, 1L], merged.Find("side")!.Values);
         Assert.Equal([1L, 7L, 8L], merged.Find("retrans_total")!.Values);
+        Assert.False(merged.Find("conn")!.IsValid(0));
+        Assert.Equal(1, merged.Find("conn")!.Values[1]);
 
         // The agent's rows were consumed: the next chunk carries only the probe's own rows
         IReadOnlyList<RawStream> second = FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, recorder).Value;
@@ -238,7 +240,7 @@ public sealed class MeasurementTests
     {
         StreamSpec info = Kinds.TcpEcho.FindStream("tcp_info")!;
         var agent = new RowStore(info, Side.Agent, 64);
-        agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000]);
+        agent.SetRow(agent.Append(), [500, 1, 40_000, 5_000, 10, 7, 100, 1_000, 1]);
 
         // A probe on a platform without kernel sampling sends the echo rows only
         RawChunk fragment = EchoFragment((0, 1000, 3000));
@@ -248,6 +250,18 @@ public sealed class MeasurementTests
         Assert.Equal(7, rows.Find("retrans_total")!.Values[0]);
     }
 
+    [Fact]
+    public void FragmentMerger_LegacyTcpSamples_KeepConnectionIdentityUnknown()
+    {
+        StreamSpec info = Kinds.TcpEcho.FindStream("tcp_info")!;
+        RawStream sample = InfoRows(info, (100, 0, 1));
+        var legacy = new RawStream(sample.Name, sample.Count, sample.Columns.Where(c => c.Name != "conn").ToArray());
+        var fragment = new RawChunk(new RawChunkHeader { Kind = "tcp_echo" }, [legacy]);
+        RawStream merged = Assert.Single(FragmentMerger.Merge(Kinds.TcpEcho, fragment, 1, new Source()).Value);
+        Assert.False(merged.Find("conn")!.IsValid(0));
+        Assert.Equal(1, merged.Find("retrans_total")!.Values[0]);
+    }
+
     [Fact]
     public void TcpInfo_OnLinux_ReadsTheKernelCountersOfALiveConnection()
     {

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

@@ -0,0 +1,77 @@
+using System.Net;
+using System.Runtime.InteropServices;
+using System.Security.Cryptography;
+using System.Security.Cryptography.X509Certificates;
+using Fleet.Core;
+
+namespace MoShiro.Measurement.Tests;
+
+public sealed class MsQuicConnectionTests
+{
+    [Theory]
+    [InlineData(1200, true)]
+    [InlineData(1280, true)]
+    [InlineData(1472, false)]
+    public void DatagramState_UsesTheNativeUint16Alignment(int maximum, bool enabled)
+    {
+        nint nativeEvent = Marshal.AllocHGlobal(IntPtr.Size + 4);
+        try
+        {
+            Marshal.WriteByte(nativeEvent, IntPtr.Size, enabled ? (byte)1 : (byte)0);
+            Marshal.WriteByte(nativeEvent, IntPtr.Size + 1, 0xA5); // padding is not part of MaxSendLength
+            Marshal.WriteInt16(nativeEvent, IntPtr.Size + 2, (short)maximum);
+            Assert.Equal((enabled, maximum), MsQuicConnection.ReadDatagramState(nativeEvent));
+        }
+        finally
+        {
+            Marshal.FreeHGlobal(nativeEvent);
+        }
+    }
+
+    [Fact]
+    public async Task DatagramConnection_CanSendCloseAndReconnect_WithoutBlockingNativeWorkers()
+    {
+        using ECDsa key = ECDsa.Create(ECCurve.NamedCurves.nistP256);
+        var request = new CertificateRequest("CN=localhost", key, HashAlgorithmName.SHA256);
+        using X509Certificate2 certificate = request.CreateSelfSigned(DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch.AddYears(100));
+        await using var listener = new MsQuicListener(IPAddress.Loopback, [0], certificate,
+            (connection, data) => connection.Send(data), NullNodeLog.Instance, null);
+        listener.Start();
+
+        for (int attempt = 0; attempt < 3; attempt++)
+        {
+            var reply = new TaskCompletionSource<byte[]>(TaskCreationOptions.RunContinuationsAsynchronously);
+            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+            MsQuicConnection client = await MsQuicConnection.ConnectAsync(IPAddress.Loopback, listener.BoundPorts[0], 0,
+                key.ExportSubjectPublicKeyInfo(), (_, data) => reply.TrySetResult(data.ToArray()), NullNodeLog.Instance, timeout.Token);
+            try
+            {
+                while (!client.IsReady)
+                {
+                    await Task.Delay(10, timeout.Token);
+                }
+
+                Assert.InRange(client.MaxDatagramBytes, 64, 1500);
+                byte[] payload = Enumerable.Range(0, 64).Select(i => (byte)i).ToArray();
+                // DATAGRAM is unreliable, and the peer can receive the first
+                // packet before its handshake-complete notification permits replies.
+                while (!reply.Task.IsCompleted)
+                {
+                    Assert.True(client.Send(payload));
+                    await Task.WhenAny(reply.Task, Task.Delay(20, timeout.Token));
+                    timeout.Token.ThrowIfCancellationRequested();
+                }
+                Assert.Equal(payload, await reply.Task.WaitAsync(timeout.Token));
+                Assert.NotNull(client.TransportSnapshot);
+            }
+            finally
+            {
+                // The old callback called RegistrationClose, deadlocking its own
+                // worker. A second connection then hung in SetParam/GetParam.
+                await client.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(3));
+                await client.WaitForShutdownAsync().WaitAsync(TimeSpan.FromSeconds(3));
+            }
+        }
+    }
+
+}

backend/shared/tests/MoShiro.Protocol.Tests/TcpWireTests.cs

@@ -0,0 +1,30 @@
+namespace MoShiro.Protocol.Tests;
+
+public sealed class TcpWireTests
+{
+    [Theory]
+    [InlineData(0)]
+    [InlineData(64)]
+    [InlineData(16384)]
+    public async Task Message_UsesOneWriteForPrefixAndBody_AndPreservesFraming(int size)
+    {
+        using var stream = new CountingStream();
+        byte[] body = Enumerable.Range(0, size).Select(i => (byte)i).ToArray();
+        await TcpWire.WriteMessageAsync(stream, body, default);
+        Assert.Equal(1, stream.Writes);
+        stream.Position = 0;
+        Assert.Equal(body, await TcpWire.ReadMessageAsync(stream, default));
+        Assert.Null(await TcpWire.ReadMessageAsync(stream, default));
+    }
+
+    private sealed class CountingStream : MemoryStream
+    {
+        public int Writes { get; private set; }
+
+        public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            Writes++;
+            return base.WriteAsync(buffer, cancellationToken);
+        }
+    }
+}

backend/src/MoShiro.Backend.Analysis/Ops/SessionOps.cs

@@ -9,7 +9,7 @@ internal static class TcpInfoTime
         side is not null && side.Values[row] == 0 ? view.ProbeToUtcMs(ts.Values[row]) : view.AgentToUtcMs(ts.Values[row]);
 }
 
-/// <summary>两次采样之间累计计数的差值（分子）与另一个累计计数的差值（分母），按 side 分别计算。</summary>
+/// <summary>两次采样之间累计计数的差值（分子）与另一个累计计数的差值（分母），按 (side, conn) 分别计算；旧数据没有 conn 时按 side 计算。</summary>
 internal sealed class KernelDeltaOp : MetricOp
 {
     private readonly string _field;
@@ -21,12 +21,13 @@ internal sealed class KernelDeltaOp : MetricOp
         _over = over;
     }
 
-    public override string Signature => $"kernel_delta/{_field}/{_over}/1";
+    public override string Signature => $"kernel_delta/{_field}/{_over}/2";
 
     public override void Evaluate(ChunkView view, RawStream stream, IObservationSink sink)
     {
         RawColumn? ts = stream.Find("ts");
         RawColumn? side = stream.Find("side");
+        RawColumn? connection = stream.Find("conn");
         RawColumn? field = stream.Find(_field);
         RawColumn? over = stream.Find(_over);
         if (ts is null || field is null || over is null)
@@ -34,21 +35,34 @@ internal sealed class KernelDeltaOp : MetricOp
             return;
         }
 
-        var last = new Dictionary<long, (long Field, long Over)>();
+        var last = new Dictionary<(long Side, long Connection), (long Ts, long Field, long Over)>();
         for (int i = 0; i < stream.Count; i++)
         {
-            long key = side?.Values[i] ?? 0;
-            if (last.TryGetValue(key, out (long Field, long Over) previous))
+            if (!ts.IsValid(i) || !field.IsValid(i) || !over.IsValid(i))
             {
-                long dField = Math.Max(0, field.Values[i] - previous.Field);
-                long dOver = Math.Max(0, over.Values[i] - previous.Over);
-                if (dOver > 0)
+                continue;
+            }
+
+            var key = (side?.Values[i] ?? 0, connection?.IsValid(i) == true ? connection.Values[i] : 0);
+            if (last.TryGetValue(key, out (long Ts, long Field, long Over) previous))
+            {
+                // Ignore delayed/duplicate samples without replacing the newer baseline.
+                if (ts.Values[i] <= previous.Ts)
+                {
+                    continue;
+                }
+
+                // Either counter falling means reset/wrap: discard the whole
+                // interval rather than mixing a new numerator with an old denominator.
+                long dField = field.Values[i] - previous.Field;
+                long dOver = over.Values[i] - previous.Over;
+                if (dField >= 0 && dOver > 0)
                 {
                     sink.Emit(TcpInfoTime.ToUtcMs(view, ts, side, i), dField, dOver);
                 }
             }
 
-            last[key] = (field.Values[i], over.Values[i]);
+            last[key] = (ts.Values[i], field.Values[i], over.Values[i]);
         }
     }
 }

backend/tests/MoShiro.Backend.Analysis.Tests/OpTests.cs

@@ -211,6 +211,55 @@ public sealed class OpTests
         Assert.Equal(100UL, retrans.Denominator);
     }
 
+    [Fact]
+    public void Retrans_InterleavedConnections_DoNotCreateFalseRetransmissions()
+    {
+        long ts = ChunkFactory.ProbeUs(0);
+        var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!);
+        builder.Add(ts, 0, 40_000, 0, 10, 10, 1000, 0, 1);
+        builder.Add(ts + 1_000_000, 0, 40_000, 0, 10, 20, 1100, 0, 2);
+        builder.Add(ts + 2_000_000, 0, 40_000, 0, 10, 11, 1100, 0, 1);
+        builder.Add(ts + 3_000_000, 0, 40_000, 0, 10, 21, 1200, 0, 2);
+        RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100,
+            _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [builder.Build()]);
+        var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"];
+        Assert.Equal(2UL, retrans.Numerator);
+        Assert.Equal(200UL, retrans.Denominator);
+    }
+
+    [Fact]
+    public void Retrans_OutOfOrderSamples_DoNotMoveTheBaselineBackwards()
+    {
+        long ts = ChunkFactory.ProbeUs(0);
+        RawStream info = ChunkFactory.TcpInfo([
+            (ts, 0, 40_000u, 0u, 100u),
+            (ts + 2_000_000, 0, 40_000u, 2u, 300u),
+            (ts + 1_000_000, 0, 40_000u, 1u, 200u),
+            (ts + 3_000_000, 0, 40_000u, 3u, 400u),
+        ]);
+        RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100,
+            _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [info]);
+        var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"];
+        Assert.Equal(3UL, retrans.Numerator);
+        Assert.Equal(300UL, retrans.Denominator);
+    }
+
+    [Fact]
+    public void Retrans_EqualConnectionIdsOnDifferentSides_AreIndependent()
+    {
+        long ts = ChunkFactory.ProbeUs(0);
+        var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!);
+        builder.Add(ts, 0, 40_000, 0, 10, 10, 1000, 0, 1);
+        builder.Add(ts, 1, 40_000, 0, 10, 100, 2000, 0, 1);
+        builder.Add(ts + 1_000_000, 0, 40_000, 0, 10, 11, 1100, 0, 1);
+        builder.Add(ts + 1_000_000, 1, 40_000, 0, 10, 102, 2200, 0, 1);
+        RawChunk chunk = ChunkFactory.Echo(s_agent, "tcp_echo", T0, 10, 100,
+            _ => new EchoPacket(false, false, 20_000, 20_000), extraStreams: [builder.Build()]);
+        var retrans = (RatioAcc)EvaluateOneBucket(chunk, "tcp.retrans").Items["tcp.retrans"];
+        Assert.Equal(3UL, retrans.Numerator);
+        Assert.Equal(300UL, retrans.Denominator);
+    }
+
     [Fact]
     public void Connect_Outcomes_AreCountedByClass()
     {

backend/tests/MoShiro.TestSupport/AgentSimulator.cs

@@ -274,6 +274,11 @@ public sealed class AgentSimulator : IAsyncDisposable
             {
                 // 已经断开
             }
+            catch (OperationCanceledException) when (_stop.IsCancellationRequested)
+            {
+                // Canceling the receive loop may abort/dispose the transport
+                // between the State check and CloseOutputAsync.
+            }
         }
 
         if (_loop is not null)

backend/tests/MoShiro.TestSupport/ChunkFactory.cs

@@ -79,7 +79,7 @@ public static class ChunkFactory
         var builder = new StreamBuilder(Kinds.TcpEcho.FindStream("tcp_info")!);
         foreach ((long ts, byte side, uint srtt, uint retrans, uint segs) in samples)
         {
-            builder.Add(ts, side, srtt, srtt / 4, 10, retrans, segs, segs * 100L);
+            builder.Add(ts, side, srtt, srtt / 4, 10, retrans, segs, segs * 100L, null);
         }
 
         return builder.Build();

docs/design/measurement.md

@@ -117,6 +117,10 @@ RawChunk
 
 > **两侧各自记录完整行的流**（`StreamSpec.Union`，例如 `tcp_info`）：Probe 和 Agent 各自采样自己的套接字（Linux `TCP_INFO`，通过 `Socket.GetRawSocketOption`，不需要 P/Invoke），每一行带 `side` 标明属于哪一侧；合并时直接拼接，不连接，Agent 一侧的行每行只交出一次。Probe 的重传反映上行方向，Agent 的重传反映下行方向（代理场景里大流量走这个方向）。非 Linux 平台没有这类采样，对应的指标在窗口里就是没有数据。
 
+`tcp_info` 的可空 `conn` 列标识当前记录器内的一条 TCP 连接，不包含地址。Backend 按 `(side, conn)` 分别求累计计数差，避免把重连前后或同时存活的套接字混为一个序列；时间不递增的样本被忽略，任一计数器回退时重新建立基线。旧数据没有 `conn`，仍按 `side` 分组，无法事后恢复丢失的连接归属。升级时先更新 Backend 和 Agent，再更新 Probe；旧 Agent 合并采样时会丢弃尚不认识的列。
+
+TCP 回显的长度头和正文合为一次写入，避免在启用 `TCP_NODELAY` 时为长度头单独发送一个小段。`tcp.retrans` 仍按内核的重传段数与发送段数求比，不由应用消息数估算。
+
 采集层只产生**原始列**（`seq`、`size`、`t1…t4`、`t_send`/`t_recv`、`outcome`……），不做任何统计，也不认识上层的方案或协议；丢包率、RTT、抖动由 Backend 的分析层从这些列推算。
 
 每个记录流（`StreamSpec`）声明：