[CI] [SKIP TESTS] Add detailed Probe-Agent connection diagnostics
This commit is contained in:
1 parent
0fd38b2367
commit
a7ebf61551
16 files changed
+376
-51
No files matched your search
@@ -23,6 +23,9 @@ public class DataSession
|
||||
|
||||
public byte[] Key { get; }
|
||||
|
||||
/// <summary>Public node identity for diagnostics; never the data-session identifier or key.</summary>
|
||||
public virtual string? DiagnosticPeerId => null;
|
||||
|
||||
public IDatagramConnection? DatagramConnection => Volatile.Read(ref _datagramConnection);
|
||||
|
||||
public void BindDatagramConnection(IDatagramConnection connection) => Volatile.Write(ref _datagramConnection, connection);
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using Fleet.Core;
|
||||
using MoShiro.Protocol;
|
||||
|
||||
namespace MoShiro.Measurement;
|
||||
@@ -46,13 +47,16 @@ internal sealed class MeasurementDatagramRouter
|
||||
private readonly IDataSessions _sessions;
|
||||
private readonly IAdmission? _admission;
|
||||
private readonly MonoClock _clock;
|
||||
private readonly INodeLog _log;
|
||||
private long _lastDropLog;
|
||||
private readonly ConcurrentDictionary<uint, QuicDatagramHandler> _handlers = new();
|
||||
|
||||
public MeasurementDatagramRouter(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters)
|
||||
public MeasurementDatagramRouter(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters, INodeLog? log = null)
|
||||
{
|
||||
_sessions = sessions;
|
||||
_admission = admission;
|
||||
_clock = clock;
|
||||
_log = log ?? NullNodeLog.Instance;
|
||||
Counters = counters;
|
||||
}
|
||||
|
||||
@@ -69,17 +73,34 @@ internal sealed class MeasurementDatagramRouter
|
||||
if (datagram.Length > payload.Length)
|
||||
{
|
||||
Counters.DropInvalid();
|
||||
LogDrop(connection, "oversize", datagram.Length);
|
||||
return;
|
||||
}
|
||||
|
||||
datagram.CopyTo(payload);
|
||||
payload = payload[..datagram.Length];
|
||||
if (!MeasurementDatagram.TryParseHeader(payload, out MeasurementDatagramHeader header)
|
||||
|| !_handlers.TryGetValue(header.StreamId, out QuicDatagramHandler? handler)
|
||||
|| _sessions.Find(header.SessionId) is not { } session
|
||||
|| !MeasurementDatagram.VerifyMac(payload, session.Key))
|
||||
if (!MeasurementDatagram.TryParseHeader(payload, out MeasurementDatagramHeader header))
|
||||
{
|
||||
Counters.DropInvalid();
|
||||
LogDrop(connection, "invalid_header", datagram.Length);
|
||||
return;
|
||||
}
|
||||
if (!_handlers.TryGetValue(header.StreamId, out QuicDatagramHandler? handler))
|
||||
{
|
||||
Counters.DropInvalid();
|
||||
LogDrop(connection, "unknown_stream", datagram.Length, header.StreamId);
|
||||
return;
|
||||
}
|
||||
if (_sessions.Find(header.SessionId) is not { } session)
|
||||
{
|
||||
Counters.DropInvalid();
|
||||
LogDrop(connection, "unknown_session", datagram.Length, header.StreamId);
|
||||
return;
|
||||
}
|
||||
if (!MeasurementDatagram.VerifyMac(payload, session.Key))
|
||||
{
|
||||
Counters.DropInvalid();
|
||||
LogDrop(connection, "invalid_mac", datagram.Length, header.StreamId);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -87,6 +108,7 @@ internal sealed class MeasurementDatagramRouter
|
||||
if (remote is null)
|
||||
{
|
||||
Counters.DropInvalid();
|
||||
LogDrop(connection, "missing_remote", datagram.Length, header.StreamId);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -97,14 +119,35 @@ internal sealed class MeasurementDatagramRouter
|
||||
{
|
||||
case Admission.Invalid:
|
||||
Counters.DropInvalid();
|
||||
LogDrop(connection, "admission_invalid", datagram.Length, header.StreamId);
|
||||
return;
|
||||
case Admission.RateLimited:
|
||||
Counters.DropRateLimited();
|
||||
LogDrop(connection, "rate_limited", datagram.Length, header.StreamId);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
session.BindDatagramConnection(connection);
|
||||
connection.NoteAuthenticatedDatagram(session.DiagnosticPeerId, header.StreamId);
|
||||
handler(new QuicDatagramDelivery(header, payload, receiveUs, session, connection, Counters));
|
||||
}
|
||||
|
||||
private void LogDrop(MsQuicConnection connection, string reason, int bytes, uint? stream = null)
|
||||
{
|
||||
// One bounded throttle for this transport, regardless of attacker-controlled sources/streams.
|
||||
long now = _clock.Time.GetTimestamp();
|
||||
long previous = Interlocked.Read(ref _lastDropLog);
|
||||
if (previous != 0 && _clock.Time.GetElapsedTime(previous - 1, now) < TimeSpan.FromSeconds(30))
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (Interlocked.CompareExchange(ref _lastDropLog, now + 1, previous) != previous)
|
||||
{
|
||||
return;
|
||||
}
|
||||
_log.Warn("measurement datagram dropped", ("phase", "measurement"), ("reason", reason),
|
||||
("quic_id", connection.DiagnosticId), ("remote", connection.RemoteEndPoint), ("bytes", bytes), ("stream", stream),
|
||||
("invalid_total", Counters.DroppedInvalid), ("rate_limited_total", Counters.DroppedRateLimited));
|
||||
}
|
||||
}
|
||||
@@ -40,7 +40,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
private readonly byte[]? _expectedPeerSpki;
|
||||
private readonly Action<MsQuicConnection>? _closed;
|
||||
private readonly Action<IPEndPoint>? _peerAddressChanged;
|
||||
private readonly INodeLog _log;
|
||||
private readonly ContextNodeLog _log;
|
||||
private readonly Dictionary<nint, SendState> _sends = [];
|
||||
private nint _registration;
|
||||
private nint _configuration;
|
||||
@@ -51,6 +51,14 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
private double? _handshakeDurationMs;
|
||||
private QuicTransportSnapshot? _transportSnapshot;
|
||||
private long _statisticsAt;
|
||||
private NativeCounters? _nativeCounters;
|
||||
private readonly long _createdAt = Stopwatch.GetTimestamp();
|
||||
private int _localPort;
|
||||
private int _receivedDatagram;
|
||||
private long _sendRejected;
|
||||
private long _sendRejectedAt;
|
||||
|
||||
private sealed record NativeCounters(ulong Sent, ulong Received, long At);
|
||||
|
||||
private MsQuicConnection(
|
||||
nint handle,
|
||||
@@ -67,7 +75,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
_configuration = configuration;
|
||||
_expectedPeerSpki = expectedPeerSpki;
|
||||
DatagramHandler = datagramHandler;
|
||||
_log = log;
|
||||
_log = new ContextNodeLog(log, ("quic_id", DiagnosticId), ("side", expectedPeerSpki is null ? "agent" : "probe"));
|
||||
_closed = closed;
|
||||
_peerAddressChanged = peerAddressChanged;
|
||||
_self = GCHandle.Alloc(this);
|
||||
@@ -75,6 +83,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
|
||||
public MsQuicDatagramCallback? DatagramHandler { get; set; }
|
||||
|
||||
internal string DiagnosticId { get; } = Guid.NewGuid().ToString("N");
|
||||
|
||||
public IPEndPoint? RemoteEndPoint { get; private set; }
|
||||
|
||||
public IPEndPoint? LocalEndPoint => Volatile.Read(ref _handle) is nint handle && handle != 0
|
||||
@@ -131,6 +141,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
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));
|
||||
Volatile.Write(ref _nativeCounters, new NativeCounters(stats.SendTotalPackets, stats.RecvTotalPackets, now));
|
||||
}
|
||||
catch (InvalidOperationException)
|
||||
{
|
||||
@@ -153,11 +164,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
long handshakeStarted = 0;
|
||||
try
|
||||
{
|
||||
log.Info("opening MsQuic client registration");
|
||||
registration = MsQuicApi.RegistrationOpen("moshiro-probe");
|
||||
log.Info("opening MsQuic client configuration");
|
||||
configuration = MsQuicApi.ConfigurationOpen(registration, QuicAlpn.Value);
|
||||
log.Info("loading MsQuic client credentials");
|
||||
LoadClientCredentials(configuration);
|
||||
var endpoint = new IPEndPoint(address.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any, localPort);
|
||||
SocketAddress serialized = endpoint.Serialize();
|
||||
@@ -171,19 +179,18 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
try
|
||||
{
|
||||
nint handle = MsQuicApi.ConnectionOpen(registration, s_callback, 0);
|
||||
log.Info("opened MsQuic client connection");
|
||||
connection = new MsQuicConnection(handle, registration, configuration, expectedPeerSpki, datagramHandler, log, null, null);
|
||||
connection.RemoteEndPoint = new IPEndPoint(address, port);
|
||||
connection._localPort = localPort;
|
||||
registration = 0;
|
||||
configuration = 0;
|
||||
MsQuicApi.SetContext(handle, GCHandle.ToIntPtr(connection._self));
|
||||
log.Info("enabling MsQuic datagram receive");
|
||||
MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1);
|
||||
log.Info("binding MsQuic local address", ("local_port", localPort));
|
||||
MsQuicApi.ConnectionBind(handle, localAddress);
|
||||
log.Info("starting MsQuic QUIC handshake", ("remote", address), ("port", port), ("local_port", localPort));
|
||||
connection._log.Info("starting MsQuic QUIC handshake", ("phase", "quic_handshake"), ("remote", connection.RemoteEndPoint),
|
||||
("local_port", localPort), ("timeout_ms", 10_000));
|
||||
handshakeStarted = Stopwatch.GetTimestamp();
|
||||
MsQuicApi.ConnectionStart(handle, connection._configuration, nativeFamily, name, checked((ushort)port));
|
||||
log.Info("MsQuic connection start returned");
|
||||
}
|
||||
finally
|
||||
{
|
||||
@@ -194,10 +201,14 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
connection._handshakeDurationMs = Stopwatch.GetElapsedTime(handshakeStarted).TotalMilliseconds;
|
||||
return connection;
|
||||
}
|
||||
catch
|
||||
catch (Exception ex)
|
||||
{
|
||||
if (connection is not null)
|
||||
{
|
||||
if (!ct.IsCancellationRequested)
|
||||
{
|
||||
connection.LogState(NodeLogLevel.Warn, "QUIC handshake failed", ("exception", ex.GetType().Name));
|
||||
}
|
||||
await connection.DisposeAsync();
|
||||
}
|
||||
|
||||
@@ -211,6 +222,10 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
MsQuicApi.RegistrationClose(registration);
|
||||
}
|
||||
|
||||
if (ex is TimeoutException)
|
||||
{
|
||||
throw new TimeoutException($"QUIC handshake timed out after 10000 ms: remote={address}:{port}, local_udp_port={localPort}", ex);
|
||||
}
|
||||
throw;
|
||||
}
|
||||
}
|
||||
@@ -226,12 +241,25 @@ 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));
|
||||
connection.RefreshRemoteEndPoint();
|
||||
connection._log.Info("accepted MsQuic QUIC connection", ("phase", "quic_handshake"), ("remote", connection.RemoteEndPoint));
|
||||
MsQuicApi.SetParam(handle, DatagramReceiveEnabled, 1);
|
||||
// Transport parameters must be set before SetConfiguration starts TLS.
|
||||
MsQuicApi.ConnectionSetConfiguration(handle, configuration);
|
||||
return connection;
|
||||
}
|
||||
|
||||
private int _authenticatedDatagram;
|
||||
|
||||
internal void NoteAuthenticatedDatagram(string? peerId, uint streamId)
|
||||
{
|
||||
if (Interlocked.Exchange(ref _authenticatedDatagram, 1) == 0)
|
||||
{
|
||||
_log.Info("first authenticated measurement datagram", ("phase", "measurement"),
|
||||
("peer", peerId), ("remote", RemoteEndPoint), ("stream", streamId));
|
||||
}
|
||||
}
|
||||
|
||||
public bool Send(ReadOnlySpan<byte> datagram)
|
||||
{
|
||||
// ConnectionClose must not free the handle or a pending send while
|
||||
@@ -247,6 +275,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
nint handle = Volatile.Read(ref _handle);
|
||||
if (handle == 0 || !IsReady || datagram.Length > MaxDatagramBytes)
|
||||
{
|
||||
LogSendRejected(handle == 0 ? "connection_closed" : !IsReady ? "datagram_not_ready" : "datagram_too_large", datagram.Length, null);
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -282,9 +311,50 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
}
|
||||
|
||||
ReleaseSend(context);
|
||||
LogSendRejected("native_send_failed", datagram.Length, status);
|
||||
return false;
|
||||
}
|
||||
|
||||
private void LogSendRejected(string reason, int bytes, uint? status)
|
||||
{
|
||||
long rejected = Interlocked.Increment(ref _sendRejected);
|
||||
long now = Stopwatch.GetTimestamp();
|
||||
long previous = Interlocked.Read(ref _sendRejectedAt);
|
||||
if (previous != 0 && Stopwatch.GetElapsedTime(previous, now) < TimeSpan.FromSeconds(30))
|
||||
{
|
||||
return;
|
||||
}
|
||||
if (Interlocked.CompareExchange(ref _sendRejectedAt, now, previous) != previous)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_log.Warn("QUIC datagram submission rejected", ("phase", "quic_data"), ("reason", reason), ("remote", RemoteEndPoint),
|
||||
("bytes", bytes), ("max_datagram", MaxDatagramBytes), ("rejected_total", rejected), ("status", status is { } s ? $"0x{s:X8}" : null));
|
||||
}
|
||||
|
||||
// Diagnostics only read managed snapshots. Never query the native worker from
|
||||
// timeout, heartbeat, or teardown paths: the worker may itself be blocked.
|
||||
private void LogState(NodeLogLevel level, string message, params (string Key, object? Value)[] details)
|
||||
{
|
||||
NativeCounters? counters = Volatile.Read(ref _nativeCounters);
|
||||
_log.Write(level, message, [
|
||||
("remote", RemoteEndPoint), ("local_port", _localPort),
|
||||
("phase", _connected.Task.IsCompletedSuccessfully ? "quic_data" : "quic_handshake"),
|
||||
("elapsed_ms", (long)Stopwatch.GetElapsedTime(_createdAt).TotalMilliseconds),
|
||||
("handshake_completed", _connected.Task.IsCompletedSuccessfully),
|
||||
("packets_sent", counters?.Sent), ("packets_received", counters?.Received),
|
||||
("stats_age_ms", counters is null ? null : (long?)Stopwatch.GetElapsedTime(counters.At).TotalMilliseconds),
|
||||
.. details,
|
||||
]);
|
||||
}
|
||||
|
||||
internal static string StatusName(uint status) => status == 0 ? "success" : !OperatingSystem.IsLinux() ? "native_error" : status switch
|
||||
{
|
||||
1 => "invalid_state", 22 => "invalid_parameter", 62 => "connection_idle", 103 => "handshake_failure",
|
||||
110 => "connection_timeout", 111 => "connection_refused", 113 => "unreachable", 125 => "aborted", _ => "native_error",
|
||||
};
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
lock (_nativeGate)
|
||||
@@ -375,22 +445,31 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
switch (eventType)
|
||||
{
|
||||
case ConnectionEventConnected:
|
||||
connection._log.Info("MsQuic QUIC handshake completed");
|
||||
connection.RefreshRemoteEndPoint();
|
||||
try
|
||||
{
|
||||
connection._localPort = connection.LocalEndPoint?.Port ?? connection._localPort;
|
||||
}
|
||||
catch (InvalidOperationException)
|
||||
{
|
||||
// An unavailable diagnostic address must not fail a successful handshake.
|
||||
}
|
||||
connection._connected.TrySetResult();
|
||||
connection.LogState(NodeLogLevel.Info, "MsQuic QUIC handshake completed");
|
||||
break;
|
||||
case ConnectionEventShutdownByTransport:
|
||||
{
|
||||
uint status = unchecked((uint)Marshal.ReadInt32(eventPointer, IntPtr.Size));
|
||||
ulong errorCode = unchecked((ulong)Marshal.ReadInt64(eventPointer, IntPtr.Size * 2));
|
||||
connection._log.Error("MsQuic QUIC transport shutdown", ("status", $"0x{status:X8}"), ("error_code", errorCode));
|
||||
connection.LogState(NodeLogLevel.Error, "MsQuic QUIC transport shutdown", ("status", $"0x{status:X8}"),
|
||||
("status_name", StatusName(status)), ("error_code", errorCode));
|
||||
connection._connected.TrySetException(new IOException($"QUIC connection closed during handshake (transport status 0x{status:X8}, error code {errorCode})."));
|
||||
break;
|
||||
}
|
||||
case ConnectionEventShutdownByPeer:
|
||||
{
|
||||
ulong errorCode = unchecked((ulong)Marshal.ReadInt64(eventPointer, IntPtr.Size));
|
||||
connection._log.Warn("MsQuic QUIC peer shutdown", ("error_code", errorCode));
|
||||
connection.LogState(NodeLogLevel.Warn, "MsQuic QUIC peer shutdown", ("error_code", errorCode));
|
||||
connection._connected.TrySetException(new IOException($"QUIC connection closed during handshake by peer (error code {errorCode})."));
|
||||
break;
|
||||
}
|
||||
@@ -403,6 +482,7 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
IPEndPoint? endpoint = ReadEndpoint(address);
|
||||
if (endpoint is not null)
|
||||
{
|
||||
connection._log.Info("QUIC peer address changed", ("previous", connection.RemoteEndPoint), ("remote", endpoint));
|
||||
connection.RemoteEndPoint = endpoint;
|
||||
connection._peerAddressChanged?.Invoke(endpoint);
|
||||
}
|
||||
@@ -446,8 +526,13 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
private void OnDatagramStateChanged(nint eventPointer)
|
||||
{
|
||||
(bool sendEnabled, int maxSendLength) = ReadDatagramState(eventPointer);
|
||||
Volatile.Write(ref _maxDatagramBytes, maxSendLength);
|
||||
Volatile.Write(ref _sendEnabled, sendEnabled ? 1 : 0);
|
||||
int previousMax = Interlocked.Exchange(ref _maxDatagramBytes, maxSendLength);
|
||||
int previousEnabled = Interlocked.Exchange(ref _sendEnabled, sendEnabled ? 1 : 0);
|
||||
if (previousMax != maxSendLength || previousEnabled != (sendEnabled ? 1 : 0))
|
||||
{
|
||||
_log.Info("QUIC datagram negotiation changed", ("phase", "datagram_negotiation"), ("remote", RemoteEndPoint),
|
||||
("send_enabled", sendEnabled), ("max_datagram", maxSendLength));
|
||||
}
|
||||
}
|
||||
|
||||
internal static (bool SendEnabled, int MaxSendLength) ReadDatagramState(nint eventPointer)
|
||||
@@ -475,6 +560,10 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
var buffer = Marshal.PtrToStructure<MsQuicApi.QuicBuffer>(bufferPointer);
|
||||
if (buffer.Buffer != 0 && buffer.Length > 0 && buffer.Length <= 65535)
|
||||
{
|
||||
if (Interlocked.Exchange(ref _receivedDatagram, 1) == 0)
|
||||
{
|
||||
_log.Info("first QUIC datagram received", ("phase", "quic_data"), ("remote", RemoteEndPoint), ("bytes", buffer.Length));
|
||||
}
|
||||
unsafe
|
||||
{
|
||||
DatagramHandler?.Invoke(this, new ReadOnlySpan<byte>((void*)buffer.Buffer, checked((int)buffer.Length)));
|
||||
@@ -581,6 +670,8 @@ internal sealed class MsQuicConnection : IAsyncDisposable, IDatagramConnection
|
||||
|
||||
private void OnShutdownComplete(nint handle, nint eventPointer)
|
||||
{
|
||||
UpdateTransportSnapshot();
|
||||
LogState(NodeLogLevel.Info, "QUIC connection shutdown completed", ("rejected_sends", Interlocked.Read(ref _sendRejected)));
|
||||
if (!_connected.Task.IsCompletedSuccessfully)
|
||||
{
|
||||
byte flags = Marshal.ReadByte(eventPointer, IntPtr.Size);
|
||||
|
||||
@@ -76,6 +76,8 @@ internal sealed class MsQuicListener : IAsyncDisposable
|
||||
MsQuicApi.ListenerStart(listener, QuicAlpn.Value, address);
|
||||
effectivePort = ReadPort(MsQuicApi.GetAddress(listener, ListenerLocalAddress));
|
||||
bound.Add(effectivePort);
|
||||
_log.Info("QUIC listener ready", ("bind", bindAddress), ("port", effectivePort),
|
||||
("ip_version", bindAddress.AddressFamily == AddressFamily.InterNetworkV6 ? "ipv6" : "ipv4"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -131,7 +133,6 @@ internal sealed class MsQuicListener : IAsyncDisposable
|
||||
}
|
||||
|
||||
nint connectionHandle = Marshal.ReadIntPtr(@event, IntPtr.Size * 2);
|
||||
owner._log.Info("accepted MsQuic QUIC connection");
|
||||
MsQuicConnection? connection = null;
|
||||
connection = MsQuicConnection.Accept(
|
||||
connectionHandle,
|
||||
|
||||
@@ -30,7 +30,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
|
||||
/// <summary>Constructs an Agent-side QUIC listener.</summary>
|
||||
public QuicDatagramTransport(IDataSessions sessions, IAdmission? admission, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPAddress bind, IReadOnlyList<int> ports, X509Certificate2 serverCertificate)
|
||||
{
|
||||
_router = new MeasurementDatagramRouter(sessions, admission, clock, counters);
|
||||
_router = new MeasurementDatagramRouter(sessions, admission, clock, counters, log);
|
||||
_log = log;
|
||||
_onDatagram = _router.Dispatch;
|
||||
_bind = bind;
|
||||
@@ -41,7 +41,7 @@ public sealed class QuicDatagramTransport : IDisposable, IAsyncDisposable
|
||||
/// <summary>Constructs a Probe-side QUIC dialer for one Agent session.</summary>
|
||||
public QuicDatagramTransport(DataSession session, MonoClock clock, DataPlaneCounters counters, INodeLog log, IPEndPoint remote, byte[] peerSpki, int previousSourcePort)
|
||||
{
|
||||
_router = new MeasurementDatagramRouter(new SingleSession(session), null, clock, counters);
|
||||
_router = new MeasurementDatagramRouter(new SingleSession(session), null, clock, counters, log);
|
||||
_log = log;
|
||||
_onDatagram = _router.Dispatch;
|
||||
_bind = remote.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
using System.Net;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Runtime.InteropServices;
|
||||
using System.Security.Cryptography;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
@@ -31,11 +32,23 @@ public sealed class MsQuicConnectionTests
|
||||
[Fact]
|
||||
public async Task DatagramConnection_CanSendCloseAndReconnect_WithoutBlockingNativeWorkers()
|
||||
{
|
||||
var log = new CaptureLog();
|
||||
var counters = new DataPlaneCounters();
|
||||
var router = new MeasurementDatagramRouter(new SingleSession(new DataSession(1, new byte[32])), null,
|
||||
new MonoClock(TimeProvider.System), counters, log);
|
||||
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);
|
||||
(connection, data) =>
|
||||
{
|
||||
// Bad measurement payloads must be counted without flooding logs.
|
||||
for (int i = 0; i < 100; i++)
|
||||
{
|
||||
router.Dispatch(connection, data);
|
||||
}
|
||||
connection.Send(data);
|
||||
}, log, null);
|
||||
listener.Start();
|
||||
|
||||
for (int attempt = 0; attempt < 3; attempt++)
|
||||
@@ -43,7 +56,7 @@ public sealed class MsQuicConnectionTests
|
||||
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);
|
||||
key.ExportSubjectPublicKeyInfo(), (_, data) => reply.TrySetResult(data.ToArray()), log, timeout.Token);
|
||||
try
|
||||
{
|
||||
while (!client.IsReady)
|
||||
@@ -72,6 +85,22 @@ public sealed class MsQuicConnectionTests
|
||||
await client.WaitForShutdownAsync().WaitAsync(TimeSpan.FromSeconds(3));
|
||||
}
|
||||
}
|
||||
Assert.True(counters.DroppedInvalid >= 300);
|
||||
var dropped = Assert.Single(log.Items, x => x.Message == "measurement datagram dropped");
|
||||
Assert.Equal("invalid_header", dropped.Fields["reason"]);
|
||||
Assert.NotNull(dropped.Fields["remote"]);
|
||||
Assert.NotNull(dropped.Fields["quic_id"]);
|
||||
var connected = log.Items.Where(x => x.Message == "MsQuic QUIC handshake completed").ToArray();
|
||||
Assert.Equal(6, connected.Select(x => x.Fields["quic_id"]).Distinct().Count());
|
||||
Assert.All(connected, entry => Assert.NotNull(entry.Fields["remote"]));
|
||||
Assert.DoesNotContain(log.Items.SelectMany(x => x.Fields.Keys), key => key is "key" or "session_id" or "mac" or "signature");
|
||||
}
|
||||
|
||||
private sealed class CaptureLog : INodeLog
|
||||
{
|
||||
public ConcurrentQueue<(string Message, Dictionary<string, object?> Fields)> Items { get; } = new();
|
||||
|
||||
public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields) =>
|
||||
Items.Enqueue((message, fields.ToDictionary(x => x.Key, x => x.Value)));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user