From 97ab3cc740f955a9f394b3f5fb9e9ce116b775b7 Mon Sep 17 00:00:00 2001 From: JustMe Date: Sat, 10 Oct 2026 14:16:57 +0800 Subject: [PATCH] feat(runtime): track node lifecycle and QUIC link state --- backend/agent/src/MoShiro.Agent/AgentApp.cs | 2 + backend/fleet/src/Fleet.Link/FleetLink.cs | 26 ++- backend/fleet/src/Fleet.Rpc/FleetMethods.cs | 9 + .../ServerContracts.cs | 2 +- .../NodeConnection.Receive.cs | 11 ++ .../Fleet.Server.Link/NodeConnection.State.cs | 2 +- .../Fleet.Server.Link/NodeConnection.Wire.cs | 1 + backend/openapi/v1.json | 172 +++++++++++++++++- backend/probe/README.md | 2 + .../src/MoShiro.Probe/Link/AgentConnection.cs | 90 ++++++++- .../Link/ProbeReconnectBackoff.cs | 31 ++++ backend/probe/src/MoShiro.Probe/ProbeApp.cs | 63 ++++++- .../ProbeReconnectBackoffTests.cs | 34 ++++ .../Rpc/MoShiroMethodTable.cs | 1 + .../shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs | 40 ++++ .../MoShiro.Protocol/Rpc/NodeRuntimeStatus.cs | 23 +++ .../NodeRuntimeStatusTests.cs | 17 ++ .../Admin/LinkEndpoints.cs | 78 +++++++- .../src/MoShiro.Backend.Api/Common/Mappers.cs | 11 +- .../Contracts/AdminDtos.cs | 6 +- .../Contracts/ApiJsonContext.cs | 1 + .../Contracts/PublicDtos.cs | 5 +- .../Agents/AgentRuntime.cs | 12 ++ .../Agents/AgentService.cs | 43 ++++- .../Agents/MoShiroFleetHooks.cs | 12 +- .../Agents/ProbeQuicRuntime.cs | 4 + .../Agents/ProbeRuntime.cs | 22 +++ .../Agents/ProbeService.cs | 40 +++- .../ApplicationServiceCollectionExtensions.cs | 1 + .../Ports/IAgentGateway.cs | 3 + .../MoShiro.Backend.Gateway/FleetGateway.cs | 3 + .../AgentEndToEndTests.cs | 14 ++ .../tests/MoShiro.Backend.Tests/LinkTests.cs | 15 ++ .../ProbeEndToEndTests.cs | 2 + docs/ops/deployment.md | 2 + web/shared/api/schema.ts | 97 ++++++++++ web/shared/api/types.ts | 2 + 37 files changed, 864 insertions(+), 35 deletions(-) create mode 100644 backend/probe/src/MoShiro.Probe/Link/ProbeReconnectBackoff.cs create mode 100644 backend/probe/tests/MoShiro.Probe.Tests/ProbeReconnectBackoffTests.cs create mode 100644 backend/shared/MoShiro.Protocol/Rpc/NodeRuntimeStatus.cs create mode 100644 backend/shared/tests/MoShiro.Protocol.Tests/NodeRuntimeStatusTests.cs create mode 100644 backend/src/MoShiro.Backend.Application/Agents/ProbeRuntime.cs diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs index 3290cb5..7e87230 100644 --- a/backend/agent/src/MoShiro.Agent/AgentApp.cs +++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs @@ -282,6 +282,7 @@ public sealed class AgentApp : IDisposable Capabilities = [MoShiroCapabilities.Base, MoShiroCapabilities.KindsPrefix + string.Join(',', BlockCatalog.AgentKinds)], HelloMeta = HelloMeta, HeartbeatExtra = HeartbeatExtra, + ShutdownReason = () => _restart.NowReason == "update" ? "updating" : "stopping", }; var link = new FleetLink(options, router, [new StateHandler(TaskSet.StateType, MoShiroJsonContext.Default.TaskSet, _tasks.ApplyAsync)], _outbox, _time, _log); @@ -335,6 +336,7 @@ public sealed class AgentApp : IDisposable long packetsOut = _counters.PacketsOut; var extra = new AgentHeartbeatExtra { + RuntimeStatus = _update?.Status().State == "applying" ? NodeRuntimeStatus.Updating : NodeRuntimeStatus.Running, Ipv6 = Ipv6Diagnostics.Sample(_ipv6Supported, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")), Probes = [.. _hub.ConnectedProbes()], Sessions = _sessions.All.Count, diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs index e31649d..5d7fbe6 100644 --- a/backend/fleet/src/Fleet.Link/FleetLink.cs +++ b/backend/fleet/src/Fleet.Link/FleetLink.cs @@ -58,6 +58,9 @@ public sealed class FleetLinkOptions /// 附加到 fleet.heartbeat 的进程内计数（不能包含主机信息）。 public Func? HeartbeatExtra { get; init; } + /// Optional reason for an orderly client shutdown, so the server can distinguish updates from service stops. + public Func? ShutdownReason { get; init; } + public TimeSpan HeartbeatInterval { get; init; } = TimeSpan.FromSeconds(10); public int MaxFrameBytes { get; init; } = 16 << 20; @@ -181,7 +184,8 @@ public sealed class FleetLink _secure = transport.Secure; var session = new RpcSession(transport, _router, new LinkContext(this), _time); - using CancellationTokenSource sessionCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + // Keep the active WebSocket receive alive until the planned shutdown marker and Close frame have been sent. + using var sessionCts = new CancellationTokenSource(); Task run = session.RunAsync(sessionCts.Token); try { @@ -192,7 +196,7 @@ public sealed class FleetLink NowUtcMs = _time.GetUtcNow().ToUnixTimeMilliseconds(), Meta = _options.HelloMeta?.Invoke(), }; - Result helloResult = await session.RequestAsync(FleetMethods.Hello, hello, FleetRpcContext.Default.HelloParams, FleetRpcContext.Default.HelloResult, sessionCts.Token); + Result helloResult = await session.RequestAsync(FleetMethods.Hello, hello, FleetRpcContext.Default.HelloParams, FleetRpcContext.Default.HelloResult, ct); if (!helloResult.IsSuccess) { return "hello failed: " + helloResult.Error.Code; @@ -208,7 +212,23 @@ public sealed class FleetLink await OnConnected(session, sessionCts.Token); } - Task finished = await Task.WhenAny(run, pump, heartbeat); + Task cancellation = Task.Delay(Timeout.Infinite, ct); + Task finished = await Task.WhenAny(run, pump, heartbeat, cancellation); + if (ct.IsCancellationRequested) + { + try + { + using var closeTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(1), _time); + string reason = _options.ShutdownReason?.Invoke() ?? "stopping"; + await session.NotifyAsync(FleetMethods.Stopping, new StoppingParams { Reason = reason }, FleetRpcContext.Default.StoppingParams, closeTimeout.Token); + await session.CloseAsync(reason, closeTimeout.Token); + } + catch (Exception ex) when (ex is OperationCanceledException or WebSocketException or IOException or ObjectDisposedException or InvalidOperationException) + { + // The socket may already be gone; shutdown must still complete. + } + } + await sessionCts.CancelAsync(); await Task.WhenAll(Quiet(run), Quiet(pump), Quiet(heartbeat)); return finished == run ? "closed by server" : "link task failed"; diff --git a/backend/fleet/src/Fleet.Rpc/FleetMethods.cs b/backend/fleet/src/Fleet.Rpc/FleetMethods.cs index 65b92f1..168eb5a 100644 --- a/backend/fleet/src/Fleet.Rpc/FleetMethods.cs +++ b/backend/fleet/src/Fleet.Rpc/FleetMethods.cs @@ -10,6 +10,7 @@ public static class FleetMethods public const string Auth = "fleet.auth"; public const string Hello = "fleet.hello"; public const string Heartbeat = "fleet.heartbeat"; + public const string Stopping = "fleet.stopping"; public const string StateSync = "fleet.state.sync"; public const string OutboxAck = "fleet.outbox.ack"; public const string Revoked = "fleet.revoked"; @@ -28,6 +29,7 @@ public static class FleetMethodTable MethodSpec.Request(FleetMethods.Auth, RpcSide.Node).AsInline(), MethodSpec.Request(FleetMethods.Hello, RpcSide.Node).AsInline(), MethodSpec.Notification(FleetMethods.Heartbeat, RpcSide.Node).AsInline(), + MethodSpec.Notification(FleetMethods.Stopping, RpcSide.Node).AsInline(), MethodSpec.Request(FleetMethods.StateSync, RpcSide.Server), MethodSpec.Notification(FleetMethods.OutboxAck, RpcSide.Server), MethodSpec.Notification(FleetMethods.Revoked, RpcSide.Server), @@ -156,6 +158,12 @@ public sealed class RevokedParams public string Reason { get; set; } = ""; } +public sealed class StoppingParams +{ + [JsonPropertyName("reason")] + public string Reason { get; set; } = "stopping"; +} + [JsonSourceGenerationOptions(DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull)] [JsonSerializable(typeof(RpcMessage))] [JsonSerializable(typeof(ChallengeParams))] @@ -164,6 +172,7 @@ public sealed class RevokedParams [JsonSerializable(typeof(HelloParams))] [JsonSerializable(typeof(HelloResult))] [JsonSerializable(typeof(HeartbeatParams))] +[JsonSerializable(typeof(StoppingParams))] [JsonSerializable(typeof(StateSyncParams))] [JsonSerializable(typeof(StateSyncResult))] [JsonSerializable(typeof(OutboxAckParams))] diff --git a/backend/fleet/src/Fleet.Server.Abstractions/ServerContracts.cs b/backend/fleet/src/Fleet.Server.Abstractions/ServerContracts.cs index bb8e4be..298896d 100644 --- a/backend/fleet/src/Fleet.Server.Abstractions/ServerContracts.cs +++ b/backend/fleet/src/Fleet.Server.Abstractions/ServerContracts.cs @@ -38,7 +38,7 @@ public enum NodeEventKind Heartbeat, } -public sealed record NodeEvent(NodeEventKind Kind, NodeRecord Node, long AtMs, HelloParams? Hello = null, JsonElement? Extra = null); + public sealed record NodeEvent(NodeEventKind Kind, NodeRecord Node, long AtMs, HelloParams? Hello = null, JsonElement? Extra = null, string? Reason = null); public interface INodeEventSink { diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs index 7df1ced..9be6688 100644 --- a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs @@ -64,6 +64,17 @@ internal sealed partial class NodeConnection case FleetMethods.Heartbeat: await HandleHeartbeatAsync(message, ct); return; + case FleetMethods.Stopping: + try + { + string? reason = message.Params?.Deserialize(FleetRpcContext.Default.StoppingParams)?.Reason; + _closeReason = reason is "updating" ? "updating" : "stopping"; + } + catch (JsonException) + { + _closeReason = "stopping"; + } + return; } if (_node!.Status != NodeStatus.Active) diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.State.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.State.cs index dafa0e5..06feb0e 100644 --- a/backend/fleet/src/Fleet.Server.Link/NodeConnection.State.cs +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.State.cs @@ -102,7 +102,7 @@ internal sealed partial class NodeConnection private async Task FireAsync(NodeEventKind kind, CancellationToken ct, HelloParams? hello = null, JsonElement? extra = null) { - var nodeEvent = new NodeEvent(kind, _node!, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), hello, extra); + var nodeEvent = new NodeEvent(kind, _node!, _host.Time.GetUtcNow().ToUnixTimeMilliseconds(), hello, extra, kind == NodeEventKind.Offline ? _closeReason : null); foreach (INodeEventSink sink in _host.Sinks) { try diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs index b528943..de4cae9 100644 --- a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Wire.cs @@ -17,6 +17,7 @@ internal sealed partial class NodeConnection ValueWebSocketReceiveResult result = await _socket.ReceiveAsync(buffer, ct); if (result.MessageType == WebSocketMessageType.Close) { + _closeReason ??= _socket.CloseStatusDescription; return null; } diff --git a/backend/openapi/v1.json b/backend/openapi/v1.json index a114f7f..1de1968 100644 --- a/backend/openapi/v1.json +++ b/backend/openapi/v1.json @@ -4821,6 +4821,76 @@ "x-moshiro-capability": "fleet.write" } }, + "/api/admin/links/{probe_id}/{agent_id}/retry": { + "post": { + "tags": [ + "admin-links" + ], + "summary": "Enable and immediately retry the Probe-to-Agent connection", + "operationId": "links_retry", + "parameters": [ + { + "name": "probe_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "agent_id", + "in": "path", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/LinkRetryDto" + } + } + } + }, + "401": { + "description": "Unauthorized", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/ProblemDto" + } + } + } + }, + "403": { + "description": "Forbidden", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/ProblemDto" + } + } + } + }, + "404": { + "description": "Not Found", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/ProblemDto" + } + } + } + } + }, + "x-moshiro-capability": "fleet.write" + } + }, "/healthz": { "get": { "tags": [ @@ -5185,6 +5255,7 @@ "rejected", "heartbeat", "endpoint_changes", + "runtime_status", "created_at", "last_seen_at" ], @@ -5325,6 +5396,9 @@ ], "format": "int32" }, + "runtime_status": { + "$ref": "#/components/schemas/NodeRuntimeStatus" + }, "created_at": { "format": "date-time" }, @@ -7007,7 +7081,9 @@ "agent_name", "linked", "override", - "connected" + "connected", + "probe_runtime_status", + "agent_runtime_status" ], "type": "object", "properties": { @@ -7034,6 +7110,46 @@ }, "connected": { "type": "boolean" + }, + "probe_runtime_status": { + "$ref": "#/components/schemas/NodeRuntimeStatus" + }, + "agent_runtime_status": { + "$ref": "#/components/schemas/NodeRuntimeStatus" + }, + "connection_state": { + "type": "string", + "default": "connecting" + }, + "connection_reason": { + "type": [ + "null", + "string" + ] + }, + "last_error": { + "type": [ + "null", + "string" + ] + }, + "retry_attempt": { + "pattern": "^-?(?:0|[1-9]\\d*)$", + "type": [ + "integer", + "string" + ], + "format": "int32", + "default": 0 + }, + "retry_in_ms": { + "pattern": "^-?(?:0|[1-9]\\d*)$", + "type": [ + "null", + "integer", + "string" + ], + "format": "int64" } } }, @@ -7055,6 +7171,21 @@ } } }, + "LinkRetryDto": { + "required": [ + "queued", + "link" + ], + "type": "object", + "properties": { + "queued": { + "type": "boolean" + }, + "link": { + "$ref": "#/components/schemas/LinkDto" + } + } + }, "LocalLimitsDto": { "required": [ "max_total_mbps", @@ -7475,6 +7606,17 @@ } } }, + "NodeRuntimeStatus": { + "enum": [ + "initializing", + "running", + "probe_paused", + "maintenance", + "stopping", + "offline", + "updating" + ] + }, "NodeUpdateDto": { "required": [ "state", @@ -8449,7 +8591,8 @@ "revoked_at", "revoked", "online", - "connections" + "connections", + "runtime_status" ], "type": "object", "properties": { @@ -8477,6 +8620,9 @@ "$ref": "#/components/schemas/ProbeConnectionDto" } }, + "runtime_status": { + "$ref": "#/components/schemas/NodeRuntimeStatus" + }, "ipv6": { "oneOf": [ { @@ -8760,6 +8906,28 @@ }, "at": { "format": "date-time" + }, + "state": { + "type": "string", + "default": "connecting" + }, + "retry_attempt": { + "pattern": "^-?(?:0|[1-9]\\d*)$", + "type": [ + "integer", + "string" + ], + "format": "int32", + "default": 0 + }, + "retry_in_ms": { + "pattern": "^-?(?:0|[1-9]\\d*)$", + "type": [ + "null", + "integer", + "string" + ], + "format": "int64" } } }, diff --git a/backend/probe/README.md b/backend/probe/README.md index 1352a16..62a5d16 100644 --- a/backend/probe/README.md +++ b/backend/probe/README.md @@ -13,6 +13,8 @@ Backend ──任务──> Agent <──测量会话──> Probe Probe 用 `probe.json` 读取面板地址（`fleet.server`）、身份目录（`fleet.identity_dir`，由 `fleet-enroll` 注册时生成）和本地测试限制。连上面板之后，面板下发要连接的 Agent 列表；Probe 将列表缓存到同目录的 `agents.json`，面板暂时连不上时靠它继续测量。身份目录里有私钥，请限制访问权限，不要提交到版本控制。 +Agent 测量连接失败时，Probe 按 1、2、4、8、16 秒递增等待，再以 32 秒间隔重试 5 次，之后最多每 60 秒重试一次；连接恢复后退避重新从 1 秒开始。管理端把没有任何已启用 Agent 目标的 Probe 标为「暂停探测」，不会将链路断开误显示为已连接。 + 容器节点的运行方式（podman + systemd）和宿主机更新器见 [`../../docs/ops/deployment.md`](../../docs/ops/deployment.md)；Probe 也可以用 OCI 镜像 `registry.oeo.one/moshiro/moshiro-probe:`，容器默认读取 `/etc/moshiro/probe.json`，命名 volume 保存身份配置和缓存，镜像以 UID/GID 1654 运行。 ## 构建 diff --git a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs index 4ef4ad0..717b3d7 100644 --- a/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs +++ b/backend/probe/src/MoShiro.Probe/Link/AgentConnection.cs @@ -19,7 +19,7 @@ public sealed record ConnectionEnv(NodeIdentity Identity, string ProbeId, ProbeL /// 每 10 秒把已完成的行切成列块放进发件箱 → 响应按需测试的阶段。会话断开时所有角色停止（凭据属于会话），按退避重连。 /// 会话内的所有状态（套接字、发件箱、角色）随会话一起创建和释放。 /// -public sealed class AgentConnection +public sealed class AgentConnection : IDisposable { private static readonly TimeSpan s_flushInterval = TimeSpan.FromSeconds(ProtocolLimits.ChunkSeconds); private static readonly TimeSpan s_clockInterval = TimeSpan.FromSeconds(30); @@ -34,6 +34,10 @@ public sealed class AgentConnection private QuicDatagramTransport? _activeDatagrams; private readonly object _handshakeGate = new(); private readonly Queue _handshakeSamplesMs = new(); + private readonly SemaphoreSlim _retrySignal = new(0, 1); + private ProbeLinkAttemptState _attemptState = ProbeLinkAttemptState.Connecting; + private string? _lastError; + private DateTimeOffset? _retryAt; private long _handshakeAttempts; private long _handshakeSuccesses; private long _handshakeFailures; @@ -53,6 +57,29 @@ public sealed class AgentConnection public long ClockOffsetUs => _session?.Clock.OffsetUs ?? 0; + /// 管理端手动配对时，跳过当前退避等待，马上再拨一次。 + public void RetryNow() + { + if (IsConnected) + { + return; + } + try + { + _retrySignal.Release(); + } + catch (SemaphoreFullException) + { + // 已经有一次手动重试排队。 + } + catch (ObjectDisposedException) + { + // 对应线路正在被移除。 + } + } + + public void Dispose() => _retrySignal.Dispose(); + public ProbeQuicLinkHealth QuicHealth() { lock (_handshakeGate) @@ -70,6 +97,12 @@ public sealed class AgentConnection AgentId = Endpoint.AgentId, IpVersion = Endpoint.IpVersion, Connected = IsConnected, + State = _attemptState, + LastError = _lastError, + RetryAttempt = _lastRetryAttempt, + RetryInMs = _attemptState == ProbeLinkAttemptState.RetryWait && _retryAt is { } retryAt + ? Math.Max(0, (long)(retryAt - _env.Time.GetUtcNow()).TotalMilliseconds) + : null, HandshakeAttempts = _handshakeAttempts, HandshakeSuccesses = _handshakeSuccesses, HandshakeFailures = _handshakeFailures, @@ -85,29 +118,55 @@ public sealed class AgentConnection } } + private int _lastRetryAttempt; + public async Task RunAsync(CancellationToken ct) { - // 重连的间隔别拉太长：每次尝试本来就要等好几秒（握手上限 30 秒），线路一恢复要尽快接上；带抖动，不会和别的节点撞在一起 - var backoff = new Backoff(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(8)); + var backoff = new ProbeReconnectBackoff(); while (!ct.IsCancellationRequested) { + SetAttempt(ProbeLinkAttemptState.Connecting, null, null, 0); try { string reason = await RunOnceAsync(backoff, ct); + SetAttempt(ProbeLinkAttemptState.RetryWait, reason, null, backoff.Attempts); _env.Log.Warn("agent session ended", ("agent", Endpoint.AgentId), ("err", reason)); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; } - catch (Exception ex) when (ex is IOException or System.Net.Sockets.SocketException or InvalidOperationException or System.Security.Authentication.AuthenticationException or OperationCanceledException or FormatException) + catch (Exception ex) { - _env.Log.Warn("agent connection failed", ("agent", Endpoint.AgentId), ("err", ex.Message)); + SetAttempt(ProbeLinkAttemptState.RetryWait, $"{ex.GetType().Name}: {ex.Message}", null, backoff.Attempts + 1); + _env.Log.Error("agent connection attempt failed; retry will continue", ("agent", Endpoint.AgentId), ("exception", ex.GetType().FullName), ("err", ex.Message)); } try { - await Task.Delay(backoff.Next(), _env.Time, ct); + TimeSpan delay = backoff.Next(); + lock (_handshakeGate) + { + _attemptState = ProbeLinkAttemptState.RetryWait; + _lastRetryAttempt = backoff.Attempts; + _retryAt = _env.Time.GetUtcNow() + 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; + _env.Log.Info("manual link retry requested", ("agent", Endpoint.AgentId)); + } + else + { + await delayTask; + try { await retryTask; } + catch (OperationCanceledException) { } + } } catch (OperationCanceledException) { @@ -116,7 +175,7 @@ public sealed class AgentConnection } } - private async Task RunOnceAsync(Backoff backoff, CancellationToken ct) + private async Task RunOnceAsync(ProbeReconnectBackoff backoff, CancellationToken ct) { AddressFamily family = Endpoint.IpVersion == IpVersions.V6 ? AddressFamily.InterNetworkV6 : AddressFamily.InterNetwork; using var resolveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(10), _env.Time); @@ -183,6 +242,7 @@ public sealed class AgentConnection try { + SetAttempt(ProbeLinkAttemptState.Connecting, null, null, 0); await datagrams.ConnectAsync(cts.Token); lock (_handshakeGate) { @@ -216,7 +276,9 @@ public sealed class AgentConnection } backoff.Reset(); + _retrySignal.Wait(0, CancellationToken.None); // 若“立即重试”恰好赶在本次拨号成功前到达，本次成功已满足请求。 _session = session; + SetAttempt(ProbeLinkAttemptState.Connected, null, null, 0); _env.Log.Info("agent session up", ("agent", Endpoint.AgentId), ("offset_us", clock.OffsetUs)); Task maintenance = MaintainAsync(rpc, session, clock, cts.Token); Task finished = await Task.WhenAny(run, receive, pump, maintenance); @@ -237,6 +299,20 @@ public sealed class AgentConnection } } + private void SetAttempt(ProbeLinkAttemptState state, string? error, DateTimeOffset? retryAt, int attempt) + { + lock (_handshakeGate) + { + _attemptState = state; + if (error is not null || state == ProbeLinkAttemptState.Connected) + { + _lastError = error; + } + _retryAt = retryAt; + _lastRetryAttempt = attempt; + } + } + private static async Task Quiet(Task task) { try diff --git a/backend/probe/src/MoShiro.Probe/Link/ProbeReconnectBackoff.cs b/backend/probe/src/MoShiro.Probe/Link/ProbeReconnectBackoff.cs new file mode 100644 index 0000000..333364a --- /dev/null +++ b/backend/probe/src/MoShiro.Probe/Link/ProbeReconnectBackoff.cs @@ -0,0 +1,31 @@ +namespace MoShiro.Probe; + +/// Reconnect cadence for a single Probe-to-Agent path: 1, 2, 4, 8, 16, then five 32-second waits, then once per minute. +public sealed class ProbeReconnectBackoff +{ + private static readonly TimeSpan[] s_delays = + [ + TimeSpan.FromSeconds(1), + TimeSpan.FromSeconds(2), + TimeSpan.FromSeconds(4), + TimeSpan.FromSeconds(8), + TimeSpan.FromSeconds(16), + TimeSpan.FromSeconds(32), + TimeSpan.FromSeconds(32), + TimeSpan.FromSeconds(32), + TimeSpan.FromSeconds(32), + TimeSpan.FromSeconds(32), + ]; + + private int _attempt; + + public int Attempts => Volatile.Read(ref _attempt); + + public TimeSpan Next() + { + int attempt = Interlocked.Increment(ref _attempt) - 1; + return attempt < s_delays.Length ? s_delays[attempt] : TimeSpan.FromMinutes(1); + } + + public void Reset() => Interlocked.Exchange(ref _attempt, 0); +} diff --git a/backend/probe/src/MoShiro.Probe/ProbeApp.cs b/backend/probe/src/MoShiro.Probe/ProbeApp.cs index 34a7549..19a4a5b 100644 --- a/backend/probe/src/MoShiro.Probe/ProbeApp.cs +++ b/backend/probe/src/MoShiro.Probe/ProbeApp.cs @@ -1,5 +1,6 @@ using Fleet.Core; using Fleet.Hosting; +using System.Collections.Concurrent; using System.Text.Json; using Fleet.Identity; using Fleet.Link; @@ -19,6 +20,7 @@ public sealed class ProbeApp : IDisposable private readonly ConnectionEnv _env; private readonly Dictionary _connections = []; + private readonly ConcurrentDictionary _pendingRetries = new(StringComparer.Ordinal); private readonly SemaphoreSlim _changed = new(0); private readonly INodeLog _log; private readonly FleetLink _link; @@ -45,7 +47,13 @@ public sealed class ProbeApp : IDisposable NodeId = identity.NodeId, NodeVersion = Version, Capabilities = [MoShiroCapabilities.Base], - HeartbeatExtra = () => JsonSerializer.SerializeToElement(new ProbeHeartbeatExtra { Update = update?.Status(), Ipv6 = Ipv6Diagnostics.Sample(null, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")), QuicLinks = SnapshotQuicLinks() }, MoShiroJsonContext.Default.ProbeHeartbeatExtra), + ShutdownReason = () => restart?.NowReason == "update" ? "updating" : "stopping", + HeartbeatExtra = () => + { + UpdateStatus? updateStatus = update?.Status(); + NodeRuntimeStatus runtimeStatus = updateStatus?.State == "applying" ? NodeRuntimeStatus.Updating : _env.Directory.Agents.Count == 0 ? NodeRuntimeStatus.ProbePaused : NodeRuntimeStatus.Running; + return JsonSerializer.SerializeToElement(new ProbeHeartbeatExtra { RuntimeStatus = runtimeStatus, Update = updateStatus, Ipv6 = Ipv6Diagnostics.Sample(null, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")), QuicLinks = SnapshotQuicLinks() }, MoShiroJsonContext.Default.ProbeHeartbeatExtra); + }, HelloMeta = () => JsonSerializer.SerializeToElement(new ProbeHelloMeta { Owners = [.. trust.Controllers], Platform = PlatformDetector.Detect(), Ipv6 = Ipv6Diagnostics.Sample(null, Environment.GetEnvironmentVariable("MOSHIRO_NETWORK_SCOPE"), Environment.GetEnvironmentVariable("HOST_PROC")) }, MoShiroJsonContext.Default.ProbeHelloMeta), }; var list = new StateHandler( @@ -57,6 +65,11 @@ public sealed class ProbeApp : IDisposable return ValueTask.FromResult(new StateSyncResult { AppliedVersion = version }); }); var router = new RpcRouter(RpcSide.Node, MoShiroMethodTable.Table, NodeRoles.Probe) + .OnNotify(MoShiroMethods.ProbeLinkRetry, MoShiroJsonContext.Default.ProbeLinkRetryParams, (_, retry, _) => + { + RetryAgent(retry.AgentId.Value); + return ValueTask.CompletedTask; + }) .OnNotify(MoShiroMethods.UpdateOffer, MoShiroJsonContext.Default.UpdateOfferParams, (_, offer, _) => { update?.Offer(offer); @@ -101,6 +114,47 @@ public sealed class ProbeApp : IDisposable } } + private void RetryAgent(string agentId, bool queueIfMissing = true) + { + AgentConnection[] matches; + lock (_connections) + { + matches = [.. _connections.Values.Select(r => r.Connection).Where(c => c.Endpoint.AgentId == agentId)]; + } + + if (matches.Length == 0) + { + if (queueIfMissing) + { + _pendingRetries[agentId] = _time.GetUtcNow().AddSeconds(30).ToUnixTimeMilliseconds(); + } + _log.Warn("manual link retry requested but no matching Agent path is configured", ("agent", agentId)); + return; + } + + _pendingRetries.TryRemove(agentId, out _); + foreach (AgentConnection connection in matches) + { + connection.RetryNow(); + } + } + + private void RetryPending() + { + long now = _time.GetUtcNow().ToUnixTimeMilliseconds(); + foreach ((string agentId, long expiresAt) in _pendingRetries.ToArray()) + { + if (expiresAt <= now) + { + _pendingRetries.TryRemove(agentId, out _); + } + else + { + RetryAgent(agentId, queueIfMissing: false); + } + } + } + private List SnapshotQuicLinks() { lock (_connections) @@ -124,6 +178,7 @@ public sealed class ProbeApp : IDisposable while (!ct.IsCancellationRequested) { await ReconcileAsync(_env.Directory.Agents, ct); + RetryPending(); await _changed.WaitAsync(ct); } } @@ -250,6 +305,10 @@ public sealed class ProbeApp : IDisposable Dispose(); } - public void Dispose() => _cts.Dispose(); + public void Dispose() + { + Connection.Dispose(); + _cts.Dispose(); + } } } diff --git a/backend/probe/tests/MoShiro.Probe.Tests/ProbeReconnectBackoffTests.cs b/backend/probe/tests/MoShiro.Probe.Tests/ProbeReconnectBackoffTests.cs new file mode 100644 index 0000000..5951e28 --- /dev/null +++ b/backend/probe/tests/MoShiro.Probe.Tests/ProbeReconnectBackoffTests.cs @@ -0,0 +1,34 @@ +using MoShiro.Probe; + +namespace MoShiro.Probe.Tests; + +public sealed class ProbeReconnectBackoffTests +{ + [Fact] + public void Next_UsesRequestedEscalationThenCapsAtOneMinute() + { + var backoff = new ProbeReconnectBackoff(); + TimeSpan[] expected = + [ + TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(4), + TimeSpan.FromSeconds(8), TimeSpan.FromSeconds(16), + TimeSpan.FromSeconds(32), TimeSpan.FromSeconds(32), TimeSpan.FromSeconds(32), + TimeSpan.FromSeconds(32), TimeSpan.FromSeconds(32), + TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(1), + ]; + + Assert.Equal(expected, expected.Select(_ => backoff.Next())); + } + + [Fact] + public void Reset_RestartsAtOneSecond() + { + var backoff = new ProbeReconnectBackoff(); + _ = backoff.Next(); + _ = backoff.Next(); + + backoff.Reset(); + + Assert.Equal(TimeSpan.FromSeconds(1), backoff.Next()); + } +} diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs index 380b7c8..febe4f1 100644 --- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs +++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroMethodTable.cs @@ -30,6 +30,7 @@ public static class MoShiroMethodTable MethodSpec.Request(MoShiroMethods.TestRun, RpcSide.Server, NodeRoles.Agent), MethodSpec.Request(MoShiroMethods.TestCancel, RpcSide.Server, NodeRoles.Agent), MethodSpec.Request(MoShiroMethods.ProbeDisconnect, RpcSide.Server, NodeRoles.Agent), + MethodSpec.Notification(MoShiroMethods.ProbeLinkRetry, RpcSide.Server, NodeRoles.Probe), MethodSpec.Notification(MoShiroMethods.UpdateOffer, RpcSide.Server), MethodSpec.Notification(MoShiroMethods.UpdateApply, RpcSide.Server), MethodSpec.Notification(MoShiroMethods.Schedule, RpcSide.Server), diff --git a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs index 742a7de..b13964b 100644 --- a/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs +++ b/backend/shared/MoShiro.Protocol/Rpc/MoShiroRpc.cs @@ -10,6 +10,7 @@ public static class MoShiroMethods public const string TestRun = "moshiro.test.run"; public const string TestCancel = "moshiro.test.cancel"; public const string ProbeDisconnect = "moshiro.probe.disconnect"; + public const string ProbeLinkRetry = "moshiro.probe.link_retry"; public const string TestStatus = "moshiro.test.status"; public const string Event = "moshiro.event"; public const string Host = "moshiro.host"; @@ -98,6 +99,13 @@ public sealed class ProbeDisconnectParams public ProbeId ProbeId { get; set; } } +/// Backend → Probe: immediately retry the selected Agent's current IP paths. +public sealed class ProbeLinkRetryParams +{ + [JsonPropertyName("agent_id")] + public NodeId AgentId { get; set; } +} + public sealed class TestStatusParams { [JsonPropertyName("test_id")] @@ -302,6 +310,10 @@ public sealed class UpdateStatus /// Probe 的 fleet.heartbeat 附加字段：更新器的状态。 public sealed class ProbeHeartbeatExtra { + /// Node operating state; registration approval and transport presence are reported separately. + [JsonPropertyName("runtime_status")] + public NodeRuntimeStatus RuntimeStatus { get; set; } = NodeRuntimeStatus.Running; + [JsonPropertyName("quic_links")] public List? QuicLinks { get; set; } @@ -324,6 +336,18 @@ public sealed class ProbeQuicLinkHealth [JsonPropertyName("connected")] public bool Connected { get; set; } + [JsonPropertyName("state")] + public ProbeLinkAttemptState State { get; set; } = ProbeLinkAttemptState.Connecting; + + [JsonPropertyName("last_error")] + public string? LastError { get; set; } + + [JsonPropertyName("retry_attempt")] + public int RetryAttempt { get; set; } + + [JsonPropertyName("retry_in_ms")] + public long? RetryInMs { get; set; } + [JsonPropertyName("handshake_attempts")] public long HandshakeAttempts { get; set; } @@ -359,6 +383,17 @@ public sealed class ProbeQuicLinkHealth public ushort? PathMtu { get; set; } } +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum ProbeLinkAttemptState +{ + [JsonStringEnumMemberName("connecting")] + Connecting, + [JsonStringEnumMemberName("retry_wait")] + RetryWait, + [JsonStringEnumMemberName("connected")] + Connected, +} + /// Probe 的 fleet.hello 附加字段：系统、架构和构建信息（只用于展示和更新）。 public sealed class ProbeHelloMeta { @@ -479,6 +514,10 @@ public sealed class ConnectedProbe /// fleet.heartbeat 的附加字段：只有进程内的计数，没有主机信息。 public sealed class AgentHeartbeatExtra { + /// Node operating state; registration approval and transport presence are reported separately. + [JsonPropertyName("runtime_status")] + public NodeRuntimeStatus RuntimeStatus { get; set; } = NodeRuntimeStatus.Running; + [JsonPropertyName("ipv6")] public Ipv6Diagnostic? Ipv6 { get; set; } @@ -645,6 +684,7 @@ public static class FragmentKeys [JsonSerializable(typeof(TestRunResult))] [JsonSerializable(typeof(TestCancelParams))] [JsonSerializable(typeof(ProbeDisconnectParams))] +[JsonSerializable(typeof(ProbeLinkRetryParams))] [JsonSerializable(typeof(HostSnapshot))] [JsonSerializable(typeof(TestStatusParams))] [JsonSerializable(typeof(AgentEventParams))] diff --git a/backend/shared/MoShiro.Protocol/Rpc/NodeRuntimeStatus.cs b/backend/shared/MoShiro.Protocol/Rpc/NodeRuntimeStatus.cs new file mode 100644 index 0000000..333ecb2 --- /dev/null +++ b/backend/shared/MoShiro.Protocol/Rpc/NodeRuntimeStatus.cs @@ -0,0 +1,23 @@ +using System.Text.Json.Serialization; + +namespace MoShiro.Protocol; + +/// Live process/measurement state. Enrollment approval remains a separate node status. +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum NodeRuntimeStatus +{ + [JsonStringEnumMemberName("initializing")] + Initializing, + [JsonStringEnumMemberName("running")] + Running, + [JsonStringEnumMemberName("probe_paused")] + ProbePaused, + [JsonStringEnumMemberName("maintenance")] + Maintenance, + [JsonStringEnumMemberName("stopping")] + Stopping, + [JsonStringEnumMemberName("offline")] + Offline, + [JsonStringEnumMemberName("updating")] + Updating, +} diff --git a/backend/shared/tests/MoShiro.Protocol.Tests/NodeRuntimeStatusTests.cs b/backend/shared/tests/MoShiro.Protocol.Tests/NodeRuntimeStatusTests.cs new file mode 100644 index 0000000..99180ab --- /dev/null +++ b/backend/shared/tests/MoShiro.Protocol.Tests/NodeRuntimeStatusTests.cs @@ -0,0 +1,17 @@ +using System.Text.Json; + +namespace MoShiro.Protocol.Tests; + +public sealed class NodeRuntimeStatusTests +{ + [Fact] + public void HeartbeatStatus_UsesStableSnakeCaseEnumValues() + { + var heartbeat = new AgentHeartbeatExtra { RuntimeStatus = NodeRuntimeStatus.ProbePaused }; + string json = JsonSerializer.Serialize(heartbeat, MoShiroJsonContext.Default.AgentHeartbeatExtra); + + Assert.Contains("\"runtime_status\":\"probe_paused\"", json, StringComparison.Ordinal); + Assert.Equal(NodeRuntimeStatus.ProbePaused, + JsonSerializer.Deserialize(json, MoShiroJsonContext.Default.AgentHeartbeatExtra)!.RuntimeStatus); + } +} diff --git a/backend/src/MoShiro.Backend.Api/Admin/LinkEndpoints.cs b/backend/src/MoShiro.Backend.Api/Admin/LinkEndpoints.cs index f459f46..b9ebeac 100644 --- a/backend/src/MoShiro.Backend.Api/Admin/LinkEndpoints.cs +++ b/backend/src/MoShiro.Backend.Api/Admin/LinkEndpoints.cs @@ -20,9 +20,12 @@ public static class LinkEndpoints group.MapPut("/links/{probe_id}/{agent_id}", SetAsync) .Api("links_set", "Force a link on or off, or give it back to the automatic setting (enabled: null)", Tag).Returns().ProducesProblems(400, 401, 403, 404) .Requires(Capability.FleetWrite); + group.MapPost("/links/{probe_id}/{agent_id}/retry", RetryAsync) + .Api("links_retry", "Enable and immediately retry the Probe-to-Agent connection", Tag).Returns().ProducesProblems(401, 403, 404) + .Requires(Capability.FleetWrite); } - private static async Task ListAsync(AgentService agents, ProbeService probes, LinkService links, CancellationToken ct) + private static async Task ListAsync(AgentService agents, ProbeService probes, LinkService links, ProbeQuicRuntime quic, TimeProvider time, CancellationToken ct) { LinkSnapshot snapshot = await links.SnapshotAsync(ct); IReadOnlyList agentViews = await agents.ListAsync(ct); @@ -31,14 +34,14 @@ public static class LinkEndpoints { foreach (AgentView agent in agentViews.OrderBy(a => a.Agent.Name, StringComparer.Ordinal)) { - items.Add(ToDto(snapshot, probe, agent)); + items.Add(ToDto(snapshot, probe, agent, quic, time)); } } return TypedResults.Json(new LinkMatrixDto(snapshot.Auto, items), ApiJsonContext.Default.LinkMatrixDto); } - private static async Task SetAsync(HttpContext http, AgentService agents, ProbeService probes, LinkService links, string probe_id, string agent_id, SetLinkRequest request, CancellationToken ct) + private static async Task SetAsync(HttpContext http, AgentService agents, ProbeService probes, LinkService links, ProbeQuicRuntime quic, TimeProvider time, string probe_id, string agent_id, SetLinkRequest request, CancellationToken ct) { if (await probes.GetAsync(probe_id, ct) is not { Revoked: false } probe) { @@ -52,10 +55,71 @@ public static class LinkEndpoints await links.SetAsync(probe_id, agent_id, request.Enabled, ct); await agents.AfterAgentSetChangedAsync(ct); // 重新下发：Probe 的 Agent 列表、Agent 允许的 Probe - return TypedResults.Json(ToDto(await links.SnapshotAsync(ct), probe, agent), ApiJsonContext.Default.LinkDto); + if (request.Enabled == true) + { + await agents.NotifyProbeLinkRetryAsync(new NodeId(probe_id), agent.Agent.Id, ct); + } + return TypedResults.Json(ToDto(await links.SnapshotAsync(ct), probe, agent, quic, time), ApiJsonContext.Default.LinkDto); } - private static LinkDto ToDto(LinkSnapshot snapshot, ProbeView probe, AgentView agent) => - new(probe.Probe.Id, probe.Probe.Name, agent.Agent.Id.Value, agent.Agent.Name, snapshot.IsLinked(probe.Probe.Id, agent.Agent.Id.Value), snapshot.OverrideOf(probe.Probe.Id, agent.Agent.Id.Value), - agent.Probes.Any(p => p.Id == probe.Probe.Id)); + private static async Task RetryAsync(HttpContext http, AgentService agents, ProbeService probes, LinkService links, ProbeQuicRuntime quic, TimeProvider time, string probe_id, string agent_id, CancellationToken ct) + { + if (await probes.GetAsync(probe_id, ct) is not { Revoked: false } probe) + { + return Problems.Create(http, MoShiroErrors.ProbeNotFound, "probe not found"); + } + if (await agents.GetAsync(new NodeId(agent_id), ct) is not { } agent) + { + return Problems.Create(http, MoShiroErrors.AgentNotFound, "agent not found"); + } + + await links.SetAsync(probe_id, agent_id, true, ct); + await agents.AfterAgentSetChangedAsync(ct); + bool queued = await agents.NotifyProbeLinkRetryAsync(new NodeId(probe_id), agent.Agent.Id, ct); + LinkDto link = ToDto(await links.SnapshotAsync(ct), probe, agent, quic, time); + return TypedResults.Json(new LinkRetryDto(queued, link), ApiJsonContext.Default.LinkRetryDto, statusCode: StatusCodes.Status202Accepted); + } + + private static LinkDto ToDto(LinkSnapshot snapshot, ProbeView probe, AgentView agent, ProbeQuicRuntime quicRuntime, TimeProvider time) + { + bool linked = snapshot.IsLinked(probe.Probe.Id, agent.Agent.Id.Value); + ProbeQuicLinkSnapshot? quicSnapshot = quicRuntime.Get(probe.Probe.Id, agent.Agent.Id.Value, IpVersions.V4); + ProbeQuicLinkHealth? health = quicSnapshot?.Health; + bool fresh = quicSnapshot is not null && time.GetUtcNow().ToUnixTimeMilliseconds() - quicSnapshot.AtMs <= 30_000; + string state; + string? reason = null; + if (!linked) { state = "disabled"; reason = "线路策略已关闭"; } + else if (probe.RuntimeStatus != NodeRuntimeStatus.Running) { state = "probe_unavailable"; reason = $"Probe 状态：{probe.RuntimeStatus}"; } + else if (agent.RuntimeStatus != NodeRuntimeStatus.Running) { state = "agent_unavailable"; reason = $"Agent 状态：{agent.RuntimeStatus}"; } + else if (health is null || !fresh) + { + state = "waiting_report"; + reason = health is null ? "Probe 尚未上报此线路的 QUIC 拨号状态" : "Probe 的线路状态上报已过期"; + } + else if (health.Connected && agent.Probes.Any(p => p.Id == probe.Probe.Id)) + { + state = "connected"; + } + else if (health.State == ProbeLinkAttemptState.RetryWait) + { + state = "retry_wait"; + reason = health.LastError is { Length: > 0 } error ? $"QUIC 连接失败：{error}" : "QUIC 连接失败，等待自动重试"; + } + else if (agent.Probes.All(p => p.Id != probe.Probe.Id)) + { + state = "control_disconnected"; + reason = health.LastError ?? "Probe 尚未建立到 Agent 的测量控制会话"; + } + else + { + state = "quic_connecting"; + reason = health.LastError ?? "控制会话已建立，正在建立 QUIC 测量会话"; + } + bool connected = state == "connected"; + return new LinkDto(probe.Probe.Id, probe.Probe.Name, agent.Agent.Id.Value, agent.Agent.Name, linked, + snapshot.OverrideOf(probe.Probe.Id, agent.Agent.Id.Value), connected, probe.RuntimeStatus, agent.RuntimeStatus, + state, reason, health?.LastError, health?.RetryAttempt ?? 0, health?.RetryInMs); + } } + +public sealed record LinkRetryDto(bool Queued, LinkDto Link); diff --git a/backend/src/MoShiro.Backend.Api/Common/Mappers.cs b/backend/src/MoShiro.Backend.Api/Common/Mappers.cs index c01ee40..ac419d0 100644 --- a/backend/src/MoShiro.Backend.Api/Common/Mappers.cs +++ b/backend/src/MoShiro.Backend.Api/Common/Mappers.cs @@ -46,7 +46,12 @@ public static class Mappers health.Connected, health.HandshakeAttempts, health.HandshakeSuccesses, health.HandshakeFailures, health.HandshakeTimeouts, Rate(health.HandshakeSuccesses), Rate(health.HandshakeFailures), Rate(health.HandshakeTimeouts), health.HandshakeP95Ms, health.HandshakeSampleCount, health.TransportPacketLossRate, health.TransportPacketsSent, health.TransportCongestionEvents, - health.TransportEcnCongestionEvents, health.PathMtu, Time(snapshot.AtMs)); + health.TransportEcnCongestionEvents, health.PathMtu, Time(snapshot.AtMs), health.State switch + { + ProbeLinkAttemptState.Connected => "connected", + ProbeLinkAttemptState.RetryWait => "retry_wait", + _ => "connecting", + }, health.RetryAttempt, health.RetryInMs); } public static HostDto ToDto(HostSnapshot h) => @@ -57,7 +62,7 @@ public static class Mappers public static ProbeDto ToDto(ProbeView v) => new( v.Probe.Id, v.Probe.Name, Time(v.Probe.CreatedAtMs), Time(v.Probe.RevokedAtMs), v.Revoked, v.Online, - [.. v.Connections.Select(c => new ProbeConnectionDto(c.AgentId.Value, c.AgentName, c.AgentOnline, c.Version, c.Remote, Time(c.SinceMs), c.IpVersion))], ProbeIpv6Of(v.Probe), v.Probe.AddedBy); + [.. v.Connections.Select(c => new ProbeConnectionDto(c.AgentId.Value, c.AgentName, c.AgentOnline, c.Version, c.Remote, Time(c.SinceMs), c.IpVersion))], v.RuntimeStatus, ProbeIpv6Of(v.Probe), v.Probe.AddedBy); public static SeriesDto ToDto(SeriesResult r, bool live) => new( @@ -119,7 +124,7 @@ public static class Mappers : null; return new AgentDto( a.Id.Value, a.Name, StatusName(a.Status), view.Online, view.ProbeConnected, [.. view.Probes.Select(ToDto)], view.Host is null ? null : ToDto(view.Host), a.PublicHost, a.ProbeTcpPort, a.ProbeUdpPorts, a.ProbeUdpPorts, a.Labels, a.PublicVisible, - a.Kinds, a.AgentVersion, limits, a.TaskVersion, a.AppliedTaskVersion, rejected, heartbeat, view.EndpointChanges, Time(a.CreatedAtMs), Time(a.LastSeenAtMs), a.Ipv6Host, a.Ipv6Supported, Ipv6Of(view), !a.PublicHostManual, !a.Ipv6HostManual, a.AddedBy); + a.Kinds, a.AgentVersion, limits, a.TaskVersion, a.AppliedTaskVersion, rejected, heartbeat, view.EndpointChanges, view.RuntimeStatus, Time(a.CreatedAtMs), Time(a.LastSeenAtMs), a.Ipv6Host, a.Ipv6Supported, Ipv6Of(view), !a.PublicHostManual, !a.Ipv6HostManual, a.AddedBy); } public static Ipv6DiagnosticDto Ipv6Of(AgentView view) diff --git a/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs b/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs index a9e00b8..2c136be 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/AdminDtos.cs @@ -1,5 +1,6 @@ using Fleet.Core; using System.Text.Json; +using MoShiro.Protocol; namespace MoShiro.Backend.Api; @@ -53,6 +54,7 @@ public sealed record AgentDto( IReadOnlyList Rejected, HeartbeatDto? Heartbeat, int EndpointChanges, + NodeRuntimeStatus RuntimeStatus, DateTimeOffset CreatedAt, DateTimeOffset? LastSeenAt, string Ipv6Host = "", @@ -123,10 +125,10 @@ public sealed record ConnectedProbeDto(string ProbeId, string? Version, string? /// 某个 Probe 现在连着的一个 Agent。 public sealed record ProbeConnectionDto(string AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, DateTimeOffset ConnectedSince, string IpVersion = "ipv4"); -public sealed record ProbeDto(string Id, string Name, DateTimeOffset CreatedAt, DateTimeOffset? RevokedAt, bool Revoked, bool Online, IReadOnlyList Connections, Ipv6DiagnosticDto? Ipv6 = null, string? AddedBy = null); +public sealed record ProbeDto(string Id, string Name, DateTimeOffset CreatedAt, DateTimeOffset? RevokedAt, bool Revoked, bool Online, IReadOnlyList Connections, NodeRuntimeStatus RuntimeStatus, Ipv6DiagnosticDto? Ipv6 = null, string? AddedBy = null); /// 一对（Probe，Agent）的线路开关：linked 是最终结果，override 是手动设置（null = 跟随自动连接），connected 是现在有没有连着。 -public sealed record LinkDto(string ProbeId, string ProbeName, string AgentId, string AgentName, bool Linked, bool? Override, bool Connected); +public sealed record LinkDto(string ProbeId, string ProbeName, string AgentId, string AgentName, bool Linked, bool? Override, bool Connected, NodeRuntimeStatus ProbeRuntimeStatus, NodeRuntimeStatus AgentRuntimeStatus, string ConnectionState = "connecting", string? ConnectionReason = null, string? LastError = null, int RetryAttempt = 0, long? RetryInMs = null); public sealed record LinkMatrixDto(bool AutoLink, IReadOnlyList Items); diff --git a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs index d466561..ee4b20f 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/ApiJsonContext.cs @@ -41,6 +41,7 @@ namespace MoShiro.Backend.Api; [JsonSerializable(typeof(AgentPatchRequest))] [JsonSerializable(typeof(LinkDto))] [JsonSerializable(typeof(LinkMatrixDto))] +[JsonSerializable(typeof(LinkRetryDto))] [JsonSerializable(typeof(SetLinkRequest))] [JsonSerializable(typeof(BuildDto))] [JsonSerializable(typeof(ReleaseDto))] diff --git a/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs b/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs index cde6f11..8dea47f 100644 --- a/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs +++ b/backend/src/MoShiro.Backend.Api/Contracts/PublicDtos.cs @@ -48,7 +48,10 @@ public sealed record QuicLineHealthDto( uint? TransportCongestionEvents, uint? TransportEcnCongestionEvents, ushort? PathMtu, - DateTimeOffset At); + DateTimeOffset At, + string State = "connecting", + int RetryAttempt = 0, + long? RetryInMs = null); public sealed record OverviewDto(string Range, DateTimeOffset From, DateTimeOffset To, string Step, string Source, bool Degraded, IReadOnlyList Lines); diff --git a/backend/src/MoShiro.Backend.Application/Agents/AgentRuntime.cs b/backend/src/MoShiro.Backend.Application/Agents/AgentRuntime.cs index 6471477..d2f6ffe 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/AgentRuntime.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/AgentRuntime.cs @@ -13,6 +13,7 @@ public sealed class AgentRuntimeState private readonly object _gate = new(); private int _endpointChanges; private IReadOnlyList _probes = []; + private string? _lastDisconnectReason; /// 当前连着这个 Agent 的 Probe（每个 Probe 一项）。 public IReadOnlyList Probes => Volatile.Read(ref _probes); @@ -60,6 +61,17 @@ public sealed class AgentRuntimeState public int EndpointChanges => Volatile.Read(ref _endpointChanges); + public string? LastDisconnectReason => Volatile.Read(ref _lastDisconnectReason); + + public void MarkOnline() + { + Volatile.Write(ref _lastDisconnectReason, null); + LastHeartbeat = null; + LastHeartbeatAtMs = 0; + } + + public void MarkOffline(string? reason) => Volatile.Write(ref _lastDisconnectReason, reason); + public void CountEndpointChange() => Interlocked.Increment(ref _endpointChanges); } diff --git a/backend/src/MoShiro.Backend.Application/Agents/AgentService.cs b/backend/src/MoShiro.Backend.Application/Agents/AgentService.cs index 9a616be..f952761 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/AgentService.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/AgentService.cs @@ -7,10 +7,46 @@ using MoShiro.Protocol; namespace MoShiro.Backend.Application; -public sealed record AgentView(AgentRecord Agent, bool Online, IReadOnlyList Probes, AgentHeartbeatExtra? Heartbeat, int EndpointChanges, long? HeartbeatAtMs = null, HostSnapshot? Host = null) +public sealed record AgentView(AgentRecord Agent, bool Online, IReadOnlyList Probes, AgentHeartbeatExtra? Heartbeat, int EndpointChanges, long? HeartbeatAtMs = null, HostSnapshot? Host = null, string? LastDisconnectReason = null) { /// 至少有一个 Probe 连着。 public bool ProbeConnected => Probes.Count > 0; + + public NodeRuntimeStatus RuntimeStatus + { + get + { + if (Agent.Status == NodeStatus.Disabled) + { + return NodeRuntimeStatus.Maintenance; + } + if (Agent.Status == NodeStatus.Pending) + { + return NodeRuntimeStatus.Initializing; + } + if (!Online) + { + if (LastDisconnectReason == "updating") + { + return NodeRuntimeStatus.Updating; + } + if (LastDisconnectReason == "stopping") + { + return Heartbeat?.Update?.State == "applying" ? NodeRuntimeStatus.Updating : NodeRuntimeStatus.Stopping; + } + return NodeRuntimeStatus.Offline; + } + if (Heartbeat?.Update?.State == "applying") + { + return NodeRuntimeStatus.Updating; + } + if (Heartbeat is null) + { + return NodeRuntimeStatus.Initializing; + } + return Heartbeat.RuntimeStatus; + } + } } /// Agent 的管理：列表、修改、审批、启停、接管、删除，以及 Probe 需要的 Agent 列表。 @@ -53,9 +89,12 @@ public sealed partial class AgentService public AgentView View(AgentRecord agent) { _runtime.TryGet(agent.Id, out AgentRuntimeState? state); - return new AgentView(agent, _gateway.IsOnline(agent.Id), state?.Probes ?? [], state?.LastHeartbeat, state?.EndpointChanges ?? 0, state?.LastHeartbeatAtMs, state?.Host); + return new AgentView(agent, _gateway.IsOnline(agent.Id), state?.Probes ?? [], state?.LastHeartbeat, state?.EndpointChanges ?? 0, state?.LastHeartbeatAtMs, state?.Host, state?.LastDisconnectReason); } + public ValueTask NotifyProbeLinkRetryAsync(NodeId probe, NodeId agent, CancellationToken ct) => + _gateway.NotifyProbeLinkRetryAsync(probe, agent, ct); + private static bool ValidHost(string host, string ipVersion) { if (IPAddress.TryParse(host, out IPAddress? address)) diff --git a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs index d1ed0ee..d4db58c 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/MoShiroFleetHooks.cs @@ -71,11 +71,13 @@ public sealed class MoShiroNodeEventSink : INodeEventSink private readonly IProbeStore _probes; private readonly NodeOwnerService _owners; private readonly ProbeQuicRuntime _quicRuntime; + private readonly ProbeRuntime _probeRuntime; - public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, ScheduleService schedule, IProbeStore probes, NodeOwnerService owners, ProbeQuicRuntime quicRuntime) + public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, ScheduleService schedule, IProbeStore probes, NodeOwnerService owners, ProbeQuicRuntime quicRuntime, ProbeRuntime probeRuntime) { _owners = owners; _quicRuntime = quicRuntime; + _probeRuntime = probeRuntime; _releases = releases; _schedule = schedule; _probes = probes; @@ -99,10 +101,12 @@ public sealed class MoShiroNodeEventSink : INodeEventSink switch (nodeEvent.Kind) { case NodeEventKind.Online: + _runtime.Get(id).MarkOnline(); _live.Publish(new NodeSignal("agent", id.Value, true, nodeEvent.AtMs)); await _events.AddAsync(id, "agent_online", nodeEvent.AtMs, null, ct); break; case NodeEventKind.Offline: + _runtime.Get(id).MarkOffline(nodeEvent.Reason); _runtime.MarkOffline(id); _releases.RecordStatus(id, null); _live.Publish(new NodeSignal("agent", id.Value, false, nodeEvent.AtMs)); @@ -143,6 +147,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink switch (nodeEvent.Kind) { case NodeEventKind.Online: + _probeRuntime.MarkOnline(id.Value); _live.Publish(new NodeSignal("probe", id.Value, true, nodeEvent.AtMs)); break; case NodeEventKind.Hello: @@ -170,6 +175,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink case NodeEventKind.Offline: _live.Publish(new NodeSignal("probe", id.Value, false, nodeEvent.AtMs)); _releases.RecordStatus(id, null); + _probeRuntime.MarkOffline(id.Value, nodeEvent.Reason); _quicRuntime.MarkProbeOffline(id.Value, nodeEvent.AtMs); break; case NodeEventKind.Heartbeat when nodeEvent.Extra is { } extra: @@ -177,6 +183,10 @@ public sealed class MoShiroNodeEventSink : INodeEventSink { ProbeHeartbeatExtra? heartbeat = extra.Deserialize(MoShiroJsonContext.Default.ProbeHeartbeatExtra); _releases.RecordStatus(id, heartbeat?.Update); + if (heartbeat is not null) + { + _probeRuntime.RecordHeartbeat(id.Value, nodeEvent.AtMs, heartbeat); + } if (heartbeat?.QuicLinks is { } quicLinks) { _quicRuntime.Replace(id.Value, nodeEvent.AtMs, quicLinks); diff --git a/backend/src/MoShiro.Backend.Application/Agents/ProbeQuicRuntime.cs b/backend/src/MoShiro.Backend.Application/Agents/ProbeQuicRuntime.cs index 4958972..951c236 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/ProbeQuicRuntime.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/ProbeQuicRuntime.cs @@ -39,6 +39,10 @@ public sealed class ProbeQuicRuntime AgentId = old.AgentId, IpVersion = old.IpVersion, Connected = false, + State = ProbeLinkAttemptState.RetryWait, + LastError = "Probe 与 Backend 的连接中断", + RetryAttempt = old.RetryAttempt, + RetryInMs = old.RetryInMs, HandshakeAttempts = old.HandshakeAttempts, HandshakeSuccesses = old.HandshakeSuccesses, HandshakeFailures = old.HandshakeFailures, diff --git a/backend/src/MoShiro.Backend.Application/Agents/ProbeRuntime.cs b/backend/src/MoShiro.Backend.Application/Agents/ProbeRuntime.cs new file mode 100644 index 0000000..f26d966 --- /dev/null +++ b/backend/src/MoShiro.Backend.Application/Agents/ProbeRuntime.cs @@ -0,0 +1,22 @@ +using System.Collections.Concurrent; +using MoShiro.Protocol; + +namespace MoShiro.Backend.Application; + +/// Volatile runtime status reported by each Probe's heartbeat and Fleet connection lifecycle. +public sealed class ProbeRuntime +{ + private readonly ConcurrentDictionary _probes = new(StringComparer.Ordinal); + + public ProbeRuntimeSnapshot? Get(string probeId) => _probes.TryGetValue(probeId, out ProbeRuntimeSnapshot? snapshot) ? snapshot : null; + + public void RecordHeartbeat(string probeId, long atMs, ProbeHeartbeatExtra heartbeat) => + _probes[probeId] = new ProbeRuntimeSnapshot(heartbeat, atMs, null); + + public void MarkOnline(string probeId) => _probes[probeId] = new ProbeRuntimeSnapshot(null, 0, null); + + public void MarkOffline(string probeId, string? reason) => + _probes.AddOrUpdate(probeId, _ => new ProbeRuntimeSnapshot(null, 0, reason), (_, current) => current with { DisconnectReason = reason }); +} + +public sealed record ProbeRuntimeSnapshot(ProbeHeartbeatExtra? Heartbeat, long AtMs, string? DisconnectReason); diff --git a/backend/src/MoShiro.Backend.Application/Agents/ProbeService.cs b/backend/src/MoShiro.Backend.Application/Agents/ProbeService.cs index 7e05d4f..d2f4d30 100644 --- a/backend/src/MoShiro.Backend.Application/Agents/ProbeService.cs +++ b/backend/src/MoShiro.Backend.Application/Agents/ProbeService.cs @@ -8,9 +8,41 @@ namespace MoShiro.Backend.Application; /// 某个 Probe 现在连着的一个 Agent（Agent 上报）。 public sealed record ProbeConnection(NodeId AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, long SinceMs, string IpVersion = IpVersions.V4); -public sealed record ProbeView(ProbeRecord Probe, bool Online, IReadOnlyList Connections) +public sealed record ProbeView(ProbeRecord Probe, bool Online, IReadOnlyList Connections, ProbeRuntimeSnapshot? Runtime = null) { public bool Revoked => Probe.RevokedAtMs is not null; + + public NodeRuntimeStatus RuntimeStatus + { + get + { + if (Revoked) + { + return NodeRuntimeStatus.Maintenance; + } + if (!Online) + { + if (Runtime?.DisconnectReason == "updating") + { + return NodeRuntimeStatus.Updating; + } + if (Runtime?.DisconnectReason == "stopping") + { + return Runtime.Heartbeat?.Update?.State == "applying" ? NodeRuntimeStatus.Updating : NodeRuntimeStatus.Stopping; + } + return NodeRuntimeStatus.Offline; + } + if (Runtime?.Heartbeat?.Update?.State == "applying") + { + return NodeRuntimeStatus.Updating; + } + if (Runtime?.Heartbeat is null) + { + return NodeRuntimeStatus.Initializing; + } + return Runtime.Heartbeat.RuntimeStatus; + } + } } /// @@ -28,13 +60,15 @@ public sealed partial class ProbeService private readonly TimeProvider _time; private readonly NodeInfoService _nodes; private readonly LinkService _links; + private readonly ProbeRuntime _probeRuntime; private readonly NodeOwnerService _owners; - public ProbeService(NodeOwnerService owners, IProbeStore probes, IAgentStore agents, AgentRuntime runtime, IAgentGateway gateway, TimeProvider time, NodeInfoService nodes, LinkService links) + public ProbeService(NodeOwnerService owners, IProbeStore probes, IAgentStore agents, AgentRuntime runtime, IAgentGateway gateway, TimeProvider time, NodeInfoService nodes, LinkService links, ProbeRuntime probeRuntime) { _owners = owners; _links = links; + _probeRuntime = probeRuntime; _nodes = nodes; _probes = probes; _agents = agents; @@ -133,7 +167,7 @@ public sealed partial class ProbeService } } - return new ProbeView(probe, _gateway.IsOnline(new NodeId(probe.Id)), [.. connections.OrderBy(c => c.AgentName, StringComparer.Ordinal)]); + return new ProbeView(probe, _gateway.IsOnline(new NodeId(probe.Id)), [.. connections.OrderBy(c => c.AgentName, StringComparer.Ordinal)], _probeRuntime.Get(probe.Id)); } [GeneratedRegex(@"^[\p{L}\p{M}\p{N}._-]{1,64}$")] diff --git a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs index 9cb9651..dc5f090 100644 --- a/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs +++ b/backend/src/MoShiro.Backend.Application/ApplicationServiceCollectionExtensions.cs @@ -16,6 +16,7 @@ public static class ApplicationServiceCollectionExtensions services.AddHostedService(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs b/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs index b330cff..5325d4e 100644 --- a/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs +++ b/backend/src/MoShiro.Backend.Application/Ports/IAgentGateway.cs @@ -30,6 +30,9 @@ public interface IAgentGateway ValueTask DisconnectAsync(NodeId agent, string reason, CancellationToken ct); + /// Notify an online Probe to immediately retry all IP paths to an Agent. + ValueTask NotifyProbeLinkRetryAsync(NodeId probe, NodeId agent, CancellationToken ct) => ValueTask.FromResult(false); + /// 把发布清单和签名原样交给节点（通知，不等回复）。 ValueTask NotifyUpdateOfferAsync(NodeId node, string manifest, string signature, CancellationToken ct); diff --git a/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs b/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs index 55d4fb4..a257f1e 100644 --- a/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs +++ b/backend/src/MoShiro.Backend.Gateway/FleetGateway.cs @@ -64,6 +64,9 @@ public sealed class FleetGateway : IAgentGateway public ValueTask DisconnectAsync(NodeId agent, string reason, CancellationToken ct) => Connections.DisconnectAsync(agent, reason, ct); + public ValueTask NotifyProbeLinkRetryAsync(NodeId probe, NodeId agent, CancellationToken ct) => + Connections.NotifyAsync(probe, MoShiroMethods.ProbeLinkRetry, new ProbeLinkRetryParams { AgentId = agent }, MoShiroJsonContext.Default.ProbeLinkRetryParams, ct); + public async ValueTask NotifyUpdateOfferAsync(NodeId node, string manifest, string signature, CancellationToken ct) => await Connections.NotifyAsync(node, MoShiroMethods.UpdateOffer, new UpdateOfferParams { Manifest = manifest, Signature = signature }, MoShiroJsonContext.Default.UpdateOfferParams, ct); diff --git a/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs b/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs index 09746c8..34d1c45 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentEndToEndTests.cs @@ -36,6 +36,20 @@ public sealed class AgentEndToEndTests : IClassFixture await using ProbeSimulator probe = await agent.ConnectProbeAsync(); await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue()); + await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["runtime_status"]!.GetValue() == "running"); + } + + [Fact] + public async Task Agent_OrderlyShutdownIsReportedAsStoppingRatherThanOffline() + { + AgentHarness agent = await AgentHarness.StartAsync(_backend); + await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["runtime_status"]!.GetValue() == "running"); + + await agent.DisposeAsync(); + + await BackendFixture.EventuallyAsync(async () => !(await AgentViewAsync(agent))["online"]!.GetValue()); + JsonNode final = await AgentViewAsync(agent); + Assert.Equal("stopping", final["runtime_status"]!.GetValue()); } [Fact] diff --git a/backend/tests/MoShiro.Backend.Tests/LinkTests.cs b/backend/tests/MoShiro.Backend.Tests/LinkTests.cs index e6eff94..9ed7519 100644 --- a/backend/tests/MoShiro.Backend.Tests/LinkTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/LinkTests.cs @@ -77,4 +77,19 @@ public sealed class LinkTests : IClassFixture using HttpResponseMessage noAgent = await SetLinkAsync(probe, "agt_missing", false); Assert.Equal(HttpStatusCode.NotFound, noAgent.StatusCode); } + + [Fact] + public async Task Links_RetryEnablesPairAndReportsWhenProbeIsOffline() + { + await using AgentSimulator agent = await _backend.ConnectAgentAsync(); + string probe = await _backend.CreateProbeIdAsync(); + + using HttpResponseMessage response = await _backend.Admin.PostAsync($"/api/admin/links/{probe}/{agent.NodeId.Value}/retry", BackendFixture.Json(new JsonObject())); + Assert.Equal(HttpStatusCode.Accepted, response.StatusCode); + JsonNode body = JsonNode.Parse(await response.Content.ReadAsStringAsync())!; + Assert.False(body["queued"]!.GetValue()); + Assert.True(body["link"]!["linked"]!.GetValue()); + Assert.Equal("probe_unavailable", body["link"]!["connection_state"]!.GetValue()); + Assert.Contains("Probe 状态", body["link"]!["connection_reason"]!.GetValue()); + } } diff --git a/backend/tests/MoShiro.Backend.Tests/ProbeEndToEndTests.cs b/backend/tests/MoShiro.Backend.Tests/ProbeEndToEndTests.cs index b43a62d..9ae38dd 100644 --- a/backend/tests/MoShiro.Backend.Tests/ProbeEndToEndTests.cs +++ b/backend/tests/MoShiro.Backend.Tests/ProbeEndToEndTests.cs @@ -48,6 +48,7 @@ public sealed class ProbeEndToEndTests : IClassFixture // 直连面板：在线，hello 里的系统、架构和版本都被记下来了 await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, path))["online"]!.GetValue(), because: "the probe links to the panel"); + await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, path))["runtime_status"]!.GetValue() == "probe_paused", because: "a Probe without any enabled Agent endpoints reports paused probing"); await BackendFixture.EventuallyAsync( async () => (await _backend.GetJsonAsync(_backend.Anonymous, "/api/public/nodes"))["items"]!.AsArray().Any(n => n!["id"]!.GetValue() == probe.ProbeId && n["os"] is not null && n["arch"] is not null), because: "the platform comes from the probe's own hello"); @@ -58,6 +59,7 @@ public sealed class ProbeEndToEndTests : IClassFixture await BackendFixture.EventuallyAsync(async () => !(await _backend.GetJsonAsync(_backend.Admin, path))["online"]!.GetValue() && !probe.App.Link.IsConnected, because: "a revoked probe is disconnected and cannot come back"); + Assert.Equal("maintenance", (await _backend.GetJsonAsync(_backend.Admin, path))["runtime_status"]!.GetValue()); } [Fact] diff --git a/docs/ops/deployment.md b/docs/ops/deployment.md index a3d971a..fde1ed9 100644 --- a/docs/ops/deployment.md +++ b/docs/ops/deployment.md @@ -18,6 +18,8 @@ Agent 的新默认监听地址为 `::`（双栈）；现有 `agent.json` 如果 上线后，在管理端的 Probe / Agent 页面查看「IPv6 诊断」。它显示检测网络范围、全局地址和默认路由；机器有 IPv6 但容器没有时，检查是否使用宿主网络。没有填 Agent IPv6 测量地址时，诊断仍会上报，但面板不会建立该 Agent 的 IPv6 测量线路。 +管理端把节点审批状态与进程运行状态分开显示。运行状态为初始化中、运行中、暂停探测、维护中、停止中、失联或更新中；Agent 停用/Probe 吊销显示为维护中，Probe 没有任何面板下发的 Agent 目标时显示暂停探测。节点通过心跳上报运行状态，正常关闭时 Fleet 连接会发送停止标记；突然断网或心跳超时显示失联。线路只有在 Probe 和 Agent 都处于运行中且 Agent 确认了 Probe 会话时才显示已连接。 + ```text scripts/publish.sh → backend/artifacts/publish/// scripts/package.sh → backend/artifacts/packages/moshiro---.tar.gz（附 .sha256） diff --git a/web/shared/api/schema.ts b/web/shared/api/schema.ts index 6a23445..a560345 100644 --- a/web/shared/api/schema.ts +++ b/web/shared/api/schema.ts @@ -1093,6 +1093,23 @@ export interface paths { patch?: never; trace?: never; }; + "/api/admin/links/{probe_id}/{agent_id}/retry": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Enable and immediately retry the Probe-to-Agent connection */ + post: operations["links_retry"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/healthz": { parameters: { query?: never; @@ -1314,6 +1331,7 @@ export interface components { heartbeat: null | components["schemas"]["HeartbeatDto"]; /** Format: int32 */ endpoint_changes: number | string; + runtime_status: components["schemas"]["NodeRuntimeStatus"]; /** Format: date-time */ created_at: unknown; /** Format: date-time */ @@ -1705,11 +1723,28 @@ export interface components { linked: boolean; override: null | boolean; connected: boolean; + probe_runtime_status: components["schemas"]["NodeRuntimeStatus"]; + agent_runtime_status: components["schemas"]["NodeRuntimeStatus"]; + /** @default connecting */ + connection_state: string; + connection_reason?: null | string; + last_error?: null | string; + /** + * Format: int32 + * @default 0 + */ + retry_attempt: number | string; + /** Format: int64 */ + retry_in_ms?: null | number | string; }; LinkMatrixDto: { auto_link: boolean; items: components["schemas"]["LinkDto"][]; }; + LinkRetryDto: { + queued: boolean; + link: components["schemas"]["LinkDto"]; + }; LocalLimitsDto: { /** Format: double */ max_total_mbps: number | string; @@ -1795,6 +1830,8 @@ export interface components { passkey: null | string; account: null | string; }; + /** @enum {unknown} */ + NodeRuntimeStatus: "initializing" | "running" | "probe_paused" | "maintenance" | "stopping" | "offline" | "updating"; NodeUpdateDto: { state: string; current_digest: null | string; @@ -2009,6 +2046,7 @@ export interface components { revoked: boolean; online: boolean; connections: components["schemas"]["ProbeConnectionDto"][]; + runtime_status: components["schemas"]["NodeRuntimeStatus"]; ipv6?: null | components["schemas"]["Ipv6DiagnosticDto"]; added_by?: null | string; }; @@ -2072,6 +2110,15 @@ export interface components { path_mtu: null | number | string; /** Format: date-time */ at: unknown; + /** @default connecting */ + state: string; + /** + * Format: int32 + * @default 0 + */ + retry_attempt: number | string; + /** Format: int64 */ + retry_in_ms?: null | number | string; }; RawChunkMetaDto: { chunk_id: string; @@ -6128,6 +6175,56 @@ export interface operations { }; }; }; + links_retry: { + parameters: { + query?: never; + header?: never; + path: { + probe_id: string; + agent_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description OK */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["LinkRetryDto"]; + }; + }; + /** @description Unauthorized */ + 401: { + headers: { + [name: string]: unknown; + }; + content: { + "application/problem+json": components["schemas"]["ProblemDto"]; + }; + }; + /** @description Forbidden */ + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/problem+json": components["schemas"]["ProblemDto"]; + }; + }; + /** @description Not Found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/problem+json": components["schemas"]["ProblemDto"]; + }; + }; + }; + }; health_get: { parameters: { query?: never; diff --git a/web/shared/api/types.ts b/web/shared/api/types.ts index 4eeb1c8..3804051 100644 --- a/web/shared/api/types.ts +++ b/web/shared/api/types.ts @@ -7,6 +7,7 @@ type NarrowValue = unknown extends V ? string : number extends V ? Exclude = { [K in keyof T]: NarrowValue } export type AgentDto = Narrow +export type NodeRuntimeStatus = S['NodeRuntimeStatus'] export type AgentIpv6Dto = Narrow export type ConnectedProbeDto = Narrow export type ProbeDto = Narrow @@ -24,6 +25,7 @@ export type ReleaseDto = Narrow export type ReleaseKeysDto = Narrow export type ReleaseKeyDto = Narrow export type LinkDto = Narrow +export type LinkRetryDto = Narrow export type LinkMatrixDto = Narrow export type OverviewDto = Narrow export type OverviewLineDto = Narrow