From 9e3d4be5cc1b056825e89f668ea538fffa09619f Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 15:17:09 +0800 Subject: [PATCH] feat: validate Worker protocol and verify process crash recovery --- README.md | 4 +- docs/05-production-acceptance.md | 4 +- docs/07-data-model.md | 2 +- docs/08-development-plan.md | 3 +- docs/README.md | 4 +- docs/api/README.md | 2 +- docs/api/worker-schema.md | 62 ++ docs/api/worker.md | 8 +- docs/reports/phase-2.md | 69 ++ docs/sdk/README.md | 2 +- docs/sdk/worker-runtime.md | 4 +- go.mod | 6 +- go.sum | 6 + .../worker_websocket_integration_test.go | 18 +- internal/testcontract/worker.go | 61 ++ internal/transport/httpserver/tasks.go | 12 + internal/transport/httpserver/worker.go | 98 +- .../httpserver/worker_contract_test.go | 213 +++++ protocol/workerv1/schema.go | 9 + protocol/workerv1/worker.schema.json | 899 ++++++++++++++++++ scripts/test-mysql.sh | 6 + sdk/go/worker_integration_test.go | 20 +- sdk/go/worker_process_integration_test.go | 394 ++++++++ 23 files changed, 1878 insertions(+), 28 deletions(-) create mode 100644 docs/api/worker-schema.md create mode 100644 docs/reports/phase-2.md create mode 100644 internal/testcontract/worker.go create mode 100644 internal/transport/httpserver/worker_contract_test.go create mode 100644 protocol/workerv1/schema.go create mode 100644 protocol/workerv1/worker.schema.json create mode 100644 sdk/go/worker_process_integration_test.go diff --git a/README.md b/README.md index 8a99fec..38fd4ec 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,9 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已实现 32 个数据模型、版本化迁移、凭据及范围校验、队列创建查询、任务幂等提交/查询/分页/待执行取消，以及对应 Go SDK 和示例。P2 服务端、WSS、底层 SDK 与自动 Worker 运行器已通过阶段内回归，支持自动心跳/续租/重连、有界排空及 [执行尝试/证据查询](docs/api/execution-history.md)；协议验收、调度器和流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已完成 P0–P2 阶段验收：32 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；安全重试、调度器和流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 -readiness 只表示数据库与迁移记录可用，不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 服务端记录](docs/reports/phase-2-server.md)、[P2 SDK 协议记录](docs/reports/phase-2-sdk-protocol.md)、[自动运行器记录](docs/reports/phase-2-worker-runtime.md) 与 [执行历史记录](docs/reports/phase-2-execution-history.md)。 +readiness 只表示数据库与迁移记录可用，不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md) 与 [P2 阶段验收](docs/reports/phase-2.md)，后者汇总服务端、SDK、执行历史及故障验证的详细记录。 ## 目录 diff --git a/docs/05-production-acceptance.md b/docs/05-production-acceptance.md index 938d9c3..0ba444f 100644 --- a/docs/05-production-acceptance.md +++ b/docs/05-production-acceptance.md @@ -1,6 +1,6 @@ # Orbit 生产验收与发布门槛 -> 版本 0.2 · 2026-09-17。本文是验收计划，所有项目当前均为待实现、待验证。没有已通过的测试或容量结论。 +> 版本 0.2 · 2026-09-17。本文是全平台生产发布门槛。P0–P2 已积累阶段证据，见 [开发计划](08-development-plan.md) 与 [P2 验收](reports/phase-2.md)；未覆盖的场景继续实施，尚无全平台生产验收或容量结论。 ## 1. 验收对象与证据 @@ -89,7 +89,7 @@ W11–W17 中的字段与暂停/计时算法按技术评审后的规范执行； ## 5. 发布前必须确定的参数 -以下数值当前均为待定：容量目标、延迟目标、可用性目标、RPO / RTO、消息和批量上限、租约/续租/心跳/重连参数、扫描与补跑预算、查询轮次/间隔/总时限与单轮重试预算、暂停继续的推进预算、保留周期、兼容版本范围。 +生产容量、延迟、可用性目标与 RPO / RTO 仍待实际拓扑验证。当前已实现的消息上限、租约/续租/心跳/重连与恢复扫描参数见 [Worker 手册](api/worker.md)，这些正确性基线不等于生产调优结论。批量/补跑预算、流程查询与暂停预算、保留及兼容范围随对应阶段落实并验证。 配置之间需要校验，例如续租间隔与安全余量必须允许在租约到期前发现失败；代理空闲超时需与 WS 活性策略协调；最长恢复补发窗口应与幂等和事件保留周期匹配。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 9cf0811..7276939 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -88,7 +88,7 @@ 外键与 `CHECK` 无法代替执行引擎。P1 已完成 HTTP 动作/队列授权、幂等摘要与原回执、受理事件，以及待执行取消的 CAS/审计事务。执行相关的以下条件继续分阶段实现： -P2 服务端已新增会话隔离、两步派发、租约、并发许可、结果回执/证据和恢复扫描，见 [执行协议](api/worker.md) 与 [阶段记录](reports/phase-2-server.md)。底层及自动 Worker SDK、尝试/证据查询已实现，完整 P2 验收仍在推进。 +P2 服务端已新增会话隔离、两步派发、租约、并发许可、结果回执/证据和恢复扫描，见 [执行协议](api/worker.md)。底层及自动 Worker SDK、尝试/证据查询、Schema 与进程故障恢复已通过 [P2 阶段验收](reports/phase-2.md)。 1. 权限验证、凭据校验、命名空间过滤和队列消费授权。 2. 同一次事务更新任务、尝试、租约、并发许可、事件与流程交付记录。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 2b30021..d0e8959 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -26,7 +26,7 @@ | --- | --- | --- | --- | | P0 基础与数据模型 | Go workspace、独立 SDK module、公共/Pulse/Weave 模型、MySQL 迁移、配置、健康检查 | 编译/竞态/静态检查；真实 MySQL 建表、隔离、唯一性、预算、不可变版本、迁移失败防护；SDK 健康调用 | 已完成并通过阶段回归（见 P0 报告） | | P1 任务受理闭环 | 应用凭据与作用域权限、队列、幂等提交、查询、待执行取消、受理事件；SDK 客户端 | API/SDK 创建查询闭环；重复键与内容冲突、响应未确认重报、跨命名空间拒绝、参数边界 | 已完成并通过阶段回归（见 P1 报告；真实断路/提交 ACK 丢失注入留后续故障阶段） | -| P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复；Worker SDK | 真 Worker 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 开发中：服务端、SDK 协议、自动运行器及尝试/证据 HTTP+SDK 查询通过阶段内回归；自动续租/重连、有界退出、权限/稳定分页和大载荷链路验证通过；完整 Schema 与进程级故障阶段验收待完成 | +| P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复；Worker SDK | 真 Worker 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 已完成并通过阶段回归：服务端、自动 SDK、历史查询、完整 Schema 与真实进程强杀恢复；测试拓扑和剩余生产故障矩阵见 P2 验收报告 | | P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 待开发 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | @@ -74,6 +74,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [生产验收总清单](05-production-acceptance.md) - [P0 验证记录](reports/phase-0.md) - [P1 验证记录](reports/phase-1.md) +- [P2 阶段验收与进程故障记录](reports/phase-2.md) - [P2 服务端阶段记录](reports/phase-2-server.md) - [P2 SDK 协议阶段记录](reports/phase-2-sdk-protocol.md) - [P2 自动 Worker 运行器记录](reports/phase-2-worker-runtime.md) diff --git a/docs/README.md b/docs/README.md index 99bf2cb..9666457 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本：0.2 - 整理日期：2026-09-17 -- 阶段：P0/P1 已完成，P2 服务端、底层 SDK、自动 Worker 运行器及执行历史查询已通过阶段内回归，协议与进程故障验收继续推进；尚未完成生产验收 +- 阶段：P0–P2 已通过阶段验收，包括协议 Schema 与真实进程故障恢复；P3–P7 继续推进，尚未完成全平台生产验收 - 适用范围：Orbit 统一平台，包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 @@ -79,4 +79,4 @@ Orbit 不内置特定行业规则；Pulse 管理执行，Weave 管理步骤依 2026-09-17，文档更新至 0.2：用户确认研发定义发布流程及 Weave 首版功能，明确纳入有界重复查询和暂停/继续等生产管理能力。技术细则与参数继续按建议或待定标注，未将功能确认等同于实现或验收通过。 -持续开发目标已建立。当前实现和验收进度以 [开发计划](08-development-plan.md)、[P0 报告](reports/phase-0.md)、[P1 报告](reports/phase-1.md)、[P2 服务端记录](reports/phase-2-server.md)、[P2 SDK 协议记录](reports/phase-2-sdk-protocol.md) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 +持续开发目标已建立。当前实现和验收进度以 [开发计划](08-development-plan.md)、[P0 报告](reports/phase-0.md)、[P1 报告](reports/phase-1.md) 与 [P2 阶段验收](reports/phase-2.md) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 diff --git a/docs/api/README.md b/docs/api/README.md index fbc4566..b2aecfb 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -13,7 +13,7 @@ | `/v1/tasks/{task_id}/attempts/{attempt_id}/evidence` 与证据详情 | P2 已实现 | [执行历史](execution-history.md) | | `/v1/queues` 创建与查询 | P1 已实现 | [队列接口](queues.md) | | 凭据校验与初始化 CLI | P1 已实现 | [鉴权](authentication.md) | -| `/v1/worker/connect` | P2 服务端、底层 SDK 与自动 Worker 已实现；完整阶段验收继续推进 | [Worker 契约](worker.md) | +| `/v1/worker/connect` | P2 已实现并通过阶段验收 | [Worker 契约](worker.md)、[Schema](worker-schema.md) | | `/v1/schedules` 与计划管理 | 计划 P3，尚未实现 | [协议草案](../03-http-websocket-protocol.md) | | `/v1/events/connect` | 计划 P3，尚未实现 | [协议草案](../03-http-websocket-protocol.md) | | `/v1/workflows` 与流程管理 | 计划 P4–P5，尚未实现 | [流程接口规划](workflows.md) | diff --git a/docs/api/worker-schema.md b/docs/api/worker-schema.md new file mode 100644 index 0000000..403e544 --- /dev/null +++ b/docs/api/worker-schema.md @@ -0,0 +1,62 @@ +# Worker v1 机器可读契约 + +子协议为 `pulse.worker.v1`，消息中 `protocol_version` 为 `"1"`。完整文件：[worker.schema.json](../../protocol/workerv1/worker.schema.json)，使用 JSON Schema Draft 2020-12。业务含义、超时及恢复方式见 [Worker 接口手册](worker.md)。 + +文件中的 $id 是契约标识，不表示该 URL 已部署下载服务。仓库文件是当前发布载体；回归使用本地嵌入文件，不访问远程 Schema。 + +## 校验入口 + +Schema 根接受双向消息，调用方应根据方向选择更严格的定义： + +| JSON Pointer | 用途 | +| --- | --- | +| `#/$defs/clientFrame` | Worker → 服务端的所有请求 | +| `#/$defs/serverFrame` | 服务端 → Worker 的所有报文 | +| `#/$defs/helloFrame` | 注册请求，不允许 session_id、reply_to 或 error | +| `#/$defs/capacityFrame` | 容量及排空声明，sequence/capacity/drain 均必填 | +| `#/$defs/resumeFrame` | 尝试归属恢复请求 | +| `#/$defs/startFrame`、`renewFrame`、`finishFrame` | 开始、续租、结果报告 | +| `#/$defs/welcomeFrame`、`assignmentFrame` | 注册确认与主动派发 | +| `#/$defs/ackFrame`、`errorFrame` | 操作确认与公开错误 | + +除首条 hello 外，请求必须提供当前 session_id。请求不得包含响应专用字段，包括 `reply_to:""` 或 `error:null`。主动派发没有 reply_to；操作响应通过 reply_to 关联原消息。error 在注册失败时不带 session_id；会话内错误带 session_id，主动致命错误不带 reply_to。 + +ACK 的 body 根据原请求选择：worker.capacity → welcome，worker.resume → lease，task.start/renew/finish → commandResult。Schema 可以检查三个 body 形状，消息关联与正确分支由连接状态检查；不能仅凭一个独立 ACK 判断它是否对应调用。 + +## 字段和兼容规则 + +- 已声明对象使用 additionalProperties=false，业务 payload/result 内部字段由接入方定义。 +- 结构体必填字段必须显式提供，包括 capacity=0、drain=false；缺少字段不能被解释为调用者明确选择零值。 +- epoch、sequence、revision 为规范十进制字符串：范围 1–18446744073709551615，无前导零和正号。SDK 使用 uint64，线上使用字符串。 +- instance_secret/lease_secret 必须是 32 字节秘密的严格无填充 base64url，长度 43；知道资源 ID 无法替代秘密。 +- 结果为非 null JSON；FAILED/UNKNOWN 必须携带稳定 error_code；SUCCEEDED 省略或使用空 error_code。 +- 返回时间使用 UTC RFC 3339，最多微秒精度。验证器应开启 format 断言以校验实际日历日期。 + +本轮收紧了开发期此前允许的省略控制字段、非规范计数字符串和数组内字段别名。当前 SDK 一直发送完整规范字段，不需迁移；其他自研客户端应按本文件更新。SDK 尚未发布首个版本。将来扩展消息类型或改变字段语义必须同时更新 Schema、服务端、SDK 与兼容回归。 + +## Schema 与运行时的边界 + +Schema 检查已解析的 JSON 结构。以下约束必须在原始字节或持久化事务层另外校验： + +| 约束 | 实际执行位置 | +| --- | --- | +| 文本帧、512 KiB 总帧限制 | WS 读写与解码入口 | +| 重复键、无效 UTF-8、孤立代理项、32 层嵌套 | 严格 JSON 扫描器，不能先交给会替换非法字符的解析器 | +| 结果 256 KiB、精确数字文本/指数范围 | 结果校验器；JSON Schema 的字符数不等于编码字节数 | +| JSON 整数字段的文本形式 | Go 类型解码，例如 1.0 不作为整数控制字段接受 | +| 当前会话、reply_to 与原操作匹配 | WS 会话及 SDK 控制请求表 | +| 身份、权限、队列、租约、版本及时间先后关系 | 数据库事务与执行状态机 | +| 原始回执与当前执行权分离 | CommandResult.receipt 与 lease；历史回执不能复活执行 | + +通过 Schema 只说明报文形状合法，不能证明请求已获授权、操作已经提交或业务副作用只发生一次。 + +## 可复现的验证 + +```sh +go test -race ./internal/transport/httpserver -run TestWorkerPublishedContract +make integration +``` + +契约测试对合法消息及变异消息同时运行 Schema 和实际应用输入校验，包含数组内字段别名、缺少字段、错误方向、容量边界、计数溢出、非规范秘密及结果分类。真实 MySQL/WSS 测试在发送和接收时校验实际报文，并覆盖注册、容量、派发、开始、续租、恢复、结果及错误。 + +测试使用固定版本 [santhosh-tekuri/jsonschema v6.0.2](https://pkg.go.dev/github.com/santhosh-tekuri/jsonschema/v6@v6.0.2)，显式开启 AssertFormat。这个验证器只用于测试工具包，不进入 Orbit 服务二进制或独立 SDK 的依赖图；运行平台仍只要求 MySQL。 diff --git a/docs/api/worker.md b/docs/api/worker.md index ce55c65..1fdcbbb 100644 --- a/docs/api/worker.md +++ b/docs/api/worker.md @@ -1,6 +1,6 @@ -# Worker 通信与执行契约 v1（开发中） +# Worker 通信与执行契约 v1 -已实现服务端执行事务、WebSocket 接入、[独立 Go 底层协议 SDK](../sdk/worker.md)、[自动 Worker 运行器](../sdk/worker-runtime.md) 与 [尝试/证据查询](execution-history.md)，通过真实 WSS、自动续租、ACK 丢失/持续断网恢复、有界退出及历史权限/分页回归。**P2 尚未完成**：完整协议 Schema 与完整阶段验收继续推进。底层 WorkerConnection 不自动管理处理器；NewWorker / Run 提供自动管理。 +已实现服务端执行事务、WebSocket 接入、[独立 Go 底层协议 SDK](../sdk/worker.md)、[自动 Worker 运行器](../sdk/worker-runtime.md) 与 [尝试/证据查询](execution-history.md)，通过 [P2 阶段验收](../reports/phase-2.md)，包含真实 WSS、协议 Schema、自动续租、ACK 丢失/断网恢复及进程强杀后的节点接管。底层 WorkerConnection 不自动管理处理器；NewWorker / Run 提供自动管理。 ## 1. 接入与身份 @@ -22,6 +22,8 @@ go run ./cmd/orbit-admin bootstrap \ 只接受 JSON 文本帧；单帧最多 512 KiB，业务结果最多 256 KiB。重复键、未知字段、无效 Unicode 和非法数值按 HTTP 相同规则拒绝。 +机器可读契约及各方向定义见 [JSON Schema 说明](worker-schema.md)。控制字段必须显式提供；epoch/sequence 等计数使用无前导零的规范十进制字符串。请求禁止响应字段，空字符串/null 也不例外。 + ```json { "protocol_version":"1", @@ -143,4 +145,4 @@ receipt 是原操作的不可变确认；lease 是本次重新核验的执行权 这些是有界的开发基线，不代表经过容量压测的生产参数。当前同队列操作使用较粗的行锁协调，吞吐与公平性需要 P6 按负载验证。服务退出会显式关闭被 HTTP Upgrade 接管的连接；断开不会直接撤销外部已发出的请求。 -运行中取消、完整 WS JSON Schema、凭据轮换和多节点进程级故障演练尚未全部完成。SDK 已实现自动重连/续租、非协作处理器的有界退出与状态报告；实际证据见 [P2 服务端阶段记录](../reports/phase-2-server.md) 和 [自动运行器回归](../reports/phase-2-worker-runtime.md)。 +协议 Schema 和真实进程强杀恢复已通过回归，见 [P2 阶段验收](../reports/phase-2.md)。运行中取消属于 P3；凭据轮换、数据库故障切换、跨主机演练与容量验收属于后续生产验证。自动运行器对非协作处理器提供有界退出与状态报告，不能强制停止其外部副作用。 diff --git a/docs/reports/phase-2.md b/docs/reports/phase-2.md new file mode 100644 index 0000000..e91fdf4 --- /dev/null +++ b/docs/reports/phase-2.md @@ -0,0 +1,69 @@ +# P2 Worker 执行闭环阶段验收 + +日期：2026-09-17。P2 已实现并通过阶段回归，此报告汇总本地正确性/故障证据。全平台首版仍需完成 P3–P7；本地多进程接管不能替代跨主机、数据库高可用和容量验收。 + +## 阶段交付范围 + +服务端已经提供持久化 Worker 会话、能力及绝对容量、两步领取/开始、续租、结果回执、排空、租约回收，以及尝试/证据查询。独立 Go SDK 提供底层连接控制和自动 Worker 运行器。机器契约、Markdown 接口、运行示例与测试一同交付。 + +文档入口：[Worker 契约](../api/worker.md)、[JSON Schema 说明](../api/worker-schema.md)、[自动 Worker](../sdk/worker-runtime.md)、[执行历史](../api/execution-history.md)。 + +## 环境及复现 + +macOS arm64、Go 1.27.1、Docker 中的单实例 MySQL 8.4.10。服务和子进程使用回环随机端口；WSS 网关使用受信任测试证书，不跳过 TLS 校验。所有故障只影响脚本创建的临时数据库、专用凭据及子进程。 + +```sh +make test +make vet +make build +make integration +cd sdk/go +GOWORK=off go test -race ./... +GOWORK=off go vet ./... +``` + +脚本先完成数据库与协议集成测试、真实服务及五个 SDK 示例，再停止原服务，进入独立进程故障测试，避免原节点的恢复扫描器掩盖新节点接管。凭据通过环境及 0600 文件传递，不进入 URL 或输出。故障通过 Process.Kill 注入，必须由 Wait 确认目标被信号终止；结束后回收所有测试子进程及容器。 + +Schema 验证器仅被测试包引用。服务二进制依赖图中不包含 jsonschema 验证器，SDK go.mod 没有新增依赖。当前协议输入由严格解码器、类型及事务校验器处理。 + +## 本轮新增证据 + +| 用例 | 注入与断言 | +| --- | --- | +| TestWorkerPublishedContract | 所有客户端类型与服务端响应形状通过 Schema；变异报文同时经 Schema/真实应用输入校验拒绝；重复键、非法 Unicode、超范围数字在原始字节层拒绝 | +| TestIntegrationWorkerWebSocket | 实际 WSS 报文逐条校验 Schema，包括注册、容量、派发、开始、续租、恢复、完成和公开错误 | +| TestIntegrationWorkerProcessACKRecovery | 三个不同 node ID 的真实 Orbit 进程：A 提交 start 后网关丢 ACK 并暂停重连，强杀 A；切到 B 后恢复同一尝试，再执行处理器；B 提交 finish 后丢 ACK 并强杀 B，启动 C 核对原回执 | +| ACK 恢复的持久化核对 | 初始开始时间及尝试 ID 不变，会话归属发生迁移；处理器仅一次，任务仅一次尝试；C 返回同一成功结果及版本，没有额外冲突证据；SDK 正常排空 | +| TestIntegrationWorkerProcessCrashUnknown | 真实 Worker 子进程写入并同步外部工作标记后被强杀，同时强杀原 Orbit；只有新 Orbit 执行恢复扫描，任务进入 NEEDS_REVIEW、尝试 UNKNOWN，未伪造结果；已用 1 次、上限 3 次，不因恢复重置 | +| 强杀后的许可核对 | 使用 max_concurrency=1 的专用队列；新 Worker 消费同能力的探针任务成功，原未知任务没有再次执行，证明许可可继续使用而原任务仍等待核查 | + +初次运行发现强杀用例的专用队列缺少必填速率值，被真实 API 正确拒绝。补齐测试配置后重新执行完整回归。没有通过绕开参数校验让用例成功。 + +## P2 要求与证据对应 + +| 要求 | 主要测试与记录 | +| --- | --- | +| WSS、协议与身份权限 | TestIntegrationWorkerWebSocket 的 handshake 子用例；TestIntegrationWorkerExecution 的 authority 子用例；本轮 Schema 契约回归 | +| 能力/队列/进程容量与排空 | multi_node_claim_obeys_queue_and_process_capacity、drain_is_durable_and_capacity_sequence_is_not_additive；自动 Worker 并发与有界收尾测试 | +| 开始前校验、旧会话隔离 | paused_gate_blocks_start_but_not_saved_result、cross_node_resume_fences_old_session_and_requires_secrets、lock_wait_does_not_reuse_an_earlier_clock_sample | +| 租约、总/单次时限与回收 | attempt_timeout_and_session_expiry_cap_renewal、recovery_is_bounded_and_unknown_effects_are_not_replayed；自动运行器跨首次租约 32 秒执行及到期取消测试 | +| 结果原身份核对、证据与事务 | durable_start_renew_and_exact_result_confirmation、result_event_failure_rolls_back_task_receipt_and_permit、finish_without_start_is_only_evidence_and_cannot_free_capacity；SDK ACK 丢失与本轮进程强杀 | +| 自动运行器恢复及执行次数 | TestIntegrationWorkerRuntimeACKLossRenewAndDrain、OutageCancelsAndReplaysEvidence、Concurrency；本轮 ProcessACKRecovery | +| 未知效果与不协作处理器 | TestIntegrationWorkerRuntimeUnknownAndUncooperativeShutdown、LeaseLossCancelsAndDoesNotRevive；本轮 ProcessCrashUnknown | +| 历史、权限、稳定分页和载荷 | TestIntegrationTaskHistory、TestIntegrationWorkerConnectionHistoryAndPayloadBounds；[执行历史回归](phase-2-execution-history.md) | +| 独立 SDK 与实际接入 | GOWORK=off 回归；health、tasks、worker-protocol、worker、history 五个例子在真实服务上运行 | + +更细的历史证据分别在 [服务端记录](phase-2-server.md)、[SDK 协议](phase-2-sdk-protocol.md)、[自动运行器](phase-2-worker-runtime.md) 和 [执行历史](phase-2-execution-history.md)。旧记录中的“下一步”描述当时阶段，本报告及持续计划反映当前进度。 + +## 契约修正 + +数组内对象现在同样拒绝字段大小写别名，例如 capabilities 中的 Handler 不再被当作 handler。Worker 控制请求必须显式提供必填字段，计数字符串必须规范且不溢出，请求不能携带响应字段的空值/null。现有 SDK 一直发送完整规范数据，所有接入示例继续通过。首版 SDK 尚未发布，对自研开发期客户端的兼容影响见 [Schema 说明](../api/worker-schema.md)。 + +## 遗留范围与后续阶段 + +- P3：策略性安全重试、运行中取消、计划触发、漏跑策略、配额/背压及可靠结果订阅。 +- P4–P5：完整 Weave 定义发布、流程推进、等待/查询、管理与恢复。 +- P6：MySQL 不可用/切主、跨主机故障、滚动升级、备份恢复、负载与容量报告、凭据轮换及运维。 +- 更广的故障矩阵继续包含真实 HTTP 受理 ACK 丢失、派发消息未送达时进程崩溃、慢消费者/重连风暴及清理竞争；当前不把相邻测试推断为这些场景已经验收。 + +进程故障测试中的外部文件仅表示已发生但未上报的工作，不模拟真实业务渠道的全部事务。SDK 未确认结果仍在进程内存；Worker 重启不能自动找回它。平台按租约和安全等级保留未知状态，后续由核查及业务幂等协议处理。没有端到端 exactly-once、生产 RPO/RTO 或峰值容量承诺。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 366021d..3c61d2b 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -176,4 +176,4 @@ task=<任务ID> state=SUCCEEDED attempts=1 evidence=1 reason=RESULT_CONFLICT dis ## 7. 后续交付 -P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) 与历史查询，继续补齐协议 Schema 及阶段验收；P3 提供计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 +P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) 与历史查询，并通过 [Schema 及进程故障阶段验收](../reports/phase-2.md)；P3 提供安全重试、计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 diff --git a/docs/sdk/worker-runtime.md b/docs/sdk/worker-runtime.md index cf06b31..a82709c 100644 --- a/docs/sdk/worker-runtime.md +++ b/docs/sdk/worker-runtime.md @@ -1,6 +1,6 @@ # 自动 Worker 接入 -更新：2026-09-17。自动 Worker 已实现，并通过真实 MySQL、实际服务命令、WSS 断线/ACK 丢失/持续断网回归。[尝试与证据查询](../api/execution-history.md) 已可用于核对结果；P2 的完整协议 Schema 和进程级故障验收仍需继续完成。 +更新：2026-09-17。自动 Worker 已实现，并通过真实 MySQL、实际服务命令、WSS 断线/ACK 丢失/持续断网及 [进程强杀恢复回归](../reports/phase-2.md)。[尝试与证据查询](../api/execution-history.md) 可核对结果，协议形状见 [Schema](../api/worker-schema.md)。 源码：[运行器](../../sdk/go/worker_runtime.go)、[配置与中文注释](../../sdk/go/worker_runtime_types.go)、[完整示例](../../sdk/go/examples/worker/main.go)。特殊传输集成可以使用 [底层协议 SDK](worker.md)，普通消费直接使用本页的 Worker。 @@ -106,4 +106,4 @@ Go 无法杀死忽略 context 的 goroutine。Run 可以有界返回，但这类 身份、尚未确认的结果和消息 ID 当前保存在内存，支持同进程重连；进程被强杀会失去这些内容。持久化租约、操作回执、最终结果和审计在 MySQL 中，业务外部效果仍需要业务幂等或查询核对。这一边界不等于端到端 exactly-once。 -本轮证据见 [自动运行器回归报告](../reports/phase-2-worker-runtime.md)。实际部署中的多主机故障、数据库复制确认与容量承诺仍需后续生产验收。 +运行器细节见 [自动运行器回归报告](../reports/phase-2-worker-runtime.md)，后续真实进程故障证据见 [P2 阶段验收](../reports/phase-2.md)。实际部署中的多主机故障、数据库复制确认与容量承诺仍需后续生产验收。 diff --git a/go.mod b/go.mod index 775b9f1..401a109 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,10 @@ go 1.27.1 require ( github.com/coder/websocket v1.8.15 github.com/go-sql-driver/mysql v1.10.1 + github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 ) -require filippo.io/edwards25519 v1.2.0 // indirect +require ( + filippo.io/edwards25519 v1.2.0 // indirect + golang.org/x/text v0.14.0 // indirect +) diff --git a/go.sum b/go.sum index f808e3b..f59ce3a 100644 --- a/go.sum +++ b/go.sum @@ -2,5 +2,11 @@ filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo= filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc= github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI= +github.com/dlclark/regexp2 v1.11.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/go-sql-driver/mysql v1.10.1 h1:arlSnNLq6a5yxGxV7qg9lF4j0C+KwD6NbQyKr9QL6ME= github.com/go-sql-driver/mysql v1.10.1/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk= +github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 h1:KRzFb2m7YtdldCEkzs6KqmJw4nqEVZGK7IN2kJkjTuQ= +github.com/santhosh-tekuri/jsonschema/v6 v6.0.2/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= diff --git a/internal/storage/mysql/worker_websocket_integration_test.go b/internal/storage/mysql/worker_websocket_integration_test.go index 5a7d86e..7f7bc9e 100644 --- a/internal/storage/mysql/worker_websocket_integration_test.go +++ b/internal/storage/mysql/worker_websocket_integration_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "git.yuntuo.hk/k2safe/orbit/internal/testcontract" "git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver" core "git.yuntuo.hk/k2safe/orbit/model" pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" @@ -75,6 +76,7 @@ func (c *testWorkerSocket) send(kind string, body any, id core.ID) core.ID { if err != nil { c.t.Fatal(err) } + testcontract.WorkerFrame(c.t, "client", frame) ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second) defer cancel() if err := c.conn.Write(ctx, websocket.MessageText, frame); err != nil { @@ -90,6 +92,7 @@ func (c *testWorkerSocket) read() testWorkerFrame { if err != nil { c.t.Fatal(err) } + testcontract.WorkerFrame(c.t, "server", raw) var frame testWorkerFrame if kind != websocket.MessageText || json.Unmarshal(raw, &frame) != nil || frame.Version != "1" || frame.MessageID.Validate() != nil { c.t.Fatal("invalid server frame") @@ -191,6 +194,10 @@ func TestIntegrationWorkerWebSocket(t *testing.T) { if err := json.Unmarshal(first.ack(startMessage), &begun); err != nil || !begun.Lease.Active || begun.Lease.State != pulse.AttemptRunning { t.Fatal("missing durable start") } + var renewed worker.CommandResult + if err := json.Unmarshal(first.ack(first.send("task.renew", ref, "")), &renewed); err != nil || !renewed.Lease.Active { + t.Fatal("missing renewal confirmation") + } // 模拟旧节点连接断开；新的 WSS 连接进入不同节点并原子迁移有效尝试。 _ = first.conn.CloseNow() second := connectTestWorker(t, ctx, serverB, fixture) @@ -220,13 +227,22 @@ func TestIntegrationWorkerWebSocket(t *testing.T) { }) t.Run("strict_worker_frame_and_hub_shutdown", func(t *testing.T) { conn := connectTestWorker(t, ctx, serverB, fixture) + secret, _, err := worker.NewSecret() + if err != nil { + t.Fatal(err) + } + message := conn.send("worker.resume", worker.AttemptRef{TaskID: executionID(t), AttemptID: executionID(t), Epoch: 1, LeaseSecret: secret}, "") + failure := conn.read() + if failure.Type != "error" || failure.ReplyTo != message || failure.Error == nil || failure.Error.Code != "NOT_FOUND" { + t.Fatal("missing structured error frame") + } raw := []byte(`{"protocol_version":"1","protocol_version":"1","type":"worker.capacity"}`) if err := conn.conn.Write(ctx, websocket.MessageText, raw); err != nil { t.Fatal(err) } readCtx, stop := context.WithTimeout(ctx, 3*time.Second) defer stop() - _, _, err := conn.conn.Read(readCtx) + _, _, err = conn.conn.Read(readCtx) if websocket.CloseStatus(err) != websocket.StatusPolicyViolation { t.Fatalf("duplicate frame not rejected: %v", err) } diff --git a/internal/testcontract/worker.go b/internal/testcontract/worker.go new file mode 100644 index 0000000..edef8e3 --- /dev/null +++ b/internal/testcontract/worker.go @@ -0,0 +1,61 @@ +// Package testcontract 在回归中校验实际报文与公开 Schema，服务二进制不依赖此包。 +package testcontract + +import ( + "strings" + "sync" + "testing" + + "git.yuntuo.hk/k2safe/orbit/protocol/workerv1" + "github.com/santhosh-tekuri/jsonschema/v6" +) + +var workerOnce sync.Once +var workerSchemas map[string]*jsonschema.Schema +var workerError error + +// WorkerSchema 编译并缓存两个方向的 Schema；开启 date-time 等格式断言。 +func WorkerSchema(t testing.TB, direction string) *jsonschema.Schema { + t.Helper() + workerOnce.Do(func() { + compiler := jsonschema.NewCompiler() + compiler.AssertFormat() + doc, err := jsonschema.UnmarshalJSON(strings.NewReader(workerv1.Schema)) + if err != nil { + workerError = err + return + } + const resource = "https://git.yuntuo.hk/k2safe/orbit/schema/worker.v1.json" + if err := compiler.AddResource(resource, doc); err != nil { + workerError = err + return + } + workerSchemas = make(map[string]*jsonschema.Schema) + for _, side := range []string{"client", "server"} { + workerSchemas[side], workerError = compiler.Compile(resource + "#/$defs/" + side + "Frame") + if workerError != nil { + return + } + } + }) + if workerError != nil { + t.Fatal(workerError) + } + schema := workerSchemas[direction] + if schema == nil { + t.Fatal("unknown Worker protocol direction") + } + return schema +} + +// WorkerFrame 验证真实传输的 JSON，不在错误中输出可能包含秘密的报文正文。 +func WorkerFrame(t testing.TB, direction string, raw []byte) { + t.Helper() + value, err := jsonschema.UnmarshalJSON(strings.NewReader(string(raw))) + if err != nil { + t.Fatal("Worker frame is not valid JSON") + } + if WorkerSchema(t, direction).Validate(value) != nil { + t.Fatalf("%s Worker frame violates published Schema", direction) + } +} diff --git a/internal/transport/httpserver/tasks.go b/internal/transport/httpserver/tasks.go index 8e061f7..53b5003 100644 --- a/internal/transport/httpserver/tasks.go +++ b/internal/transport/httpserver/tasks.go @@ -214,6 +214,18 @@ func exactFields(raw []byte, t reflect.Type) error { if t == reflect.TypeOf(time.Time{}) || t == reflect.TypeOf(json.RawMessage{}) { return nil } + if t.Kind() == reflect.Slice || t.Kind() == reflect.Array { + var values []json.RawMessage + if err := json.Unmarshal(raw, &values); err != nil { + return err + } + for _, value := range values { + if err := exactFields(value, t.Elem()); err != nil { + return err + } + } + return nil + } if t.Kind() != reflect.Struct { return nil } diff --git a/internal/transport/httpserver/worker.go b/internal/transport/httpserver/worker.go index 0c60795..9f9c502 100644 --- a/internal/transport/httpserver/worker.go +++ b/internal/transport/httpserver/worker.go @@ -7,6 +7,7 @@ import ( "errors" "net/http" "reflect" + "strconv" "strings" "sync" "time" @@ -142,9 +143,9 @@ func (h *WorkerHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { if err != nil { return } - var hello workerFrame + hello, decodeErr := decodeWorkerRequest(raw, true) var input worker.ConnectInput - if kind != websocket.MessageText || decodeWorkerJSON(raw, &hello) != nil || hello.Version != "1" || hello.Type != "worker.hello" || hello.MessageID.Validate() != nil || hello.SessionID != "" || hello.ReplyTo != "" || hello.Error != nil || decodeWorkerJSON(hello.Body, &input) != nil { + if kind != websocket.MessageText || decodeErr != nil || decodeWorkerJSON(hello.Body, &input) != nil { _ = conn.Close(websocket.StatusPolicyViolation, "invalid worker hello") return } @@ -177,8 +178,8 @@ func (h *WorkerHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { if err != nil { return } - var frame workerFrame - if kind != websocket.MessageText || decodeWorkerJSON(raw, &frame) != nil || frame.Version != "1" || frame.MessageID.Validate() != nil || frame.SessionID != binding.SessionID || frame.ReplyTo != "" || frame.Error != nil { + frame, decodeErr := decodeWorkerRequest(raw, false) + if kind != websocket.MessageText || decodeErr != nil || frame.SessionID != binding.SessionID { _ = conn.Close(websocket.StatusPolicyViolation, "invalid worker frame") return } @@ -393,6 +394,9 @@ func decodeWorkerJSON(raw []byte, out any) error { if err := exactFields(raw, reflect.TypeOf(out).Elem()); err != nil { return fault.Invalid("unknown or invalid worker field") } + if err := requiredWorkerFields(raw, reflect.TypeOf(out).Elem()); err != nil { + return err + } decoder := json.NewDecoder(bytes.NewReader(raw)) decoder.DisallowUnknownFields() if err := decoder.Decode(out); err != nil { @@ -400,3 +404,89 @@ func decodeWorkerJSON(raw []byte, out any) error { } return nil } + +// decodeWorkerRequest 拒绝请求中伪装为响应的字段；空字符串/null 不能绕过方向约束。 +func decodeWorkerRequest(raw []byte, hello bool) (workerFrame, error) { + var frame workerFrame + if err := decodeWorkerJSON(raw, &frame); err != nil { + return frame, err + } + if frame.Version != "1" || frame.MessageID.Validate() != nil || len(frame.Body) == 0 { + return frame, fault.Invalid("invalid worker envelope") + } + var fields map[string]json.RawMessage + _ = json.Unmarshal(raw, &fields) + if fields["error"] != nil || fields["reply_to"] != nil { + return frame, fault.Invalid("response fields are not allowed in worker requests") + } + if hello { + if frame.Type != "worker.hello" || fields["session_id"] != nil { + return frame, fault.Invalid("invalid worker hello envelope") + } + } else if frame.SessionID.Validate() != nil { + return frame, fault.Invalid("worker session is required") + } + return frame, nil +} + +// requiredWorkerFields 校验必填字段及规范 uint64 字符串，不把省略布尔值等同于显式控制意图。 +func requiredWorkerFields(raw []byte, t reflect.Type) error { + if t == reflect.TypeOf(json.RawMessage{}) || t == reflect.TypeOf(time.Time{}) { + return nil + } + if t.Kind() == reflect.Pointer { + return nil + } + if t.Kind() == reflect.Array || t.Kind() == reflect.Slice { + var values []json.RawMessage + if err := json.Unmarshal(raw, &values); err != nil { + return fault.Invalid("invalid worker array") + } + for _, value := range values { + if err := requiredWorkerFields(value, t.Elem()); err != nil { + return err + } + } + return nil + } + if t.Kind() != reflect.Struct { + return nil + } + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil { + return fault.Invalid("invalid worker object") + } + for i := 0; i < t.NumField(); i++ { + field := t.Field(i) + tag := strings.Split(field.Tag.Get("json"), ",") + if tag[0] == "" || tag[0] == "-" { + continue + } + value, found := fields[tag[0]] + optional, numberString := false, false + for _, flag := range tag[1:] { + optional = optional || flag == "omitempty" + numberString = numberString || flag == "string" + } + if !found { + if optional { + continue + } + return fault.Invalid("missing required worker field") + } + if field.Type.Kind() == reflect.Uint64 && numberString { + var text string + if json.Unmarshal(value, &text) != nil { + return fault.Invalid("worker counter must be a decimal string") + } + n, err := strconv.ParseUint(text, 10, 64) + if err != nil || n == 0 || strconv.FormatUint(n, 10) != text { + return fault.Invalid("invalid worker counter") + } + } + if err := requiredWorkerFields(value, field.Type); err != nil { + return err + } + } + return nil +} diff --git a/internal/transport/httpserver/worker_contract_test.go b/internal/transport/httpserver/worker_contract_test.go new file mode 100644 index 0000000..199b96e --- /dev/null +++ b/internal/transport/httpserver/worker_contract_test.go @@ -0,0 +1,213 @@ +package httpserver + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" + "git.yuntuo.hk/k2safe/orbit/internal/testcontract" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" + "github.com/santhosh-tekuri/jsonschema/v6" +) + +// 此存储只接收已通过应用校验的命令，测试不伪造执行成功或当前租约。 +type contractStore struct{ worker.Store } + +func (contractStore) Connect(context.Context, auth.Token, worker.ConnectInput) (worker.Binding, worker.Welcome, error) { + return worker.Binding{}, worker.Welcome{}, nil +} +func (contractStore) Capacity(context.Context, auth.Token, worker.Binding, worker.CapacityInput) (worker.Welcome, error) { + return worker.Welcome{}, nil +} +func (contractStore) Resume(context.Context, auth.Token, worker.Binding, worker.AttemptRef) (worker.LeaseStatus, error) { + return worker.LeaseStatus{}, nil +} +func (contractStore) Start(context.Context, auth.Token, worker.Binding, core.ID, worker.AttemptRef) (worker.CommandResult, error) { + return worker.CommandResult{}, nil +} +func (contractStore) Renew(context.Context, auth.Token, worker.Binding, core.ID, worker.AttemptRef) (worker.CommandResult, error) { + return worker.CommandResult{}, nil +} +func (contractStore) Finish(context.Context, auth.Token, worker.Binding, core.ID, worker.FinishInput) (worker.CommandResult, error) { + return worker.CommandResult{}, nil +} + +func validateClientContract(raw []byte) error { + var kind struct { + Type string `json:"type"` + } + if err := json.Unmarshal(raw, &kind); err != nil { + return err + } + f, err := decodeWorkerRequest(raw, kind.Type == "worker.hello") + if err != nil { + return err + } + svc := worker.New(contractStore{}) + ctx := context.Background() + token, binding := auth.Token{}, worker.Binding{} + switch f.Type { + case "worker.hello": + var in worker.ConnectInput + if err := decodeWorkerJSON(f.Body, &in); err != nil { + return err + } + _, _, err = svc.Connect(ctx, token, in) + case "worker.capacity": + var in worker.CapacityInput + if err := decodeWorkerJSON(f.Body, &in); err != nil { + return err + } + _, err = svc.Capacity(ctx, token, binding, in) + case "worker.resume", "task.start", "task.renew": + var in worker.AttemptRef + if err := decodeWorkerJSON(f.Body, &in); err != nil { + return err + } + switch f.Type { + case "worker.resume": + _, err = svc.Resume(ctx, token, binding, in) + case "task.start": + _, err = svc.Start(ctx, token, binding, f.MessageID, in) + case "task.renew": + _, err = svc.Renew(ctx, token, binding, f.MessageID, in) + } + case "task.finish": + var in worker.FinishInput + if err := decodeWorkerJSON(f.Body, &in); err != nil { + return err + } + _, err = svc.Finish(ctx, token, binding, f.MessageID, in) + default: + return fault.Invalid("unsupported worker message type") + } + return err +} + +func TestWorkerPublishedContract(t *testing.T) { + id := core.ID(strings.Repeat("1", 32)) + secret := strings.Repeat("A", 43) + ref := worker.AttemptRef{TaskID: id, AttemptID: id, Epoch: 1, LeaseSecret: secret} + hello := worker.ConnectInput{InstanceID: id, InstanceSecret: secret, QueueID: id, Capacity: 1, Capabilities: []worker.Capability{{Handler: "text.normalize", HandlerVersion: "v1", PayloadVersion: "v1"}}} + cases := []struct { + kind string + body any + }{ + {"worker.hello", hello}, + {"worker.capacity", worker.CapacityInput{Sequence: 1, Capacity: 0, Drain: true}}, + {"worker.resume", ref}, {"task.start", ref}, {"task.renew", ref}, + {"task.finish", worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{"n":9007199254740993}`)}}, + {"task.finish", worker.FinishInput{Ref: ref, Outcome: "FAILED", ErrorCode: "OBJECT_REJECTED", Result: json.RawMessage(`{}`)}}, + {"task.finish", worker.FinishInput{Ref: ref, Outcome: "UNKNOWN", ErrorCode: "LEASE_LOST", Result: json.RawMessage(`{}`)}}, + } + var valid [][]byte + for _, tc := range cases { + body, _ := jsonwire.Marshal(tc.body) + f := workerFrame{Version: "1", Type: tc.kind, MessageID: id, SessionID: id, Body: body} + if tc.kind == "worker.hello" { + f.SessionID = "" + } + raw, err := jsonwire.Marshal(f) + if err != nil { + t.Fatal(err) + } + testcontract.WorkerFrame(t, "client", raw) + if err := validateClientContract(raw); err != nil { + t.Fatalf("valid %s rejected: %v", tc.kind, err) + } + valid = append(valid, raw) + } + // 修改报文而非重新构造 Go DTO，覆盖不同语言接入时的字段别名、缺失和类型边界。 + invalid := []struct { + name string + raw string + }{ + {"hello_has_session", strings.Replace(string(valid[0]), `"body":`, `"session_id":"", "body":`, 1)}, + {"response_null", strings.Replace(string(valid[0]), `"body":`, `"error":null, "body":`, 1)}, + {"response_reply", strings.Replace(string(valid[0]), `"body":`, `"reply_to":"", "body":`, 1)}, + {"capability_alias", strings.Replace(string(valid[0]), `"handler":`, `"Handler":`, 1)}, + {"capability_conflicting_alias", strings.Replace(string(valid[0]), `"handler":`, `"Handler":"wrong", "handler":`, 1)}, + {"missing_drain", strings.Replace(string(valid[1]), `,"drain":true`, ``, 1)}, + {"counter_numeric", strings.Replace(string(valid[1]), `"sequence":"1"`, `"sequence":1`, 1)}, + {"counter_leading_zero", strings.Replace(string(valid[1]), `"sequence":"1"`, `"sequence":"01"`, 1)}, + {"counter_overflow", strings.Replace(string(valid[1]), `"sequence":"1"`, `"sequence":"18446744073709551616"`, 1)}, + {"counter_zero", strings.Replace(string(valid[1]), `"sequence":"1"`, `"sequence":"0"`, 1)}, + {"capacity_limit", strings.Replace(string(valid[1]), `"capacity":0`, `"capacity":257`, 1)}, + {"secret_noncanonical", strings.Replace(string(valid[2]), secret, strings.Repeat("A", 42)+"B", 1)}, + {"null_result", strings.Replace(string(valid[5]), `{"n":9007199254740993}`, `null`, 1)}, + {"failure_missing_code", strings.Replace(string(valid[6]), `,"error_code":"OBJECT_REJECTED"`, ``, 1)}, + {"success_with_code", strings.Replace(string(valid[6]), `"FAILED"`, `"SUCCEEDED"`, 1)}, + } + for _, tc := range invalid { + t.Run(tc.name, func(t *testing.T) { + value, err := jsonschema.UnmarshalJSON(strings.NewReader(tc.raw)) + if err != nil { + t.Fatal("fixture must remain valid JSON") + } + if testcontract.WorkerSchema(t, "client").Validate(value) == nil { + t.Fatal("Schema accepted invalid message") + } + if validateClientContract([]byte(tc.raw)) == nil { + t.Fatal("runtime accepted invalid message") + } + }) + } + // 标准 Schema 无法表达原始字节层重复键和 Unicode 替换，运行时必须独立拒绝。 + for _, raw := range []string{ + strings.Replace(string(valid[1]), `"drain":true`, `"drain":true,"drain":true`, 1), + strings.Replace(string(valid[5]), `{"n":9007199254740993}`, `{"text":"\ud800"}`, 1), + strings.Replace(string(valid[5]), `{"n":9007199254740993}`, `{"n":1e999}`, 1), + } { + if validateClientContract([]byte(raw)) == nil { + t.Fatal("runtime lost byte-level guard") + } + } + maximum := strings.Replace(string(valid[1]), `"sequence":"1"`, `"sequence":"18446744073709551615"`, 1) + testcontract.WorkerFrame(t, "client", []byte(maximum)) + if err := validateClientContract([]byte(maximum)); err != nil { + t.Fatal(err) + } + + now := time.Date(2026, 9, 17, 8, 0, 0, 0, time.UTC) + welcome := worker.Welcome{SessionID: id, Epoch: 1, ServerTime: now, ExpiresAt: now.Add(45 * time.Second), LeaseDurationMS: 30000} + lease := worker.LeaseStatus{Active: true, State: pulse.AttemptRunning, LeaseUntil: now.Add(30 * time.Second), ServerTime: now} + receipt := worker.Receipt{Operation: "start", TaskState: pulse.TaskRunning, AttemptState: pulse.AttemptRunning, TaskRevision: 3, LeaseUntil: lease.LeaseUntil, Disposition: "APPLIED"} + task := service.TaskView{ID: id, NamespaceID: id, QueueID: id, State: pulse.TaskLeased, Revision: 2, Handler: "text.normalize", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), AvailableAt: now, DeadlineAt: now.Add(time.Hour), RecoverySafety: pulse.RecoveryReconcileOnly, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 30000}, AttemptCount: 1, CreatedAt: now, UpdatedAt: now} + assignment := worker.Assignment{Task: task, AttemptID: id, Epoch: 1, LeaseSecret: secret, LeaseUntil: lease.LeaseUntil, ServerTime: now} + for _, tc := range []struct { + kind string + body any + }{ + {"worker.welcome", welcome}, {"ack", welcome}, {"ack", lease}, + {"ack", worker.CommandResult{Receipt: receipt, Lease: lease, Deduplicated: true}}, {"task.assign", assignment}, + } { + raw, _ := jsonwire.Marshal(tc.body) + f := workerFrame{Version: "1", Type: tc.kind, MessageID: id, SessionID: id, ReplyTo: id, Body: raw} + if tc.kind == "task.assign" { + f.ReplyTo = "" + } + frame, _ := jsonwire.Marshal(f) + testcontract.WorkerFrame(t, "server", frame) + } + for _, f := range []workerFrame{ + {Version: "1", Type: "error", MessageID: id, ReplyTo: id, Error: fault.Unauthorized}, + {Version: "1", Type: "error", MessageID: id, SessionID: id, Error: worker.StaleSession}, + {Version: "1", Type: "error", MessageID: id, SessionID: id, ReplyTo: id, Error: worker.LeaseLost}, + } { + raw, _ := jsonwire.Marshal(f) + testcontract.WorkerFrame(t, "server", raw) + } + invalidResponse, _ := jsonwire.Marshal(workerFrame{Version: "1", Type: "task.assign", MessageID: id, SessionID: id, ReplyTo: id, Body: json.RawMessage(`{}`)}) + value, _ := jsonschema.UnmarshalJSON(strings.NewReader(string(invalidResponse))) + if testcontract.WorkerSchema(t, "server").Validate(value) == nil { + t.Fatal("invalid server assignment accepted") + } +} diff --git a/protocol/workerv1/schema.go b/protocol/workerv1/schema.go new file mode 100644 index 0000000..9d62b8b --- /dev/null +++ b/protocol/workerv1/schema.go @@ -0,0 +1,9 @@ +// Package workerv1 提供 Worker v1 的机器可读契约，运行时仍由类型校验和持久化状态机执行约束。 +package workerv1 + +import _ "embed" + +// Schema 是 Draft 2020-12 的协议文档；clientFrame 与 serverFrame 分别限定通信方向。 +// +//go:embed worker.schema.json +var Schema string diff --git a/protocol/workerv1/worker.schema.json b/protocol/workerv1/worker.schema.json new file mode 100644 index 0000000..a8c6711 --- /dev/null +++ b/protocol/workerv1/worker.schema.json @@ -0,0 +1,899 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://git.yuntuo.hk/k2safe/orbit/schema/worker.v1.json", + "title": "Orbit Pulse Worker v1", + "description": "HTTP Upgrade 子协议 pulse.worker.v1 的 JSON 文本帧。根接受双向帧；clientFrame/serverFrame 可单独校验方向。状态、权限、消息关联、长度与租约约束仍由运行时执行。", + "oneOf": [ + { + "$ref": "#/$defs/clientFrame" + }, + { + "$ref": "#/$defs/serverFrame" + } + ], + "$defs": { + "id": { + "type": "string", + "pattern": "^[0-9a-f]{32}$", + "description": "32 位小写十六进制资源/操作身份。" + }, + "uint64": { + "type": "string", + "pattern": "^(?:[1-9][0-9]{0,18}|1[0-7][0-9]{18}|18[0-3][0-9]{17}|184[0-3][0-9]{16}|1844[0-5][0-9]{15}|18446[0-6][0-9]{14}|184467[0-3][0-9]{13}|1844674[0-3][0-9]{12}|184467440[0-6][0-9]{10}|1844674407[0-2][0-9]{9}|18446744073[0-6][0-9]{8}|1844674407370[0-8][0-9]{6}|18446744073709[0-4][0-9]{5}|184467440737095[0-4][0-9]{4}|18446744073709550[0-9]{3}|18446744073709551[0-5][0-9]{2}|1844674407370955160[0-9]{1}|1844674407370955161[0-4]|18446744073709551615)$", + "description": "1 至 18446744073709551615 的规范十进制字符串，不允许前导零。" + }, + "secret": { + "type": "string", + "pattern": "^[A-Za-z0-9_-]{42}[AEIMQUYcgkosw048]$", + "description": "32 字节秘密的严格无填充 base64url，43 字符；禁止日志记录。" + }, + "time": { + "type": "string", + "format": "date-time", + "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(?:\\.[0-9]{1,6})?Z$", + "description": "服务端 UTC RFC 3339，最多微秒精度。" + }, + "json": { + "not": { + "type": "null" + }, + "description": "非 null 业务 JSON；256 KiB、数值文本和完整帧深度限制由运行时校验。" + }, + "capability": { + "type": "object", + "properties": { + "handler": { + "type": "string", + "minLength": 1, + "maxLength": 128, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + }, + "handler_version": { + "type": "string", + "minLength": 1, + "maxLength": 64, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + }, + "payload_version": { + "type": "string", + "minLength": 1, + "maxLength": 32, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + } + }, + "required": [ + "handler", + "handler_version", + "payload_version" + ], + "additionalProperties": false + }, + "attemptState": { + "enum": [ + "LEASED", + "RUNNING", + "SUCCEEDED", + "FAILED", + "CANCELLED", + "EXPIRED", + "UNKNOWN" + ] + }, + "taskState": { + "enum": [ + "WAITING", + "LEASED", + "RUNNING", + "SUCCEEDED", + "FAILED", + "CANCELLED", + "NEEDS_REVIEW" + ] + }, + "hello": { + "type": "object", + "properties": { + "instance_id": { + "$ref": "#/$defs/id" + }, + "instance_secret": { + "$ref": "#/$defs/secret" + }, + "queue_id": { + "$ref": "#/$defs/id" + }, + "capabilities": { + "type": "array", + "minItems": 1, + "maxItems": 32, + "uniqueItems": true, + "items": { + "$ref": "#/$defs/capability" + } + }, + "capacity": { + "type": "integer", + "minimum": 1, + "maximum": 256 + } + }, + "required": [ + "instance_id", + "instance_secret", + "queue_id", + "capabilities", + "capacity" + ], + "additionalProperties": false, + "description": "重连保留身份、凭据、队列和能力，进程重启换新身份。" + }, + "capacity": { + "type": "object", + "properties": { + "sequence": { + "$ref": "#/$defs/uint64" + }, + "capacity": { + "type": "integer", + "minimum": 0, + "maximum": 256 + }, + "drain": { + "type": "boolean" + } + }, + "required": [ + "sequence", + "capacity", + "drain" + ], + "additionalProperties": false, + "description": "覆盖容量，零暂不领取；drain 一旦生效不能清除。" + }, + "attemptRef": { + "type": "object", + "properties": { + "task_id": { + "$ref": "#/$defs/id" + }, + "attempt_id": { + "$ref": "#/$defs/id" + }, + "epoch": { + "$ref": "#/$defs/uint64" + }, + "lease_secret": { + "$ref": "#/$defs/secret" + } + }, + "required": [ + "task_id", + "attempt_id", + "epoch", + "lease_secret" + ], + "additionalProperties": false + }, + "finish": { + "type": "object", + "properties": { + "ref": { + "$ref": "#/$defs/attemptRef" + }, + "outcome": { + "enum": [ + "SUCCEEDED", + "FAILED", + "UNKNOWN" + ] + }, + "result": { + "$ref": "#/$defs/json" + }, + "error_code": { + "type": "string", + "maxLength": 64 + } + }, + "required": [ + "ref", + "outcome", + "result" + ], + "additionalProperties": false, + "allOf": [ + { + "if": { + "properties": { + "outcome": { + "const": "SUCCEEDED" + } + } + }, + "then": { + "properties": { + "error_code": { + "const": "" + } + } + }, + "else": { + "required": [ + "error_code" + ], + "properties": { + "error_code": { + "type": "string", + "minLength": 1, + "maxLength": 64, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + } + } + } + } + ] + }, + "welcome": { + "type": "object", + "properties": { + "session_id": { + "$ref": "#/$defs/id" + }, + "epoch": { + "$ref": "#/$defs/uint64" + }, + "server_time": { + "$ref": "#/$defs/time" + }, + "expires_at": { + "$ref": "#/$defs/time" + }, + "lease_duration_ms": { + "type": "integer", + "minimum": 1000, + "maximum": 120000 + }, + "draining": { + "type": "boolean" + } + }, + "required": [ + "session_id", + "epoch", + "server_time", + "expires_at", + "lease_duration_ms", + "draining" + ], + "additionalProperties": false + }, + "lease": { + "type": "object", + "properties": { + "active": { + "type": "boolean" + }, + "state": { + "$ref": "#/$defs/attemptState" + }, + "lease_until": { + "$ref": "#/$defs/time" + }, + "server_time": { + "$ref": "#/$defs/time" + } + }, + "required": [ + "active", + "state", + "lease_until", + "server_time" + ], + "additionalProperties": false, + "description": "本次重新核验的授权；只有 active 且 RUNNING 并在本地保守期限内才能执行。", + "allOf": [ + { + "if": { + "properties": { + "active": { + "const": true + } + } + }, + "then": { + "properties": { + "state": { + "enum": [ + "LEASED", + "RUNNING" + ] + } + } + } + } + ] + }, + "receipt": { + "type": "object", + "properties": { + "operation": { + "enum": [ + "start", + "renew", + "finish" + ] + }, + "task_state": { + "$ref": "#/$defs/taskState" + }, + "attempt_state": { + "$ref": "#/$defs/attemptState" + }, + "task_revision": { + "$ref": "#/$defs/uint64" + }, + "lease_until": { + "$ref": "#/$defs/time" + }, + "disposition": { + "enum": [ + "APPLIED", + "EVIDENCE_ONLY" + ] + } + }, + "required": [ + "operation", + "task_state", + "attempt_state", + "task_revision", + "lease_until", + "disposition" + ], + "additionalProperties": false, + "description": "稳定操作身份的原始不可变回执，不提供当前执行权。" + }, + "commandResult": { + "type": "object", + "properties": { + "receipt": { + "$ref": "#/$defs/receipt" + }, + "lease": { + "$ref": "#/$defs/lease" + }, + "deduplicated": { + "type": "boolean" + } + }, + "required": [ + "receipt", + "lease", + "deduplicated" + ], + "additionalProperties": false + }, + "retry": { + "type": "object", + "properties": { + "max_attempts": { + "type": "integer", + "minimum": 1, + "maximum": 100 + }, + "initial_delay_ms": { + "type": "integer", + "minimum": 0, + "maximum": 9223372036854 + }, + "max_delay_ms": { + "type": "integer", + "minimum": 0, + "maximum": 9223372036854 + }, + "attempt_timeout_ms": { + "type": "integer", + "minimum": 1, + "maximum": 2592000000 + } + }, + "required": [ + "max_attempts", + "initial_delay_ms", + "max_delay_ms", + "attempt_timeout_ms" + ], + "additionalProperties": false + }, + "assignedTask": { + "type": "object", + "properties": { + "id": { + "$ref": "#/$defs/id" + }, + "namespace_id": { + "$ref": "#/$defs/id" + }, + "queue_id": { + "$ref": "#/$defs/id" + }, + "state": { + "const": "LEASED" + }, + "revision": { + "$ref": "#/$defs/uint64" + }, + "handler": { + "type": "string", + "minLength": 1, + "maxLength": 128, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + }, + "handler_version": { + "type": "string", + "minLength": 1, + "maxLength": 64, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + }, + "payload_version": { + "type": "string", + "minLength": 1, + "maxLength": 32, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + }, + "payload": { + "$ref": "#/$defs/json" + }, + "result": { + "$ref": "#/$defs/json" + }, + "business_key": { + "type": "string", + "maxLength": 191 + }, + "available_at": { + "$ref": "#/$defs/time" + }, + "start_deadline_at": { + "$ref": "#/$defs/time" + }, + "deadline_at": { + "$ref": "#/$defs/time" + }, + "recovery_safety": { + "enum": [ + "IDEMPOTENT", + "RECONCILE_ONLY" + ] + }, + "retry": { + "$ref": "#/$defs/retry" + }, + "attempt_count": { + "type": "integer", + "minimum": 1, + "maximum": 100 + }, + "cancel_requested_at": { + "$ref": "#/$defs/time" + }, + "completed_at": { + "$ref": "#/$defs/time" + }, + "created_at": { + "$ref": "#/$defs/time" + }, + "updated_at": { + "$ref": "#/$defs/time" + } + }, + "required": [ + "id", + "namespace_id", + "queue_id", + "state", + "revision", + "handler", + "handler_version", + "payload_version", + "payload", + "available_at", + "deadline_at", + "recovery_safety", + "retry", + "attempt_count", + "created_at", + "updated_at" + ], + "additionalProperties": false + }, + "assignment": { + "type": "object", + "properties": { + "task": { + "$ref": "#/$defs/assignedTask" + }, + "attempt_id": { + "$ref": "#/$defs/id" + }, + "epoch": { + "$ref": "#/$defs/uint64" + }, + "lease_secret": { + "$ref": "#/$defs/secret" + }, + "lease_until": { + "$ref": "#/$defs/time" + }, + "server_time": { + "$ref": "#/$defs/time" + } + }, + "required": [ + "task", + "attempt_id", + "epoch", + "lease_secret", + "lease_until", + "server_time" + ], + "additionalProperties": false + }, + "error": { + "type": "object", + "properties": { + "code": { + "type": "string", + "minLength": 1, + "maxLength": 64, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_.:-]*$" + }, + "message": { + "type": "string", + "maxLength": 1024 + } + }, + "required": [ + "code", + "message" + ], + "additionalProperties": false, + "description": "公开稳定分类，不包含 SQL、原始异常或秘密。" + }, + "helloFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "worker.hello" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/hello" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "body" + ], + "additionalProperties": false + }, + "capacityFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "worker.capacity" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/capacity" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "body" + ], + "additionalProperties": false + }, + "resumeFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "worker.resume" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/attemptRef" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "body" + ], + "additionalProperties": false + }, + "startFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "task.start" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/attemptRef" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "body" + ], + "additionalProperties": false + }, + "renewFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "task.renew" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/attemptRef" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "body" + ], + "additionalProperties": false + }, + "finishFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "task.finish" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/finish" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "body" + ], + "additionalProperties": false + }, + "welcomeFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "worker.welcome" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "reply_to": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/welcome" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "reply_to", + "body" + ], + "additionalProperties": false + }, + "assignmentFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "task.assign" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "body": { + "$ref": "#/$defs/assignment" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "body" + ], + "additionalProperties": false + }, + "ackFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "ack" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "reply_to": { + "$ref": "#/$defs/id" + }, + "body": { + "oneOf": [ + { + "$ref": "#/$defs/welcome" + }, + { + "$ref": "#/$defs/lease" + }, + { + "$ref": "#/$defs/commandResult" + } + ] + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "session_id", + "reply_to", + "body" + ], + "additionalProperties": false + }, + "errorFrame": { + "type": "object", + "properties": { + "protocol_version": { + "const": "1" + }, + "type": { + "const": "error" + }, + "message_id": { + "$ref": "#/$defs/id" + }, + "session_id": { + "$ref": "#/$defs/id" + }, + "error": { + "$ref": "#/$defs/error" + }, + "reply_to": { + "$ref": "#/$defs/id" + } + }, + "required": [ + "protocol_version", + "type", + "message_id", + "error" + ], + "additionalProperties": false, + "description": "注册失败可省略 session_id；会话内错误必须带当前会话。异步致命错误无 reply_to，其余与请求关联。" + }, + "clientFrame": { + "oneOf": [ + { + "$ref": "#/$defs/helloFrame" + }, + { + "$ref": "#/$defs/capacityFrame" + }, + { + "$ref": "#/$defs/resumeFrame" + }, + { + "$ref": "#/$defs/startFrame" + }, + { + "$ref": "#/$defs/renewFrame" + }, + { + "$ref": "#/$defs/finishFrame" + } + ] + }, + "serverFrame": { + "oneOf": [ + { + "$ref": "#/$defs/welcomeFrame" + }, + { + "$ref": "#/$defs/assignmentFrame" + }, + { + "$ref": "#/$defs/ackFrame" + }, + { + "$ref": "#/$defs/errorFrame" + } + ] + } + } +} diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 6811b41..d4efd7d 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -53,3 +53,9 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/worker-protocol) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/worker) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/history) + +# 停止原服务后再做进程故障演练，避免其他恢复扫描器掩盖接管验证。 +kill "$TASK_SERVER_PID" +wait "$TASK_SERVER_PID" +TASK_SERVER_PID= +(cd sdk/go && ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" ORBIT_TEST_WORKER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_WORKER_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" test -race -count=1 -run '^TestIntegrationWorkerProcess' -v .) diff --git a/sdk/go/worker_integration_test.go b/sdk/go/worker_integration_test.go index 5869a55..7493b1d 100644 --- a/sdk/go/worker_integration_test.go +++ b/sdk/go/worker_integration_test.go @@ -48,18 +48,21 @@ func integrationWorkerClient(t *testing.T) (*Client, string, string) { // workerFaultGateway 在实际服务端确认后丢弃指定 ACK；SDK 到网关使用真实 TLS。 // 只丢响应，不伪造任务状态，测试凭据不进入 URL 或输出。 type workerFaultGateway struct { - drop atomic.Value - offline atomic.Bool - mu sync.Mutex - active map[*websocket.Conn]struct{} - dropped chan string - server *httptest.Server + drop atomic.Value + target atomic.Value + dropOffline atomic.Bool + offline atomic.Bool + mu sync.Mutex + active map[*websocket.Conn]struct{} + dropped chan string + server *httptest.Server } func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { t.Helper() g := &workerFaultGateway{dropped: make(chan string, 4), active: make(map[*websocket.Conn]struct{})} g.drop.Store("") + g.target.Store(upstream) ctx, cancel := context.WithCancel(context.Background()) var handlers sync.WaitGroup g.server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -73,7 +76,7 @@ func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { defer stop() header := http.Header{} header.Set("Authorization", r.Header.Get("Authorization")) - up, _, err := websocket.Dial(bridge, upstream+"/v1/worker/connect", &websocket.DialOptions{HTTPHeader: header, Subprotocols: []string{workerSubprotocol}}) + up, _, err := websocket.Dial(bridge, g.target.Load().(string)+"/v1/worker/connect", &websocket.DialOptions{HTTPHeader: header, Subprotocols: []string{workerSubprotocol}}) if err != nil { http.Error(w, "upstream unavailable", 503) return @@ -135,6 +138,9 @@ func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { delete(operations, f.ReplyTo) mu.Unlock() if f.Type == "ack" && operation != "" && g.drop.CompareAndSwap(operation, "") { + if g.dropOffline.Load() { + g.offline.Store(true) + } g.dropped <- operation return } diff --git a/sdk/go/worker_process_integration_test.go b/sdk/go/worker_process_integration_test.go new file mode 100644 index 0000000..2bb6c8e --- /dev/null +++ b/sdk/go/worker_process_integration_test.go @@ -0,0 +1,394 @@ +package orbit + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +// orbitTestProcess 持有真实子进程，必须 Wait 确认退出才算完成故障注入。 +type orbitTestProcess struct { + cmd *exec.Cmd + done chan struct{} + waitErr error + url, nodeID string +} + +func processEnv(overrides map[string]string) []string { + env := make([]string, 0, len(os.Environ())+len(overrides)) + for _, entry := range os.Environ() { + key, _, _ := strings.Cut(entry, "=") + if _, replace := overrides[key]; !replace { + env = append(env, entry) + } + } + for k, v := range overrides { + env = append(env, k+"="+v) + } + return env +} +func launchTestProcess(t *testing.T, cmd *exec.Cmd) *orbitTestProcess { + t.Helper() + p := &orbitTestProcess{cmd: cmd, done: make(chan struct{})} + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + go func() { p.waitErr = cmd.Wait(); close(p.done) }() + t.Cleanup(func() { + select { + case <-p.done: + return + default: + } + _ = p.cmd.Process.Kill() + select { + case <-p.done: + case <-time.After(5 * time.Second): + t.Error("owned child process did not exit") + } + }) + return p +} +func (p *orbitTestProcess) kill(t *testing.T) { + t.Helper() + select { + case <-p.done: + t.Fatal("fault target exited before injection") + default: + } + if err := p.cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + select { + case <-p.done: + var exit *exec.ExitError + if !errors.As(p.waitErr, &exit) || exit.ExitCode() != -1 { + t.Fatal("owned process was not terminated by a signal") + } + case <-time.After(5 * time.Second): + t.Fatal("killed process did not exit") + } +} +func startOrbitProcess(t *testing.T) *orbitTestProcess { + t.Helper() + binary, dsn := os.Getenv("ORBIT_TEST_SERVER_BINARY"), os.Getenv("ORBIT_TEST_MYSQL_DSN") + if binary == "" || dsn == "" { + t.Skip("requires isolated database and compiled Orbit binary") + } + if !filepath.IsAbs(binary) || !strings.Contains(dsn, "@tcp(127.0.0.1:") { + t.Fatal("process faults require disposable local resources") + } + cmd := exec.Command(binary) + cmd.Env = processEnv(map[string]string{"ORBIT_MYSQL_DSN": dsn, "ORBIT_HTTP_ADDR": "127.0.0.1:0"}) + output, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + cmd.Stderr = io.Discard + address := make(chan [2]string, 1) + go func() { + scanner := bufio.NewScanner(output) + for scanner.Scan() { + var line struct { + Address string `json:"address"` + Node string `json:"node_id"` + } + if json.Unmarshal(scanner.Bytes(), &line) == nil && strings.HasPrefix(line.Address, "127.0.0.1:") && validID(line.Node) { + select { + case address <- [2]string{line.Address, line.Node}: + default: + } + } + } + }() + p := launchTestProcess(t, cmd) + select { + case value := <-address: + p.url = "http://" + value[0] + p.nodeID = value[1] + case <-p.done: + t.Fatal("isolated Orbit exited during startup") + case <-time.After(10 * time.Second): + t.Fatal("isolated Orbit did not publish its bound address") + } + return p +} +func processClient(t *testing.T, p *orbitTestProcess, token string) *Client { + t.Helper() + c, err := NewClient(p.url, Options{Token: token}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if _, err := c.Readiness(ctx); err != nil { + t.Fatal(err) + } + return c +} +func awaitProcessDrop(t *testing.T, ctx context.Context, g *workerFaultGateway, operation string) { + t.Helper() + select { + case got := <-g.dropped: + if got != operation { + t.Fatal("wrong process fault boundary") + } + case <-ctx.Done(): + t.Fatal("process fault boundary not reached") + } +} + +// 两次在真实数据库提交后丢 ACK 并强杀节点，由另外的真实进程核对原操作。 +func TestIntegrationWorkerProcessACKRecovery(t *testing.T) { + a := startOrbitProcess(t) + _, queue, token := integrationWorkerClient(t) + apiA := processClient(t, a, token) + b := startOrbitProcess(t) + apiB := processClient(t, b, token) + if a.nodeID == b.nodeID { + t.Fatal("server processes reused node identity") + } + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + g := newWorkerFaultGateway(t, a.url) + g.dropOffline.Store(true) + g.drop.Store("task.start") + client, err := NewClient(g.server.URL, Options{Token: token, HTTPClient: g.server.Client()}) + if err != nil { + t.Fatal(err) + } + name := "process.ack." + workerTestID(t) + entered, release := make(chan WorkerExecution, 2), make(chan struct{}) + var called atomic.Int32 + w, err := NewWorker(client, WorkerOptions{QueueID: queue, Capacity: 1, ReconnectMaxDelay: time.Second, Handlers: []WorkerHandler{runtimeHandler(name, func(work context.Context, e WorkerExecution) (json.RawMessage, error) { + called.Add(1) + entered <- e + select { + case <-release: + return json.RawMessage(`{"value":9007199254740993}`), nil + case <-work.Done(): + return nil, work.Err() + } + })}}) + if err != nil { + t.Fatal(err) + } + task := runtimeTask(t, ctx, apiA, queue, name, 30000) + run, stop := context.WithCancel(ctx) + defer stop() + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + awaitProcessDrop(t, ctx, g, "task.start") + if called.Load() != 0 { + t.Fatal("handler ran before recovering lost start confirmation") + } + before, err := apiB.ListTaskAttempts(ctx, task.ID, HistoryPageRequest{}) + if err != nil || len(before.Items) != 1 || before.Items[0].AuthorizedAt == nil { + t.Fatal("start was not durably committed before kill") + } + initial, err := apiB.GetTaskAttempt(ctx, task.ID, before.Items[0].ID) + if err != nil { + t.Fatal(err) + } + a.kill(t) + g.target.Store(b.url) + g.offline.Store(false) + var execution WorkerExecution + select { + case execution = <-entered: + case err := <-done: + t.Fatalf("Worker stopped during node recovery: %v", err) + case <-ctx.Done(): + t.Fatal("handler did not start after failover") + } + moved, err := apiB.GetTaskAttempt(ctx, task.ID, execution.AttemptID) + if err != nil || moved.ID != initial.ID || moved.SessionID == initial.SessionID || moved.WorkerInstanceID != initial.WorkerInstanceID || moved.AuthorizedAt == nil || !moved.AuthorizedAt.Equal(*initial.AuthorizedAt) { + t.Fatal("node failover changed attempt or start identity") + } + g.drop.Store("task.finish") + close(release) + awaitProcessDrop(t, ctx, g, "task.finish") + committed, err := apiB.GetTask(ctx, task.ID) + if err != nil || committed.State != TaskSucceeded { + t.Fatal("finish was not committed before second kill") + } + b.kill(t) + c := startOrbitProcess(t) + apiC := processClient(t, c, token) + if c.nodeID == a.nodeID || c.nodeID == b.nodeID { + t.Fatal("replacement reused node identity") + } + g.target.Store(c.url) + g.offline.Store(false) + for w.Stats().ReportsConfirmed != 1 { + select { + case err := <-done: + t.Fatalf("Worker stopped before saved result confirmation: %v", err) + case <-ctx.Done(): + t.Fatal("saved result was not recovered") + case <-time.After(20 * time.Millisecond): + } + } + stop() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("Worker did not drain after failover") + } + current, err := apiC.GetTask(ctx, task.ID) + if err != nil || current.State != TaskSucceeded || current.AttemptCount != 1 || current.Revision != committed.Revision || string(current.Result) != `{"value":9007199254740993}` { + t.Fatal("lost ACK recovery changed authoritative result") + } + attempts, err := apiC.ListTaskAttempts(ctx, task.ID, HistoryPageRequest{}) + if err != nil || len(attempts.Items) != 1 || attempts.Items[0].ID != initial.ID { + t.Fatal("node crash created duplicate execution") + } + evidence, err := apiC.ListAttemptEvidence(ctx, task.ID, initial.ID, HistoryPageRequest{}) + if err != nil || len(evidence.Items) != 0 || called.Load() != 1 || w.Stats().Started != 1 || w.Stats().Reconnects < 2 { + t.Fatal("node recovery replayed handler or produced false evidence") + } +} + +// 此测试只供父测试启动真实 Worker 子进程，父进程负责强杀与 Wait。 +func TestWorkerProcessCrashChild(t *testing.T) { + if os.Getenv("ORBIT_TEST_WORKER_CHILD") != "1" { + t.Skip("owned subprocess only") + } + api, queue, _ := integrationWorkerClient(t) + queue = os.Getenv("ORBIT_TEST_WORKER_CHILD_QUEUE") + if !validID(queue) { + t.Fatal("invalid child queue") + } + name, marker := os.Getenv("ORBIT_TEST_WORKER_HANDLER"), os.Getenv("ORBIT_TEST_WORKER_MARKER") + if !workerName(name, 128) || !filepath.IsAbs(marker) { + t.Fatal("invalid child fixture") + } + w, err := NewWorker(api, WorkerOptions{QueueID: queue, Capacity: 1, Handlers: []WorkerHandler{runtimeHandler(name, func(ctx context.Context, e WorkerExecution) (json.RawMessage, error) { + // 文件代表已发起的外部工作标记，不能推断为可安全重试的明确失败。 + file, err := os.OpenFile(marker, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return nil, err + } + _, err = file.WriteString(e.Task.ID) + syncErr := file.Sync() + closeErr := file.Close() + if err != nil || syncErr != nil || closeErr != nil { + return nil, errors.New("child marker failed") + } + <-ctx.Done() + return nil, ctx.Err() + })}}) + if err != nil { + t.Fatal(err) + } + if err := w.Run(context.Background()); err != nil { + t.Fatal(err) + } + t.Fatal("crash child unexpectedly drained") +} +func TestIntegrationWorkerProcessCrashUnknown(t *testing.T) { + a := startOrbitProcess(t) + _, queue, token := integrationWorkerClient(t) + apiA := processClient(t, a, token) + ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second) + defer cancel() + dedicated, err := apiA.CreateQueue(ctx, "crash-queue-"+workerTestID(t), CreateQueueRequest{Name: "crash-" + workerTestID(t), MaxConcurrency: 1, RatePerSecond: 10}) + if err != nil { + t.Fatal(err) + } + queue = dedicated.Queue.ID + name := "process.crash." + workerTestID(t) + task, err := apiA.SubmitTask(ctx, "process-crash-"+workerTestID(t), SubmitTaskRequest{QueueID: queue, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), TaskTimeoutMS: 30000, Retry: RetryPolicy{MaxAttempts: 3, AttemptTimeoutMS: 6000}, RecoverySafety: RecoveryReconcileOnly}) + if err != nil { + t.Fatal(err) + } + marker := filepath.Join(t.TempDir(), "external-work-started") + executable, err := os.Executable() + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(executable, "-test.run=^TestWorkerProcessCrashChild$", "-test.timeout=30s") + cmd.Env = processEnv(map[string]string{"ORBIT_TEST_WORKER_CHILD": "1", "ORBIT_TEST_WORKER_URL": a.url, "ORBIT_TEST_WORKER_CHILD_QUEUE": queue, "ORBIT_TEST_WORKER_HANDLER": name, "ORBIT_TEST_WORKER_MARKER": marker}) + cmd.Stdout = io.Discard + cmd.Stderr = io.Discard + child := launchTestProcess(t, cmd) + for { + raw, err := os.ReadFile(marker) + if err == nil && string(raw) == task.Task.ID { + break + } + if err != nil && !os.IsNotExist(err) { + t.Fatal(err) + } + select { + case <-child.done: + t.Fatal("Worker child exited before starting") + case <-ctx.Done(): + t.Fatal("Worker child did not start") + case <-time.After(20 * time.Millisecond): + } + } + running, err := apiA.GetTask(ctx, task.Task.ID) + if err != nil || running.State != TaskRunning { + t.Fatal("child did not receive durable start authorization") + } + child.kill(t) + a.kill(t) + b := startOrbitProcess(t) + apiB := processClient(t, b, token) + unknown := waitRuntimeState(t, ctx, apiB, task.Task.ID, TaskNeedsReview) + if unknown.AttemptCount != 1 || unknown.Retry.MaxAttempts != 3 || len(unknown.Result) != 0 { + t.Fatal("crash recovery reset budget or fabricated success") + } + attempts, err := apiB.ListTaskAttempts(ctx, task.Task.ID, HistoryPageRequest{}) + if err != nil || len(attempts.Items) != 1 || attempts.Items[0].State != AttemptUnknown || attempts.Items[0].EndedAt == nil { + t.Fatal("replacement did not preserve unknown attempt") + } + // 相同能力的新 Worker 消费探针，原任务保持核查，验证许可已回收。 + var called atomic.Int32 + w, err := NewWorker(apiB, WorkerOptions{QueueID: queue, Capacity: 1, Handlers: []WorkerHandler{runtimeHandler(name, func(_ context.Context, e WorkerExecution) (json.RawMessage, error) { + called.Add(1) + if e.Task.ID == task.Task.ID { + return nil, errors.New("unknown task was replayed") + } + return json.RawMessage(`{"probe":true}`), nil + })}}) + if err != nil { + t.Fatal(err) + } + probe := runtimeTask(t, ctx, apiB, queue, name, 5000) + run, stop := context.WithCancel(ctx) + defer stop() + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + waitRuntimeState(t, ctx, apiB, probe.ID, TaskSucceeded) + stop() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("replacement Worker did not drain") + } + final, err := apiB.GetTask(ctx, task.Task.ID) + if err != nil || final.State != TaskNeedsReview || final.AttemptCount != 1 || called.Load() != 1 { + t.Fatal("replacement replayed unknown work") + } + raw, err := os.ReadFile(marker) + if err != nil || string(raw) != task.Task.ID { + t.Fatal("external work marker lost") + } +}