zcdsj
/
zcbot
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(node): 修复重连状态分裂与界面假未注册

This commit is contained in:
caoqianming 2026-08-19 14:53:36 +08:00
parent 67ee7c244e
commit 1aa79ae0b1
10 changed files with 177 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
RUN.md
View File

@ -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。

2
tests/test_software_nodes.py
View File

@ -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:

11
tests/test_windows_node_source.py
View File

@ -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)

11
web/routers/software_nodes.py
View File

@ -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}

2
windows-node/Zcbot.WindowsNode/ConfigurationForm.cs
View File

@ -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,
};

18
windows-node/Zcbot.WindowsNode/JobInboxStore.cs
View File

@ -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)

3
windows-node/Zcbot.WindowsNode/JobMonitorModels.cs
View File

@ -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");
}

67
windows-node/Zcbot.WindowsNode/JobOutputUploader.cs
View File

@ -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");

97
windows-node/Zcbot.WindowsNode/NodeConnectionLoop.cs
View File

@ -32,46 +32,59 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
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<NodeStatus>?
{
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<NodeStatus>?
{
if (!job.UploadComplete)
{
StartJobPipeline(socket, job);
if (!job.CloudTerminal) StartJobPipeline(socket, job);
}
}
else
@ -468,6 +482,11 @@ internal sealed class NodeConnectionLoop(NodeConfig config, Action<NodeStatus>?
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
? "结果已上传并由云端确认"

6
windows-node/Zcbot.WindowsNode/TrayApplicationContext.cs
View File

@ -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();
}