using Fleet.Core;
using Fleet.Link;
using MoShiro.Measurement;
using MoShiro.Protocol;
namespace MoShiro.Agent;
///
/// 一条线路(本节点拨号到一个监听节点的一个 IP 族)的状态:控制层和 QUIC 数据通道各一个状态机,状态迁移只在这里发生、只在这里记日志。
/// 控制层每次迁移写一条 link state;数据通道只在“可用 ↔ 不可用”的边界上写,重拨过程中的失败只计数。
///
public sealed class LinkStatus
{
private const int MaxHandshakeSamples = 128;
private readonly object _gate = new();
private readonly string _listenerId;
private readonly string _ipVersion;
private readonly TimeProvider _time;
private readonly Queue _handshakeMs = new();
private ControlState _control = ControlState.Dialing;
private long _controlSince;
private LinkFault? _controlFault;
private int _controlAttempt;
private long? _controlRetryAt;
private QuicState _quic = QuicState.Idle;
private long _quicSince;
private long? _quicDownSince;
private LinkFault? _quicFault;
private int _quicAttempt;
private long? _quicRetryAt;
private long _handshakeAttempts;
private long _handshakeSuccesses;
private long _handshakeFailures;
private long _handshakeTimeouts;
public LinkStatus(string listenerId, string ipVersion, TimeProvider time, INodeLog log)
{
_listenerId = listenerId;
_ipVersion = ipVersion;
_time = time;
Log = log;
_controlSince = _quicSince = Now();
}
/// 当前这次尝试的日志(带 attempt_id 等上下文);控制层每次重拨前由连接更新。
public INodeLog Log { get; set; }
public ControlState Control
{
get
{
lock (_gate)
{
return _control;
}
}
}
public QuicState Quic
{
get
{
lock (_gate)
{
return _quic;
}
}
}
public bool Ready => Control == ControlState.Up && Quic == QuicState.Ready;
public void SetControl(ControlState state, LinkFault? fault = null, int attempt = 0, TimeSpan? retryIn = null)
{
ControlState from;
long inStateMs;
lock (_gate)
{
from = _control;
long now = Now();
inStateMs = now - _controlSince;
if (state != from)
{
_controlSince = now;
}
_control = state;
_controlFault = state == ControlState.Up ? null : fault ?? _controlFault;
_controlAttempt = attempt;
_controlRetryAt = retryIn is { } delay ? now + (long)delay.TotalMilliseconds : null;
}
if (state == from)
{
return;
}
string reason = state == ControlState.Backoff && fault is not null ? fault.ToString() : "";
Log.Write(state == ControlState.Backoff ? NodeLogLevel.Warn : NodeLogLevel.Info, "link state",
[
("layer", "control"), ("listener", _listenerId), ("ip_version", _ipVersion), ("from", Name(from)), ("to", Name(state)),
("in_state_ms", inStateMs), ("attempt", attempt), ("retry_in_ms", retryIn is { } r ? (long)r.TotalMilliseconds : null),
("reason", reason.Length == 0 ? null : reason),
]);
}
public void SetQuic(QuicState state, LinkFault? fault = null, int attempt = 0, TimeSpan? retryIn = null)
{
QuicState from;
long? downMs = null;
bool logged;
lock (_gate)
{
from = _quic;
long now = Now();
if (state != from)
{
_quicSince = now;
}
if (state == QuicState.Ready)
{
downMs = _quicDownSince is { } down ? now - down : null;
_quicDownSince = null;
_quicFault = null;
}
else if (state is QuicState.Retrying or QuicState.Connecting)
{
_quicDownSince ??= now;
_quicFault = fault ?? _quicFault;
}
else
{
_quicDownSince = null;
}
// 数据通道抖动时不逐次打印:恢复可用、首次失败时各一条,重拨中的失败只计数。
logged = state == QuicState.Ready || (state == QuicState.Retrying && (from == QuicState.Ready || attempt == 1));
_quic = state;
_quicAttempt = attempt;
_quicRetryAt = retryIn is { } delay ? now + (long)delay.TotalMilliseconds : null;
}
if (logged)
{
Log.Write(state == QuicState.Retrying ? NodeLogLevel.Warn : NodeLogLevel.Info, "link state",
[
("layer", "task_quic"), ("listener", _listenerId), ("ip_version", _ipVersion), ("from", Name(from)), ("to", Name(state)),
("down_ms", downMs), ("attempt", attempt), ("retry_in_ms", retryIn is { } r ? (long)r.TotalMilliseconds : null),
("reason", state == QuicState.Retrying ? fault?.ToString() : null),
]);
}
}
/// 记录一次 QUIC 握手的结果。
public void Handshake(bool succeeded, double? elapsedMs, bool timedOut)
{
lock (_gate)
{
_handshakeAttempts++;
if (succeeded)
{
_handshakeSuccesses++;
if (elapsedMs is { } ms)
{
_handshakeMs.Enqueue(ms);
while (_handshakeMs.Count > MaxHandshakeSamples)
{
_handshakeMs.Dequeue();
}
}
}
else
{
_handshakeFailures++;
if (timedOut)
{
_handshakeTimeouts++;
}
}
}
}
public LinkHealth Snapshot(Outbox outbox, int tcpBlocks, QuicTransportSnapshot? transport)
{
lock (_gate)
{
long now = Now();
double? p95 = null;
if (_handshakeMs.Count > 0)
{
double[] sorted = [.. _handshakeMs.Order()];
p95 = sorted[(int)Math.Ceiling(sorted.Length * 0.95) - 1];
}
return new LinkHealth
{
ListenerId = _listenerId,
IpVersion = _ipVersion,
Control = new ControlHealth
{
State = _control,
SinceMs = _controlSince,
LastError = _controlFault,
RetryAttempt = _controlAttempt,
RetryInMs = _control == ControlState.Backoff ? RetryIn(_controlRetryAt, now) : null,
OutboxBytes = outbox.PendingBytes,
OutboxPending = outbox.PendingCount,
},
Task = new TaskHealth
{
TcpBlocks = tcpBlocks,
Quic = new QuicHealth
{
State = _quic,
SinceMs = _quicSince,
LastError = _quicFault,
RetryAttempt = _quicAttempt,
RetryInMs = _quic == QuicState.Retrying ? RetryIn(_quicRetryAt, now) : null,
HandshakeAttempts = _handshakeAttempts,
HandshakeSuccesses = _handshakeSuccesses,
HandshakeFailures = _handshakeFailures,
HandshakeTimeouts = _handshakeTimeouts,
HandshakeP95Ms = p95,
HandshakeSampleCount = _handshakeMs.Count,
TransportPacketLossRate = transport?.PacketLossRate,
TransportPacketsSent = transport?.PacketsSent,
TransportCongestionEvents = transport?.CongestionEvents,
TransportEcnCongestionEvents = transport?.EcnCongestionEvents,
PathMtu = transport?.PathMtu,
},
},
};
}
}
private static long? RetryIn(long? at, long now) => at is { } t ? Math.Max(0, t - now) : null;
private long Now() => _time.GetUtcNow().ToUnixTimeMilliseconds();
private static string Name(ControlState state) => state.ToString().ToLowerInvariant();
private static string Name(QuicState state) => state.ToString().ToLowerInvariant();
}