diff --git a/PROGRESS.md b/PROGRESS.md index b091d4f..eb2c7ab 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -22,7 +22,7 @@ ### 2026-08-14 -- **08-14 / Unreleased / Windows Node 输出上传收敛与诊断**:恢复中的成功任务优先向云端重放完成确认，已发布结果不再重新打开可能被 Origin 占用的 OPJU；文件共享冲突按 0.5/1/2/5 秒有界退避，本地完成标记独立重试，并按 Job 写入带阶段、产物、重试次数和 HRESULT 的 1 MiB 轮转诊断日志。同步修正本机更新时间早于接收时间的展示边界；相关 45 项专项 unittest、.NET build 与 diff 检查通过，未写入生产 DB。 +- **08-14 / Unreleased / Windows Node 输出上传收敛与诊断**:恢复中的成功任务优先向云端重放完成确认，已发布结果不再重新打开可能被 Origin 占用的 OPJU；文件共享冲突按 0.5/1/2/5 秒有界退避，本地完成标记使用可跨心跳复用的固定 pending 文件，只重试重命名以避开系统程序对每个新文件的重复扫描，并按 Job 写入带阶段、产物、重试次数和 HRESULT 的 1 MiB 轮转诊断日志。同步修正本机更新时间早于接收时间的展示边界；相关 45 项专项 unittest、.NET build 与 diff 检查通过，未写入生产 DB。 - **08-14 / Unreleased / Windows Node 本机任务监控 + Origin 多输入多输出 v2**:配置窗口新增只读本机任务列表与详情，每秒从 `%ProgramData%/Zcbot/WindowsNode/jobs` 合并 request/state/terminal/upload-complete，展示标题、全部输入、阶段、执行耗时、时间、Job ID 与错误，重启后可恢复且不查询云端未派发 Job。按用户明确授权直接切换到 `origin.plot@v2`：通用 `inputs[] + operation + outputs[]` 支持 1–16 个输入、跨输入系列和多个显式产物，Node 下载到 keyed 目录，Origin Worker 建立多工作表后合并绘图；本地阶段统一为 `downloading_inputs/software_running`，不兼容 v1 请求或旧本地任务。相关专项 71 项 unittest、Python/JavaScript 语法、Ruff 致命规则、.NET build 与 diff 检查通过；完整测试 593 项中非数据库测试通过，3 个数据库集成模块因显式测试库未迁移、缺少 `users` 表而未通过，未连接或写入生产 DB。 diff --git a/tests/test_windows_node_source.py b/tests/test_windows_node_source.py index 06f590b..607f232 100644 --- a/tests/test_windows_node_source.py +++ b/tests/test_windows_node_source.py @@ -276,6 +276,11 @@ class WindowsNodeSourceTests(unittest.TestCase): self.assertIn('"node-output-upload.log"', uploader) self.assertIn("MaxDiagnosticLogBytes", uploader) self.assertIn("outputUploader.RecordDeferred(current, exception)", connection) + self.assertIn('completionPath + ".pending"', uploader) + self.assertIn("PreparePendingCompletion(pendingPath, responseBody)", uploader) + self.assertIn("File.Move(pendingPath, completionPath, overwrite: false)", uploader) + self.assertIn("operation=move_pending", uploader) + self.assertNotIn('".tmp-" + Guid.NewGuid()', uploader) self.assertNotIn("Process.Start", uploader) def test_local_job_monitor_never_shows_update_before_acceptance(self) -> None: diff --git a/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs b/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs index c29a91e..72c2718 100644 --- a/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs +++ b/windows-node/Zcbot.WindowsNode/JobOutputUploader.cs @@ -180,26 +180,86 @@ internal sealed class JobOutputUploader(NodeConfig config) string completionPath, byte[] responseBody) { + var jobDirectory = Path.GetDirectoryName(completionPath)!; + if (Directory.Exists(completionPath)) + { + throw new InvalidDataException( + "本地完成标记路径被同名目录占用：upload-complete.json。"); + } + if (File.Exists(completionPath)) + { + return; + } + var pendingPath = completionPath + ".pending"; + PreparePendingCompletion(pendingPath, responseBody); for (var attempt = 0; ; attempt++) { + if (File.Exists(completionPath)) + { + return; + } try { - AtomicWrite(completionPath, responseBody); + File.Move(pendingPath, completionPath, overwrite: false); return; } catch (IOException exception) when ( IsSharingViolation(exception) && attempt < SharingViolationBackoff.Length) { var delay = SharingViolationBackoff[attempt]; - Log(Path.GetDirectoryName(completionPath)!, "WARN", + Log(jobDirectory, "WARN", $"Completion marker busy job={job.JobId:D} " + + "operation=move_pending " + $"attempt={attempt + 1} retry_ms={(int)delay.TotalMilliseconds} " + $"hresult=0x{exception.HResult:X8}", error: true); await Task.Delay(delay); } + catch (IOException exception) when (IsSharingViolation(exception)) + { + Log(jobDirectory, "WARN", + $"Completion marker busy job={job.JobId:D} " + + "operation=move_pending retries=exhausted " + + $"hresult=0x{exception.HResult:X8}", error: true); + throw new IOException( + "本地完成标记 upload-complete.json 在重命名阶段持续被占用" + + $"（HRESULT=0x{exception.HResult:X8}）。", + exception); + } } } + private static void PreparePendingCompletion(string pendingPath, byte[] responseBody) + { + if (File.Exists(pendingPath)) + { + try + { + using var pending = JsonDocument.Parse(File.ReadAllBytes(pendingPath)); + return; + } + catch (JsonException) + { + } + catch (IOException exception) when (IsSharingViolation(exception)) + { + // 上一轮已经完整落盘但暂时被系统程序扫描时，保留同一个 pending + // 文件；重试重命名不会再次触发“新文件关闭后立即扫描”。 + return; + } + } + using var response = JsonDocument.Parse(responseBody); + var content = JsonSerializer.SerializeToUtf8Bytes(new + { + completed_at = DateTimeOffset.UtcNow, + response = response.RootElement, + }, JsonOptions); + using var stream = new FileStream( + pendingPath, FileMode.Create, FileAccess.Write, FileShare.None, + 4096, FileOptions.WriteThrough); + stream.Write(content); + stream.Flush(flushToDisk: true); + } + private static bool IsSharingViolation(IOException exception) => (exception.HResult & 0xFFFF) is 32 or 33; @@ -243,27 +303,4 @@ internal sealed class JobOutputUploader(NodeConfig config) } } - private static void AtomicWrite(string path, byte[] responseBody) - { - using var response = JsonDocument.Parse(responseBody); - var content = JsonSerializer.SerializeToUtf8Bytes(new - { - completed_at = DateTimeOffset.UtcNow, - response = response.RootElement, - }, JsonOptions); - var temporary = path + ".tmp-" + Guid.NewGuid().ToString("N"); - try - { - using var stream = new FileStream( - temporary, FileMode.CreateNew, FileAccess.Write, FileShare.None, - 4096, FileOptions.WriteThrough); - stream.Write(content); - stream.Flush(flushToDisk: true); - File.Move(temporary, path, overwrite: false); - } - finally - { - if (File.Exists(temporary)) File.Delete(temporary); - } - } }