diff --git a/README.md b/README.md index 8358c62..7b94a10 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,9 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已实现 32 个数据模型、版本化迁移、凭据及范围校验、队列创建查询、任务幂等提交/查询/分页/待执行取消，以及对应 Go SDK 和示例。P2 服务端执行内核和 WSS 已通过阶段内回归；Worker SDK 自动运行器、调度器和流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已实现 32 个数据模型、版本化迁移、凭据及范围校验、队列创建查询、任务幂等提交/查询/分页/待执行取消，以及对应 Go SDK 和示例。P2 服务端执行内核、WSS 和独立 Worker 底层 SDK 已通过阶段内回归，可实际完成文本处理任务；自动 Worker 运行器、调度器和流程引擎按 [持续开发计划](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)。 +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)。 ## 目录 @@ -53,7 +53,7 @@ cd sdk/go ORBIT_URL=http://127.0.0.1:8080 go run ./examples/health ``` -任务接入从 [初始化凭据](docs/api/authentication.md) 开始，再按 [SDK 接入](docs/sdk/README.md) 运行任务示例。它会实际提交、重复确认、查询并取消一个任务；当前阶段尚不执行业务处理器。 +任务接入从 [初始化凭据](docs/api/authentication.md) 开始，再按 [SDK 接入](docs/sdk/README.md) 运行受理/取消示例。另有 [Worker 协议示例](docs/sdk/worker.md)，实际领取并处理文本、提交结果与排空；自动运行器仍在开发。 ## 开发与回归 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index f0a2fc8..dda87a9 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 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 开发中：服务端执行事务和 WSS 已通过阶段内回归；Worker SDK 自动运行/重连/退出及完整验收待完成 | +| P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复；Worker SDK | 真 Worker 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 开发中：服务端、独立 SDK 底层协议及实际 ACK 丢失恢复回归通过；文本执行示例可运行；自动运行/重连/退出、查询与完整验收待完成 | | P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 待开发 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | @@ -74,3 +74,4 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P0 验证记录](reports/phase-0.md) - [P1 验证记录](reports/phase-1.md) - [P2 服务端阶段记录](reports/phase-2-server.md) +- [P2 SDK 协议阶段记录](reports/phase-2-sdk-protocol.md) diff --git a/docs/README.md b/docs/README.md index cef0b53..cc3374f 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本：0.2 - 整理日期：2026-09-17 -- 阶段：P0/P1 已完成，P2 服务端执行/WSS 已通过阶段内回归，Worker SDK 继续开发；尚未完成生产验收 +- 阶段：P0/P1 已完成，P2 服务端与 Worker 底层 SDK 已通过阶段内回归，自动 Worker 运行器继续开发；尚未完成生产验收 - 适用范围：Orbit 统一平台，包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 @@ -56,7 +56,7 @@ Orbit 不内置特定行业规则；Pulse 管理执行，Weave 管理步骤依 | --- | --- | --- | | HTTP / WS 的最终字段、路径、错误码 | 以本稿作为协议草案，评审后冻结 v1 | [协议](03-http-websocket-protocol.md) | | Weave 技术细则 | 功能范围已确认；SDK/定义 Schema、暂停门控与计时、查询预算、并行失败策略、人工恢复身份仍需评审 | [流程](06-workflow-spec.md) | -| WS 库及后续依赖版本 | 当前基础使用 Go 1.27.1、MySQL 8.4+ 与显式 SQL；WS 库随 P2 实现锁定 | [模型基线](07-data-model.md) | +| WS 库及后续依赖版本 | 当前基础使用 Go 1.27.1、MySQL 8.4+ 与显式 SQL；WS 已锁定 coder/websocket v1.8.15，后续依赖需单独验证 | [模型基线](07-data-model.md) | | 人工重试的任务身份 | 建议创建关联的新任务，自动重试保持原任务 ID | [功能](01-functional-spec.md) | | Cron、固定间隔与夏令时细节 | 建议五段 Cron、固定间隔首版采用 fixed-rate；夏令时行为需明确 | [功能](01-functional-spec.md) | | 批量大小、消息上限、心跳、租约、扫描周期 | 均需设置有限上限；数值通过原型与压测确定 | [验收](05-production-acceptance.md) | @@ -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) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 +持续开发目标已建立。当前实现和验收进度以 [开发计划](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) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 diff --git a/docs/api/README.md b/docs/api/README.md index adcecf8..d311c15 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -11,7 +11,7 @@ | `/v1/tasks` 提交、查询、分页与待执行取消 | P1 已实现 | [任务接口](tasks.md) | | `/v1/queues` 创建与查询 | P1 已实现 | [队列接口](queues.md) | | 凭据校验与初始化 CLI | P1 已实现 | [鉴权](authentication.md) | -| `/v1/worker/connect` | P2 服务端已实现，Worker SDK 仍在开发 | [Worker 契约](worker.md) | +| `/v1/worker/connect` | P2 服务端及底层 SDK 已实现，自动 Worker 运行器开发中 | [Worker 契约](worker.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.md b/docs/api/worker.md index 657e5a0..97236f2 100644 --- a/docs/api/worker.md +++ b/docs/api/worker.md @@ -1,6 +1,6 @@ # Worker 通信与执行契约 v1（开发中） -已实现服务端执行事务、WebSocket 接入和真实 WSS 回归。**P2 尚未完成**：独立 Go SDK 的 Worker 运行器、自动重连/续租、处理器生命周期与优雅退出还在开发。本页记录当前可验证的服务端契约，不把测试客户端当作可交付 Worker SDK。 +已实现服务端执行事务、WebSocket 接入、[独立 Go 底层协议 SDK](../sdk/worker.md) 及真实 WSS/ACK 丢失回归。**P2 尚未完成**：自动 Worker 运行器、自动重连/续租、处理器生命周期与优雅退出还在开发。本页记录当前可验证的协议；底层 SDK 不自动管理处理器。 ## 1. 接入与身份 @@ -76,7 +76,7 @@ worker.welcome 返回 session_id、epoch、server_time、expires_at、lease_dura | worker.capacity | sequence 字符串、capacity 0–256、drain bool | 更新后的 welcome 字段 | | task.assign | 服务端主动发送 | task 完整任务、attempt_id、epoch、lease_secret、lease_until、server_time | -capacity 使用单调递增序号和覆盖值。同序号同内容可重复，同序号不同内容或旧序号冲突；capacity=0 不释放已有任务的并发许可。drain=true 是进程级持久化排空，重连保留，不能通过后续消息清除；新进程使用新实例身份。 +capacity 使用单调递增序号和覆盖值。同序号同内容可重复，同序号不同内容或旧序号冲突；capacity=0 不释放已有任务的并发许可。drain=true 是进程级持久化排空，重连保留；后续必须保持 drain=true，尝试清除返回 STATE_CONFLICT。新进程使用新实例身份。 一个 Worker 进程当前绑定一个队列。可以运行多个 Worker 进程或后续 SDK 实例消费不同队列；不能通过重连静默切换队列。 diff --git a/docs/examples/README.md b/docs/examples/README.md index 4fe216d..e1b3106 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -19,6 +19,15 @@ - 异常：凭据失效、权限不足、内容冲突、版本竞争和未知网络结果，见 [任务接口](../api/tasks.md)。 - 清理：示例以取消结束；任务和审计历史保留。重复运行会生成独立示例任务。 +## 已可运行：文本规范化执行 + +- 源码：[Worker 底层协议示例](../../sdk/go/examples/worker-protocol/main.go)。 +- 前置：专用 Worker 凭据，启动步骤见 [Worker SDK 文档](../sdk/worker.md)。 +- 输入与步骤：提交 `{"text":" Hello Orbit "}` → 注册/容量 → 领取 → 开始授权 → 实际处理 → 确认结果 → 查询 → 排空。 +- 预期：SUCCEEDED、一次尝试、结果 `{"text":"HELLO ORBIT"}`。 +- 异常：控制 ACK 丢失使用原身份有限重报；业务函数不随重连再次调用；迟到证据不能当作成功结果。自动续租与持续消费仍待运行器实现。 +- 清理：任务历史保留；运行结束确认排空并关闭连接。每次示例使用独立任务/处理器身份。 + ## 计划场景与验收标准 | 场景 | 输入与步骤 | 预期结果 | 必测异常 | 状态 | diff --git a/docs/reports/phase-2-sdk-protocol.md b/docs/reports/phase-2-sdk-protocol.md new file mode 100644 index 0000000..9ff1259 --- /dev/null +++ b/docs/reports/phase-2-sdk-protocol.md @@ -0,0 +1,55 @@ +# P2 Worker SDK 协议阶段回归 + +日期：2026-09-17。此记录覆盖独立 SDK 的底层连接和一次性执行闭环，**不代表 P2 完成，也不代表自动 Worker 运行器已交付**。 + +## 实现范围 + +- 独立 module 内实现 WS/WSS 注册、容量/心跳、派发读取、Start、Renew、Finish、Resume；不引用服务端模型/存储。 +- 固定子协议、授权头、受信任 TLS、禁用重定向/压缩、握手/控制/读取时限及缓冲上限。 +- 并发 ACK 关联、稳定消息身份、公开错误分类；秘密和业务结果在默认格式化输出中隐藏。 +- 当前租约与原回执分离，保守单调时钟期限；精确 JSON 数字及字符串 uint64 版本。 +- 可运行的纯文本规范化示例，有限手动重连核对控制操作，不重复调用处理器。 +- 新增字段均有中文契约注释；无持久化模型或迁移变更。 + +## 验证环境与命令 + +macOS arm64，Go 1.27.1，Docker 隔离 MySQL 8.4.10。连接依赖锁定 `github.com/coder/websocket v1.8.15`。 + +```sh +make test +make vet +make build +make integration +cd sdk/go +GOWORK=off go test -race ./... +GOWORK=off go vet ./... +GOWORK=off go test -run '^$' -fuzz FuzzWorkerJSON -fuzztime=15s . +``` + +测试脚本在一次性数据库先执行原服务端迁移、任务、执行、WSS 回归，再启动真实 orbit 命令及独立凭据，用 SDK 接入该服务；随后执行真实示例。测试不连接已有业务库，凭据不放入 URL、命令参数或输出。WSS 故障网关使用 `httptest.NewTLSServer` 及受信任测试证书，没有跳过证书校验。 + +## 通过的场景 + +| 测试层 | 场景与断言 | +| --- | --- | +| SDK 单元/WSS | 派发与 ACK 交错，多个请求逆序响应仍正确关联；业务整数/高精度小数和 uint64 版本不转为 float64 | +| SDK 授权判断 | 数据库墙钟偏差 24 小时不改变本地有效窗口；扣除全请求耗时和 250 毫秒余量；历史 RUNNING 回执和自行解码的 Lease 均不能授权 | +| SDK 有界传输 | 32 个未确认控制请求后拒绝新增；同 ID 同时在途拒绝；取消让所有等待者退出；派发缓冲耗尽关闭；关闭后不返回缓存任务 | +| SDK 协议防护 | 重定向不转交凭据；非法/未知回复、错误会话控制、失效凭据错误关闭；拒绝重复键、非法 Unicode、过深 JSON；默认格式化隐藏秘密 | +| 真实服务 + MySQL + WSS 故障网关 | 服务端完成 start 后丢 ACK；新连接 Resume 并重报原 start ID 得到原回执，原尝试保持不变 | +| 真实服务 + MySQL + WSS 故障网关 | 服务端完成 finish 后丢 ACK；新连接核对原结果，保持 SUCCEEDED、一次尝试及原始精确数值；旧 start ACK 不能再次授权 | +| 真实服务 + MySQL | 同消息续租返回原版本/期限；冲突结果只保存 EVIDENCE_ONLY；持久化排空跨重连，尝试清除被拒绝 | +| 真实服务 + MySQL | 短尝试时限驱动本地 context 到期；过期后旧回执不授权、迟到报告仅证据；服务端恢复为 NEEDS_REVIEW，不生成第二次尝试 | +| 真实命令行示例 | 提交文本 → 开始授权 → 实际规范化 → 上报 → 查询 → 排空；输出 SUCCEEDED、attempts=1、HELLO ORBIT | + +15 秒 JSON 模糊测试本次执行 738,707 个输入，通过。实际执行数只描述本次运行，不是稳定吞吐或性能基准。完整隔离回归中原服务端测试通过，SDK 故障测试通过，三类 SDK 示例通过。 + +首次新增集成测试把“已排空后发送 Drain=false”误写为可成功但保持排空；实际服务端按既定契约返回 STATE_CONFLICT。已修正测试为确认拒绝，再以 Drain=true 心跳验证，重新执行完整隔离回归通过。没有为通过测试放宽服务端排空语义。 + +## 尚未覆盖与下一步 + +本轮故障恢复由测试驱动器/示例显式调用底层 SDK，不应描述为自动重连。业务函数调用一次的断言限定在示例驱动流程，不能推导任意外部系统的 exactly-once 效果。 + +下一步实现自动 Worker 运行器：处理器注册与并发限制、自动心跳/续租、单调期限取消、同进程断线恢复、稳定结果重报、处理器 panic/未知效果分类、非协作处理器与有界排空退出。之后继续补齐尝试/证据查询、完整 WS Schema 和 P2 阶段验收。 + +进程崩溃、宿主故障、数据库故障转移、生产 TLS 终止及高并发容量验证仍在后续范围。本地受控 WSS 中断不等同于跨故障域高可用验收。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 8342f63..00f9047 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -1,8 +1,8 @@ # Go SDK 接入说明 -已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询。Worker 和流程构建接口按 [开发计划](../08-development-plan.md) 继续实现。 +已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询，以及 [Worker 底层协议 SDK](worker.md)。自动 Worker 运行器和流程构建接口按 [开发计划](../08-development-plan.md) 继续实现。 -module：`git.yuntuo.hk/k2safe/orbit/sdk/go`。SDK 使用独立 go.mod，当前仅依赖 Go 标准库，不引用服务端模型或数据库实现。 +module：`git.yuntuo.hk/k2safe/orbit/sdk/go`。SDK 使用独立 go.mod，依赖 Go 标准库与 `github.com/coder/websocket v1.8.15`，不引用服务端模型或数据库实现。 ## 1. 从初始化到运行示例 @@ -133,4 +133,4 @@ go mod tidy ## 6. 后续交付 -P2 提供 Worker 注册、WSS、能力/容量、执行确认、续租和排空；P3 提供计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 +P2 已提供 Worker 底层注册、WSS、能力/容量、执行确认、续租、恢复和排空接口及 [可运行示例](worker.md)，继续开发自动 Worker 运行器；P3 提供计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 diff --git a/docs/sdk/worker.md b/docs/sdk/worker.md new file mode 100644 index 0000000..9ecb70d --- /dev/null +++ b/docs/sdk/worker.md @@ -0,0 +1,123 @@ +# Go Worker 底层协议 SDK + +更新：2026-09-17。已实现可独立构建的 `WorkerConnection`，配合真实 Orbit 服务和 MySQL 通过控制操作、WSS、ACK 丢失与租约过期回归。**自动 Worker 运行器仍在开发**：本页的方法不会自动启动处理器、心跳、续租、重连或排空。需要持续消费的接入方应等待运行器交付，不能把一次性示例当作完整托管运行时。 + +源码：[连接与控制方法](../../sdk/go/worker_connection.go)、[中文注释的数据契约](../../sdk/go/worker_types.go)。传输字段、权限和事务语义见 [Worker 接口](../api/worker.md)。SDK 独立 module，只依赖标准库和锁定版本的 `github.com/coder/websocket v1.8.15`，不导入服务端包。 + +## 1. 可运行的文本规范化示例 + +需要已完成迁移、已启动的 Orbit 和具有 `workers:connect` / `workers:consume` 权限的专用凭据。在仓库根目录运行： + +```sh +mkdir -p .local +chmod 700 .local +go run ./cmd/orbit-admin bootstrap \ + --application worker-example --namespace default --environment development \ + --queue normalization --worker-access --output .local/worker.credentials.json +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/worker.credentials.json \ + go run ./examples/worker-protocol +``` + +预期输出： + +```text +task=<32位任务ID> final_state=SUCCEEDED attempts=1 result={"text":"HELLO ORBIT"} +``` + +[完整源码](../../sdk/go/examples/worker-protocol/main.go) 实际提交 `{"text":" Hello Orbit "}`，领取并取得开始授权，执行去首尾空格/转大写，报告结果、查询当前状态并排空。没有生成文件或调用外部服务。每次运行创建独立任务与临时处理器名，避免消费前次示例遗留任务；历史和审计保留，不硬删除。 + +示例最多对开始/结果操作重连重报两次，保持原进程身份、尝试证明、消息 ID 和内容；业务函数只在首次取得有效开始确认后调用一次。网络结果仍不明时非零退出，不能推断服务端回滚。身份与待确认结果只保留在当前进程内，进程被终止后的未知结果由服务端回收进入核查。本示例处理很短的纯计算，不演示长任务自动续租。 + +## 2. 创建进程身份与连接 + +以下为接入函数内片段；完整错误处理见示例。 + +```go +registration, err := orbit.NewWorkerRegistration(queueID, 4, []orbit.WorkerCapability{ + {Handler: "text.normalize", HandlerVersion: "v1", PayloadVersion: "v1"}, +}) +if err != nil { return err } + +conn, err := client.DialWorker(ctx, registration) +if err != nil { return err } +defer conn.Close() +``` + +同一进程只生成一次 registration；重连复用，进程重启换新值。它固定凭据、队列、能力集合和实例秘密。不要把身份和秘密复制给另一个活跃进程：新会话会隔离旧会话。 + +`DialWorker` 的 ctx 只控制握手，最长 8 秒，调用方配置的 HTTP 超时可以更短；成功返回后需主动 Close。URL 基于 `NewClient` 的 HTTP(S) 地址，分别建立 WS/WSS，禁止重定向，不关闭 TLS 证书校验，不把 token 放入 URL。只支持 `pulse.worker.v1` 子协议，禁用压缩。 + +`Welcome()` 返回首次连接快照；`Done()` 关闭表示读取循环已退出；`Err()` 返回关闭原因。`Close()` 立即停止传输，不撤销外部已发出的请求，也不代替排空确认。 + +注册和尝试证明包含秘密，JSON 序列化只用于协议。SDK 的 `String` / `GoString` 隐藏注册、尝试、派发和结果中的秘密/业务数据；直接读字段、JSON 日志和调试器仍可能泄露，接入方应只记录任务 ID、尝试 ID 和错误分类。 + +## 3. 方法清单 + +除 Welcome/Done/Err/Close，所有连接操作第一个参数都是 context。控制操作同时要求调用方提供 32 位小写十六进制 message ID。 + +| 方法 | 输入 | 返回 | 注意事项 | +| --- | --- | --- | --- | +| `NewWorkerRegistration` | queueID、capacity、capabilities | WorkerRegistration | 生成随机进程身份与秘密，复制/排序能力，拒绝重复能力 | +| `NewWorkerMessageID` | 无 | string | 只为新的逻辑操作生成；未知结果重报复用旧值 | +| `client.DialWorker` | registration | *WorkerConnection | 握手后尚未启用派发 | +| `conn.Capacity` | messageID、WorkerCapacity | WorkerWelcome | 总容量覆盖值；同时续期会话 | +| `conn.NextAssignment` | ctx | WorkerAssignment | 只表示预占，不能直接运行处理器 | +| `conn.Start` | messageID、WorkerAttempt | WorkerCommandResult | 持久化开始后确认，检查 Lease 的当前授权 | +| `conn.Renew` | messageID、WorkerAttempt | WorkerCommandResult | 新续租用新消息；原消息只核对原操作 | +| `conn.Finish` | messageID、WorkerFinish | WorkerCommandResult | 区分权威结果 APPLIED 与证据 EVIDENCE_ONLY | +| `conn.Resume` | messageID、WorkerAttempt | WorkerLease | 重连核对原尝试，不创建尝试、不延长租约 | + +同连接内 `Capacity.Sequence` 从 1 开始递增；同序号必须同容量/排空标志，不能用增量值。容量为零停止新领取，不释放在途许可。首次确认容量才开启派发，所以重连时可以先 Resume 所有已知尝试，再发布容量。会话与任务续期是两件事：Capacity 不延长任务租约，Renew 不代替会话心跳。 + +排空使用 `Drain:true`。排空在同进程重连后保留，后续心跳也必须带 true；尝试清除会返回 `STATE_CONFLICT`。排空不禁止收集已发请求结果。 + +## 4. 当前授权与本地期限 + +```go +assignment, err := conn.NextAssignment(ctx) +if err != nil { return err } +ref := assignment.Attempt() +startID, err := orbit.NewWorkerMessageID() +if err != nil { return err } +started, err := conn.Start(ctx, startID, ref) +if err != nil { return err } +deadline, active := started.Lease.ExecutionDeadline() +if !active { return errors.New("attempt has no current execution authority") } +work, cancel := context.WithDeadline(ctx, deadline) +defer cancel() +// 将 work 与业务输入传给处理器，不传执行秘密。 +``` + +`ExecutionDeadline()` 只认可 `Active=true`、`State=RUNNING` 和仍未到期的本地期限。SDK 以服务端 `lease_until-server_time` 为窗口，扣除从调用开始起的全部耗时（含等待写入）及 250 毫秒安全余量，用本地单调时钟表示。数据库墙钟与客户端相差一天也不会额外获得一天授权。 + +只有 SDK 控制响应会设置内部本地期限；自行构造/JSON 解码的 WorkerLease 不会得到执行授权。旧 start 回执即使包含历史 RUNNING 和未来 lease_until，也不能越过当前 Lease 的失效判断。 + +此底层接口不会替调用方更新处理器上下文。长任务必须有独立的续租与到期取消机制，断线期间也要在最后确认的本地期限停止推进。新的 Resume/Start 回执不能复活一个已因租约到期而停止的处理器。Go 无法强杀不理会 context 的 goroutine，SDK 更不能撤回外部请求；业务幂等/执行代次隔离仍需在接入方落实。 + +## 5. 重报与结果分类 + +| 情况 | 正确处理 | +| --- | --- | +| Start 请求结果不明，业务尚未开始 | 保留 startID；重连后 Resume，再用原 startID 核对；核验当前授权后最多开始一次 | +| 处理器已在当前进程运行，随后断线 | 不再次调用处理器；保留它的已确认租约期限，重连核对原尝试 | +| Finish 请求结果不明 | 保留原 finishID、Outcome、Result、ErrorCode，重连核对后原样重报 | +| 明确业务成功 | `WorkerSucceeded`，不带 ErrorCode | +| 可确认的业务失败 | `WorkerFailed`，带稳定 ErrorCode | +| 超时、失联或不能确认外部效果 | `WorkerUnknown`，带稳定 ErrorCode，进入核查 | +| `Disposition=EVIDENCE_ONLY` | 内容已留作证据；不能当作任务成功，另查 GetTask 的权威状态 | + +Result 是不超过 256 KiB 的非 null JSON，可传对象引用；原始异常不应写入稳定错误分类。SDK 拒绝非法 Unicode、重复键和过深 JSON，保留数字文本；数值范围仍由服务端最终校验。 + +## 6. 错误与有界行为 + +- 握手 HTTP 拒绝返回 `*HTTPError`；业务控制拒绝返回 `*WorkerError`，通过 Code 分支处理。Error() 只显示分类，避免未知错误正文进入日志。 +- 传输中断返回 `ErrWorkerDisconnected`；消息格式/版本不兼容返回 `ErrWorkerProtocol`；主动关闭为 `ErrWorkerClosed`；超时/取消支持标准 context 错误。 +- 最多 32 个在途控制请求，达到上限返回 `ErrWorkerBusy`，请求未发送。同一消息 ID 已在途时也拒绝第二个并发等待者。 +- 派发缓冲最多 8 条；读取不及时会以 `ErrWorkerBackpressure` 关闭连接。持久化预占由服务端恢复，不无限积压内存。 +- 控制操作最长 10 秒，包含排队、发送与等待 ACK；调用方 context 可以缩短。已发送后取消/超时会关闭连接，操作是否提交需用原身份核对。 +- `NextAssignment` 的 context 取消只停止等待，不关闭连接。已关闭连接不会继续交付缓冲中的派发。 +- 入站单帧最多 512 KiB；读空闲期限 60 秒。需要调用方持续心跳，不能用此期限代替默认 45 秒会话期限。 + +以上是实现上限与正确性保护，不是容量承诺。测试命令与真实覆盖见 [P2 SDK 协议回归](../reports/phase-2-sdk-protocol.md)。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 9108074..0209d51 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -48,3 +48,6 @@ done if [ -z "$TASK_HTTP_PORT" ]; then printf '%s\n' 'Bootstrap HTTP server did not start.' >&2; exit 1; fi (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" "$TASK_GO" run ./examples/health) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/sdk.credentials.json" "$TASK_GO" run ./examples/tasks) +ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap --application sdk-worker --namespace default --environment test --queue worker-examples --worker-access --output "$TASK_TEST_DIR/worker.credentials.json" +(cd sdk/go && 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 IntegrationWorkerConnection -v .) +(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) diff --git a/sdk/go/doc.go b/sdk/go/doc.go index 4ceff13..23a6803 100644 --- a/sdk/go/doc.go +++ b/sdk/go/doc.go @@ -1,4 +1,4 @@ -// Package orbit is the standalone Go SDK for Orbit. This foundation release -// implements health calls; task, Worker and workflow APIs are planned next. -// The SDK does not import server models, migrations or storage dependencies. +// Package orbit 是 Orbit 的独立 Go SDK，提供健康、任务、队列与 Worker 底层协议。 +// WorkerConnection 不自动重连、续租或运行处理器；自动运行器与流程 SDK 仍在开发。 +// SDK 不导入服务端模型、迁移或数据库依赖。 package orbit diff --git a/sdk/go/examples/worker-protocol/main.go b/sdk/go/examples/worker-protocol/main.go new file mode 100644 index 0000000..6295d46 --- /dev/null +++ b/sdk/go/examples/worker-protocol/main.go @@ -0,0 +1,214 @@ +// worker-protocol 演示一次文本规范化任务的真实执行与原消息重报。 +// 这是底层协议示例；持续运行、自动续租与优雅退出由后续 Worker 运行器提供。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run() error { + token, queue := os.Getenv("ORBIT_TOKEN"), os.Getenv("ORBIT_QUEUE_ID") + if path := os.Getenv("ORBIT_CREDENTIALS_FILE"); path != "" { + raw, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read private credentials: %w", err) + } + var credentials struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if json.Unmarshal(raw, &credentials) != nil { + return errors.New("invalid credentials file") + } + if token == "" { + token = credentials.Token + } + if queue == "" { + queue = credentials.QueueID + } + } + client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + if err != nil { + return err + } + // 每次运行使用独立能力名，避免接走前一次示例遗留的任务。 + operation, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + handler := "text.normalize." + operation + registration, err := orbit.NewWorkerRegistration(queue, 1, []orbit.WorkerCapability{{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + accepted, err := client.SubmitTask(ctx, "normalize-"+operation, orbit.SubmitTaskRequest{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"text":" Hello Orbit "}`), RecoverySafety: orbit.RecoveryIdempotent, TaskTimeoutMS: 60000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 20000}}) + if err != nil { + return err + } + session := &demoSession{client: client, registration: registration} + defer session.close() + if err := session.connect(ctx); err != nil { + return err + } + capacityID, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + if _, err := session.conn.Capacity(ctx, capacityID, orbit.WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + return err + } + assignment, err := session.conn.NextAssignment(ctx) + if err != nil { + return err + } + if assignment.Task.ID != accepted.Task.ID { + return errors.New("unexpected task in isolated example") + } + ref := assignment.Attempt() + session.ref = &ref + startID, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + started, err := session.confirm(ctx, func(c *orbit.WorkerConnection) (orbit.WorkerCommandResult, error) { return c.Start(ctx, startID, ref) }) + if err != nil { + return err + } + deadline, active := started.Lease.ExecutionDeadline() + if !active { + return errors.New("start receipt does not grant current execution authority") + } + // 业务代码只拿任务输入；不接收注册秘密或尝试秘密。处理器只调用这一次。 + work, stop := context.WithDeadline(ctx, deadline) + result, handlerErr := normalize(work, assignment.Task.Payload) + stop() + finish := orbit.WorkerFinish{Ref: ref, Outcome: orbit.WorkerSucceeded, Result: result} + if handlerErr != nil { + // 普通异常不能证明外部效果未发生；默认报告 UNKNOWN，保留核查路径。 + finish.Outcome = orbit.WorkerUnknown + finish.ErrorCode = "HANDLER_UNCONFIRMED" + finish.Result = json.RawMessage(`{"reason":"handler did not confirm completion"}`) + } + finishID, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + confirmed, err := session.confirm(ctx, func(c *orbit.WorkerConnection) (orbit.WorkerCommandResult, error) { + return c.Finish(ctx, finishID, finish) + }) + if err != nil { + return err + } + if confirmed.Receipt.Disposition != "APPLIED" { + return errors.New("result saved as evidence only; inspect authoritative task state") + } + current, err := client.GetTask(ctx, assignment.Task.ID) + if err != nil { + return err + } + if current.State != orbit.TaskSucceeded || current.AttemptCount != 1 || string(current.Result) != string(result) { + return errors.New("task did not complete as expected") + } + drainID, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + // 新连接尚未开启容量时，Sequence=2 仍是合法递增值。 + if _, err := session.conn.Capacity(ctx, drainID, orbit.WorkerCapacity{Sequence: 2, Capacity: 0, Drain: true}); err != nil { + return err + } + fmt.Printf("task=%s final_state=%s attempts=%d result=%s\n", current.ID, current.State, current.AttemptCount, current.Result) + return nil +} + +func normalize(ctx context.Context, payload json.RawMessage) (json.RawMessage, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + var in struct { + Text string `json:"text"` + } + if err := json.Unmarshal(payload, &in); err != nil { + return nil, err + } + return json.Marshal(struct { + Text string `json:"text"` + }{Text: strings.ToUpper(strings.TrimSpace(in.Text))}) +} + +type demoSession struct { + client *orbit.Client + registration orbit.WorkerRegistration + conn *orbit.WorkerConnection + ref *orbit.WorkerAttempt +} + +func (s *demoSession) close() { + if s.conn != nil { + s.conn.Close() + } +} +func (s *demoSession) connect(ctx context.Context) error { + s.close() + c, err := s.client.DialWorker(ctx, s.registration) + if err != nil { + return err + } + s.conn = c + if s.ref != nil { + id, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + // 即使租约已失效，也可以重报原结果取得原回执或保存迟到证据。 + if _, err := c.Resume(ctx, id, *s.ref); err != nil { + return err + } + } + return nil +} + +// confirm 只重报控制操作，不再次调用业务函数；同一操作闭包固定消息 ID 与内容。 +func (s *demoSession) confirm(ctx context.Context, call func(*orbit.WorkerConnection) (orbit.WorkerCommandResult, error)) (orbit.WorkerCommandResult, error) { + var out orbit.WorkerCommandResult + var err error + for attempt := 0; attempt < 3; attempt++ { + if attempt > 0 { + select { + case <-ctx.Done(): + return out, ctx.Err() + case <-time.After(time.Duration(attempt) * 100 * time.Millisecond): + } + if err = s.connect(ctx); err != nil { + return out, err + } + } + out, err = call(s.conn) + if err == nil { + return out, nil + } + var control *orbit.WorkerError + retry := errors.Is(err, orbit.ErrWorkerDisconnected) || errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil || errors.As(err, &control) && control.Code == "STALE_SESSION" + if !retry { + return out, err + } + } + return out, fmt.Errorf("control result still unconfirmed after bounded replay: %w", err) +} diff --git a/sdk/go/go.mod b/sdk/go/go.mod index a0811c6..5e5bae5 100644 --- a/sdk/go/go.mod +++ b/sdk/go/go.mod @@ -1,3 +1,5 @@ module git.yuntuo.hk/k2safe/orbit/sdk/go go 1.27.1 + +require github.com/coder/websocket v1.8.15 diff --git a/sdk/go/go.sum b/sdk/go/go.sum new file mode 100644 index 0000000..4e0a48d --- /dev/null +++ b/sdk/go/go.sum @@ -0,0 +1,2 @@ +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= diff --git a/sdk/go/worker_codec.go b/sdk/go/worker_codec.go new file mode 100644 index 0000000..70eae93 --- /dev/null +++ b/sdk/go/worker_codec.go @@ -0,0 +1,138 @@ +package orbit + +import ( + "bytes" + "encoding/json" + "io" + "strconv" + "unicode/utf8" + + "github.com/coder/websocket" +) + +func parseWorkerFrame(kind websocket.MessageType, raw []byte) (workerFrame, error) { + var f workerFrame + if kind != websocket.MessageText || len(raw) > 512<<10 || validateWorkerJSON(raw) != nil { + return f, ErrWorkerProtocol + } + // Envelope 字段严格区分大小写，防止 encoding/json 的大小写别名覆盖控制字段。 + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil || fields == nil { + return f, ErrWorkerProtocol + } + for k := range fields { + switch k { + case "protocol_version", "type", "message_id", "reply_to", "session_id", "body", "error": + default: + return f, ErrWorkerProtocol + } + } + if json.Unmarshal(raw, &f) != nil || f.Version != "1" || !validID(f.MessageID) || f.ReplyTo != "" && !validID(f.ReplyTo) || f.SessionID != "" && !validID(f.SessionID) { + return f, ErrWorkerProtocol + } + if f.Type == "error" { + if f.Error == nil || !workerName(f.Error.Code, 64) || len(f.Error.Message) > 1024 || len(f.Body) > 0 { + return f, ErrWorkerProtocol + } + } else if f.Error != nil || len(f.Body) == 0 || bytes.Equal(f.Body, []byte("null")) { + return f, ErrWorkerProtocol + } + return f, nil +} + +// validateWorkerJSON 拒绝重复键、过深嵌套及非法 Unicode，保留数字原文。 +// 数值的业务范围仍由服务端规范化器最终验证；SDK 不经由 float64 重编码。 +func validateWorkerJSON(raw []byte) error { + if len(raw) == 0 || !utf8.Valid(raw) || !json.Valid(raw) || !workerUnicode(raw) { + return ErrWorkerProtocol + } + d := json.NewDecoder(bytes.NewReader(raw)) + d.UseNumber() + var value func(int) error + value = func(depth int) error { + if depth > 32 { + return ErrWorkerProtocol + } + t, err := d.Token() + if err != nil { + return err + } + switch t { + case json.Delim('{'): + keys := map[string]bool{} + for d.More() { + k, err := d.Token() + if err != nil { + return err + } + s, ok := k.(string) + if !ok || keys[s] { + return ErrWorkerProtocol + } + keys[s] = true + if err := value(depth + 1); err != nil { + return err + } + } + _, err = d.Token() + return err + case json.Delim('['): + for d.More() { + if err := value(depth + 1); err != nil { + return err + } + } + _, err = d.Token() + return err + } + return nil + } + if err := value(0); err != nil { + return err + } + if _, err := d.Token(); err != io.EOF { + return ErrWorkerProtocol + } + return nil +} + +// JSON 的未配对 UTF-16 转义不能被静默替换为 U+FFFD。 +func workerUnicode(raw []byte) bool { + inside := false + for i := 0; i < len(raw); i++ { + if raw[i] == '"' { + inside = !inside + continue + } + if !inside || raw[i] != '\\' { + continue + } + i++ + if i >= len(raw) { + return false + } + if raw[i] != 'u' { + continue + } + if i+4 >= len(raw) { + return false + } + n, err := strconv.ParseUint(string(raw[i+1:i+5]), 16, 16) + if err != nil { + return false + } + i += 4 + if n < 0xd800 || n > 0xdfff { + continue + } + if n > 0xdbff || i+6 >= len(raw) || raw[i+1] != '\\' || raw[i+2] != 'u' { + return false + } + low, err := strconv.ParseUint(string(raw[i+3:i+7]), 16, 16) + if err != nil || low < 0xdc00 || low > 0xdfff { + return false + } + i += 6 + } + return !inside +} diff --git a/sdk/go/worker_connection.go b/sdk/go/worker_connection.go new file mode 100644 index 0000000..fa9cf53 --- /dev/null +++ b/sdk/go/worker_connection.go @@ -0,0 +1,418 @@ +package orbit + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "sync" + "time" + + "github.com/coder/websocket" +) + +const workerSubprotocol = "pulse.worker.v1" +const workerMaxPending = 32 +const workerSafetyMargin = 250 * time.Millisecond + +var ( + ErrWorkerClosed = errors.New("Worker connection closed") + ErrWorkerDisconnected = errors.New("Worker connection interrupted; reconcile before replay") + ErrWorkerProtocol = errors.New("invalid Worker protocol response") + ErrWorkerBusy = errors.New("Worker control request limit reached") + ErrWorkerBackpressure = errors.New("Worker assignment buffer exhausted") +) + +// WorkerError 是服务端公开的控制错误；不会把错误解释为业务没有生效。 +type WorkerError struct { + Code string `json:"code"` // Code 稳定分类，例如 LEASE_LOST、STALE_SESSION。 + Message string `json:"message"` // Message 公开说明，不应包含内部异常正文。 +} + +func (e *WorkerError) Error() string { return "orbit Worker " + e.Code } + +type workerFrame struct { + Version string `json:"protocol_version"` + Type string `json:"type"` + MessageID string `json:"message_id"` + ReplyTo string `json:"reply_to,omitempty"` + SessionID string `json:"session_id,omitempty"` + Body json.RawMessage `json:"body,omitempty"` + Error *WorkerError `json:"error,omitempty"` +} + +// WorkerConnection 是一次连接的底层协议客户端，可并发调用控制方法。 +// 本类型不自动重连、不自动续租、不运行处理器。关闭连接不等于取消业务效果。 +// 最多 32 个在途控制请求、8 个未读派发；超时关闭连接以便按原消息身份核对。 +type WorkerConnection struct { + conn *websocket.Conn + ctx context.Context + cancel context.CancelFunc + welcome WorkerWelcome + registration WorkerRegistration + assignments chan WorkerAssignment + done chan struct{} + mu sync.Mutex + err error + pending map[string]chan workerFrame +} + +// DialWorker 使用 Authorization 和固定子协议连接；ctx 只控制握手。 +// 返回后调用方必须 Close。重连使用同一 registration，再逐个 Resume 旧尝试， +// 完成核对后才发送第一次 Capacity 开启新派发。 +func (c *Client) DialWorker(ctx context.Context, registration WorkerRegistration) (*WorkerConnection, error) { + if c.token == "" { + return nil, errors.New("Orbit token is required for Worker APIs") + } + r, err := normalizeRegistration(registration) + if err != nil { + return nil, err + } + handshake, stop := context.WithTimeout(ctx, 8*time.Second) + defer stop() + header := http.Header{} + header.Set("Authorization", "Bearer "+c.token) + conn, response, err := websocket.Dial(handshake, c.baseURL+"/v1/worker/connect", &websocket.DialOptions{HTTPClient: c.http, HTTPHeader: header, Subprotocols: []string{workerSubprotocol}, CompressionMode: websocket.CompressionDisabled}) + if err != nil { + if response != nil && response.StatusCode != http.StatusSwitchingProtocols { + return nil, &HTTPError{StatusCode: response.StatusCode} + } + return nil, workerIOError(err) + } + ok := false + defer func() { + if !ok { + conn.CloseNow() + } + }() + if conn.Subprotocol() != workerSubprotocol { + return nil, ErrWorkerProtocol + } + conn.SetReadLimit(512 << 10) + id, err := NewWorkerMessageID() + if err != nil { + return nil, err + } + body, _ := json.Marshal(r) + frame, _ := json.Marshal(workerFrame{Version: "1", Type: "worker.hello", MessageID: id, Body: body}) + if err = conn.Write(handshake, websocket.MessageText, frame); err != nil { + return nil, workerIOError(err) + } + kind, raw, err := conn.Read(handshake) + if err != nil { + return nil, workerIOError(err) + } + f, err := parseWorkerFrame(kind, raw) + if err != nil || f.ReplyTo != id { + return nil, ErrWorkerProtocol + } + if f.Type == "error" { + return nil, f.Error + } + var welcome WorkerWelcome + if f.Type != "worker.welcome" || json.Unmarshal(f.Body, &welcome) != nil || !validWelcome(welcome) || welcome.SessionID != f.SessionID { + return nil, ErrWorkerProtocol + } + lifetime, cancel := context.WithCancel(context.Background()) + w := &WorkerConnection{conn: conn, ctx: lifetime, cancel: cancel, welcome: welcome, registration: r, assignments: make(chan WorkerAssignment, 8), done: make(chan struct{}), pending: make(map[string]chan workerFrame)} + ok = true + go w.readLoop() + return w, nil +} + +// Welcome 返回本次连接最初的确认快照；最新心跳确认由 Capacity 返回。 +func (w *WorkerConnection) Welcome() WorkerWelcome { return w.welcome } + +// Done 在读取循环退出并关闭底层连接后关闭；可用于触发重新连接。 +func (w *WorkerConnection) Done() <-chan struct{} { return w.done } + +// Err 返回连接关闭的原因；未关闭时返回 nil，不含帧正文或秘密。 +func (w *WorkerConnection) Err() error { w.mu.Lock(); defer w.mu.Unlock(); return w.err } + +// Close 立即关闭传输并等待读取循环退出；排空需先调用 Capacity(Drain:true)。 +func (w *WorkerConnection) Close() error { w.fail(ErrWorkerClosed); <-w.done; return nil } + +func (w *WorkerConnection) fail(err error) { + w.mu.Lock() + if w.err != nil { + w.mu.Unlock() + return + } + w.err = err + w.cancel() + w.mu.Unlock() + _ = w.conn.CloseNow() +} + +func (w *WorkerConnection) readLoop() { + defer close(w.done) + for { + read, stop := context.WithTimeout(w.ctx, 60*time.Second) + kind, raw, err := w.conn.Read(read) + stop() + if err != nil { + w.fail(workerIOError(err)) + return + } + f, err := parseWorkerFrame(kind, raw) + if err != nil || f.SessionID != w.welcome.SessionID { + w.fail(ErrWorkerProtocol) + return + } + if f.Type == "task.assign" { + var a WorkerAssignment + if f.ReplyTo != "" || json.Unmarshal(f.Body, &a) != nil || !w.validAssignment(a) { + w.fail(ErrWorkerProtocol) + return + } + select { + case w.assignments <- a: + default: + w.fail(ErrWorkerBackpressure) + return + } + continue + } + if f.Type != "ack" && f.Type != "error" { + w.fail(ErrWorkerProtocol) + return + } + if f.ReplyTo == "" && f.Type == "error" { + w.fail(f.Error) + return + } + w.mu.Lock() + waiter := w.pending[f.ReplyTo] + w.mu.Unlock() + if waiter == nil { + w.fail(ErrWorkerProtocol) + return + } + select { + case waiter <- f: + default: + w.fail(ErrWorkerProtocol) + return + } + if f.Type == "error" && (f.Error.Code == "STALE_SESSION" || f.Error.Code == "UNAUTHENTICATED" || f.Error.Code == "PERMISSION_DENIED") { + w.fail(f.Error) + return + } + } +} + +// NextAssignment 等待预占通知；即使返回成功，也必须通过 Start 确认当前执行权。 +// ctx 到期只中断等待，不关闭连接；读取过慢触发有界缓冲保护时连接会关闭。 +func (w *WorkerConnection) NextAssignment(ctx context.Context) (WorkerAssignment, error) { + if err := ctx.Err(); err != nil { + return WorkerAssignment{}, err + } + select { + case <-ctx.Done(): + return WorkerAssignment{}, ctx.Err() + case <-w.ctx.Done(): + return WorkerAssignment{}, w.Err() + case a := <-w.assignments: + if err := w.Err(); err != nil { + return WorkerAssignment{}, err + } + return a, nil + } +} + +func (w *WorkerConnection) call(ctx context.Context, kind, id string, in, out any) (time.Time, error) { + started := time.Now() + if !validID(id) { + return started, errors.New("invalid Worker message id") + } + if err := ctx.Err(); err != nil { + return started, err + } + waiter := make(chan workerFrame, 1) + w.mu.Lock() + if w.err != nil { + err := w.err + w.mu.Unlock() + return started, err + } + if len(w.pending) >= workerMaxPending { + w.mu.Unlock() + return started, ErrWorkerBusy + } + if _, ok := w.pending[id]; ok { + w.mu.Unlock() + return started, errors.New("Worker message id is already in flight") + } + w.pending[id] = waiter + w.mu.Unlock() + defer func() { w.mu.Lock(); delete(w.pending, id); w.mu.Unlock() }() + // 先占用有界槽再分配编码缓冲，超额调用不先复制大结果。 + body, err := json.Marshal(in) + if err != nil { + return started, errors.New("invalid Worker request JSON") + } + raw, err := json.Marshal(workerFrame{Version: "1", Type: kind, MessageID: id, SessionID: w.welcome.SessionID, Body: body}) + if err != nil || len(raw) > 512<<10 { + return started, errors.New("Worker request exceeds frame limit") + } + bounded, stop := context.WithDeadline(ctx, started.Add(10*time.Second)) + defer stop() + // coder/websocket 串行化并发 Write；排队时间包含在操作总时限和租约余量中。 + if err = w.conn.Write(bounded, websocket.MessageText, raw); err != nil { + err = workerIOError(err) + w.fail(err) + return started, err + } + var f workerFrame + select { + case f = <-waiter: + case <-w.ctx.Done(): + // 致命控制错误先交付给相应调用，再关闭连接。 + select { + case f = <-waiter: + default: + return started, w.Err() + } + case <-bounded.Done(): + w.fail(bounded.Err()) + return started, bounded.Err() + } + if f.Error != nil { + return started, f.Error + } + if json.Unmarshal(f.Body, out) != nil { + w.fail(ErrWorkerProtocol) + return started, ErrWorkerProtocol + } + return started, nil +} + +// Capacity 确认绝对容量，同时续期会话。消息身份和 Sequence 在重报时保持不变。 +func (w *WorkerConnection) Capacity(ctx context.Context, id string, in WorkerCapacity) (WorkerWelcome, error) { + var out WorkerWelcome + if in.Sequence == 0 || in.Capacity > 256 { + return out, errors.New("invalid Worker capacity or sequence") + } + _, err := w.call(ctx, "worker.capacity", id, in, &out) + if err == nil && (!validWelcome(out) || out.SessionID != w.welcome.SessionID || out.Epoch != w.welcome.Epoch) { + err = ErrWorkerProtocol + w.fail(err) + } + return out, err +} + +// Resume 核对原尝试并迁移到当前会话，不创建尝试、不续期，也不授权再次运行处理器。 +func (w *WorkerConnection) Resume(ctx context.Context, id string, ref WorkerAttempt) (WorkerLease, error) { + var out WorkerLease + if !validWorkerAttempt(ref) { + return out, errors.New("invalid Worker attempt proof") + } + started, err := w.call(ctx, "worker.resume", id, ref, &out) + if err == nil { + err = out.confirm(started) + if err != nil { + w.fail(err) + } + } + return out, err +} + +// Start 持久化开始授权；只在返回 Lease.ExecutionDeadline 为真时运行一次处理器。 +func (w *WorkerConnection) Start(ctx context.Context, id string, ref WorkerAttempt) (WorkerCommandResult, error) { + return w.command(ctx, "start", id, ref, ref) +} + +// Renew 续期已运行尝试；新续租用新消息身份，响应不明时原身份只核对原操作。 +func (w *WorkerConnection) Renew(ctx context.Context, id string, ref WorkerAttempt) (WorkerCommandResult, error) { + return w.command(ctx, "renew", id, ref, ref) +} + +// Finish 保存业务结果/证据；迟到结果可返回 EVIDENCE_ONLY，不能视为任务成功终结。 +func (w *WorkerConnection) Finish(ctx context.Context, id string, in WorkerFinish) (WorkerCommandResult, error) { + var out WorkerCommandResult + if (in.Outcome != WorkerSucceeded && in.Outcome != WorkerFailed && in.Outcome != WorkerUnknown) || len(in.Result) > 256<<10 || bytes.Equal(bytes.TrimSpace(in.Result), []byte("null")) || validateWorkerJSON(in.Result) != nil { + return out, errors.New("invalid Worker result outcome or JSON") + } + if in.Outcome == WorkerSucceeded && in.ErrorCode != "" || in.Outcome != WorkerSucceeded && !workerName(in.ErrorCode, 64) { + return out, errors.New("Worker error code must match outcome") + } + return w.command(ctx, "finish", id, in.Ref, in) +} +func (w *WorkerConnection) command(ctx context.Context, op, id string, ref WorkerAttempt, in any) (WorkerCommandResult, error) { + var out WorkerCommandResult + if !validWorkerAttempt(ref) { + return out, errors.New("invalid Worker attempt proof") + } + started, err := w.call(ctx, "task."+op, id, in, &out) + if err == nil && (out.Receipt.Operation != op || out.Receipt.TaskRevision == 0 || !workerTaskState(out.Receipt.TaskState) || !workerAttemptState(out.Receipt.AttemptState) || out.Receipt.Disposition != "APPLIED" && out.Receipt.Disposition != "EVIDENCE_ONLY") { + err = ErrWorkerProtocol + } + if err == nil { + err = out.Lease.confirm(started) + } + if errors.Is(err, ErrWorkerProtocol) { + w.fail(err) + } + return out, err +} + +func (w *WorkerConnection) validAssignment(a WorkerAssignment) bool { + if !validWorkerAttempt(a.Attempt()) || a.Task.QueueID != w.registration.QueueID || !validID(a.Task.NamespaceID) || a.Task.State != TaskLeased || a.Task.Revision == 0 || a.LeaseUntil.IsZero() || a.ServerTime.IsZero() || len(a.Task.Payload) == 0 || bytes.Equal(bytes.TrimSpace(a.Task.Payload), []byte("null")) { + return false + } + for _, c := range w.registration.Capabilities { + if c.Handler == a.Task.Handler && c.HandlerVersion == a.Task.HandlerVersion && c.PayloadVersion == a.Task.PayloadVersion { + return true + } + } + return false +} +func validWelcome(w WorkerWelcome) bool { + return validID(w.SessionID) && w.Epoch > 0 && !w.ServerTime.IsZero() && w.ExpiresAt.After(w.ServerTime) && w.ExpiresAt.Sub(w.ServerTime) <= 5*time.Minute && w.LeaseDurationMS >= 1000 && w.LeaseDurationMS <= 120000 +} +func (l *WorkerLease) confirm(started time.Time) error { + l.deadline = time.Time{} + if l.ServerTime.IsZero() || !workerAttemptState(l.State) { + return ErrWorkerProtocol + } + if !l.Active { + return nil + } + window := l.LeaseUntil.Sub(l.ServerTime) + if (l.State != "LEASED" && l.State != "RUNNING") || window <= 0 || window > 2*time.Minute { + return ErrWorkerProtocol + } + // 整个往返（包含 SDK 排队）都扣除；只使用服务端时间差，不比较两端墙钟。 + l.deadline = started.Add(window - workerSafetyMargin) + return nil +} +func workerAttemptState(s string) bool { + switch s { + case "LEASED", "RUNNING", "SUCCEEDED", "FAILED", "UNKNOWN", "EXPIRED", "CANCELLED": + return true + } + return false +} +func workerTaskState(s TaskState) bool { + switch s { + case TaskWaiting, TaskLeased, TaskRunning, TaskSucceeded, TaskFailed, TaskCancelled, TaskNeedsReview: + return true + } + return false +} +func workerIOError(err error) error { + if errors.Is(err, context.Canceled) { + return context.Canceled + } + if errors.Is(err, context.DeadlineExceeded) { + return context.DeadlineExceeded + } + return ErrWorkerDisconnected +} + +func (w *WorkerConnection) String() string { + return fmt.Sprintf("WorkerConnection{SessionID:%s secrets:redacted}", w.welcome.SessionID) +} +func (w *WorkerConnection) GoString() string { return w.String() } diff --git a/sdk/go/worker_connection_test.go b/sdk/go/worker_connection_test.go new file mode 100644 index 0000000..8687258 --- /dev/null +++ b/sdk/go/worker_connection_test.go @@ -0,0 +1,424 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +const workerTestQueue = "11111111111111111111111111111111" +const workerTestSession = "22222222222222222222222222222222" +const workerTestTask = "33333333333333333333333333333333" +const workerTestAttempt = "44444444444444444444444444444444" + +func workerTestID(t *testing.T) string { + t.Helper() + s, err := NewWorkerMessageID() + if err != nil { + t.Fatal(err) + } + return s +} +func workerTestRegistration(t *testing.T) WorkerRegistration { + t.Helper() + r, err := NewWorkerRegistration(workerTestQueue, 2, []WorkerCapability{{Handler: "text.normalize", HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + t.Fatal(err) + } + return r +} +func workerTestWelcome() WorkerWelcome { + return WorkerWelcome{SessionID: workerTestSession, Epoch: 1, ServerTime: time.Now().UTC(), ExpiresAt: time.Now().UTC().Add(45 * time.Second), LeaseDurationMS: 30000} +} +func workerTestAssignment(r WorkerRegistration) WorkerAssignment { + return WorkerAssignment{Task: Task{ID: workerTestTask, NamespaceID: workerTestQueue, QueueID: workerTestQueue, State: TaskLeased, Revision: 2, Handler: r.Capabilities[0].Handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"amount":9007199254740993}`)}, AttemptID: workerTestAttempt, Epoch: 1, LeaseSecret: r.InstanceSecret, ServerTime: time.Now().UTC(), LeaseUntil: time.Now().UTC().Add(30 * time.Second)} +} +func workerTestRead(t *testing.T, ctx context.Context, c *websocket.Conn) workerFrame { + t.Helper() + kind, raw, err := c.Read(ctx) + if err != nil { + t.Errorf("read test frame: %v", err) + return workerFrame{} + } + var f workerFrame + if kind != websocket.MessageText || json.Unmarshal(raw, &f) != nil { + t.Error("invalid client frame") + } + return f +} +func workerTestWrite(t *testing.T, ctx context.Context, c *websocket.Conn, f workerFrame) { + t.Helper() + f.Version = "1" + f.SessionID = workerTestSession + f.MessageID = workerTestID(t) + raw, err := json.Marshal(f) + if err != nil { + t.Error(err) + return + } + if err := c.Write(ctx, websocket.MessageText, raw); err != nil { + t.Errorf("write test frame: %v", err) + } +} +func workerTestBody(t *testing.T, v any) json.RawMessage { + t.Helper() + raw, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + return raw +} + +func workerMock(t *testing.T, r WorkerRegistration, serve func(context.Context, *websocket.Conn)) (*Client, *WorkerConnection) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.URL.Path != "/v1/worker/connect" || req.Header.Get("Authorization") != "Bearer unit-test-token" || req.URL.RawQuery != "" { + t.Error("unexpected handshake") + } + c, err := websocket.Accept(w, req, &websocket.AcceptOptions{Subprotocols: []string{workerSubprotocol}}) + if err != nil { + t.Error(err) + return + } + defer c.CloseNow() + f := workerTestRead(t, ctx, c) + var got WorkerRegistration + if f.Type != "worker.hello" || f.SessionID != "" || !validID(f.MessageID) || json.Unmarshal(f.Body, &got) != nil || got.InstanceSecret != r.InstanceSecret || got.InstanceID != r.InstanceID { + t.Error("invalid hello") + } + workerTestWrite(t, ctx, c, workerFrame{Type: "worker.welcome", ReplyTo: f.MessageID, Body: workerTestBody(t, workerTestWelcome())}) + serve(ctx, c) + })) + t.Cleanup(func() { cancel(); server.Close() }) + client, err := NewClient(server.URL, Options{Token: "unit-test-token", HTTPClient: server.Client()}) + if err != nil { + t.Fatal(err) + } + w, err := client.DialWorker(ctx, r) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { w.Close() }) + return client, w +} + +func TestWorkerProtocolMultiplexAndMonotonicLease(t *testing.T) { + r := workerTestRegistration(t) + a := workerTestAssignment(r) + _, w := workerMock(t, r, func(ctx context.Context, c *websocket.Conn) { + capacity := workerTestRead(t, ctx, c) + workerTestWrite(t, ctx, c, workerFrame{Type: "task.assign", Body: workerTestBody(t, a)}) + workerTestWrite(t, ctx, c, workerFrame{Type: "ack", ReplyTo: capacity.MessageID, Body: workerTestBody(t, workerTestWelcome())}) + first, second := workerTestRead(t, ctx, c), workerTestRead(t, ctx, c) + for _, f := range []workerFrame{second, first} { + // 模拟数据库墙钟相差 24 小时；SDK 只能根据时间差确定单调期限。 + now := time.Now().UTC().Add(24 * time.Hour) + result := WorkerCommandResult{Receipt: WorkerReceipt{Operation: strings.TrimPrefix(f.Type, "task."), TaskState: TaskRunning, AttemptState: "RUNNING", TaskRevision: 9007199254740993, Disposition: "APPLIED", LeaseUntil: now.Add(30 * time.Second)}, Lease: WorkerLease{Active: true, State: "RUNNING", ServerTime: now, LeaseUntil: now.Add(30 * time.Second)}} + workerTestWrite(t, ctx, c, workerFrame{Type: "ack", ReplyTo: f.MessageID, Body: workerTestBody(t, result)}) + } + <-ctx.Done() + }) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 2}); err != nil { + t.Fatal(err) + } + got, err := w.NextAssignment(ctx) + if err != nil { + t.Fatal(err) + } + if string(got.Task.Payload) != string(a.Task.Payload) { + t.Fatal("payload precision changed") + } + var wg sync.WaitGroup + for range 2 { + wg.Add(1) + go func() { + defer wg.Done() + out, err := w.Start(ctx, workerTestID(t), a.Attempt()) + if err != nil { + t.Error(err) + return + } + deadline, active := out.Lease.ExecutionDeadline() + remaining := time.Until(deadline) + if !active || remaining < 29*time.Second || remaining > 30*time.Second || out.Receipt.TaskRevision != 9007199254740993 { + t.Error("wrong ACK correlation, precise revision or monotonic deadline") + } + }() + } + wg.Wait() +} + +func TestWorkerLateACKDoesNotAuthorize(t *testing.T) { + r := workerTestRegistration(t) + a := workerTestAssignment(r) + _, w := workerMock(t, r, func(ctx context.Context, c *websocket.Conn) { + f := workerTestRead(t, ctx, c) + now := time.Now().UTC() + result := WorkerCommandResult{Receipt: WorkerReceipt{Operation: "start", TaskState: TaskRunning, AttemptState: "RUNNING", TaskRevision: 3, Disposition: "APPLIED", LeaseUntil: now.Add(time.Hour)}, Lease: WorkerLease{State: "UNKNOWN", ServerTime: now, LeaseUntil: now.Add(-time.Second)}, Deduplicated: true} + workerTestWrite(t, ctx, c, workerFrame{Type: "ack", ReplyTo: f.MessageID, Body: workerTestBody(t, result)}) + <-ctx.Done() + }) + out, err := w.Start(context.Background(), workerTestID(t), a.Attempt()) + if err != nil { + t.Fatal(err) + } + if !out.Deduplicated { + t.Fatal("missing original receipt") + } + if _, active := out.Lease.ExecutionDeadline(); active { + t.Fatal("historical ACK authorized execution") + } + manual := WorkerLease{Active: true, State: "RUNNING", LeaseUntil: time.Now().Add(time.Minute), ServerTime: time.Now()} + if _, active := manual.ExecutionDeadline(); active { + t.Fatal("unconfirmed lease authorized execution") + } +} + +func TestWorkerBoundedRequestAndCancellation(t *testing.T) { + r := workerTestRegistration(t) + received := make(chan workerFrame, 64) + _, w := workerMock(t, r, func(ctx context.Context, c *websocket.Conn) { + for { + _, raw, err := c.Read(ctx) + if err != nil { + return + } + var f workerFrame + if json.Unmarshal(raw, &f) != nil { + return + } + received <- f + } + }) + ctx, cancel := context.WithCancel(context.Background()) + errs := make(chan error, workerMaxPending) + id := workerTestID(t) + go func() { _, err := w.Capacity(ctx, id, WorkerCapacity{Sequence: 1, Capacity: 1}); errs <- err }() + <-received + if _, err := w.Capacity(ctx, id, WorkerCapacity{Sequence: 1, Capacity: 1}); err == nil { + t.Fatal("duplicate in-flight message accepted") + } + for i := 1; i < workerMaxPending; i++ { + go func(seq uint64) { + _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: seq, Capacity: 1}) + errs <- err + }(uint64(i + 1)) + } + for i := 1; i < workerMaxPending; i++ { + <-received + } + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 100, Capacity: 1}); !errors.Is(err, ErrWorkerBusy) { + t.Fatalf("want bounded queue, got %v", err) + } + cancel() + for range workerMaxPending { + select { + case err := <-errs: + if err == nil { + t.Error("pending operation unexpectedly confirmed") + } + case <-time.After(time.Second): + t.Fatal("pending operation did not exit") + } + } + select { + case <-w.Done(): + case <-time.After(time.Second): + t.Fatal("connection leaked on cancellation") + } +} + +func TestWorkerNextAssignmentCancelAndBackpressure(t *testing.T) { + r := workerTestRegistration(t) + send := make(chan struct{}) + _, w := workerMock(t, r, func(ctx context.Context, c *websocket.Conn) { + <-send + for i := 0; i < 9; i++ { + workerTestWrite(t, ctx, c, workerFrame{Type: "task.assign", Body: workerTestBody(t, workerTestAssignment(r))}) + } + <-ctx.Done() + }) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := w.NextAssignment(ctx); !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if w.Err() != nil { + t.Fatal("cancelled wait closed connection") + } + close(send) + select { + case <-w.Done(): + case <-time.After(time.Second): + t.Fatal("unbounded assignment reader") + } + if !errors.Is(w.Err(), ErrWorkerBackpressure) { + t.Fatal(w.Err()) + } + if _, err := w.NextAssignment(context.Background()); !errors.Is(err, ErrWorkerBackpressure) { + t.Fatal("returned buffered work after close") + } +} + +func TestWorkerRedirectDoesNotForwardCredential(t *testing.T) { + var reached atomic.Int32 + destination := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { reached.Add(1) })) + defer destination.Close() + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, destination.URL, http.StatusTemporaryRedirect) + })) + defer origin.Close() + c, _ := NewClient(origin.URL, Options{Token: "private-token"}) + _, err := c.DialWorker(context.Background(), workerTestRegistration(t)) + var httpErr *HTTPError + if !errors.As(err, &httpErr) || httpErr.StatusCode != 307 || reached.Load() != 0 { + t.Fatalf("redirect followed or wrong error: %v", err) + } +} + +func TestWorkerCodecAndValidation(t *testing.T) { + for _, raw := range []string{`{"a":1,"a":2}`, `{"x":{"a":1,"a":2}}`, `{"x":"\ud800"}`, `{"x":"\udc00"}`, `{"x":"\ud800\u0041"}`, `{"x":"` + string([]byte{0xff}) + `"}`, strings.Repeat("[", 34) + "0" + strings.Repeat("]", 34), `{} {}`} { + if validateWorkerJSON([]byte(raw)) == nil { + t.Errorf("invalid JSON accepted: %q", raw) + } + } + for _, raw := range []string{`{"x":"\ud83d\ude00"}`, `{"x":"literal \\uD800 and \"quote"}`, `{"a":9007199254740993,"b":0.1234567890123456789012345,"c":[true,null]}`} { + if validateWorkerJSON([]byte(raw)) != nil { + t.Errorf("valid JSON rejected: %s", raw) + } + } + r := workerTestRegistration(t) + for _, v := range []any{r, workerTestAssignment(r), workerTestAssignment(r).Attempt(), WorkerFinish{Ref: workerTestAssignment(r).Attempt(), Result: json.RawMessage(`{"secret":"business-private"}`)}} { + for _, format := range []string{"%v", "%+v", "%#v"} { + out := fmt.Sprintf(format, v) + if strings.Contains(out, r.InstanceSecret) || strings.Contains(out, "business-private") { + t.Fatal("secret in formatted Worker value") + } + } + } + before := []WorkerCapability{{Handler: "b", HandlerVersion: "v1", PayloadVersion: "v1"}, {Handler: "a", HandlerVersion: "v1", PayloadVersion: "v1"}} + normalized, err := NewWorkerRegistration(workerTestQueue, 1, before) + if err != nil || normalized.Capabilities[0].Handler != "a" || before[0].Handler != "b" { + t.Fatal("normalization mutated caller capability slice") + } + if _, err := NewWorkerRegistration(workerTestQueue, 1, []WorkerCapability{before[0], before[0]}); err == nil { + t.Fatal("duplicate capability") + } + for _, window := range []time.Duration{0, -time.Second, 3 * time.Minute} { + lease := WorkerLease{Active: true, State: "RUNNING", ServerTime: time.Now()} + lease.LeaseUntil = lease.ServerTime.Add(window) + if lease.confirm(time.Now()) == nil { + t.Fatal("invalid active lease") + } + } + lease := WorkerLease{Active: true, State: "RUNNING", ServerTime: time.Now()} + lease.LeaseUntil = lease.ServerTime.Add(time.Second) + if err := lease.confirm(time.Now().Add(-time.Second)); err != nil { + t.Fatal(err) + } + if _, ok := lease.ExecutionDeadline(); ok { + t.Fatal("request duration not deducted") + } +} + +func TestWorkerFatalAndInvalidFrames(t *testing.T) { + for _, test := range []struct { + name string + frame workerFrame + want error + }{ + {"revoked", workerFrame{Type: "error", Error: &WorkerError{Code: "UNAUTHENTICATED", Message: "credential expired"}}, &WorkerError{Code: "UNAUTHENTICATED"}}, + {"unknown_reply", workerFrame{Type: "ack", ReplyTo: workerTestTask, Body: json.RawMessage(`{}`)}, ErrWorkerProtocol}, + {"invalid_body", workerFrame{Type: "task.assign", Body: json.RawMessage(`{}`)}, ErrWorkerProtocol}, + } { + t.Run(test.name, func(t *testing.T) { + r := workerTestRegistration(t) + _, w := workerMock(t, r, func(ctx context.Context, c *websocket.Conn) { workerTestWrite(t, ctx, c, test.frame); <-ctx.Done() }) + select { + case <-w.Done(): + case <-time.After(time.Second): + t.Fatal("invalid frame did not close") + } + var serverErr *WorkerError + if want, ok := test.want.(*WorkerError); ok { + if !errors.As(w.Err(), &serverErr) || serverErr.Code != want.Code { + t.Fatal(w.Err()) + } + } else if !errors.Is(w.Err(), test.want) { + t.Fatal(w.Err()) + } + }) + } +} + +func FuzzWorkerJSON(f *testing.F) { + for _, s := range []string{`{"x":"\ud83d\ude00"}`, `{"n":9007199254740993}`, `[true,null,{}]`, `{"a":1,"a":2}`} { + f.Add([]byte(s)) + } + f.Fuzz(func(t *testing.T, raw []byte) { + if len(raw) > 512<<10 { + t.Skip() + } + if validateWorkerJSON(raw) == nil && !json.Valid(raw) { + t.Fatal("accepted invalid JSON") + } + }) +} + +func TestWorkerConnectionRejectsWrongSessionAndBinaryFrames(t *testing.T) { + for _, kind := range []websocket.MessageType{websocket.MessageText, websocket.MessageBinary} { + t.Run(fmt.Sprint(kind), func(t *testing.T) { + r := workerTestRegistration(t) + _, w := workerMock(t, r, func(ctx context.Context, c *websocket.Conn) { + f := workerFrame{Version: "1", Type: "task.assign", MessageID: workerTestID(t), SessionID: workerTestTask, Body: workerTestBody(t, workerTestAssignment(r))} + if kind == websocket.MessageBinary { + f.SessionID = workerTestSession + } + if err := c.Write(ctx, kind, workerTestBody(t, f)); err != nil { + t.Error(err) + } + <-ctx.Done() + }) + select { + case <-w.Done(): + case <-time.After(time.Second): + t.Fatal("invalid session or binary frame did not close") + } + if !errors.Is(w.Err(), ErrWorkerProtocol) { + t.Fatal(w.Err()) + } + }) + } +} + +func TestWorkerHandshakeRequiresSubprotocol(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer c.CloseNow() + ctx, cancel := context.WithTimeout(r.Context(), time.Second) + defer cancel() + _, _, _ = c.Read(ctx) + })) + defer server.Close() + c, _ := NewClient(server.URL, Options{Token: "test-token"}) + if _, err := c.DialWorker(context.Background(), workerTestRegistration(t)); !errors.Is(err, ErrWorkerProtocol) { + t.Fatalf("missing subprotocol accepted: %v", err) + } +} diff --git a/sdk/go/worker_integration_test.go b/sdk/go/worker_integration_test.go new file mode 100644 index 0000000..586e346 --- /dev/null +++ b/sdk/go/worker_integration_test.go @@ -0,0 +1,385 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +// 由 scripts/test-mysql.sh 启动真实服务与隔离 MySQL；SDK 不导入服务端内部包。 +func integrationWorkerClient(t *testing.T) (*Client, string, string) { + t.Helper() + path := os.Getenv("ORBIT_TEST_WORKER_CREDENTIALS_FILE") + base := os.Getenv("ORBIT_TEST_WORKER_URL") + if path == "" || base == "" { + t.Skip("requires isolated Orbit server and Worker credentials") + } + if !strings.HasPrefix(base, "http://127.0.0.1:") { + t.Fatal("Worker integration requires an isolated loopback server") + } + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var credentials struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" || !validID(credentials.QueueID) { + t.Fatal("invalid integration credentials") + } + c, err := NewClient(base, Options{Token: credentials.Token}) + if err != nil { + t.Fatal(err) + } + return c, credentials.QueueID, credentials.Token +} + +// workerFaultGateway 在实际服务端确认后丢弃指定 ACK；SDK 到网关使用真实 TLS。 +// 只丢响应，不伪造任务状态，测试凭据不进入 URL 或输出。 +type workerFaultGateway struct { + drop atomic.Value + dropped chan string + server *httptest.Server +} + +func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { + t.Helper() + g := &workerFaultGateway{dropped: make(chan string, 4)} + g.drop.Store("") + ctx, cancel := context.WithCancel(context.Background()) + var handlers sync.WaitGroup + g.server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + handlers.Add(1) + defer handlers.Done() + bridge, stop := context.WithCancel(ctx) + 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}}) + if err != nil { + http.Error(w, "upstream unavailable", 503) + return + } + defer up.CloseNow() + up.SetReadLimit(512 << 10) + down, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{workerSubprotocol}}) + if err != nil { + return + } + defer down.CloseNow() + down.SetReadLimit(512 << 10) + var mu sync.Mutex + var forwarding sync.WaitGroup + forwarding.Add(2) + operations := map[string]string{} + finished := make(chan struct{}, 1) + go func() { + defer forwarding.Done() + defer func() { finished <- struct{}{} }() + for { + kind, raw, err := down.Read(bridge) + if err != nil { + return + } + var f workerFrame + if json.Unmarshal(raw, &f) != nil { + return + } + mu.Lock() + operations[f.MessageID] = f.Type + mu.Unlock() + if up.Write(bridge, kind, raw) != nil { + return + } + } + }() + go func() { + defer forwarding.Done() + defer stop() + for { + kind, raw, err := up.Read(bridge) + if err != nil { + return + } + var f workerFrame + if json.Unmarshal(raw, &f) != nil { + return + } + mu.Lock() + operation := operations[f.ReplyTo] + delete(operations, f.ReplyTo) + mu.Unlock() + if f.Type == "ack" && operation != "" && g.drop.CompareAndSwap(operation, "") { + g.dropped <- operation + return + } + if down.Write(bridge, kind, raw) != nil { + return + } + } + }() + select { + case <-bridge.Done(): + case <-finished: + } + stop() + down.CloseNow() + up.CloseNow() + forwarding.Wait() + })) + t.Cleanup(func() { cancel(); g.server.Close(); handlers.Wait() }) + return g +} +func (g *workerFaultGateway) expectDrop(t *testing.T, operation string) { + t.Helper() + select { + case got := <-g.dropped: + if got != operation { + t.Fatal("wrong dropped operation") + } + case <-time.After(2 * time.Second): + t.Fatal("ACK was not dropped after upstream confirmation") + } +} + +func TestIntegrationWorkerConnectionACKLoss(t *testing.T) { + api, queue, token := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + r, err := NewWorkerRegistration(queue, 1, []WorkerCapability{{Handler: "sdk.ack-loss", HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + t.Fatal(err) + } + g := newWorkerFaultGateway(t, api.baseURL) + client, err := NewClient(g.server.URL, Options{Token: token, HTTPClient: g.server.Client()}) + if err != nil { + t.Fatal(err) + } + dial := func() *WorkerConnection { + t.Helper() + w, err := client.DialWorker(ctx, r) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { w.Close() }) + return w + } + accepted, err := api.SubmitTask(ctx, "sdk-ack-"+workerTestID(t), SubmitTaskRequest{QueueID: queue, Handler: r.Capabilities[0].Handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"value":9007199254740993}`), TaskTimeoutMS: 60000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 30000}}) + if err != nil { + t.Fatal(err) + } + w := dial() + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + t.Fatal(err) + } + a, err := w.NextAssignment(ctx) + if err != nil { + t.Fatal(err) + } + if a.Task.ID != accepted.Task.ID { + t.Fatal("wrong assigned task") + } + startID := workerTestID(t) + g.drop.Store("task.start") + if _, err := w.Start(ctx, startID, a.Attempt()); err == nil { + t.Fatal("lost start ACK reported success") + } + g.expectDrop(t, "task.start") + w.Close() + w = dial() + lease, err := w.Resume(ctx, workerTestID(t), a.Attempt()) + if err != nil { + t.Fatal(err) + } + if _, active := lease.ExecutionDeadline(); !active { + t.Fatal("committed start did not resume") + } + started, err := w.Start(ctx, startID, a.Attempt()) + if err != nil { + t.Fatal(err) + } + if !started.Deduplicated { + t.Fatal("start receipt was recreated") + } + if _, active := started.Lease.ExecutionDeadline(); !active { + t.Fatal("resume missing current authority") + } + // 处理器仅在首次确认当前授权后运行。之后连接恢复只核对，不再次运行它。 + var handlerCalls int + handlerCalls++ + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + t.Fatal(err) + } + renewID := workerTestID(t) + first, err := w.Renew(ctx, renewID, a.Attempt()) + if err != nil { + t.Fatal(err) + } + second, err := w.Renew(ctx, renewID, a.Attempt()) + if err != nil { + t.Fatal(err) + } + if first.Deduplicated || !second.Deduplicated || first.Receipt.TaskRevision != second.Receipt.TaskRevision || !first.Receipt.LeaseUntil.Equal(second.Receipt.LeaseUntil) { + t.Fatal("duplicate renew changed original receipt") + } + finishID := workerTestID(t) + result := json.RawMessage(`{"normalized":"hello","sequence":9007199254740993,"ratio":0.1234567890123456789012345}`) + finish := WorkerFinish{Ref: a.Attempt(), Outcome: WorkerSucceeded, Result: result} + g.drop.Store("task.finish") + if _, err := w.Finish(ctx, finishID, finish); err == nil { + t.Fatal("lost finish ACK reported success") + } + g.expectDrop(t, "task.finish") + w.Close() + w = dial() + lease, err = w.Resume(ctx, workerTestID(t), a.Attempt()) + if err != nil { + t.Fatal(err) + } + if lease.Active || lease.State != "SUCCEEDED" { + t.Fatal("completed attempt resumed as executable") + } + confirmed, err := w.Finish(ctx, finishID, finish) + if err != nil { + t.Fatal(err) + } + if !confirmed.Deduplicated || confirmed.Receipt.Disposition != "APPLIED" || confirmed.Receipt.TaskState != TaskSucceeded { + t.Fatal("missing durable finish receipt") + } + stale, err := w.Start(ctx, startID, a.Attempt()) + if err != nil { + t.Fatal(err) + } + if !stale.Deduplicated { + t.Fatal("missing old start receipt") + } + if _, active := stale.Lease.ExecutionDeadline(); active { + t.Fatal("old start ACK resurrected execution") + } + conflict := finish + conflict.Result = json.RawMessage(`{"normalized":"conflicting evidence"}`) + evidence, err := w.Finish(ctx, workerTestID(t), conflict) + if err != nil { + t.Fatal(err) + } + if evidence.Receipt.Disposition != "EVIDENCE_ONLY" { + t.Fatal("conflicting result replaced authority") + } + current, err := api.GetTask(ctx, a.Task.ID) + if err != nil { + t.Fatal(err) + } + if current.State != TaskSucceeded || current.AttemptCount != 1 || string(current.Result) != string(result) || handlerCalls != 1 { + t.Fatal("result, precision or execution count changed") + } + drain, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 0, Drain: true}) + if err != nil || !drain.Draining { + t.Fatalf("drain: %v", err) + } + w.Close() + w = dial() + if !w.Welcome().Draining { + t.Fatal("drain did not persist across reconnect") + } + _, err = w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}) + var control *WorkerError + if !errors.As(err, &control) || control.Code != "STATE_CONFLICT" { + t.Fatalf("clearing persistent drain was not rejected: %v", err) + } + drain, err = w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 0, Drain: true}) + if err != nil || !drain.Draining { + t.Fatalf("draining heartbeat failed: %v", err) + } +} + +func TestIntegrationWorkerConnectionExpiredLease(t *testing.T) { + api, queue, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second) + defer cancel() + r, err := NewWorkerRegistration(queue, 1, []WorkerCapability{{Handler: "sdk.expiry", HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + t.Fatal(err) + } + _, err = api.SubmitTask(ctx, "sdk-expiry-"+workerTestID(t), SubmitTaskRequest{QueueID: queue, Handler: "sdk.expiry", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), TaskTimeoutMS: 30000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 1100}}) + if err != nil { + t.Fatal(err) + } + w, err := api.DialWorker(ctx, r) + if err != nil { + t.Fatal(err) + } + defer w.Close() + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + t.Fatal(err) + } + a, err := w.NextAssignment(ctx) + if err != nil { + t.Fatal(err) + } + startID := workerTestID(t) + out, err := w.Start(ctx, startID, a.Attempt()) + if err != nil { + t.Fatal(err) + } + deadline, active := out.Lease.ExecutionDeadline() + if !active { + t.Fatal("short valid lease was rejected") + } + // 用单调期限驱动协作取消；额外等待安全余量确保服务端授权也已过期。 + work, stop := context.WithDeadline(ctx, deadline) + defer stop() + <-work.Done() + if !errors.Is(work.Err(), context.DeadlineExceeded) { + t.Fatal(work.Err()) + } + select { + case <-ctx.Done(): + t.Fatal(ctx.Err()) + case <-time.After(350 * time.Millisecond): + } + stale, err := w.Start(ctx, startID, a.Attempt()) + if err != nil { + t.Fatal(err) + } + if !stale.Deduplicated { + t.Fatal("missing historical receipt") + } + if _, active := stale.Lease.ExecutionDeadline(); active { + t.Fatal("expired historical ACK authorized work") + } + late, err := w.Finish(ctx, workerTestID(t), WorkerFinish{Ref: a.Attempt(), Outcome: WorkerUnknown, Result: json.RawMessage(`{"reason":"handler_cancelled"}`), ErrorCode: "LEASE_EXPIRED"}) + if err != nil { + t.Fatal(err) + } + if late.Receipt.Disposition != "EVIDENCE_ONLY" { + t.Fatal("late report changed authoritative result") + } + for { + current, err := api.GetTask(ctx, a.Task.ID) + if err != nil { + t.Fatal(err) + } + if current.State == TaskNeedsReview { + if current.AttemptCount != 1 || len(current.Result) != 0 { + t.Fatal("expired execution was replayed or evidence promoted") + } + break + } + select { + case <-ctx.Done(): + t.Fatal("server did not recover expired execution") + case <-time.After(100 * time.Millisecond): + } + } +} diff --git a/sdk/go/worker_types.go b/sdk/go/worker_types.go new file mode 100644 index 0000000..1718e5f --- /dev/null +++ b/sdk/go/worker_types.go @@ -0,0 +1,205 @@ +package orbit + +import ( + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "time" +) + +// WorkerCapability 精确声明一种处理器与参数版本，不进行静默版本降级。 +type WorkerCapability struct { + Handler string `json:"handler"` // Handler 处理器名称，1–128 位 ASCII 标识。 + HandlerVersion string `json:"handler_version"` // HandlerVersion 处理器版本，1–64 位。 + PayloadVersion string `json:"payload_version"` // PayloadVersion 输入契约版本，1–32 位。 +} + +// WorkerRegistration 是同一进程重连时必须复用的身份；进程重启使用新身份。 +// 包含秘密，禁止以 JSON 等形式写入日志；SDK 的 String/GoString 会隐藏秘密。 +type WorkerRegistration struct { + InstanceID string `json:"instance_id"` // InstanceID 本进程随机标识，32 位小写十六进制。 + InstanceSecret string `json:"instance_secret"` // InstanceSecret 32 字节随机秘密的 base64url 编码，重连保留。 + QueueID string `json:"queue_id"` // QueueID 绑定队列，重连不可更换。 + Capabilities []WorkerCapability `json:"capabilities"` // Capabilities 1–32 种精确能力，重连集合固定。 + Capacity uint32 `json:"capacity"` // Capacity 总执行槽数量，1–256；不是空闲槽的增量。 +} + +// NewWorkerRegistration 生成进程身份；同进程只创建一次，并在重连时复用返回值。 +func NewWorkerRegistration(queueID string, capacity uint32, capabilities []WorkerCapability) (WorkerRegistration, error) { + id, err := NewWorkerMessageID() + if err != nil { + return WorkerRegistration{}, err + } + var secret [32]byte + if _, err := rand.Read(secret[:]); err != nil { + return WorkerRegistration{}, err + } + return normalizeRegistration(WorkerRegistration{InstanceID: id, InstanceSecret: base64.RawURLEncoding.EncodeToString(secret[:]), QueueID: queueID, Capacity: capacity, Capabilities: capabilities}) +} + +// NewWorkerMessageID 生成一次控制操作的身份；响应不明时重报必须复用原值。 +func NewWorkerMessageID() (string, error) { + var id [16]byte + if _, err := rand.Read(id[:]); err != nil { + return "", err + } + return hex.EncodeToString(id[:]), nil +} + +// WorkerWelcome 是连接/心跳确认，不提供任务执行授权。 +type WorkerWelcome struct { + SessionID string `json:"session_id"` // SessionID 当前连接身份，重连变化。 + Epoch uint64 `json:"epoch,string"` // Epoch 会话代次，JSON 使用字符串防止精度丢失。 + ServerTime time.Time `json:"server_time"` // ServerTime 服务端本次确认时的数据库时钟。 + ExpiresAt time.Time `json:"expires_at"` // ExpiresAt 会话期限，不能替代任务租约。 + LeaseDurationMS uint64 `json:"lease_duration_ms"` // LeaseDurationMS 单次租约的最大窗口，毫秒。 + Draining bool `json:"draining"` // Draining 是否已持久化排空；同进程重连保留。 +} + +// WorkerCapacity 使用绝对容量和单调序号；降低容量不撤销在途执行。 +type WorkerCapacity struct { + Sequence uint64 `json:"sequence,string"` // Sequence 当前连接内从 1 开始递增，同序号必须同内容。 + Capacity uint32 `json:"capacity"` // Capacity 总槽数，0–256；零停止新增领取。 + Drain bool `json:"drain"` // Drain 为真后不可在同一进程身份下清除。 +} + +// WorkerAssignment 是预占通知；必须 Start 并核验当前租约后才能运行处理器。 +// 含业务输入和执行秘密，禁止序列化到普通日志。 +type WorkerAssignment struct { + Task Task `json:"task"` // Task 任务受理快照及业务输入。 + AttemptID string `json:"attempt_id"` // AttemptID 本次尝试标识。 + Epoch uint64 `json:"epoch,string"` // Epoch 任务执行代次，隔离过时尝试。 + LeaseSecret string `json:"lease_secret"` // LeaseSecret 本次尝试的秘密，业务处理器不应接收它。 + LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 预占期限；不能独立授权执行。 + ServerTime time.Time `json:"server_time"` // ServerTime 领取时数据库时间。 +} + +// Attempt 返回控制操作所需的证明；不要把证明转交业务处理器。 +func (a WorkerAssignment) Attempt() WorkerAttempt { + return WorkerAttempt{TaskID: a.Task.ID, AttemptID: a.AttemptID, Epoch: a.Epoch, LeaseSecret: a.LeaseSecret} +} + +// WorkerAttempt 证明对某次尝试的控制权，不包含连接身份；重连保留原值。 +type WorkerAttempt struct { + TaskID string `json:"task_id"` // TaskID 目标任务标识。 + AttemptID string `json:"attempt_id"` // AttemptID 原尝试标识，重连不新建。 + Epoch uint64 `json:"epoch,string"` // Epoch 原执行代次。 + LeaseSecret string `json:"lease_secret"` // LeaseSecret 原执行秘密，禁止普通日志输出。 +} + +// WorkerFinish 将明确结果与未知效果分开；普通网络错误不等于业务明确失败。 +type WorkerFinish struct { + Ref WorkerAttempt `json:"ref"` // Ref 对目标尝试的控制证明。 + Outcome WorkerOutcome `json:"outcome"` // Outcome 明确成功、明确失败或效果未知。 + Result json.RawMessage `json:"result"` // Result 非 null JSON，最多 256 KiB；保留精确数字文本。 + ErrorCode string `json:"error_code,omitempty"` // ErrorCode 失败/未知时必填稳定分类，成功时为空；禁止原始异常正文。 +} + +// WorkerOutcome 描述业务效果；UNKNOWN 必须进入核查而不是盲目重跑。 +type WorkerOutcome string + +const ( + WorkerSucceeded WorkerOutcome = "SUCCEEDED" + WorkerFailed WorkerOutcome = "FAILED" + WorkerUnknown WorkerOutcome = "UNKNOWN" +) + +// WorkerReceipt 是不可变历史回执，不能作为当前执行权。 +type WorkerReceipt struct { + Operation string `json:"operation"` // Operation 原控制操作 start、renew 或 finish。 + TaskState TaskState `json:"task_state"` // TaskState 原操作提交时的任务状态。 + AttemptState string `json:"attempt_state"` // AttemptState 原操作提交时的尝试状态。 + TaskRevision uint64 `json:"task_revision,string"` // TaskRevision 原操作提交版本。 + LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 历史期限；即便尚未到期也不能独立授权。 + Disposition string `json:"disposition"` // Disposition 为 APPLIED 或 EVIDENCE_ONLY。 +} + +// WorkerLease 是本次响应重新核验的授权；必须调用 ExecutionDeadline 判断可否执行。 +type WorkerLease struct { + Active bool `json:"active"` // Active 当前会话是否仍拥有有效尝试；LEASED 仅代表预占。 + State string `json:"state"` // State 当前尝试状态，只有 RUNNING 可执行。 + LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 当前服务端期限，不直接和客户端墙钟比较。 + ServerTime time.Time `json:"server_time"` // ServerTime 本次核验的数据库时钟。 + deadline time.Time // deadline 扣除全请求耗时和安全余量后的本地单调期限，不参与 JSON。 +} + +// ExecutionDeadline 返回保守的本地执行截止时间。false 时不能运行或继续处理器。 +// 仅 Start/Renew/Resume 的 SDK 响应具有本地期限；自行 JSON 解码的值不会授权执行。 +func (l WorkerLease) ExecutionDeadline() (time.Time, bool) { + return l.deadline, l.Active && l.State == "RUNNING" && !l.deadline.IsZero() && time.Now().Before(l.deadline) +} + +// WorkerCommandResult 把历史回执与当前权限分开；Deduplicated 不表示可再次执行。 +type WorkerCommandResult struct { + Receipt WorkerReceipt `json:"receipt"` // Receipt 对应稳定消息身份的原始回执。 + Lease WorkerLease `json:"lease"` // Lease 本次重新验证的当前授权。 + Deduplicated bool `json:"deduplicated"` // Deduplicated 是否返回已保存的回执。 +} + +func (r WorkerRegistration) String() string { + return fmt.Sprintf("WorkerRegistration{InstanceID:%s QueueID:%s Capacity:%d secrets:redacted}", r.InstanceID, r.QueueID, r.Capacity) +} +func (r WorkerRegistration) GoString() string { return r.String() } +func (a WorkerAttempt) String() string { + return fmt.Sprintf("WorkerAttempt{TaskID:%s AttemptID:%s Epoch:%d secret:redacted}", a.TaskID, a.AttemptID, a.Epoch) +} +func (a WorkerAttempt) GoString() string { return a.String() } +func (a WorkerAssignment) String() string { return a.Attempt().String() } +func (a WorkerAssignment) GoString() string { return a.String() } +func (f WorkerFinish) String() string { + return fmt.Sprintf("WorkerFinish{%s Outcome:%s data:redacted}", f.Ref, f.Outcome) +} +func (f WorkerFinish) GoString() string { return f.String() } + +func workerName(s string, max int) bool { + if len(s) < 1 || len(s) > max { + return false + } + for i, c := range s { + if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || i > 0 && (c == '_' || c == '.' || c == ':' || c == '-')) { + return false + } + } + return true +} +func validWorkerSecret(s string) bool { + if len(s) != 43 { + return false + } + raw, err := base64.RawURLEncoding.Strict().DecodeString(s) + return err == nil && len(raw) == 32 +} +func normalizeRegistration(r WorkerRegistration) (WorkerRegistration, error) { + if !validID(r.InstanceID) || !validID(r.QueueID) || !validWorkerSecret(r.InstanceSecret) || r.Capacity < 1 || r.Capacity > 256 || len(r.Capabilities) < 1 || len(r.Capabilities) > 32 { + return WorkerRegistration{}, errors.New("invalid Worker identity, queue, secret, capacity or capabilities") + } + r.Capabilities = append([]WorkerCapability(nil), r.Capabilities...) + for _, c := range r.Capabilities { + if !workerName(c.Handler, 128) || !workerName(c.HandlerVersion, 64) || !workerName(c.PayloadVersion, 32) { + return WorkerRegistration{}, errors.New("invalid Worker capability") + } + } + sort.Slice(r.Capabilities, func(i, j int) bool { + a, b := r.Capabilities[i], r.Capabilities[j] + if a.Handler != b.Handler { + return a.Handler < b.Handler + } + if a.HandlerVersion != b.HandlerVersion { + return a.HandlerVersion < b.HandlerVersion + } + return a.PayloadVersion < b.PayloadVersion + }) + for i := 1; i < len(r.Capabilities); i++ { + if r.Capabilities[i] == r.Capabilities[i-1] { + return WorkerRegistration{}, errors.New("duplicate Worker capability") + } + } + return r, nil +} +func validWorkerAttempt(a WorkerAttempt) bool { + return validID(a.TaskID) && validID(a.AttemptID) && a.Epoch > 0 && validWorkerSecret(a.LeaseSecret) +}