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.
387 lines
17 KiB
C#
387 lines
17 KiB
C#
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Text.Json;
|
|
using Fleet.Core;
|
|
using Fleet.Identity;
|
|
using Fleet.Link;
|
|
using Fleet.Rpc;
|
|
using MoShiro.Measurement;
|
|
using MoShiro.Protocol;
|
|
|
|
namespace MoShiro.Agent;
|
|
|
|
/// <summary>所有连接共用的东西。</summary>
|
|
public sealed record DialEnv(NodeIdentity Identity, string NodeId, LimitsSection Limits, DialDirectory Directory, MonoClock Clock, TimeProvider Time, INodeLog Log);
|
|
|
|
/// <summary>
|
|
/// 与一个 Agent 的点对点连接，分两层：
|
|
/// <list type="bullet">
|
|
/// <item>控制层：TCP+TLS 长连接。认证握手 → 同步时钟 → 按 Agent 下发的 <c>line_task_set</c> 运行积木 → 每 10 秒把已完成的行切成列块放进发件箱，
|
|
/// 发件箱和控制 RPC 共用这条连接。控制层断开才算与 Agent 失联：积木停止（数据面凭据属于控制会话），按 <see cref="ControlBackoff"/> 重拨。</item>
|
|
/// <item>任务层：TCP 类积木各自拨号；QUIC 数据通道由 <see cref="TaskLink"/> 在控制层在线期间单独维护，断开只重拨 QUIC。</item>
|
|
/// </list>
|
|
/// 发件箱属于这条连接，不属于某次控制会话：控制层重连之后重发还没确认的数据块。状态与日志由 <see cref="LinkStatus"/> 统一维护。
|
|
/// </summary>
|
|
public sealed class DialConnection : IDisposable
|
|
{
|
|
private static readonly TimeSpan s_flushInterval = TimeSpan.FromSeconds(ProtocolLimits.ChunkSeconds);
|
|
private static readonly TimeSpan s_clockInterval = TimeSpan.FromSeconds(30);
|
|
|
|
/// <summary>Agent 的任何帧（时钟回复、任务集）都算存活；比时钟同步间隔长得多，容得下差线路上的黑洞，但不会让一个 Agent 已经丢掉的会话永远挂着。</summary>
|
|
private static readonly TimeSpan s_idleLimit = TimeSpan.FromSeconds(120);
|
|
|
|
private static readonly RpcRouter<DialSession> s_router = BuildRouter();
|
|
|
|
/// <summary>对等面入站方法的路由器（只读，测试用）。</summary>
|
|
public static RpcRouter<DialSession> Router => s_router;
|
|
|
|
private readonly DialEnv _env;
|
|
private readonly LinkStatus _status;
|
|
private readonly Outbox _outbox = new(8 << 20);
|
|
private readonly SemaphoreSlim _retrySignal = new(0, 1);
|
|
private INodeLog _attemptLog;
|
|
private string _phase = "idle";
|
|
private long _attemptStarted;
|
|
private volatile DialSession? _session;
|
|
private volatile TaskLink? _task;
|
|
private QuicDatagramTransport? _activeDatagrams;
|
|
|
|
public DialConnection(DialEndpoint endpoint, DialEnv env)
|
|
{
|
|
Endpoint = endpoint;
|
|
_env = env;
|
|
_attemptLog = env.Log;
|
|
_status = new LinkStatus(endpoint.NodeId, endpoint.IpVersion, env.Time, env.Log);
|
|
}
|
|
|
|
public DialEndpoint Endpoint { get; }
|
|
|
|
/// <summary>控制层在线，且 QUIC 数据通道可发送。</summary>
|
|
public bool IsConnected => _session is not null && Volatile.Read(ref _activeDatagrams)?.IsConnected == true;
|
|
|
|
public int RoleCount => _session?.RoleCount ?? 0;
|
|
|
|
public long ClockOffsetUs => _session?.Clock.OffsetUs ?? 0;
|
|
|
|
/// <summary>当前线路状态（随心跳上报给面板）。</summary>
|
|
public LinkHealth Health() =>
|
|
_status.Snapshot(_outbox, _session?.TcpBlockCount ?? 0, Volatile.Read(ref _activeDatagrams)?.GetTransportSnapshot());
|
|
|
|
/// <summary>管理端手动配对：控制层在退避就马上再拨；控制层在线而 QUIC 在重拨，就跳过 QUIC 的等待。</summary>
|
|
public void RetryNow()
|
|
{
|
|
if (_status.Control == ControlState.Up)
|
|
{
|
|
if (_status.Quic == QuicState.Retrying)
|
|
{
|
|
_task?.RetryNow();
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
_retrySignal.Release();
|
|
}
|
|
catch (SemaphoreFullException)
|
|
{
|
|
// 已经有一次手动重试排队。
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
// 对应线路正在被移除。
|
|
}
|
|
}
|
|
|
|
public void Dispose() => _retrySignal.Dispose();
|
|
|
|
/// <summary>控制层循环：拨号，会话结束后按退避重拨。</summary>
|
|
public async Task RunAsync(CancellationToken ct)
|
|
{
|
|
var backoff = new ControlBackoff();
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
_attemptStarted = _env.Time.GetTimestamp();
|
|
_attemptLog = new ContextNodeLog(_env.Log, ("agent", Endpoint.NodeId), ("ip_version", Endpoint.IpVersion),
|
|
("attempt_id", Guid.NewGuid().ToString("N")));
|
|
_status.Log = _attemptLog;
|
|
_phase = "dns";
|
|
_status.SetControl(ControlState.Dialing, attempt: backoff.Attempts);
|
|
LinkFault fault;
|
|
try
|
|
{
|
|
fault = await RunOnceAsync(backoff, ct);
|
|
}
|
|
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
|
{
|
|
return;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
fault = LinkFaults.From(ex, _phase);
|
|
}
|
|
|
|
try
|
|
{
|
|
TimeSpan delay = backoff.Next();
|
|
_status.SetControl(ControlState.Backoff, fault, backoff.Attempts, delay);
|
|
using var waitCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
|
Task delayTask = Task.Delay(delay, _env.Time, waitCts.Token);
|
|
Task retryTask = _retrySignal.WaitAsync(waitCts.Token);
|
|
Task completed = await Task.WhenAny(delayTask, retryTask);
|
|
await waitCts.CancelAsync();
|
|
if (completed == retryTask)
|
|
{
|
|
await retryTask;
|
|
_attemptLog.Info("manual link retry requested");
|
|
}
|
|
else
|
|
{
|
|
await delayTask;
|
|
try { await retryTask; }
|
|
catch (OperationCanceledException) { }
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>一次控制会话：建立 → 运行 → 结束。建立失败抛出异常；建立之后结束返回结束原因。</summary>
|
|
private async Task<LinkFault> RunOnceAsync(ControlBackoff backoff, CancellationToken ct)
|
|
{
|
|
_attemptLog.Info("agent connection attempt started", ("phase", _phase), ("host", Endpoint.Host),
|
|
("tcp_port", Endpoint.TcpPort), ("quic_ports", string.Join(',', Endpoint.QuicPorts)));
|
|
IPAddress address = await ResolveAsync(ct);
|
|
byte[] agentKey = Convert.FromBase64String(Endpoint.PublicKey);
|
|
(PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(
|
|
address.ToString(), Endpoint.TcpPort, agentKey, _env.Identity, _env.NodeId, AgentApp.Version, 8 << 20, _env.Time, _attemptLog, ct,
|
|
onPhase: OnPeerPhase);
|
|
var data = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key);
|
|
var remote = (IPEndPoint)info.Remote!;
|
|
List<int> quicPorts = Endpoint.QuicPorts;
|
|
int quicPort = quicPorts.Count == 0 ? 7443 : quicPorts[0];
|
|
var datagrams = new QuicDatagramTransport(
|
|
data,
|
|
_env.Clock,
|
|
new DataPlaneCounters(),
|
|
_attemptLog,
|
|
new IPEndPoint(remote.Address, quicPort),
|
|
agentKey);
|
|
Volatile.Write(ref _activeDatagrams, datagrams);
|
|
using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
|
|
|
_outbox.ResetSent();
|
|
var clock = new ClockEstimator();
|
|
var shipper = new FragmentShipper(_outbox, new NodeId(Endpoint.NodeId), new NodeId(_env.NodeId), clock, _env.Time);
|
|
var blockEnv = new BlockEnv
|
|
{
|
|
Clock = _env.Clock,
|
|
Log = _attemptLog,
|
|
Session = () => data,
|
|
Datagrams = datagrams,
|
|
PeerAddress = remote.Address,
|
|
PeerSpkiSha256 = NodeIdentity.Fingerprint(agentKey),
|
|
KnockKey = Knock.DeriveKey(_env.Identity, agentKey),
|
|
Fragments = shipper,
|
|
MaxPacketsPerSecond = _env.Limits.MaxTotalPps,
|
|
MaxBulkMbps = _env.Limits.MaxTotalMbps,
|
|
};
|
|
var session = new DialSession(blockEnv, clock, _outbox, _env);
|
|
var rpc = new RpcSession<DialSession>(transport, s_router, session, _env.Time);
|
|
rpc.OnBinary = (_, _) => ValueTask.CompletedTask;
|
|
|
|
Task run = rpc.RunAsync(cts.Token);
|
|
Task pump = PumpAsync(rpc, _outbox, cts.Token);
|
|
Task quic = Task.CompletedTask;
|
|
Task maintenance = Task.CompletedTask;
|
|
TaskLink? taskLink = null;
|
|
try
|
|
{
|
|
// 先同步时钟（块头里要带偏移）；Agent 随后推送线路任务集
|
|
_phase = "clock_sync";
|
|
_status.SetControl(ControlState.Syncing);
|
|
await SyncClockAsync(rpc, clock, rounds: 5, cts.Token);
|
|
_attemptLog.Info("agent clock synchronized", ("phase", _phase), ("offset_us", clock.OffsetUs));
|
|
|
|
backoff.Reset();
|
|
_retrySignal.Wait(0, CancellationToken.None); // 若“立即重试”恰好赶在本次拨号成功前到达，本次成功已满足请求。
|
|
_session = session;
|
|
_phase = "control_session";
|
|
_status.SetControl(ControlState.Up);
|
|
taskLink = new TaskLink(datagrams, _status, _env.Time);
|
|
_task = taskLink;
|
|
quic = taskLink.RunAsync(cts.Token);
|
|
maintenance = MaintainAsync(rpc, session, clock, cts.Token);
|
|
_attemptLog.Info("agent control session up", ("phase", _phase), ("remote", remote), ("quic_port", quicPort), ("offset_us", clock.OffsetUs));
|
|
Task finished = await Task.WhenAny(run, pump, maintenance, quic);
|
|
LinkFault fault = finished.IsFaulted && finished.Exception?.GetBaseException() is { } error
|
|
? LinkFaults.From(error, _phase)
|
|
: LinkFaults.Closed(_phase, finished == run
|
|
? "closed by the Agent control connection"
|
|
: finished == pump
|
|
? "outbox pump stopped"
|
|
: finished == quic
|
|
? "QUIC link loop stopped"
|
|
: "connection maintenance stopped");
|
|
|
|
await cts.CancelAsync();
|
|
await BoundedAsync("stop tasks", () => Task.WhenAll(Quiet(run), Quiet(pump), Quiet(maintenance), Quiet(quic)));
|
|
return fault;
|
|
}
|
|
finally
|
|
{
|
|
_task = null;
|
|
_session = null;
|
|
await cts.CancelAsync();
|
|
session.FlushLineBlocks(); // 停止前把已稳定的行交给发件箱，重连之后补传
|
|
await BoundedAsync("stop blocks", session.StopAllAsync);
|
|
await BoundedAsync("wait control", () => Quiet(run));
|
|
await BoundedAsync("wait pump", () => Quiet(pump));
|
|
await BoundedAsync("wait quic", () => Quiet(quic));
|
|
taskLink?.Dispose();
|
|
await BoundedAsync("dispose control", () => rpc.DisposeAsync().AsTask());
|
|
Volatile.Write(ref _activeDatagrams, null);
|
|
await BoundedAsync("dispose quic", () => datagrams.DisposeAsync().AsTask());
|
|
_status.SetQuic(QuicState.Idle);
|
|
}
|
|
}
|
|
|
|
private async Task<IPAddress> ResolveAsync(CancellationToken ct)
|
|
{
|
|
AddressFamily family = Endpoint.IpVersion == IpVersions.V6 ? AddressFamily.InterNetworkV6 : AddressFamily.InterNetwork;
|
|
using var resolveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(10), _env.Time);
|
|
using CancellationTokenSource resolveCt = CancellationTokenSource.CreateLinkedTokenSource(ct, resolveTimeout.Token);
|
|
IPAddress[] addresses = IPAddress.TryParse(Endpoint.Host, out IPAddress? literal)
|
|
? [literal]
|
|
: await Dns.GetHostAddressesAsync(Endpoint.Host, family, resolveCt.Token);
|
|
IPAddress address = addresses.FirstOrDefault(a => a.AddressFamily == family && !a.IsIPv4MappedToIPv6)
|
|
?? throw new SocketException((int)SocketError.HostNotFound);
|
|
_attemptLog.Info("agent address resolved", ("phase", _phase), ("remote", address),
|
|
("elapsed_ms", (long)_env.Time.GetElapsedTime(_attemptStarted).TotalMilliseconds));
|
|
return address;
|
|
}
|
|
|
|
private void OnPeerPhase(string phase)
|
|
{
|
|
_phase = phase;
|
|
if (phase == "peer_auth")
|
|
{
|
|
_status.SetControl(ControlState.Authenticating);
|
|
}
|
|
}
|
|
|
|
private async Task BoundedAsync(string step, Func<Task> work)
|
|
{
|
|
if (!await Bounded.WaitAsync(work(), Bounded.Teardown, _env.Time))
|
|
{
|
|
_attemptLog.Warn("agent session teardown step timed out, moving on", ("phase", _phase), ("step", step), ("timeout_ms", Bounded.Teardown.TotalMilliseconds));
|
|
}
|
|
}
|
|
|
|
private static async Task Quiet(Task task)
|
|
{
|
|
try
|
|
{
|
|
await task;
|
|
}
|
|
catch (Exception ex) when (ex is OperationCanceledException or IOException or ObjectDisposedException or System.Net.Sockets.SocketException or InvalidOperationException or TimeoutException)
|
|
{
|
|
// 会话结束时的正常噪声
|
|
}
|
|
}
|
|
|
|
/// <summary>回传数据：发件箱里的块经控制连接发给 Agent。</summary>
|
|
private static async Task PumpAsync(RpcSession<DialSession> rpc, Outbox outbox, CancellationToken ct)
|
|
{
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
byte[] frame = await outbox.TakeNextAsync(2 << 20, ct);
|
|
await rpc.SendBinaryAsync(frame, ct);
|
|
}
|
|
}
|
|
|
|
private async Task SyncClockAsync(RpcSession<DialSession> rpc, ClockEstimator clock, int rounds, CancellationToken ct)
|
|
{
|
|
for (int i = 0; i < rounds; i++)
|
|
{
|
|
long sent = _env.Clock.NowUs();
|
|
Result<ClockResult> reply = await rpc.RequestAsync(MoShiroMethods.Clock, new ClockParams { DialerMonoUs = sent }, MoShiroJsonContext.Default.ClockParams, MoShiroJsonContext.Default.ClockResult, ct);
|
|
if (reply.TryGet(out ClockResult? value, out _))
|
|
{
|
|
clock.Add(sent, value.ListenerMonoUs, _env.Clock.NowUs());
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>切块（每 10 秒）、时钟同步（每 30 秒）和控制连接的存活检查。</summary>
|
|
private async Task MaintainAsync(RpcSession<DialSession> rpc, DialSession session, ClockEstimator clock, CancellationToken ct)
|
|
{
|
|
using var flush = new PeriodicTimer(s_flushInterval, _env.Time);
|
|
long lastClockUs = _env.Clock.NowUs();
|
|
while (await flush.WaitForNextTickAsync(ct))
|
|
{
|
|
if (rpc.IdleFor > s_idleLimit)
|
|
{
|
|
// 监听方已经不认这个会话（它重启、重连、或者空闲超时关了会话，而 FIN 没有到达）：拨号方这边结束它，让外层按退避重拨
|
|
throw new ControlSilentException($"the Agent sent nothing on this session for {(long)rpc.IdleFor.TotalSeconds}s");
|
|
}
|
|
|
|
session.FlushLineBlocks();
|
|
if (_env.Clock.NowUs() - lastClockUs >= s_clockInterval.TotalMicroseconds)
|
|
{
|
|
lastClockUs = _env.Clock.NowUs();
|
|
await SyncClockAsync(rpc, clock, rounds: 1, ct);
|
|
}
|
|
}
|
|
}
|
|
|
|
private static RpcRouter<DialSession> BuildRouter() =>
|
|
new RpcRouter<DialSession>(RpcSide.Dialer, MoShiroMethodTable.Table)
|
|
.On(
|
|
FleetMethods.StateSync,
|
|
FleetRpcContext.Default.StateSyncParams,
|
|
FleetRpcContext.Default.StateSyncResult,
|
|
async (session, sync, _) =>
|
|
{
|
|
var result = new StateSyncResult { AppliedVersion = sync.Version };
|
|
if (sync.Type == LineTaskSet.StateType)
|
|
{
|
|
LineTaskSet? set = sync.Doc.Deserialize(MoShiroJsonContext.Default.LineTaskSet);
|
|
if (set is null)
|
|
{
|
|
return Result<StateSyncResult>.Fail(FleetErrors.InvalidParams, "line_task_set could not be parsed");
|
|
}
|
|
|
|
result.Rejected.AddRange(await session.ApplyAsync(set));
|
|
}
|
|
|
|
return Result<StateSyncResult>.Ok(result);
|
|
})
|
|
.On(
|
|
MoShiroMethods.TestPhase,
|
|
MoShiroJsonContext.Default.TestPhaseStartParams,
|
|
MoShiroJsonContext.Default.TestPhaseStartResult,
|
|
(session, start, _) => ValueTask.FromResult(Result<TestPhaseStartResult>.Ok(new TestPhaseStartResult { Accepted = session.StartPhase(start) })))
|
|
.On(
|
|
MoShiroMethods.TestPhaseEnd,
|
|
MoShiroJsonContext.Default.TestPhaseEndParams,
|
|
MoShiroJsonContext.Default.TestPhaseEndResult,
|
|
async (session, end, ct) => Result<TestPhaseEndResult>.Ok(new TestPhaseEndResult { Fragments = await session.EndPhaseAsync(end, ct) }))
|
|
.OnNotify(
|
|
MoShiroMethods.TestAbort,
|
|
MoShiroJsonContext.Default.TestAbortParams,
|
|
async (session, abort, _) => await session.AbortAsync(abort.TestId))
|
|
.OnNotify(
|
|
FleetMethods.OutboxAck,
|
|
FleetRpcContext.Default.OutboxAckParams,
|
|
(session, ack, _) =>
|
|
{
|
|
session.Outbox.Acknowledge(ack.Ids.Concat(ack.Duplicates).Select(Guid.Parse));
|
|
return ValueTask.CompletedTask;
|
|
});
|
|
}
|