justme
ad5efc62ff
Replace Probe with Agent across node runtime, Fleet contracts, backend state, API and storage. Report incoming sessions and outgoing link health through each Agent uplink, and retain directed link controls and retry. Update protocol fixtures and end-to-end coverage for the unified model; accept requested cancellation during the Fleet timeout test cleanup.
307 lines
13 KiB
C#
307 lines
13 KiB
C#
using System.Text.Json;
|
|
using Fleet.Core;
|
|
using Fleet.Link;
|
|
using Fleet.Rpc;
|
|
using MoShiro.Measurement;
|
|
using MoShiro.Protocol;
|
|
|
|
namespace MoShiro.Agent;
|
|
|
|
/// <summary>Agent → Backend 的上行：事件和测试状态是尽力而为的通知（链路断开时丢弃；列块走发件箱，不会丢）。</summary>
|
|
public sealed class BackendUplink
|
|
{
|
|
private readonly TimeProvider _time;
|
|
private volatile FleetLink? _link;
|
|
|
|
public BackendUplink(TimeProvider time)
|
|
{
|
|
_time = time;
|
|
}
|
|
|
|
public void Attach(FleetLink link) => _link = link;
|
|
|
|
public bool IsConnected => _link?.CurrentSession is not null;
|
|
|
|
public async ValueTask EmitEventAsync(string kind, string? detail = null, ConnectedDialer? dialer = null)
|
|
{
|
|
RpcSession<LinkContext>? session = _link?.CurrentSession;
|
|
if (session is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var parameters = new AgentEventParams { Kind = kind, AtUtcMs = _time.GetUtcNow().ToUnixTimeMilliseconds(), Detail = detail, Dialer = dialer };
|
|
await SafeAsync(() => session.NotifyAsync(MoShiroMethods.Event, parameters, MoShiroJsonContext.Default.AgentEventParams, CancellationToken.None));
|
|
}
|
|
|
|
public async ValueTask SendHostAsync(HostSnapshot snapshot)
|
|
{
|
|
RpcSession<LinkContext>? session = _link?.CurrentSession;
|
|
if (session is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await SafeAsync(() => session.NotifyAsync(MoShiroMethods.Host, snapshot, MoShiroJsonContext.Default.HostSnapshot, CancellationToken.None));
|
|
}
|
|
|
|
public async ValueTask SendTestStatusAsync(TestId test, string status, int? phase = null, string? errorCode = null)
|
|
{
|
|
RpcSession<LinkContext>? session = _link?.CurrentSession;
|
|
if (session is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var parameters = new TestStatusParams { TestId = test, Status = status, Phase = phase, ErrorCode = errorCode };
|
|
await SafeAsync(() => session.NotifyAsync(MoShiroMethods.TestStatus, parameters, MoShiroJsonContext.Default.TestStatusParams, CancellationToken.None));
|
|
}
|
|
|
|
private static async ValueTask SafeAsync(Func<ValueTask> send)
|
|
{
|
|
try
|
|
{
|
|
await send();
|
|
}
|
|
catch (Exception ex) when (ex is ObjectDisposedException or InvalidOperationException or IOException or System.Net.WebSockets.WebSocketException)
|
|
{
|
|
// 链路刚好断开：状态在重连之后由 hello 与下一次事件补上
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 拨号方 的点对点会话：每次握手成功之后登记会话（同一个拨号方的新会话取代它自己的旧会话，不同拨号方的会话并存）、为这个会话起一组线路任务的流并下发线路任务集、接收拨号方的列块并交给
|
|
/// <see cref="ChunkAssembler"/>。会话结束时通知 Backend，并让正在运行的测试以 <c>dialer_offline</c> 结束。
|
|
/// </summary>
|
|
public sealed class PeerHub
|
|
{
|
|
private const int DedupeWindow = 1024;
|
|
|
|
/// <summary>超过这么久没有收到拨号方的任何帧就认为会话已死（拨号方 每 30 秒至少发一次时钟同步）。
|
|
/// 要比线路可能的中断长得多：差的线路上几十秒甚至一两分钟的黑洞时有发生（TCP RTT 能到 30 秒），75 秒就关会话会把一次中断变成几分钟的数据空白（要重新握手，而握手在差线路上也慢）。</summary>
|
|
private static readonly TimeSpan IdleLimit = TimeSpan.FromSeconds(180);
|
|
|
|
private readonly RpcRouter<PeerContext> _router;
|
|
private readonly SessionRegistry _sessions;
|
|
private readonly MonoClock _clock;
|
|
private readonly ChunkAssembler _chunks;
|
|
private readonly TaskManager _tasks;
|
|
private readonly BackendUplink _uplink;
|
|
private readonly TimeProvider _time;
|
|
private readonly INodeLog _log;
|
|
|
|
/// <summary>对等面入站方法的路由器（只读，测试用）。</summary>
|
|
public RpcRouter<PeerContext> Router => _router;
|
|
|
|
public PeerHub(SessionRegistry sessions, MonoClock clock, ChunkAssembler chunks, TaskManager tasks, BackendUplink uplink, TimeProvider time, INodeLog log)
|
|
{
|
|
_router = BuildRouter();
|
|
_sessions = sessions;
|
|
_clock = clock;
|
|
_chunks = chunks;
|
|
_tasks = tasks;
|
|
_uplink = uplink;
|
|
_time = time;
|
|
_log = log;
|
|
}
|
|
|
|
/// <summary>拨号方 会话断开（这个拨号方上的测试据此中止）。</summary>
|
|
public Action<DialerSession>? OnDialerLost { get; set; }
|
|
|
|
public async Task OnSessionAsync(PeerSessionInfo info, IRpcTransport transport, CancellationToken ct)
|
|
{
|
|
var dialer = new DialerSession(info) { ConnectedAtUtcMs = _time.GetUtcNow().ToUnixTimeMilliseconds() };
|
|
var log = new ContextNodeLog(_log, ("dialer", dialer.DialerId), ("ip_version", dialer.IpVersion), ("control_id", Guid.NewGuid().ToString("N")));
|
|
var rpc = new RpcSession<PeerContext>(transport, _router, new PeerContext(dialer, _clock), _time);
|
|
dialer.Rpc = rpc;
|
|
|
|
var seen = new HashSet<Guid>();
|
|
var order = new Queue<Guid>();
|
|
rpc.OnBinary = async (frame, token) =>
|
|
{
|
|
if (!OutboxFrame.TryParse(frame, out byte channel, out Guid id, out ReadOnlyMemory<byte> payload) || channel != MoShiroChannels.ColumnBlock)
|
|
{
|
|
return;
|
|
}
|
|
|
|
// 至少一次投递：重复到达的列块只处理一次，但每次都要确认
|
|
if (seen.Add(id))
|
|
{
|
|
order.Enqueue(id);
|
|
if (order.Count > DedupeWindow)
|
|
{
|
|
seen.Remove(order.Dequeue());
|
|
}
|
|
|
|
Result<RawChunk> decoded = RawChunkCodec.Decode(payload.Span);
|
|
if (decoded.TryGet(out RawChunk? fragment, out FleetError? error))
|
|
{
|
|
_chunks.Accept(fragment, dialer);
|
|
}
|
|
else
|
|
{
|
|
_log.Warn("fragment undecodable", ("err", error.Code), ("detail", error.Message));
|
|
}
|
|
}
|
|
|
|
var ack = new OutboxAckParams { Ids = [id.ToString()] };
|
|
await rpc.NotifyAsync(FleetMethods.OutboxAck, ack, FleetRpcContext.Default.OutboxAckParams, token);
|
|
};
|
|
|
|
DialerSession? previous = _sessions.Replace(dialer);
|
|
if (previous?.Rpc is { } old)
|
|
{
|
|
previous.CloseReason = "replaced";
|
|
log.Info("replacing previous dialer control session");
|
|
await old.CloseAsync("replaced", CancellationToken.None);
|
|
}
|
|
|
|
// 这个会话自己的一组流：序号从 0 开始，上一个会话留下的记录不会算到它头上
|
|
LineTaskSet initial = _tasks.OpenSession(dialer);
|
|
|
|
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
|
Task run = rpc.RunAsync(linked.Token);
|
|
Task watchdog = WatchIdleAsync(dialer, rpc, log, linked.Token);
|
|
log.Info("dialer connected", ("phase", "control_session"), ("version", info.Version ?? ""), ("remote", info.Remote));
|
|
await _uplink.EmitEventAsync(AgentEventKinds.DialerConnected, info.PeerId, Describe(dialer));
|
|
await PushAsync(dialer, initial, linked.Token);
|
|
try
|
|
{
|
|
await run;
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// 停止
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
dialer.CloseReason ??= "control_error";
|
|
log.Warn("dialer control session failed", ("exception", ex.GetType().Name), ("err", ex.Message));
|
|
throw;
|
|
}
|
|
finally
|
|
{
|
|
_sessions.Remove(dialer);
|
|
await linked.CancelAsync();
|
|
await watchdog;
|
|
await rpc.DisposeAsync();
|
|
await _tasks.CloseSessionAsync(dialer);
|
|
log.Info("dialer disconnected", ("phase", "control_session"),
|
|
("reason", dialer.CloseReason ?? (ct.IsCancellationRequested ? "service_stopping" : "peer_closed")),
|
|
("duration_ms", _time.GetUtcNow().ToUnixTimeMilliseconds() - dialer.ConnectedAtUtcMs),
|
|
("datagram_bound", dialer.DatagramConnection is not null), ("data_remote", dialer.DatagramRemoteEndPoint));
|
|
OnDialerLost?.Invoke(dialer);
|
|
await _uplink.EmitEventAsync(AgentEventKinds.DialerDisconnected, info.PeerId, Describe(dialer));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// 拨号方 每 30 秒发一次时钟同步请求，所以会话里至少每 30 秒有一帧：超过 <see cref="IdleLimit"/> 没有收到任何东西，
|
|
/// 说明拨号方已经静默死亡（断电、进程被冻结），关闭会话，让 Backend 看到 dialer_disconnected，而不是一直显示在线。
|
|
/// </summary>
|
|
private async Task WatchIdleAsync(DialerSession dialer, RpcSession<PeerContext> rpc, INodeLog log, CancellationToken ct)
|
|
{
|
|
long lastSummary = _time.GetTimestamp();
|
|
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(10), _time);
|
|
try
|
|
{
|
|
while (await timer.WaitForNextTickAsync(ct))
|
|
{
|
|
if (rpc.IdleFor > IdleLimit)
|
|
{
|
|
dialer.CloseReason = "idle_timeout";
|
|
log.Warn("dialer session is silent, closing it", ("phase", "control_session"), ("idle_s", (long)rpc.IdleFor.TotalSeconds), ("limit_s", IdleLimit.TotalSeconds));
|
|
await rpc.CloseAsync("idle", CancellationToken.None);
|
|
return;
|
|
}
|
|
if (_time.GetElapsedTime(lastSummary) >= TimeSpan.FromSeconds(60))
|
|
{
|
|
lastSummary = _time.GetTimestamp();
|
|
log.Info("dialer session summary", ("phase", "control_session"), ("remote", dialer.Info.Remote),
|
|
("control_idle_s", (long)rpc.IdleFor.TotalSeconds), ("datagram_bound", dialer.DatagramConnection is not null),
|
|
("data_remote", dialer.DatagramRemoteEndPoint));
|
|
}
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// 会话结束
|
|
}
|
|
}
|
|
|
|
/// <summary>当前连着的拨号方（心跳和 Backend 重连之后的补报都用它）。</summary>
|
|
public IReadOnlyList<ConnectedDialer> ConnectedDialers() => [.. _sessions.All.Select(Describe)];
|
|
|
|
private static ConnectedDialer Describe(DialerSession p) => new() { Id = new NodeId(p.DialerId), Version = p.Version, Remote = p.RemoteAddress, SinceUtcMs = p.ConnectedAtUtcMs, IpVersion = p.IpVersion };
|
|
|
|
/// <summary>向一个拨号方下发它的线路任务集（拨号方 连上时由 <see cref="OnSessionAsync"/> 先发一份）。</summary>
|
|
public async ValueTask PushToAsync(DialerSession dialer, LineTaskSet set) => await PushAsync(dialer, set, CancellationToken.None);
|
|
|
|
/// <summary>Backend 要求断开某个拨号方（它会按退避重连，除非已经被吊销）。没有这个会话时返回 false。</summary>
|
|
public async ValueTask<bool> DisconnectAsync(string dialerId)
|
|
{
|
|
bool disconnected = false;
|
|
foreach (DialerSession dialer in _sessions.All)
|
|
{
|
|
if (dialer.DialerId == dialerId && dialer.Rpc is { } rpc)
|
|
{
|
|
dialer.CloseReason = "disconnected";
|
|
await rpc.CloseAsync("disconnected", CancellationToken.None);
|
|
disconnected = true;
|
|
}
|
|
}
|
|
|
|
return disconnected;
|
|
}
|
|
|
|
/// <summary>名单（期望状态里的 peers）里已经没有的拨号方：立刻关闭它的会话（吊销生效）。</summary>
|
|
public void CloseNotIn(IReadOnlySet<string> allowed)
|
|
{
|
|
foreach (DialerSession dialer in _sessions.All)
|
|
{
|
|
if (!allowed.Contains(dialer.DialerId) && dialer.Rpc is { } rpc)
|
|
{
|
|
dialer.CloseReason = "revoked";
|
|
_log.Info("dialer is no longer allowed, closing its session", ("dialer", dialer.DialerId));
|
|
_ = rpc.CloseAsync("revoked", CancellationToken.None).AsTask();
|
|
}
|
|
}
|
|
}
|
|
|
|
private async ValueTask PushAsync(DialerSession dialer, LineTaskSet set, CancellationToken ct)
|
|
{
|
|
if (dialer.Rpc is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var parameters = new StateSyncParams
|
|
{
|
|
Type = LineTaskSet.StateType,
|
|
Version = set.Version,
|
|
Doc = JsonSerializer.SerializeToElement(set, MoShiroJsonContext.Default.LineTaskSet),
|
|
};
|
|
try
|
|
{
|
|
Result<StateSyncResult> result = await dialer.Rpc.RequestAsync(FleetMethods.StateSync, parameters, FleetRpcContext.Default.StateSyncParams, FleetRpcContext.Default.StateSyncResult, ct);
|
|
if (!result.IsSuccess)
|
|
{
|
|
_log.Warn("dialer rejected the line task set", ("err", result.Error.Code));
|
|
}
|
|
}
|
|
catch (Exception ex) when (ex is ObjectDisposedException or InvalidOperationException or IOException or OperationCanceledException)
|
|
{
|
|
// 会话已经结束：下一个会话连上时补发
|
|
}
|
|
}
|
|
|
|
private static RpcRouter<PeerContext> BuildRouter() =>
|
|
new RpcRouter<PeerContext>(RpcSide.Listener, MoShiroMethodTable.Table)
|
|
.On(
|
|
MoShiroMethods.Clock,
|
|
MoShiroJsonContext.Default.ClockParams,
|
|
MoShiroJsonContext.Default.ClockResult,
|
|
(context, _, _) => ValueTask.FromResult(Result<ClockResult>.Ok(new ClockResult { ListenerMonoUs = context.Clock.NowUs() })));
|
|
}
|