388 lines
17 KiB
C#
388 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, QuicSettings Quic, 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,
|
|
_env.Quic);
|
|
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;
|
|
});
|
|
}
|