From 1aa79ae0b1e175456ff39cc16afcc40bcf91a44f Mon Sep 17 00:00:00 2001 From: caoqianming Date: Wed, 19 Aug 2026 14:53:36 +0800 Subject: [PATCH] =?UTF-8?q?fix(node):=20=E4=BF=AE=E5=A4=8D=E9=87=8D?= =?UTF-8?q?=E8=BF=9E=E7=8A=B6=E6=80=81=E5=88=86=E8=A3=82=E4=B8=8E=E7=95=8C?= =?UTF-8?q?=E9=9D=A2=E5=81=87=E6=9C=AA=E6=B3=A8=E5=86=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- RUN.md | 2 + tests/test_software_nodes.py | 2 + tests/test_windows_node_source.py | 11 +++ web/routers/software_nodes.py | 11 +++ .../Zcbot.WindowsNode/ConfigurationForm.cs | 2 + .../Zcbot.WindowsNode/JobInboxStore.cs | 18 +++- .../Zcbot.WindowsNode/JobMonitorModels.cs | 3 +- .../Zcbot.WindowsNode/JobOutputUploader.cs | 67 +++++++++++++ .../Zcbot.WindowsNode/NodeConnectionLoop.cs | 97 +++++++++++-------- .../TrayApplicationContext.cs | 6 ++ 10 files changed, 177 insertions(+), 42 deletions(-) diff --git a/RUN.md b/RUN.md index 7ccf97e..8050524 100644 --- a/RUN.md +++ b/RUN.md @@ -1147,6 +1147,8 @@ Web 用户登录后，文件栏 Job 中心会聚合本人最近任务。活动 直接双击 EXE 启动托盘 UI：红点为未注册/身份失效，黄点为连接中，绿点为在线；双击托盘图标打开配置窗。原 CLI 注册入口继续保留，无 UI 模式使用 `Zcbot.WindowsNode.exe run --headless`。 +若执行中的 Job 遇到 Node 重连，客户端会先等待已接收的本地执行管线收尾，再建立新连接，避免同一 Worker 被新旧连接同时恢复。若云端已将 Job 判为失败或取消、但本地 Worker 随后仍生成了结果，本机任务会显示“云端已终止”并保留工作区文件，不再永久停在 90% 重传；这些本地结果不会反向覆盖云端终态。 + 经 nginx 反代时，`/v1/software-nodes/connect` 必须单独透传 WebSocket Upgrade/Connection 头并设置长连接超时，配置见 `deploy/nginx/zcbot.conf.example`。若注册成功后节点持续显示“连接中断，等待重连”，先用 WebSocket 握手检查该路径；返回普通 HTTP 404 通常表示请求落入了清空 `Connection` 头的默认 location。 diff --git a/tests/test_software_nodes.py b/tests/test_software_nodes.py index 11ae2d4..9f872e8 100644 --- a/tests/test_software_nodes.py +++ b/tests/test_software_nodes.py @@ -758,6 +758,8 @@ class SoftwareJobProtocolTests(unittest.TestCase): self.assertIn('digest.hexdigest() != item["sha256"]', source) self.assertIn('contract.expected_outputs(context["request"])', source) self.assertIn("artifact_id not in requested_ids", source) + self.assertIn('"code": "JOB_ALREADY_TERMINAL"', source) + self.assertIn('context["status"] in {"failed", "cancelled"}', source) @patch("core.software_jobs.session_scope") def test_job_state_restores_disconnected_job(self, session_scope) -> None: diff --git a/tests/test_windows_node_source.py b/tests/test_windows_node_source.py index 7cade72..16ee3b6 100644 --- a/tests/test_windows_node_source.py +++ b/tests/test_windows_node_source.py @@ -315,6 +315,8 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn('stage = "software_running"', connection) self.assertNotIn('stage = "origin_running"', connection) self.assertIn("&& !job.UploadComplete", connection) + self.assertIn("&& !job.CloudTerminal", connection) + self.assertIn("await Task.WhenAll(jobPipelines.Values.ToArray())", connection) self.assertIn("StartJobPipeline(socket, job)", connection) self.assertIn('stage = "downloading_inputs"', connection) self.assertIn('Path.Combine(jobDirectory, "terminal.json")', inbox) @@ -392,6 +394,8 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn("adapter.WorkspaceStateFilename", connection) self.assertIn("WorkspaceSize(jobDirectory), allowConflict: true", uploader) self.assertIn("completeResponse.StatusCode == HttpStatusCode.Conflict", uploader) + self.assertIn('"JOB_ALREADY_TERMINAL"', uploader) + self.assertIn('"cloud-terminal.json"', uploader) self.assertIn("OpenOutputWithRetryAsync", uploader) self.assertIn("IsSharingViolation", uploader) self.assertIn("(exception.HResult & 0xFFFF) is 32 or 33", uploader) @@ -407,6 +411,13 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertNotIn('".tmp-" + Guid.NewGuid()', uploader) self.assertNotIn("Process.Start", uploader) + tray = (PROJECT / "TrayApplicationContext.cs").read_text(encoding="utf-8") + self.assertIn("_ = form.Handle", tray) + self.assertIn( + 'UpdateStatus(NodeStatus.Create(NodeState.Connecting, "正在连接 zcbot…"))', + tray, + ) + def test_local_job_monitor_never_shows_update_before_acceptance(self) -> None: inbox = (PROJECT / "JobInboxStore.cs").read_text(encoding="utf-8") self.assertIn("state.UpdatedAt > acceptedAt", inbox) diff --git a/web/routers/software_nodes.py b/web/routers/software_nodes.py index bad8a18..75aa7b4 100644 --- a/web/routers/software_nodes.py +++ b/web/routers/software_nodes.py @@ -600,6 +600,15 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: context["capability"], context["request"], body.get("artifact_manifest") ) contract = get_contract(context["capability"]) + if context["status"] in {"failed", "cancelled"}: + raise HTTPException( + 409, + { + "code": "JOB_ALREADY_TERMINAL", + "status": context["status"], + "message": "cloud job is already terminal; local outputs were retained", + }, + ) if contract.workspace is not None: if context["status"] == "succeeded": return { @@ -652,6 +661,8 @@ def register_software_node_routes(app, *, require_user, require_admin) -> None: } await asyncio.to_thread(record_job_terminal, node_id, terminal) await dispatch_followup(request.app, job_id) + except HTTPException: + raise except (SoftwareJobError, KeyError, TypeError) as exc: raise HTTPException(409, str(exc)) from exc return {"status": "succeeded", "artifact_manifest": published} diff --git a/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs b/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs index 143107a..214595b 100644 --- a/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs +++ b/windows-node/Zcbot.WindowsNode/ConfigurationForm.cs @@ -526,6 +526,7 @@ internal sealed class ConfigurationForm : Form row.DefaultCellStyle.ForeColor = snapshot.Stage switch { "failed" => Color.Firebrick, + "cloud_terminal" => Color.Firebrick, "cancelled" => Color.DimGray, "succeeded" => Color.ForestGreen, _ => Color.FromArgb(30, 41, 59), @@ -589,6 +590,7 @@ internal sealed class ConfigurationForm : Form "uploading_outputs" => "上传结果", "succeeded" => "成功", "failed" => "失败", + "cloud_terminal" => "云端已终止", "cancelled" => "已取消", _ => stage, }; diff --git a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs index dd6b13e..02d639f 100644 --- a/windows-node/Zcbot.WindowsNode/JobInboxStore.cs +++ b/windows-node/Zcbot.WindowsNode/JobInboxStore.cs @@ -60,7 +60,8 @@ internal sealed class JobInboxStore(string jobsDirectory) root.TryGetProperty("input_transfers", out var transfers) ? transfers.Clone() : null, ReadTerminal(Path.Combine(jobDirectory, "terminal.json")), - File.Exists(Path.Combine(jobDirectory, "upload-complete.json")))); + File.Exists(Path.Combine(jobDirectory, "upload-complete.json")), + File.Exists(Path.Combine(jobDirectory, "cloud-terminal.json")))); } catch (Exception exception) when ( exception is JsonException or IOException or UnauthorizedAccessException) @@ -109,6 +110,8 @@ internal sealed class JobInboxStore(string jobsDirectory) var terminal = ReadTerminal(Path.Combine(jobDirectory, "terminal.json")); var uploadPath = Path.Combine(jobDirectory, "upload-complete.json"); var uploadComplete = File.Exists(uploadPath); + var cloudTerminalPath = Path.Combine(jobDirectory, "cloud-terminal.json"); + var cloudTerminal = ReadTerminal(cloudTerminalPath); var stage = state?.Stage ?? "accepted"; var progress = state?.Progress ?? 0; var detail = state?.Detail ?? "任务已由本机接收"; @@ -145,6 +148,16 @@ internal sealed class JobInboxStore(string jobsDirectory) : "结果已上传并由云端确认"; updatedAt = LatestWrite(updatedAt, uploadPath); } + else if (cloudTerminal is JsonElement cloudValue) + { + var cloudStatus = ReadString(cloudValue, "status", "failed"); + stage = "cloud_terminal"; + progress = 100; + detail = cloudStatus == "cancelled" + ? "云端任务已取消，本地生成的结果仍保留" + : "云端任务已终止，本地生成的结果仍保留"; + updatedAt = LatestWrite(updatedAt, cloudTerminalPath); + } snapshots.Add(new JobDisplaySnapshot( jobId, capability, @@ -602,7 +615,8 @@ internal sealed record RecoverableJob( WorkspaceBinding? Workspace, JsonElement? InputTransfers, JsonElement? Terminal, - bool UploadComplete); + bool UploadComplete, + bool CloudTerminal); internal sealed record JobOfferResult( bool Accepted, Guid JobId, Guid LeaseId, string RequestDigest, string Reason) diff --git a/windows-node/Zcbot.WindowsNode/JobMonitorModels.cs b/windows-node/Zcbot.WindowsNode/JobMonitorModels.cs index 80383f2..bd827d5 100644 --- a/windows-node/Zcbot.WindowsNode/JobMonitorModels.cs +++ b/windows-node/Zcbot.WindowsNode/JobMonitorModels.cs @@ -18,5 +18,6 @@ internal sealed record JobDisplaySnapshot( DateTimeOffset UpdatedAt, bool UploadComplete) { - internal bool IsActive => Stage is not ("succeeded" or "failed" or "cancelled"); + internal bool IsActive => Stage is not ( + "succeeded" or "failed" or "cancelled" or "cloud_terminal"); } diff --git a/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs b/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs index 757f54c..e3833ef 100644 --- a/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs +++ b/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs @@ -29,6 +29,8 @@ internal sealed class JobOutputUploader(NodeConfig config) NodePaths.ForCurrentMachine().JobsDirectory, job.JobId.ToString("D")); var completionPath = Path.Combine(jobDirectory, "upload-complete.json"); if (File.Exists(completionPath)) return; + var cloudTerminalPath = Path.Combine(jobDirectory, "cloud-terminal.json"); + if (File.Exists(cloudTerminalPath)) return; var terminalPath = Path.Combine(jobDirectory, "terminal.json"); using var terminal = JsonDocument.Parse(await File.ReadAllBytesAsync(terminalPath)); if (terminal.RootElement.GetProperty("status").GetString() != "succeeded") return; @@ -44,6 +46,7 @@ internal sealed class JobOutputUploader(NodeConfig config) { var replay = await TryCompleteAsync( client, jobDirectory, job, manifest, WorkspaceSize(jobDirectory), allowConflict: true); + if (File.Exists(cloudTerminalPath)) return; if (replay is not null) { await WriteCompletionWithRetryAsync(job, completionPath, replay); @@ -218,6 +221,15 @@ internal sealed class JobOutputUploader(NodeConfig config) $"/v1/software-jobs/{job.JobId:D}/outputs/complete", completeContent); if (allowConflict && completeResponse.StatusCode == HttpStatusCode.Conflict) { + var body = await completeResponse.Content.ReadAsByteArrayAsync(); + if (TryReadCloudTerminal(body, out var status, out var detail)) + { + WriteCloudTerminal(jobDirectory, status, detail); + Log(jobDirectory, "WARN", + $"Output upload stopped job={job.JobId:D} " + + $"reason=cloud_terminal status={status}", error: true); + return null; + } Log(jobDirectory, "INFO", $"Output upload reconciliation pending job={job.JobId:D} " + "reason=cloud_not_complete"); @@ -227,6 +239,61 @@ internal sealed class JobOutputUploader(NodeConfig config) return await completeResponse.Content.ReadAsByteArrayAsync(); } + private static bool TryReadCloudTerminal( + byte[] responseBody, out string status, out string detail) + { + status = "failed"; + detail = "云端任务已终止，本地生成的结果仍保留"; + try + { + using var response = JsonDocument.Parse(responseBody); + if (!response.RootElement.TryGetProperty("detail", out var value) + || value.ValueKind != JsonValueKind.Object + || !value.TryGetProperty("code", out var code) + || code.GetString() != "JOB_ALREADY_TERMINAL") + { + return false; + } + if (value.TryGetProperty("status", out var statusValue) + && statusValue.GetString() is { Length: > 0 } cloudStatus) + { + status = cloudStatus; + } + if (value.TryGetProperty("message", out var messageValue) + && messageValue.GetString() is { Length: > 0 } message) + { + detail = message; + } + return true; + } + catch (JsonException) + { + return false; + } + } + + private static void WriteCloudTerminal(string jobDirectory, string status, string detail) + { + var path = Path.Combine(jobDirectory, "cloud-terminal.json"); + if (File.Exists(path)) return; + var pending = path + ".pending"; + var content = JsonSerializer.SerializeToUtf8Bytes(new + { + status, + detail, + recorded_at = DateTimeOffset.UtcNow, + }, JsonOptions); + File.WriteAllBytes(pending, content); + try + { + File.Move(pending, path, overwrite: false); + } + catch (IOException) when (File.Exists(path)) + { + File.Delete(pending); + } + } + private static long? WorkspaceSize(string jobDirectory) { var path = Path.Combine(jobDirectory, "workspace-result.json"); diff --git a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs index 67859fc..8f672a8 100644 --- a/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs +++ b/windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs @@ -32,46 +32,59 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? internal async Task RunAsync(CancellationToken cancellationToken) { - var attempt = 0; - while (!cancellationToken.IsCancellationRequested) + try { - try + var attempt = 0; + while (!cancellationToken.IsCancellationRequested) { - Report(NodeState.Connecting, "正在连接 zcbot…"); - await ConnectOnceAsync(cancellationToken); - attempt = 0; - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) - { - throw; - } - catch (WebSocketException exception) - { - Report(NodeState.Offline, "连接中断，等待重连"); - Console.Error.WriteLine($"[WARN] WebSocket disconnected: {exception.Message}"); - } - catch (IOException exception) - { - Report(NodeState.Offline, "网络不可用，等待重连"); - Console.Error.WriteLine($"[WARN] Connection I/O failed: {exception.Message}"); - } - catch (JsonException exception) - { - Report(NodeState.Offline, "服务端消息无效，等待重连"); - Console.Error.WriteLine($"[WARN] Invalid server message: {exception.Message}"); - } - catch (NodeEndpointException exception) - { - Report(NodeState.Offline, "WebSocket 握手被拒绝，请检查服务端或反向代理"); - Console.Error.WriteLine($"[WARN] WebSocket handshake rejected: {exception.Message}"); - } + try + { + Report(NodeState.Connecting, "正在连接 zcbot…"); + await ConnectOnceAsync(cancellationToken); + attempt = 0; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (WebSocketException exception) + { + Report(NodeState.Offline, "连接中断，等待重连"); + Console.Error.WriteLine($"[WARN] WebSocket disconnected: {exception.Message}"); + } + catch (IOException exception) + { + Report(NodeState.Offline, "网络不可用，等待重连"); + Console.Error.WriteLine($"[WARN] Connection I/O failed: {exception.Message}"); + } + catch (JsonException exception) + { + Report(NodeState.Offline, "服务端消息无效，等待重连"); + Console.Error.WriteLine($"[WARN] Invalid server message: {exception.Message}"); + } + catch (NodeEndpointException exception) + { + Report(NodeState.Offline, "WebSocket 握手被拒绝，请检查服务端或反向代理"); + Console.Error.WriteLine($"[WARN] WebSocket handshake rejected: {exception.Message}"); + } - var baseDelay = Backoff[Math.Min(attempt, Backoff.Length - 1)]; - attempt++; - var jitter = TimeSpan.FromMilliseconds(Random.Shared.Next(0, 750)); - var delay = baseDelay + jitter; - Console.WriteLine($"[INFO] Reconnecting in {delay.TotalSeconds:F1}s."); - await Task.Delay(delay, cancellationToken); + var baseDelay = Backoff[Math.Min(attempt, Backoff.Length - 1)]; + attempt++; + var jitter = TimeSpan.FromMilliseconds(Random.Shared.Next(0, 750)); + var delay = baseDelay + jitter; + Console.WriteLine($"[INFO] Reconnecting in {delay.TotalSeconds:F1}s."); + await Task.Delay(delay, cancellationToken); + } + } + finally + { + // A manual reconnect replaces this connection loop. Let any accepted job + // finish its local pipeline first so the new loop cannot recover the same + // worker concurrently and manufacture a conflicting terminal state. + while (!jobPipelines.IsEmpty) + { + await Task.WhenAll(jobPipelines.Values.ToArray()); + } } } @@ -132,7 +145,8 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? { if (job.Terminal is JsonElement terminal && terminal.GetProperty("status").GetString() == "succeeded" - && !job.UploadComplete) + && !job.UploadComplete + && !job.CloudTerminal) { StartJobPipeline(socket, job); } @@ -151,7 +165,7 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? { if (!job.UploadComplete) { - StartJobPipeline(socket, job); + if (!job.CloudTerminal) StartJobPipeline(socket, job); } } else @@ -468,6 +482,11 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action? recoveringOutputs, adapter.WorkspaceStateFilename, adapter.PreviewOutputIds); + if (jobInbox.ReadRecoverableJobs() + .Single(item => item.JobId == job.JobId).CloudTerminal) + { + return; + } jobInbox.WriteState(refreshed, "succeeded", 100, adapter.WorkspaceStateFilename is null ? "结果已上传并由云端确认" diff --git a/windows-node/Zcbot.WindowsNode/TrayApplicationContext.cs b/windows-node/Zcbot.WindowsNode/TrayApplicationContext.cs index 2289a1e..124a3e6 100644 --- a/windows-node/Zcbot.WindowsNode/TrayApplicationContext.cs +++ b/windows-node/Zcbot.WindowsNode/TrayApplicationContext.cs @@ -15,6 +15,10 @@ internal sealed class TrayApplicationContext : ApplicationContext { this.store = store; form = new ConfigurationForm(); + // Create the hidden form handle before the connection loop starts. Fast local + // connections can otherwise report Connecting/Online before a handle exists, + // and PostStatus would discard both updates until the user reconnects manually. + _ = form.Handle; form.RegisterRequested += RegisterAsync; form.ReconnectRequested += RestartConnection; form.ResetIdentityRequested += ResetIdentity; @@ -42,6 +46,7 @@ internal sealed class TrayApplicationContext : ApplicationContext try { config = store.Load(); + UpdateStatus(NodeStatus.Create(NodeState.Connecting, "正在连接 zcbot…")); StartConnection(); } catch (NodeConfigurationException exception) @@ -126,6 +131,7 @@ internal sealed class TrayApplicationContext : ApplicationContext ShowConfiguration(); return; } + UpdateStatus(NodeStatus.Create(NodeState.Connecting, "等待本机任务收尾后重连…")); connectionStop?.Cancel(); _ = RestartAfterStopAsync(); }