diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs index 8028c52..4691aa4 100644 --- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs +++ b/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, - State = _attemptState, - LastError = _lastError, + Connected = connected, + State = state, + 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 { diff --git a/backend/shared/MoShiro.Measurement/Blocks/Echo.cs b/backend/shared/MoShiro.Measurement/Blocks/Echo.cs index e29fe4f..de07a7d 100644 --- a/backend/shared/MoShiro.Measurement/Blocks/Echo.cs +++ b/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); - MeasurementDatagram.Write(packet, new MeasurementDatagramHeader(MeasurementDatagramType.EchoRequest, session.SessionId, Spec.StreamId, seq, t1), session.Key); - datagrams.Send(session, target, packet); + lock (_gate) + { + 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); + } } } } diff --git a/backend/shared/MoShiro.Measurement/Blocks/Train.cs b/backend/shared/MoShiro.Measurement/Blocks/Train.cs index f5dee5c..647ce09 100644 --- a/backend/shared/MoShiro.Measurement/Blocks/Train.cs +++ b/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 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); diff --git a/backend/shared/MoShiro.Measurement/Core/RowStore.cs b/backend/shared/MoShiro.Measurement/Core/RowStore.cs index 77845b2..dc8cb8d 100644 --- a/backend/shared/MoShiro.Measurement/Core/RowStore.cs +++ b/backend/shared/MoShiro.Measurement/Core/RowStore.cs @@ -138,6 +138,36 @@ public sealed class RowStore } } + /// Remove a row that was reserved locally but never accepted by the transport. + 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; + } + } + /// 读一个值；行不存在或该列没有设置（可空列）返回 false。 public bool TryGet(long key, int column, out long value) { diff --git a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs b/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs index fe50448..9e03202 100644 --- a/backend/shared/MoShiro.Measurement/Net/MsQuicConnection.cs +++ b/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); diff --git a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs b/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs index f403e0b..bfe4bc5 100644 --- a/backend/shared/MoShiro.Measurement/Net/QuicDatagramTransport.cs +++ b/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); /// Sends one measurement datagram on the connection for the supplied session. - public void Send(DataSession session, EndPoint target, ReadOnlySpan packet) + public bool Send(DataSession session, EndPoint target, ReadOnlySpan 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) diff --git a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs b/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs index d0f301b..53bd828 100644 --- a/backend/shared/tests/MoShiro.Measurement.Tests/MeasurementTests.cs +++ b/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]