From da123bba69a38ec26b435549ac9f73e20850f847 Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 14:29:28 +0800 Subject: [PATCH] feat: add automatic Worker runtime with lease recovery and bounded shutdown --- README.md | 6 +- docs/08-development-plan.md | 3 +- docs/README.md | 2 +- docs/api/README.md | 2 +- docs/api/worker.md | 4 +- docs/examples/README.md | 10 +- docs/reports/phase-2-sdk-protocol.md | 2 +- docs/reports/phase-2-worker-runtime.md | 50 ++ docs/sdk/README.md | 4 +- docs/sdk/worker-runtime.md | 109 +++++ docs/sdk/worker.md | 2 +- scripts/test-mysql.sh | 3 +- sdk/go/doc.go | 3 +- sdk/go/examples/worker-protocol/main.go | 2 +- sdk/go/examples/worker/main.go | 133 ++++++ sdk/go/worker_handler.go | 130 ++++++ sdk/go/worker_integration_test.go | 27 +- sdk/go/worker_lease_guard.go | 76 +++ sdk/go/worker_runtime.go | 543 ++++++++++++++++++++++ sdk/go/worker_runtime_integration_test.go | 357 ++++++++++++++ sdk/go/worker_runtime_test.go | 181 ++++++++ sdk/go/worker_runtime_types.go | 152 ++++++ 22 files changed, 1784 insertions(+), 17 deletions(-) create mode 100644 docs/reports/phase-2-worker-runtime.md create mode 100644 docs/sdk/worker-runtime.md create mode 100644 sdk/go/examples/worker/main.go create mode 100644 sdk/go/worker_handler.go create mode 100644 sdk/go/worker_lease_guard.go create mode 100644 sdk/go/worker_runtime.go create mode 100644 sdk/go/worker_runtime_integration_test.go create mode 100644 sdk/go/worker_runtime_test.go create mode 100644 sdk/go/worker_runtime_types.go diff --git a/README.md b/README.md index 7b94a10..a1c3940 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,9 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已实现 32 个数据模型、版本化迁移、凭据及范围校验、队列创建查询、任务幂等提交/查询/分页/待执行取消，以及对应 Go SDK 和示例。P2 服务端执行内核、WSS 和独立 Worker 底层 SDK 已通过阶段内回归，可实际完成文本处理任务；自动 Worker 运行器、调度器和流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已实现 32 个数据模型、版本化迁移、凭据及范围校验、队列创建查询、任务幂等提交/查询/分页/待执行取消，以及对应 Go SDK 和示例。P2 服务端、WSS、底层 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) 与 [P2 SDK 协议记录](docs/reports/phase-2-sdk-protocol.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)。 ## 目录 @@ -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) 运行受理/取消示例。另有 [Worker 协议示例](docs/sdk/worker.md)，实际领取并处理文本、提交结果与排空；自动运行器仍在开发。 +任务接入从 [初始化凭据](docs/api/authentication.md) 开始，再按 [SDK 接入](docs/sdk/README.md) 运行受理/取消示例。[自动 Worker 示例](docs/sdk/worker-runtime.md) 演示处理器注册、并发消费、结果确认与排空；特殊接入也可使用 [底层协议示例](docs/sdk/worker.md)。 ## 开发与回归 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index dda87a9..b9da778 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 底层协议及实际 ACK 丢失恢复回归通过；文本执行示例可运行；自动运行/重连/退出、查询与完整验收待完成 | +| P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复；Worker SDK | 真 Worker 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 开发中：服务端、SDK 协议和自动运行器通过阶段内回归；跨初始租约执行、自动重连/证据重报、有界退出及并发验证通过；尝试/证据查询、完整 Schema 和阶段验收待完成 | | P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 待开发 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | @@ -75,3 +75,4 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P1 验证记录](reports/phase-1.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 cc3374f..627bb0b 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本：0.2 - 整理日期：2026-09-17 -- 阶段：P0/P1 已完成，P2 服务端与 Worker 底层 SDK 已通过阶段内回归，自动 Worker 运行器继续开发；尚未完成生产验收 +- 阶段：P0/P1 已完成，P2 服务端、底层 SDK 与自动 Worker 运行器已通过阶段内回归，查询/协议验收继续推进；尚未完成生产验收 - 适用范围：Orbit 统一平台，包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 diff --git a/docs/api/README.md b/docs/api/README.md index d311c15..65e4c7d 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 服务端及底层 SDK 已实现，自动 Worker 运行器开发中 | [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 97236f2..1e0cd82 100644 --- a/docs/api/worker.md +++ b/docs/api/worker.md @@ -1,6 +1,6 @@ # Worker 通信与执行契约 v1（开发中） -已实现服务端执行事务、WebSocket 接入、[独立 Go 底层协议 SDK](../sdk/worker.md) 及真实 WSS/ACK 丢失回归。**P2 尚未完成**：自动 Worker 运行器、自动重连/续租、处理器生命周期与优雅退出还在开发。本页记录当前可验证的协议；底层 SDK 不自动管理处理器。 +已实现服务端执行事务、WebSocket 接入、[独立 Go 底层协议 SDK](../sdk/worker.md) 和 [自动 Worker 运行器](../sdk/worker-runtime.md)，通过真实 WSS、自动续租、ACK 丢失/持续断网恢复及有界退出回归。**P2 尚未完成**：尝试/证据查询、完整协议 Schema 与完整阶段验收继续推进。底层 WorkerConnection 不自动管理处理器；NewWorker / Run 提供自动管理。 ## 1. 接入与身份 @@ -143,4 +143,4 @@ receipt 是原操作的不可变确认；lease 是本次重新核验的执行权 这些是有界的开发基线，不代表经过容量压测的生产参数。当前同队列操作使用较粗的行锁协调，吞吐与公平性需要 P6 按负载验证。服务退出会显式关闭被 HTTP Upgrade 接管的连接；断开不会直接撤销外部已发出的请求。 -运行中取消、SDK 自动重连/续租、非协作处理器处理、完整 WS JSON Schema、凭据轮换和多节点进程级故障演练尚未全部完成。实际证据见 [P2 服务端阶段记录](../reports/phase-2-server.md)。 +运行中取消、完整 WS JSON Schema、凭据轮换和多节点进程级故障演练尚未全部完成。SDK 已实现自动重连/续租、非协作处理器的有界退出与状态报告；实际证据见 [P2 服务端阶段记录](../reports/phase-2-server.md) 和 [自动运行器回归](../reports/phase-2-worker-runtime.md)。 diff --git a/docs/examples/README.md b/docs/examples/README.md index e1b3106..3ef7477 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -25,9 +25,17 @@ - 前置：专用 Worker 凭据，启动步骤见 [Worker SDK 文档](../sdk/worker.md)。 - 输入与步骤：提交 `{"text":" Hello Orbit "}` → 注册/容量 → 领取 → 开始授权 → 实际处理 → 确认结果 → 查询 → 排空。 - 预期：SUCCEEDED、一次尝试、结果 `{"text":"HELLO ORBIT"}`。 -- 异常：控制 ACK 丢失使用原身份有限重报；业务函数不随重连再次调用；迟到证据不能当作成功结果。自动续租与持续消费仍待运行器实现。 +- 异常：底层示例使用原身份有限重报控制 ACK；业务函数不随重连再次调用；迟到证据不能当作成功结果。自动续租与持续消费见下方运行器示例。 - 清理：任务历史保留；运行结束确认排空并关闭连接。每次示例使用独立任务/处理器身份。 +## 已可运行：自动 Worker 并发消费 + +- 源码：[自动 Worker 示例](../../sdk/go/examples/worker/main.go)，[接入与启动命令](../sdk/worker-runtime.md)。 +- 步骤：注册版本化处理器 → Run → 以容量 2 消费三条文本任务 → 查询结果 → 取消 Run context 并等待排空。 +- 预期：三条规范化结果、started=3、confirmed=3、running=0；输入与结果均由实际代码处理。 +- 异常：自动心跳/续租与原身份重连；普通错误/panic 按未知效果报告，超时退出不声称杀死 goroutine；完整故障证据见 [回归记录](../reports/phase-2-worker-runtime.md)。 +- 清理：正常排空关闭连接，保留历史；进程被强杀后的在途状态由租约恢复与核查衔接。 + ## 计划场景与验收标准 | 场景 | 输入与步骤 | 预期结果 | 必测异常 | 状态 | diff --git a/docs/reports/phase-2-sdk-protocol.md b/docs/reports/phase-2-sdk-protocol.md index 9ff1259..eba84d8 100644 --- a/docs/reports/phase-2-sdk-protocol.md +++ b/docs/reports/phase-2-sdk-protocol.md @@ -1,6 +1,6 @@ # P2 Worker SDK 协议阶段回归 -日期：2026-09-17。此记录覆盖独立 SDK 的底层连接和一次性执行闭环，**不代表 P2 完成，也不代表自动 Worker 运行器已交付**。 +日期：2026-09-17。此历史阶段记录覆盖独立 SDK 的底层连接和一次性执行闭环，**不代表 P2 完成，也不覆盖自动 Worker 运行器**。后续运行器的实现与验证见 [独立报告](phase-2-worker-runtime.md)。 ## 实现范围 diff --git a/docs/reports/phase-2-worker-runtime.md b/docs/reports/phase-2-worker-runtime.md new file mode 100644 index 0000000..3ca0f38 --- /dev/null +++ b/docs/reports/phase-2-worker-runtime.md @@ -0,0 +1,50 @@ +# P2 自动 Worker 运行器回归 + +日期：2026-09-17。自动运行器已实现并通过本轮回归；P2 仍需补齐尝试/证据查询、协议 Schema 及完整阶段验收。没有修改已有迁移或持久化模型。 + +## 本轮实现 + +独立 SDK 新增 NewWorker / Run / Stats / Events，支持能力注册、容量限制、独立处理器执行、自动会话心跳、自动任务续租、稳定身份重连、结果重报及有界排空。业务明确失败与未知效果分开；panic、Goexit 和不可接受的结果都有稳定分类。配置及数据结构提供中文注释。 + +网络控制和处理器执行分离；租约监视器独立工作，不能因控制请求阻塞而延长权限。过期取消不可逆，新的较短会话期限也会缩短本地授权。当前控制请求按连接串行，吞吐取决于数据库/网络与配置容量，需要 P6 负载验证。 + +## 验证环境与入口 + +macOS arm64、Go 1.27.1、Docker MySQL 8.4.10、锁定 coder/websocket v1.8.15。真实 Orbit 命令运行在随机回环端口，SDK WSS 故障网关使用受信任测试证书，不跳过 TLS 验证。 + +```sh +make test +make vet +make build +make integration +cd sdk/go +GOWORK=off go test -race ./... +GOWORK=off go vet ./... +``` + +隔离集成脚本先运行既有模型/迁移/任务/Worker/WSS 回归，再启动真实服务器与 Worker 专用凭据，执行底层 SDK 和自动运行器测试及全部示例。凭据不进入 URL、命令行参数或输出，结束后移除隔离测试容器。 + +## 通过的场景 + +| 场景 | 实际断言 | +| --- | --- | +| 自动重连与双重 ACK 丢失 | 网关分别在实际服务提交 start / finish 后丢 ACK；运行器自动重连核对，处理器只调用一次，服务端仅一次尝试，结果保持精确数值 | +| 跨越初始租约的执行 | 处理器执行 32 秒，跨越服务端首次 30 秒租约，至少两次自动续租确认后仍取得权威 SUCCEEDED | +| 收尾期间保持执行 | 处理器开始后立即取消 Run context，运行器持久化排空并在 45 秒 grace 内继续续租/收集结果，正常返回且 Running=0 | +| 业务失败与未知效果 | 明确 WorkerFailure → FAILED；普通 error 和 panic → NEEDS_REVIEW；SDK 不泄露异常正文 | +| 不协作的处理器 | 函数忽略 context 时，250 毫秒收尾期限后 Run 有界返回，错误记录 1 个未确认尝试和仍运行的处理器，没有声称杀死 goroutine | +| 自动租约到期取消 | 达到单次尝试时限后处理器观察到 ErrExecutionLeaseLost，只调用一次，任务进入 NEEDS_REVIEW | +| 持续断网与证据重报 | 断开活跃 WSS 并让网关拒绝重连，处理器仍按本地期限取消；恢复服务后自动上报待确认结果，收到 EVIDENCE_ONLY，权威状态和尝试数不变 | +| 实际业务并发上限 | 6 个任务、容量 2，处理器实际峰值恰为 2，所有任务一次尝试完成并排空 | +| 租约监视器 | 续期后旧期限不取消处理器；新会话缩短期限被采用；到期后更新不能复活 | +| 卡住的握手与认证失败 | HTTP 握手不返回时，50 毫秒 grace 仍能中断并收尾；HTTP 401 立即终止且不执行处理器，不输出上游原始正文 | +| 处理器异常保护 | panic、runtime.Goexit、重复 JSON 键、超范围数字与超深结果均产生受控 UNKNOWN 报告；执行计数正确释放 | +| 实际命令行示例 | 自动 Worker 执行三条文本规范化任务；started=3、confirmed=3、running=0，结果按预期转换 | + +以上为正确性和受控故障证据，不是性能基准。实际自动恢复测试运行在同一真实服务进程后面的 WSS 故障网关；已有服务端测试覆盖不同 node ID 的租约迁移，但本轮未把这两层合并描述成 SDK 多主机宕机验收。 + +## 限制与下一步 + +未确认结果只在 Worker 进程内保留，强杀进程后由服务端租约恢复与业务核查衔接。业务副作用幂等、资源端代次隔离、宿主机/数据库故障与部署拓扑仍需对应验证。运行器不提供业务自动重试；策略性重试在 P3。 + +继续补齐 P2 尝试/证据查询和协议 Schema；再进入 P3 的重试、计划、配额、可靠事件订阅。不会以本轮 SDK 回归代替全平台生产验收。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 00f9047..b167f87 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -1,6 +1,6 @@ # Go SDK 接入说明 -已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询，以及 [Worker 底层协议 SDK](worker.md)。自动 Worker 运行器和流程构建接口按 [开发计划](../08-development-plan.md) 继续实现。 +已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询、[自动 Worker](worker-runtime.md) 和 [Worker 底层协议 SDK](worker.md)。流程构建等后续接口按 [开发计划](../08-development-plan.md) 继续实现。 module：`git.yuntuo.hk/k2safe/orbit/sdk/go`。SDK 使用独立 go.mod，依赖 Go 标准库与 `github.com/coder/websocket v1.8.15`，不引用服务端模型或数据库实现。 @@ -133,4 +133,4 @@ go mod tidy ## 6. 后续交付 -P2 已提供 Worker 底层注册、WSS、能力/容量、执行确认、续租、恢复和排空接口及 [可运行示例](worker.md)，继续开发自动 Worker 运行器；P3 提供计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 +P2 已提供底层协议与 [自动 Worker 运行器及示例](worker-runtime.md)，继续补齐查询与阶段验收；P3 提供计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 diff --git a/docs/sdk/worker-runtime.md b/docs/sdk/worker-runtime.md new file mode 100644 index 0000000..62d66fa --- /dev/null +++ b/docs/sdk/worker-runtime.md @@ -0,0 +1,109 @@ +# 自动 Worker 接入 + +更新：2026-09-17。自动 Worker 已实现，并通过真实 MySQL、实际服务命令、WSS 断线/ACK 丢失/持续断网回归。P2 的尝试与证据查询、完整协议 Schema 和进程级故障验收仍需继续完成。 + +源码：[运行器](../../sdk/go/worker_runtime.go)、[配置与中文注释](../../sdk/go/worker_runtime_types.go)、[完整示例](../../sdk/go/examples/worker/main.go)。特殊传输集成可以使用 [底层协议 SDK](worker.md)，普通消费直接使用本页的 Worker。 + +## 1. 运行完整示例 + +按 [底层 SDK 的初始化步骤](worker.md#1-可运行的文本规范化示例) 创建具有 Worker 权限的专用凭据，在仓库根目录执行： + +```sh +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/worker.credentials.json \ + go run ./examples/worker +``` + +示例提交三条文本任务，以容量 2 启动自动 Worker，真实执行文本规范化，查询结果后取消 Run context 并等待排空。预期输出： + +```text +task=<任务ID> state=SUCCEEDED result={"text":"HELLO ORBIT"} +task=<任务ID> state=SUCCEEDED result={"text":"REUSABLE WORKER"} +task=<任务ID> state=SUCCEEDED result={"text":"DURABLE RESULT"} +worker_drained=true started=3 confirmed=3 running=0 +``` + +每次示例使用独立操作和能力名称，避免接走以前运行留下的任务；任务与审计历史保留。生产通常使用固定处理器版本，由独立长驻 Worker 消费；本示例把生产者和消费者放在一个命令里方便复现。 + +## 2. 注册处理器与启动 + +以下为接入函数片段；函数收到的 ctx 应来自进程信号或应用生命周期，并等待 Run 返回后再退出进程。 + +```go +worker, err := orbit.NewWorker(client, orbit.WorkerOptions{ + QueueID: queueID, + Capacity: 4, + ShutdownTimeout: 30 * time.Second, + Handlers: []orbit.WorkerHandler{{ + Capability: orbit.WorkerCapability{ + Handler: "text.normalize", HandlerVersion: "v1", PayloadVersion: "v1", + }, + Handle: func(ctx context.Context, execution orbit.WorkerExecution) (json.RawMessage, error) { + // 实际 HTTP/数据库/文件操作应使用这个 ctx，并实现业务效果幂等。 + if err := ctx.Err(); err != nil { return nil, err } + return json.RawMessage(`{"processed":true}`), nil + }, + }}, +}) +if err != nil { return err } +return worker.Run(ctx) +``` + +任务的 handler、handler_version、payload_version 必须和注册能力精确匹配。处理器收到 Task、AttemptID 与执行 Epoch，不收到实例/尝试秘密。Task.ID 或接入方保存的业务操作键用于效果去重；AttemptID 标识一次尝试，不能作为跨尝试的业务幂等键。资源侧支持执行代次校验时，可拒绝旧 Epoch 的写入。 + +SDK 的 Run 每个实例只能调用一次，包括已经正常退出的实例。重连由内部保留原身份处理；不要通过反复 NewWorker 来代替断线恢复。进程重启会生成新身份，不能继续使用旧进程的内存状态来声称已经完成。 + +## 3. 配置和接口 + +| 配置 | 默认 / 范围 | 含义 | +| --- | --- | --- | +| QueueID | 必填，32 位 ID | 当前凭据有消费权限的队列 | +| Capacity | 4；1–256 | 同时尚未返回的处理器上限；不响应取消的函数仍占物理槽 | +| Handlers | 1–32 项 | 精确能力，重复能力和空函数在构造时拒绝 | +| ShutdownTimeout | 30 秒；1 毫秒–5 分钟 | 从 Run context 取消起计算的整体收尾期限 | +| ReconnectMinDelay | 200 毫秒；至少 10 毫秒 | 连接失败退避起点 | +| ReconnectMaxDelay | 5 秒；不小于起点，最多 30 秒 | 指数退避上限，区间抖动；不增加任务尝试次数 | + +`Run(ctx)` 正常排空返回 nil；同实例再次运行返回 ErrWorkerAlreadyRun。`Stats()` 返回进程内 Running、Started、RenewalsConfirmed、ReportsConfirmed、EvidenceConfirmed、Reconnects、DroppedEvents 快照，不是持久化业务成功率。 + +`Events()` 是 128 条缓冲的只读本地通知流，不读取也不会阻塞运行；缓冲满时丢弃并计数，Run 返回时关闭。通知只含类型、任务/尝试 ID 与稳定分类，没有业务结果或秘密。它不能代替审计、可靠结果订阅或任务查询。 + +## 4. 执行、续租与恢复 + +- 只有 Start 当前授权有效才调用处理器；同进程中的已知尝试最多调用一次。重复确认与重新连接只核对控制操作。 +- 处理器在独立 goroutine 执行，网络控制由事件循环管理；每连接当前只保留一个执行器控制请求在途。不同处理器可以并行，控制请求顺序与业务并发上限分开。 +- 会话心跳按服务端 TTL 的三分之一计算，上限 5 秒。任务续租按当前保守租约剩余时间的三分之一安排，下限 20 毫秒；服务器仍限制总截止与单次尝试时限。 +- 每个处理器的租约监视器独立取消 context，断网或控制请求阻塞也不会延长本地执行权限。最新会话若缩短租约，监视器会采用较短期限。已到期的处理器不会被后续确认复活。 +- 重连复用实例和尝试身份，在开启容量之前 Resume 所有已知尝试；待确认 start/renew/finish 使用原消息 ID。认证、权限及协议错误终止 Run；临时服务/连接错误按退避核对。 +- 结果未确认或处理器失去租约时，运行器降低对外声明的容量。仍未返回的函数占用真实并发槽；已在传输中的预占会有界接收，已知尝试最多为配置容量的两倍。 + +Run 的重连是控制消息恢复，不是业务自动重试。当前明确失败仍成为 FAILED；策略化业务重试在 P3 实现。 + +## 5. 结果与错误分类 + +| 处理器结果 | SDK 报告 / 服务端行为 | +| --- | --- | +| 合法非 null JSON + nil error | SUCCEEDED | +| `&orbit.WorkerFailure{Code:"OBJECT_REJECTED", Details:...}` | FAILED；仅用于能确认的业务失败，Details 为空时为 `{}` | +| 普通 error | UNKNOWN / HANDLER_ERROR，进入 NEEDS_REVIEW | +| panic / runtime.Goexit | UNKNOWN / HANDLER_PANIC 或 HANDLER_ABORTED | +| 非法 JSON、超限结果、不支持数值或嵌套 | UNKNOWN / INVALID_HANDLER_RESULT | +| 处理器因租约取消返回错误 | UNKNOWN / LEASE_LOST | +| 报告到达时租约已失效、或结果与权威状态冲突 | 服务端可只保存 EVIDENCE_ONLY；不把证据计为任务成功 | + +原始异常和 panic 内容不写入结果或通知。Result/Details 最多 256 KiB，应优先使用对象引用。SDK 保留精确数字文本，并按服务端数字范围及完整帧的嵌套上限预检，避免产生永远无法接受的报告。 + +普通网络错误不能证明副作用未发生，因此不会自动解释为明确失败。UNKNOWN 和迟到证据的后续核查由平台管理能力与业务对账共同处理。 + +## 6. 退出与故障边界 + +取消 Run context 后，运行器请求持久化排空，收集排空 ACK 之前已派发的工作，继续为在途任务续租并确认结果。进程应等待 Run 返回，而不是取消 context 后立即退出。 + +ShutdownTimeout 从取消信号发生时开始计时，包含网络等待和处理器收尾。到期后取消控制请求及处理器 context，返回 `*WorkerRunError`；可用 `errors.Is(err, orbit.ErrWorkerShutdownTimeout)` 判断，Unconfirmed 和 Running 给出终止收尾时的快照。其他致命错误也带这些计数，便于识别未核对的执行。 + +Go 无法杀死忽略 context 的 goroutine。Run 可以有界返回，但这类函数可能仍在进程中执行；不得把返回错误解释为副作用已停止，也不要在同一进程盲目重建 Worker 叠加执行。服务端对失联执行按租约回收并进入核查。 + +身份、尚未确认的结果和消息 ID 当前保存在内存，支持同进程重连；进程被强杀会失去这些内容。持久化租约、操作回执、最终结果和审计在 MySQL 中，业务外部效果仍需要业务幂等或查询核对。这一边界不等于端到端 exactly-once。 + +本轮证据见 [自动运行器回归报告](../reports/phase-2-worker-runtime.md)。实际部署中的多主机故障、数据库复制确认与容量承诺仍需后续生产验收。 diff --git a/docs/sdk/worker.md b/docs/sdk/worker.md index 9ecb70d..007fad1 100644 --- a/docs/sdk/worker.md +++ b/docs/sdk/worker.md @@ -1,6 +1,6 @@ # Go Worker 底层协议 SDK -更新：2026-09-17。已实现可独立构建的 `WorkerConnection`，配合真实 Orbit 服务和 MySQL 通过控制操作、WSS、ACK 丢失与租约过期回归。**自动 Worker 运行器仍在开发**：本页的方法不会自动启动处理器、心跳、续租、重连或排空。需要持续消费的接入方应等待运行器交付，不能把一次性示例当作完整托管运行时。 +更新：2026-09-17。已实现可独立构建的 `WorkerConnection`，配合真实 Orbit 服务和 MySQL 通过控制操作、WSS、ACK 丢失与租约过期回归。本页是底层接口，不自动启动处理器、心跳、续租、重连或排空；普通持续消费使用已实现的 [自动 Worker 运行器](worker-runtime.md)。 源码：[连接与控制方法](../../sdk/go/worker_connection.go)、[中文注释的数据契约](../../sdk/go/worker_types.go)。传输字段、权限和事务语义见 [Worker 接口](../api/worker.md)。SDK 独立 module，只依赖标准库和锁定版本的 `github.com/coder/websocket v1.8.15`，不导入服务端包。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 0209d51..1dc9fa5 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -49,5 +49,6 @@ if [ -z "$TASK_HTTP_PORT" ]; then printf '%s\n' 'Bootstrap HTTP server did not s (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_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 'IntegrationWorker(Connection|Runtime)' -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) +(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) diff --git a/sdk/go/doc.go b/sdk/go/doc.go index 23a6803..e81d17c 100644 --- a/sdk/go/doc.go +++ b/sdk/go/doc.go @@ -1,4 +1,5 @@ // Package orbit 是 Orbit 的独立 Go SDK，提供健康、任务、队列与 Worker 底层协议。 -// WorkerConnection 不自动重连、续租或运行处理器；自动运行器与流程 SDK 仍在开发。 +// WorkerConnection 是底层接口；NewWorker/Run 自动管理处理器、续租、恢复与排空。 +// 流程 SDK 等后续能力按阶段继续开发。 // SDK 不导入服务端模型、迁移或数据库依赖。 package orbit diff --git a/sdk/go/examples/worker-protocol/main.go b/sdk/go/examples/worker-protocol/main.go index 6295d46..dbe623a 100644 --- a/sdk/go/examples/worker-protocol/main.go +++ b/sdk/go/examples/worker-protocol/main.go @@ -1,5 +1,5 @@ // worker-protocol 演示一次文本规范化任务的真实执行与原消息重报。 -// 这是底层协议示例；持续运行、自动续租与优雅退出由后续 Worker 运行器提供。 +// 这是底层协议示例；持续运行、自动续租与优雅退出见 examples/worker 的运行器示例。 package main import ( diff --git a/sdk/go/examples/worker/main.go b/sdk/go/examples/worker/main.go new file mode 100644 index 0000000..bac6b0a --- /dev/null +++ b/sdk/go/examples/worker/main.go @@ -0,0 +1,133 @@ +// worker 演示自动 Worker：声明处理器，Run 管理连接/心跳/续租/恢复，取消后排空。 +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.runtime." + operation + w, err := orbit.NewWorker(client, orbit.WorkerOptions{QueueID: queue, Capacity: 2, ShutdownTimeout: 5 * time.Second, Handlers: []orbit.WorkerHandler{{Capability: orbit.WorkerCapability{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: normalize}}}) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + // 示例先受理，再启动消费者；生产通常是独立的长驻 Worker 进程。 + var tasks []orbit.Task + for i, text := range []string{" hello orbit ", " reusable worker ", " durable result "} { + payload, _ := json.Marshal(struct { + Text string `json:"text"` + }{text}) + accepted, err := client.SubmitTask(ctx, fmt.Sprintf("runtime-%s-%d", operation, i), orbit.SubmitTaskRequest{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, TaskTimeoutMS: 60000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 10000}, RecoverySafety: orbit.RecoveryIdempotent}) + if err != nil { + return err + } + tasks = append(tasks, accepted.Task) + } + workerCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(workerCtx) }() + // 任一调用失败时也等待有界排空，不在 main 返回时静默丢掉执行中的 Worker。 + waited := false + defer func() { + stop() + if !waited { + <-done + } + }() + for _, task := range tasks { + for { + current, err := client.GetTask(ctx, task.ID) + if err != nil { + return err + } + if current.State == orbit.TaskSucceeded { + fmt.Printf("task=%s state=%s result=%s\n", current.ID, current.State, current.Result) + break + } + if current.State == orbit.TaskFailed || current.State == orbit.TaskNeedsReview { + return fmt.Errorf("task %s requires attention: %s", current.ID, current.State) + } + select { + case err := <-done: + waited = true + return fmt.Errorf("Worker stopped before completion: %v", err) + case <-ctx.Done(): + return ctx.Err() + case <-time.After(100 * time.Millisecond): + } + } + } + stop() + err = <-done + waited = true + if err != nil { + return err + } + stats := w.Stats() + if stats.Started != 3 || stats.ReportsConfirmed != 3 || stats.Running != 0 { + return errors.New("unexpected Worker completion counts") + } + fmt.Printf("worker_drained=true started=%d confirmed=%d running=%d\n", stats.Started, stats.ReportsConfirmed, stats.Running) + return nil +} +func normalize(ctx context.Context, execution orbit.WorkerExecution) (json.RawMessage, error) { + var in struct { + Text string `json:"text"` + } + if err := json.Unmarshal(execution.Task.Payload, &in); err != nil { + return nil, &orbit.WorkerFailure{Code: "INVALID_TEXT"} + } + // 模拟可取消的短工作；真实 I/O 也必须使用传入的 ctx。 + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(200 * time.Millisecond): + } + return json.Marshal(struct { + Text string `json:"text"` + }{Text: strings.ToUpper(strings.TrimSpace(in.Text))}) +} diff --git a/sdk/go/worker_handler.go b/sdk/go/worker_handler.go new file mode 100644 index 0000000..a0e0632 --- /dev/null +++ b/sdk/go/worker_handler.go @@ -0,0 +1,130 @@ +package orbit + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "strconv" + "strings" +) + +type handlerCompletion struct { + attemptID string + result WorkerFinish +} + +func (w *Worker) invoke(ctx context.Context, handler WorkerHandler, execution WorkerExecution, ref WorkerAttempt, results chan<- handlerCompletion, runDone <-chan struct{}) { + out := WorkerFinish{Ref: ref, Outcome: WorkerUnknown, ErrorCode: "HANDLER_ABORTED", Result: json.RawMessage(`{}`)} + completed := false + defer func() { + if recover() != nil { + out = unknownHandlerResult(ref, "HANDLER_PANIC") + } else if !completed { + out = unknownHandlerResult(ref, "HANDLER_ABORTED") + } + w.running.Add(-1) + select { + case results <- handlerCompletion{attemptID: ref.AttemptID, result: out}: + case <-runDone: + } + }() + if ctx.Err() != nil { + out = unknownHandlerResult(ref, "LEASE_LOST") + completed = true + return + } + w.started.Add(1) + result, err := handler.Handle(ctx, execution) + if err == nil { + result = append(json.RawMessage(nil), result...) + if validHandlerResult(result) { + out = WorkerFinish{Ref: ref, Outcome: WorkerSucceeded, Result: result} + } else { + out = unknownHandlerResult(ref, "INVALID_HANDLER_RESULT") + } + } else { + var known *WorkerFailure + if errors.As(err, &known) && known != nil && workerName(known.Code, 64) { + details := append(json.RawMessage(nil), known.Details...) + if len(details) == 0 { + details = json.RawMessage(`{}`) + } + if validHandlerResult(details) { + out = WorkerFinish{Ref: ref, Outcome: WorkerFailed, ErrorCode: known.Code, Result: details} + } else { + out = unknownHandlerResult(ref, "INVALID_HANDLER_RESULT") + } + } else if errors.Is(context.Cause(ctx), ErrExecutionLeaseLost) { + out = unknownHandlerResult(ref, "LEASE_LOST") + } else { + out = unknownHandlerResult(ref, "HANDLER_ERROR") + } + } + completed = true +} +func unknownHandlerResult(ref WorkerAttempt, code string) WorkerFinish { + return WorkerFinish{Ref: ref, Outcome: WorkerUnknown, ErrorCode: code, Result: json.RawMessage(`{}`)} +} + +// validHandlerResult 提前拒绝服务端无法持久化的结果，避免失效报告无限重报。 +// 只检验数值范围；不把精确业务数字改写成 float64。 +func validHandlerResult(raw json.RawMessage) bool { + if len(raw) == 0 || len(raw) > 256<<10 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) { + return false + } + // 预留 task.finish 的信封/body 嵌套，避免合法结果单独校验后在帧校验阶段超深。 + frame := append([]byte(`{"body":{"result":`), raw...) + frame = append(frame, []byte(`}}`)...) + if validateWorkerJSON(frame) != nil { + return false + } + d := json.NewDecoder(bytes.NewReader(raw)) + d.UseNumber() + for { + value, err := d.Token() + if err == io.EOF { + return true + } + if err != nil { + return false + } + number, ok := value.(json.Number) + if !ok { + continue + } + s := string(number) + if len(s) > 256 { + return false + } + if _, err := strconv.ParseFloat(s, 64); err != nil { + return false + } + s = strings.TrimPrefix(s, "-") + exponent := 0 + if i := strings.IndexAny(s, "eE"); i >= 0 { + var err error + exponent, err = strconv.Atoi(s[i+1:]) + if err != nil || exponent < -308 || exponent > 308 { + return false + } + s = s[:i] + } + if i := strings.IndexByte(s, '.'); i >= 0 { + exponent -= len(s) - i - 1 + s = s[:i] + s[i+1:] + } + s = strings.TrimLeft(s, "0") + if s == "" { + continue + } + for strings.HasSuffix(s, "0") { + s = s[:len(s)-1] + exponent++ + } + if exponent < -308 || exponent > 308 || exponent+len(s)-1 > 308 { + return false + } + } +} diff --git a/sdk/go/worker_integration_test.go b/sdk/go/worker_integration_test.go index 586e346..5869a55 100644 --- a/sdk/go/worker_integration_test.go +++ b/sdk/go/worker_integration_test.go @@ -49,17 +49,24 @@ func integrationWorkerClient(t *testing.T) (*Client, string, string) { // 只丢响应，不伪造任务状态，测试凭据不进入 URL 或输出。 type workerFaultGateway struct { drop atomic.Value + 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)} + g := &workerFaultGateway{dropped: make(chan string, 4), active: make(map[*websocket.Conn]struct{})} 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) { + if g.offline.Load() { + http.Error(w, "gateway offline", 503) + return + } handlers.Add(1) defer handlers.Done() bridge, stop := context.WithCancel(ctx) @@ -78,6 +85,13 @@ func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { return } defer down.CloseNow() + g.mu.Lock() + g.active[down] = struct{}{} + g.mu.Unlock() + defer func() { g.mu.Lock(); delete(g.active, down); g.mu.Unlock() }() + if g.offline.Load() { + return + } down.SetReadLimit(512 << 10) var mu sync.Mutex var forwarding sync.WaitGroup @@ -141,6 +155,17 @@ func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { t.Cleanup(func() { cancel(); g.server.Close(); handlers.Wait() }) return g } +func (g *workerFaultGateway) disconnectActive() { + g.mu.Lock() + active := make([]*websocket.Conn, 0, len(g.active)) + for c := range g.active { + active = append(active, c) + } + g.mu.Unlock() + for _, c := range active { + c.CloseNow() + } +} func (g *workerFaultGateway) expectDrop(t *testing.T, operation string) { t.Helper() select { diff --git a/sdk/go/worker_lease_guard.go b/sdk/go/worker_lease_guard.go new file mode 100644 index 0000000..7b0517e --- /dev/null +++ b/sdk/go/worker_lease_guard.go @@ -0,0 +1,76 @@ +package orbit + +import ( + "context" + "sync" + "time" +) + +// executionGuard 独立于网络控制取消处理器，已到期/取消的上下文永不恢复。 +type executionGuard struct { + ctx context.Context + cancel context.CancelCauseFunc + mu sync.Mutex + deadline time.Time + wake chan struct{} + done chan struct{} +} + +func newExecutionGuard(parent context.Context, lease WorkerLease) (*executionGuard, bool) { + deadline, ok := lease.ExecutionDeadline() + if !ok { + return nil, false + } + ctx, cancel := context.WithCancelCause(parent) + g := &executionGuard{ctx: ctx, cancel: cancel, deadline: deadline, wake: make(chan struct{}, 1), done: make(chan struct{})} + go g.watch() + return g, true +} +func (g *executionGuard) watch() { + defer close(g.done) + timer := time.NewTimer(time.Until(g.until())) + defer timer.Stop() + for { + select { + case <-g.ctx.Done(): + return + case <-g.wake: + case <-timer.C: + } + g.mu.Lock() + remaining := time.Until(g.deadline) + if remaining <= 0 { + g.cancel(ErrExecutionLeaseLost) + g.mu.Unlock() + return + } + timer.Reset(remaining) + g.mu.Unlock() + } +} +func (g *executionGuard) update(lease WorkerLease) bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.ctx.Err() != nil || !time.Now().Before(g.deadline) { + g.cancel(ErrExecutionLeaseLost) + return false + } + deadline, ok := lease.ExecutionDeadline() + if !lease.Active || lease.State != "RUNNING" { + g.cancel(ErrExecutionLeaseLost) + return false + } + // Lease 是新鲜核验，不是历史 Receipt。新会话可能缩短有效窗口，必须遵守。 + if !ok { + g.cancel(ErrExecutionLeaseLost) + return false + } + g.deadline = deadline + select { + case g.wake <- struct{}{}: + default: + } + return true +} +func (g *executionGuard) until() time.Time { g.mu.Lock(); defer g.mu.Unlock(); return g.deadline } +func (g *executionGuard) stop(cause error) { g.cancel(cause); <-g.done } diff --git a/sdk/go/worker_runtime.go b/sdk/go/worker_runtime.go new file mode 100644 index 0000000..2935c18 --- /dev/null +++ b/sdk/go/worker_runtime.go @@ -0,0 +1,543 @@ +package orbit + +import ( + "context" + "errors" + "math/rand/v2" + "sync" + "sync/atomic" + "time" +) + +type runtimeAttempt struct { + assignment WorkerAssignment + startID, renewID, finishID string + guard *executionGuard + invoked, returned, lost bool + result WorkerFinish + nextTry, nextRenew time.Time + retryDelay time.Duration +} +type runtimeCommand struct { + kind string + id string + job *runtimeAttempt + capacity WorkerCapacity + conn *WorkerConnection +} +type runtimeReply struct { + command *runtimeCommand + result WorkerCommandResult + welcome WorkerWelcome + err error +} +type runtimeConnected struct { + conn *WorkerConnection + leases map[string]WorkerLease + err error +} + +type workerRuntime struct { + worker *Worker + ctx context.Context + jobs map[string]*runtimeAttempt + order []string + cursor int + conn *WorkerConnection + command *runtimeCommand + connecting bool + connectedOnce bool + draining, drainConfirmed bool + capacity uint32 + sequence uint64 + heartbeat, dialAt time.Time + backoff time.Duration + results chan handlerCompletion + replies chan runtimeReply + connections chan runtimeConnected + wg sync.WaitGroup +} + +// Run 持续消费直到 ctx 取消或不可恢复的认证/协议错误。取消后先排空并在 +// ShutdownTimeout 内收集在途结果；正常收尾返回 nil，超时返回 *WorkerRunError。 +// 已到期的处理器上下文不会因重连复活；每个已知尝试在本进程最多调用一次处理器。 +func (w *Worker) Run(parent context.Context) error { + if !w.used.CompareAndSwap(false, true) { + return ErrWorkerAlreadyRun + } + defer close(w.events) + if parent.Err() != nil { + return nil + } + ctx, cancel := context.WithCancel(context.WithoutCancel(parent)) + var shutdownExpired atomic.Bool + watchDone := make(chan struct{}) + go func() { + defer close(watchDone) + select { + case <-ctx.Done(): + return + case <-parent.Done(): + } + timer := time.NewTimer(w.options.ShutdownTimeout) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + shutdownExpired.Store(true) + cancel() + } + }() + r := &workerRuntime{worker: w, ctx: ctx, jobs: make(map[string]*runtimeAttempt), backoff: w.options.ReconnectMinDelay, results: make(chan handlerCompletion, w.options.Capacity), replies: make(chan runtimeReply, 2), connections: make(chan runtimeConnected, 1)} + defer func() { + cancel() + if r.conn != nil { + r.conn.Close() + } + for _, j := range r.jobs { + if j.guard != nil { + j.guard.stop(context.Canceled) + } + } + r.wg.Wait() + // 连接可能在退出与交付结果之间成功，仍需关闭；不留下新会话连接。 + select { + case c := <-r.connections: + if c.conn != nil { + c.conn.Close() + } + default: + } + <-watchDone + }() + tick := time.NewTicker(20 * time.Millisecond) + defer tick.Stop() + stop := parent.Done() + for { + if ctx.Err() != nil { + cause := ctx.Err() + if shutdownExpired.Load() { + cause = ErrWorkerShutdownTimeout + } + return r.failure(cause) + } + if parent.Err() != nil && !r.draining { + r.draining = true + stop = nil + w.emit(WorkerEvent{Type: "draining"}) + } + r.observeExpired() + if r.conn != nil && r.conn.Err() != nil { + if err := r.disconnect(r.conn.Err()); err != nil { + return r.failure(err) + } + } + // drain ACK 之前的派发已被底层读取器排进队列，收齐后才能宣布排空。 + if r.conn != nil { + for len(r.conn.assignments) > 0 { + if err := r.accept(<-r.conn.assignments); err != nil { + return r.failure(err) + } + } + } + if r.draining && len(r.jobs) == 0 && (r.drainConfirmed || !r.connectedOnce && !r.connecting) { + return nil + } + if r.conn == nil && !r.connecting && !time.Now().Before(r.dialAt) { + r.connect() + } + if r.conn != nil && r.command == nil { + if err := r.dispatch(); err != nil { + return r.failure(err) + } + } + var assignments <-chan WorkerAssignment + var closed <-chan struct{} + if r.conn != nil { + assignments = r.conn.assignments + closed = r.conn.Done() + } + select { + case <-ctx.Done(): + case <-stop: + stop = nil + case <-tick.C: + case <-closed: + case a := <-assignments: + if r.conn.Err() == nil { + if err := r.accept(a); err != nil { + return r.failure(err) + } + } + case done := <-r.results: + if j := r.jobs[done.attemptID]; j != nil { + j.returned = true + j.result = done.result + j.nextTry = time.Time{} + } + case reply := <-r.replies: + if err := r.receive(reply); err != nil { + return r.failure(err) + } + case connected := <-r.connections: + if err := r.establish(connected); err != nil { + return r.failure(err) + } + } + } +} + +func (r *workerRuntime) failure(err error) error { + return &WorkerRunError{Cause: err, Unconfirmed: len(r.jobs), Running: r.worker.running.Load()} +} +func (r *workerRuntime) event(j *runtimeAttempt, kind, code string) { + r.worker.emit(WorkerEvent{Type: kind, TaskID: j.assignment.Task.ID, AttemptID: j.assignment.AttemptID, Code: code}) +} +func (r *workerRuntime) accept(a WorkerAssignment) error { + if j := r.jobs[a.AttemptID]; j != nil { + if j.assignment.Attempt() != a.Attempt() { + return ErrWorkerProtocol + } + return nil + } + // 运行中失去授权但不退出的处理器仍占物理槽；允许接收一轮已在网络中的预占， + // 不突破处理器容量，也不让待核对身份无限累积。 + if len(r.jobs) >= 2*int(r.worker.options.Capacity) { + return ErrWorkerBackpressure + } + id, err := NewWorkerMessageID() + if err != nil { + return err + } + r.jobs[a.AttemptID] = &runtimeAttempt{assignment: a, startID: id} + r.order = append(r.order, a.AttemptID) + return nil +} +func (r *workerRuntime) remove(j *runtimeAttempt) { + if j.guard != nil { + j.guard.stop(context.Canceled) + } + delete(r.jobs, j.assignment.AttemptID) + for i, id := range r.order { + if id == j.assignment.AttemptID { + r.order = append(r.order[:i], r.order[i+1:]...) + break + } + } + if len(r.order) > 0 { + r.cursor %= len(r.order) + } else { + r.cursor = 0 + } +} +func (r *workerRuntime) observeExpired() { + for _, j := range r.jobs { + if j.guard != nil && j.guard.ctx.Err() != nil && !j.lost { + j.lost = true + r.event(j, "lease_lost", "LEASE_LOST") + } + } +} +func (r *workerRuntime) desiredCapacity() uint32 { + if r.draining { + return 0 + } + ghosts := uint32(0) + for _, j := range r.jobs { + if j.returned || j.lost { + ghosts++ + } + } + if ghosts >= r.worker.options.Capacity { + return 0 + } + return r.worker.options.Capacity - ghosts +} + +func (r *workerRuntime) connect() { + r.connecting = true + if r.connectedOnce { + r.worker.reconnects.Add(1) + r.worker.emit(WorkerEvent{Type: "reconnecting"}) + } + refs := make([]WorkerAttempt, 0, len(r.jobs)) + for _, j := range r.jobs { + refs = append(refs, j.assignment.Attempt()) + } + r.wg.Add(1) + go func() { + defer r.wg.Done() + ctx, stop := context.WithTimeout(r.ctx, 30*time.Second) + defer stop() + c, err := r.worker.client.DialWorker(ctx, r.worker.registration) + result := runtimeConnected{conn: c, err: err, leases: make(map[string]WorkerLease, len(refs))} + if err == nil { + for _, ref := range refs { + id, e := NewWorkerMessageID() + if e != nil { + result.err = e + break + } + lease, e := c.Resume(ctx, id, ref) + if e != nil { + result.err = e + break + } + result.leases[ref.AttemptID] = lease + } + } + if result.err != nil && c != nil { + c.Close() + result.conn = nil + } + select { + case r.connections <- result: + case <-r.ctx.Done(): + if result.conn != nil { + result.conn.Close() + } + } + }() +} +func (r *workerRuntime) establish(c runtimeConnected) error { + r.connecting = false + if c.err != nil { + if !workerRetryable(c.err) { + return c.err + } + r.deferReconnect() + return nil + } + r.conn = c.conn + r.connectedOnce = true + r.sequence = 0 + r.heartbeat = time.Time{} + r.capacity = ^uint32(0) + r.drainConfirmed = false + r.draining = r.draining || c.conn.Welcome().Draining + for id, lease := range c.leases { + j := r.jobs[id] + if j == nil { + continue + } + if j.guard != nil { + if !j.guard.update(lease) { + j.lost = true + r.event(j, "lease_lost", "LEASE_LOST") + } + j.nextRenew = time.Time{} + } + if !j.invoked && !lease.Active { + r.event(j, "lease_lost", "NOT_STARTED") + r.remove(j) + } + } + r.worker.emit(WorkerEvent{Type: "connected"}) + return nil +} +func (r *workerRuntime) deferReconnect() { + min := r.worker.options.ReconnectMinDelay + r.dialAt = time.Now().Add(min + time.Duration(rand.Int64N(int64(r.backoff-min)+1))) + r.backoff = minDuration(r.backoff*2, r.worker.options.ReconnectMaxDelay) +} +func (r *workerRuntime) disconnect(err error) error { + if r.conn != nil { + r.conn.Close() + r.conn = nil + } + r.command = nil + if !workerRetryable(err) { + return err + } + r.deferReconnect() + return nil +} +func workerRetryable(err error) bool { + if errors.Is(err, ErrWorkerDisconnected) || errors.Is(err, ErrWorkerClosed) || errors.Is(err, ErrWorkerBackpressure) || errors.Is(err, context.DeadlineExceeded) { + return true + } + var control *WorkerError + if errors.As(err, &control) { + return control.Code == "STALE_SESSION" || control.Code == "UNAVAILABLE" + } + var httpErr *HTTPError + return errors.As(err, &httpErr) && (httpErr.StatusCode == 429 || httpErr.StatusCode >= 500) +} +func minDuration(a, b time.Duration) time.Duration { + if a < b { + return a + } + return b +} + +func (r *workerRuntime) dispatch() error { + now := time.Now() + capacity := r.desiredCapacity() + if !now.Before(r.heartbeat) || capacity != r.capacity || r.draining && !r.drainConfirmed { + id, err := NewWorkerMessageID() + if err != nil { + return err + } + r.sequence++ + r.launch(&runtimeCommand{kind: "capacity", id: id, capacity: WorkerCapacity{Sequence: r.sequence, Capacity: capacity, Drain: r.draining}}) + return nil + } + // 结果优先释放占用；其后照顾临近续租，再批准新处理器。每类内部轮转。 + for _, kind := range []string{"finish", "renew", "start"} { + for n := 0; n < len(r.order); n++ { + i := (r.cursor + n) % len(r.order) + j := r.jobs[r.order[i]] + if now.Before(j.nextTry) { + continue + } + var id *string + switch kind { + case "finish": + if !j.returned { + continue + } + id = &j.finishID + case "renew": + if !j.invoked || j.lost || j.returned || now.Before(j.nextRenew) { + continue + } + id = &j.renewID + case "start": + if j.invoked || r.worker.running.Load() >= int64(r.worker.options.Capacity) { + continue + } + id = &j.startID + } + if *id == "" { + var err error + *id, err = NewWorkerMessageID() + if err != nil { + return err + } + } + r.cursor = (i + 1) % len(r.order) + r.launch(&runtimeCommand{kind: kind, id: *id, job: j}) + return nil + } + } + return nil +} +func (r *workerRuntime) launch(command *runtimeCommand) { + command.conn = r.conn + r.command = command + // 操作所需值在 goroutine 启动前复制；运行器事件循环独占可变尝试状态。 + var ref WorkerAttempt + var finish WorkerFinish + if command.job != nil { + ref = command.job.assignment.Attempt() + finish = command.job.result + } + r.wg.Add(1) + go func() { + defer r.wg.Done() + reply := runtimeReply{command: command} + switch command.kind { + case "capacity": + reply.welcome, reply.err = command.conn.Capacity(r.ctx, command.id, command.capacity) + case "start": + reply.result, reply.err = command.conn.Start(r.ctx, command.id, ref) + case "renew": + reply.result, reply.err = command.conn.Renew(r.ctx, command.id, ref) + case "finish": + reply.result, reply.err = command.conn.Finish(r.ctx, command.id, finish) + } + select { + case r.replies <- reply: + case <-r.ctx.Done(): + } + }() +} + +func (r *workerRuntime) receive(reply runtimeReply) error { + command := reply.command + if r.command != command || command.conn != r.conn { + return nil + } + r.command = nil + j := command.job + if reply.err != nil { + var control *WorkerError + if j != nil && errors.As(reply.err, &control) { + switch control.Code { + case "LEASE_LOST": + if command.kind == "finish" { + return reply.err + } + if j.guard != nil { + j.guard.stop(ErrExecutionLeaseLost) + j.lost = true + r.event(j, "lease_lost", control.Code) + } else { + r.remove(j) + } + return nil + case "EXECUTION_PAUSED", "UNAVAILABLE": + if j.retryDelay == 0 { + j.retryDelay = 200 * time.Millisecond + } else { + j.retryDelay = minDuration(j.retryDelay*2, 5*time.Second) + } + j.nextTry = time.Now().Add(j.retryDelay) + return nil + } + } + if workerRetryable(reply.err) { + return r.disconnect(reply.err) + } + return reply.err + } + r.backoff = r.worker.options.ReconnectMinDelay + if command.kind == "capacity" { + r.capacity = command.capacity.Capacity + r.drainConfirmed = command.capacity.Drain && reply.welcome.Draining + r.heartbeat = time.Now().Add(minDuration(5*time.Second, reply.welcome.ExpiresAt.Sub(reply.welcome.ServerTime)/3)) + return nil + } + if command.kind == "finish" { + if reply.result.Receipt.Disposition == "APPLIED" { + r.worker.reports.Add(1) + r.event(j, "result_applied", string(reply.result.Receipt.TaskState)) + } else { + r.worker.evidence.Add(1) + r.event(j, "result_evidence", string(reply.result.Receipt.TaskState)) + } + r.remove(j) + return nil + } + if command.kind == "start" { + guard, ok := newExecutionGuard(r.ctx, reply.result.Lease) + if !ok { + r.event(j, "lease_lost", "START_NOT_AUTHORIZED") + r.remove(j) + return nil + } + j.guard = guard + j.invoked = true + r.worker.running.Add(1) + task := j.assignment.Task + task.Payload = append([]byte(nil), task.Payload...) + task.Result = append([]byte(nil), task.Result...) + execution := WorkerExecution{Task: task, AttemptID: j.assignment.AttemptID, Epoch: j.assignment.Epoch} + handler := r.worker.handlers[WorkerCapability{Handler: task.Handler, HandlerVersion: task.HandlerVersion, PayloadVersion: task.PayloadVersion}] + r.event(j, "handler_started", "") + go r.worker.invoke(guard.ctx, handler, execution, j.assignment.Attempt(), r.results, r.ctx.Done()) + } else { + r.worker.renewals.Add(1) + j.renewID = "" + if !j.guard.update(reply.result.Lease) { + j.lost = true + r.event(j, "lease_lost", "LEASE_LOST") + } + } + j.nextTry = time.Time{} + j.retryDelay = 0 + if !j.lost { + j.nextRenew = time.Now().Add(max(20*time.Millisecond, time.Until(j.guard.until())/3)) + } + return nil +} diff --git a/sdk/go/worker_runtime_integration_test.go b/sdk/go/worker_runtime_integration_test.go new file mode 100644 index 0000000..8b2047d --- /dev/null +++ b/sdk/go/worker_runtime_integration_test.go @@ -0,0 +1,357 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "sync/atomic" + "testing" + "time" +) + +func runtimeTask(t *testing.T, ctx context.Context, c *Client, queue, handler string, timeout uint64) Task { + t.Helper() + out, err := c.SubmitTask(ctx, "runtime-"+workerTestID(t), SubmitTaskRequest{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"sequence":9007199254740993}`), TaskTimeoutMS: 120000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: timeout}}) + if err != nil { + t.Fatal(err) + } + return out.Task +} +func runtimeHandler(name string, handle func(context.Context, WorkerExecution) (json.RawMessage, error)) WorkerHandler { + return WorkerHandler{Capability: WorkerCapability{Handler: name, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: handle} +} +func waitRuntimeState(t *testing.T, ctx context.Context, c *Client, id string, state TaskState) Task { + t.Helper() + for { + task, err := c.GetTask(ctx, id) + if err != nil { + t.Fatal(err) + } + if task.State == state { + return task + } + select { + case <-ctx.Done(): + t.Fatalf("task did not reach %s", state) + case <-time.After(50 * time.Millisecond): + } + } +} + +func TestIntegrationWorkerRuntimeACKLossRenewAndDrain(t *testing.T) { + api, queue, token := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 55*time.Second) + defer cancel() + g := newWorkerFaultGateway(t, api.baseURL) + client, err := NewClient(g.server.URL, Options{Token: token, HTTPClient: g.server.Client()}) + if err != nil { + t.Fatal(err) + } + name := "runtime.renew." + workerTestID(t) + entered := make(chan struct{}) + var calls atomic.Int32 + w, err := NewWorker(client, WorkerOptions{QueueID: queue, Capacity: 1, ShutdownTimeout: 45 * time.Second, Handlers: []WorkerHandler{runtimeHandler(name, func(work context.Context, e WorkerExecution) (json.RawMessage, error) { + calls.Add(1) + close(entered) + // 必须跨过服务端最初的 30 秒租约；只有自动续租成功才能获得 APPLIED。 + select { + case <-work.Done(): + return nil, work.Err() + case <-time.After(32 * time.Second): + return json.RawMessage(`{"sequence":9007199254740993,"done":true}`), nil + } + })}}) + if err != nil { + t.Fatal(err) + } + task := runtimeTask(t, ctx, api, queue, name, 90000) + g.drop.Store("task.start") + runCtx, stop := context.WithCancel(ctx) + defer stop() + done := make(chan error, 1) + go func() { done <- w.Run(runCtx) }() + g.expectDrop(t, "task.start") + select { + case <-entered: + case err := <-done: + t.Fatalf("Worker stopped before handler: %v", err) + case <-ctx.Done(): + t.Fatal("handler did not start") + } + g.drop.Store("task.finish") + stop() // 取消 Run 只进入排空，不能立即取消正在执行的函数。 + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("Worker did not finish within grace") + } + g.expectDrop(t, "task.finish") + current := waitRuntimeState(t, ctx, api, task.ID, TaskSucceeded) + stats := w.Stats() + if calls.Load() != 1 || stats.Started != 1 || stats.Running != 0 || stats.ReportsConfirmed != 1 || stats.Reconnects < 2 || stats.RenewalsConfirmed < 2 || current.AttemptCount != 1 || string(current.Result) != `{"sequence":9007199254740993,"done":true}` { + t.Fatalf("automatic execution invariant failed: calls=%d stats=%+v attempts=%d", calls.Load(), stats, current.AttemptCount) + } + // 仍在等待的同类工作不能被已排空的进程领取。 + waiting := runtimeTask(t, ctx, api, queue, name, 10000) + observed, err := api.GetTask(ctx, waiting.ID) + if err != nil { + t.Fatal(err) + } + if observed.State != TaskWaiting { + t.Fatal("drained Worker accepted another task") + } + if _, err := api.CancelTask(ctx, observed.ID, "cleanup-"+workerTestID(t), CancelTaskRequest{ExpectedRevision: observed.Revision, Reason: "finish runtime drain regression"}); err != nil { + t.Fatal(err) + } +} + +func TestIntegrationWorkerRuntimeUnknownAndUncooperativeShutdown(t *testing.T) { + api, queue, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + for _, test := range []struct { + name string + handle func(context.Context, WorkerExecution) (json.RawMessage, error) + state TaskState + }{ + {"known", func(context.Context, WorkerExecution) (json.RawMessage, error) { + return nil, &WorkerFailure{Code: "INVALID_OBJECT"} + }, TaskFailed}, + {"unknown", func(context.Context, WorkerExecution) (json.RawMessage, error) { + return nil, errors.New("external request timed out with private details") + }, TaskNeedsReview}, + {"panic", func(context.Context, WorkerExecution) (json.RawMessage, error) { panic("private panic") }, TaskNeedsReview}, + } { + t.Run(test.name, func(t *testing.T) { + name := "runtime." + test.name + "." + workerTestID(t) + w, err := NewWorker(api, WorkerOptions{QueueID: queue, Capacity: 1, Handlers: []WorkerHandler{runtimeHandler(name, test.handle)}}) + if err != nil { + t.Fatal(err) + } + task := runtimeTask(t, ctx, api, queue, name, 5000) + run, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + defer stop() + waitRuntimeState(t, ctx, api, task.ID, test.state) + stop() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("classification Worker did not stop") + } + }) + } + t.Run("uncooperative", func(t *testing.T) { + name := "runtime.uncooperative." + workerTestID(t) + entered, release, returned := make(chan struct{}), make(chan struct{}), make(chan struct{}) + defer close(release) + w, err := NewWorker(api, WorkerOptions{QueueID: queue, Capacity: 1, ShutdownTimeout: 250 * time.Millisecond, Handlers: []WorkerHandler{runtimeHandler(name, func(context.Context, WorkerExecution) (json.RawMessage, error) { + close(entered) + <-release + close(returned) + return json.RawMessage(`{}`), nil + })}}) + if err != nil { + t.Fatal(err) + } + runtimeTask(t, ctx, api, queue, name, 1500) + run, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + defer stop() + select { + case <-entered: + case <-ctx.Done(): + t.Fatal("handler did not enter") + } + began := time.Now() + stop() + select { + case err := <-done: + var stopped *WorkerRunError + if !errors.As(err, &stopped) || !errors.Is(err, ErrWorkerShutdownTimeout) || stopped.Running != 1 || stopped.Unconfirmed != 1 { + t.Fatalf("uncooperative handler hidden: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("shutdown waited forever for handler") + } + if time.Since(began) > time.Second { + t.Fatal("shutdown exceeded bounded grace") + } + select { + case <-returned: + t.Fatal("SDK claimed to kill uncooperative handler") + default: + } + }) +} + +func TestIntegrationWorkerRuntimeLeaseLossCancelsAndDoesNotRevive(t *testing.T) { + api, queue, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 12*time.Second) + defer cancel() + name := "runtime.expired." + workerTestID(t) + cause := make(chan error, 1) + w, err := NewWorker(api, WorkerOptions{QueueID: queue, Capacity: 1, Handlers: []WorkerHandler{runtimeHandler(name, func(work context.Context, _ WorkerExecution) (json.RawMessage, error) { + <-work.Done() + cause <- context.Cause(work) + return nil, work.Err() + })}}) + if err != nil { + t.Fatal(err) + } + task := runtimeTask(t, ctx, api, queue, name, 1400) + run, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + defer stop() + select { + case err := <-cause: + if !errors.Is(err, ErrExecutionLeaseLost) { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("handler did not observe lease cancellation") + } + waitRuntimeState(t, ctx, api, task.ID, TaskNeedsReview) + stop() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("Worker did not stop") + } + if w.Stats().Started != 1 { + t.Fatal("expired handler was invoked again") + } +} + +func TestIntegrationWorkerRuntimeOutageCancelsAndReplaysEvidence(t *testing.T) { + api, queue, token := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 18*time.Second) + defer cancel() + g := newWorkerFaultGateway(t, api.baseURL) + client, err := NewClient(g.server.URL, Options{Token: token, HTTPClient: g.server.Client()}) + if err != nil { + t.Fatal(err) + } + name := "runtime.outage." + workerTestID(t) + entered, cancelled := make(chan struct{}), make(chan error, 1) + w, err := NewWorker(client, WorkerOptions{QueueID: queue, Capacity: 1, ReconnectMaxDelay: time.Second, Handlers: []WorkerHandler{runtimeHandler(name, func(work context.Context, _ WorkerExecution) (json.RawMessage, error) { + close(entered) + <-work.Done() + cancelled <- context.Cause(work) + return nil, work.Err() + })}}) + if err != nil { + t.Fatal(err) + } + task := runtimeTask(t, ctx, api, queue, name, 2000) + run, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + defer stop() + select { + case <-entered: + case <-ctx.Done(): + t.Fatal("handler did not start") + } + g.offline.Store(true) + g.disconnectActive() + select { + case cause := <-cancelled: + if !errors.Is(cause, ErrExecutionLeaseLost) { + t.Fatal(cause) + } + case <-ctx.Done(): + t.Fatal("disconnected handler ignored local expiry") + } + waitRuntimeState(t, ctx, api, task.ID, TaskNeedsReview) + g.offline.Store(false) + for w.Stats().EvidenceConfirmed != 1 { + select { + case err := <-done: + t.Fatalf("Worker stopped before evidence replay: %v", err) + case <-ctx.Done(): + t.Fatal("pending result was not replayed") + case <-time.After(50 * time.Millisecond): + } + } + stop() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("Worker did not stop") + } + current, err := api.GetTask(ctx, task.ID) + if err != nil { + t.Fatal(err) + } + if w.Stats().Started != 1 || current.AttemptCount != 1 || current.State != TaskNeedsReview || len(current.Result) != 0 { + t.Fatal("outage revived handler or promoted late evidence") + } +} + +func TestIntegrationWorkerRuntimeConcurrency(t *testing.T) { + api, queue, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 18*time.Second) + defer cancel() + name := "runtime.concurrency." + workerTestID(t) + var running, peak atomic.Int32 + w, err := NewWorker(api, WorkerOptions{QueueID: queue, Capacity: 2, Handlers: []WorkerHandler{runtimeHandler(name, func(work context.Context, _ WorkerExecution) (json.RawMessage, error) { + value := running.Add(1) + defer running.Add(-1) + for old := peak.Load(); value > old; old = peak.Load() { + if peak.CompareAndSwap(old, value) { + break + } + } + select { + case <-work.Done(): + return nil, work.Err() + case <-time.After(300 * time.Millisecond): + return json.RawMessage(`{"done":true}`), nil + } + })}}) + if err != nil { + t.Fatal(err) + } + var tasks []Task + for range 6 { + tasks = append(tasks, runtimeTask(t, ctx, api, queue, name, 10000)) + } + run, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + defer stop() + for _, task := range tasks { + current := waitRuntimeState(t, ctx, api, task.ID, TaskSucceeded) + if current.AttemptCount != 1 { + t.Fatal("unexpected new attempt") + } + } + stop() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("Worker did not drain") + } + if peak.Load() != 2 || running.Load() != 0 || w.Stats().Started != 6 || w.Stats().ReportsConfirmed != 6 { + t.Fatalf("concurrency invariant failed: peak=%d stats=%+v", peak.Load(), w.Stats()) + } +} diff --git a/sdk/go/worker_runtime_test.go b/sdk/go/worker_runtime_test.go new file mode 100644 index 0000000..80baf02 --- /dev/null +++ b/sdk/go/worker_runtime_test.go @@ -0,0 +1,181 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "testing" + "time" +) + +func guardLease(duration time.Duration) WorkerLease { + return WorkerLease{Active: true, State: "RUNNING", deadline: time.Now().Add(duration)} +} +func TestExecutionGuardRenewsShortensAndNeverRevives(t *testing.T) { + parent, cancel := context.WithCancel(context.Background()) + defer cancel() + g, ok := newExecutionGuard(parent, guardLease(80*time.Millisecond)) + if !ok { + t.Fatal("missing guard") + } + defer g.stop(context.Canceled) + if !g.update(guardLease(500 * time.Millisecond)) { + t.Fatal("renew rejected") + } + select { + case <-g.ctx.Done(): + t.Fatal("old lease cancelled a renewed handler") + case <-time.After(100 * time.Millisecond): + } + if !g.update(guardLease(35 * time.Millisecond)) { + t.Fatal("new short session rejected") + } + select { + case <-g.ctx.Done(): + case <-time.After(200 * time.Millisecond): + t.Fatal("shorter session did not cancel handler") + } + if !errors.Is(context.Cause(g.ctx), ErrExecutionLeaseLost) { + t.Fatal(context.Cause(g.ctx)) + } + if g.update(guardLease(time.Minute)) { + t.Fatal("expired handler revived") + } +} + +func TestHandlerOutcomesAndAbort(t *testing.T) { + ref := WorkerAttempt{TaskID: workerTestTask, AttemptID: workerTestAttempt, Epoch: 1} + for _, test := range []struct { + name string + fn func(context.Context, WorkerExecution) (json.RawMessage, error) + out WorkerOutcome + code string + }{ + {"success", func(context.Context, WorkerExecution) (json.RawMessage, error) { + return json.RawMessage(`{"n":9007199254740993}`), nil + }, WorkerSucceeded, ""}, + {"known_failure", func(context.Context, WorkerExecution) (json.RawMessage, error) { + return nil, &WorkerFailure{Code: "INPUT_REJECTED", Details: json.RawMessage(`{"field":"name"}`)} + }, WorkerFailed, "INPUT_REJECTED"}, + {"ordinary_error", func(context.Context, WorkerExecution) (json.RawMessage, error) { + return nil, errors.New("private error secret") + }, WorkerUnknown, "HANDLER_ERROR"}, + {"panic", func(context.Context, WorkerExecution) (json.RawMessage, error) { panic("private panic secret") }, WorkerUnknown, "HANDLER_PANIC"}, + {"goexit", func(context.Context, WorkerExecution) (json.RawMessage, error) { runtime.Goexit(); return nil, nil }, WorkerUnknown, "HANDLER_ABORTED"}, + {"bad_json", func(context.Context, WorkerExecution) (json.RawMessage, error) { + return json.RawMessage(`{"n":1,"n":2}`), nil + }, WorkerUnknown, "INVALID_HANDLER_RESULT"}, + {"number_range", func(context.Context, WorkerExecution) (json.RawMessage, error) { return json.RawMessage(`1e999`), nil }, WorkerUnknown, "INVALID_HANDLER_RESULT"}, + } { + t.Run(test.name, func(t *testing.T) { + w := &Worker{} + w.running.Add(1) + done := make(chan handlerCompletion, 1) + go w.invoke(context.Background(), WorkerHandler{Handle: test.fn}, WorkerExecution{}, ref, done, make(chan struct{})) + select { + case got := <-done: + if got.result.Outcome != test.out || got.result.ErrorCode != test.code || strings.Contains(string(got.result.Result), "secret") { + t.Fatalf("unexpected classification: %s/%s", got.result.Outcome, got.result.ErrorCode) + } + case <-time.After(time.Second): + t.Fatal("handler lost result") + } + if w.running.Load() != 0 || w.started.Load() != 1 { + t.Fatal("handler counters leaked") + } + }) + } + for _, raw := range []string{`null`, `0.1e-308`, strings.Repeat("[", 31) + "1" + strings.Repeat("]", 31), strings.Repeat("9", 257)} { + if validHandlerResult(json.RawMessage(raw)) { + t.Fatal("unsupported result passed runtime validation") + } + } +} + +func TestWorkerOptionsAndSingleRun(t *testing.T) { + c, _ := NewClient("http://127.0.0.1:1", Options{Token: "test-token"}) + opts := WorkerOptions{QueueID: workerTestQueue, Handlers: []WorkerHandler{{Capability: WorkerCapability{Handler: "test", HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(context.Context, WorkerExecution) (json.RawMessage, error) { return json.RawMessage(`{}`), nil }}}} + w, err := NewWorker(c, opts) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := w.Run(ctx); err != nil { + t.Fatal(err) + } + if err := w.Run(ctx); !errors.Is(err, ErrWorkerAlreadyRun) { + t.Fatal("Worker identity reused after Run") + } + if _, open := <-w.Events(); open { + t.Fatal("event channel not closed") + } + for _, invalid := range []WorkerOptions{{QueueID: workerTestQueue}, {QueueID: workerTestQueue, Handlers: append(opts.Handlers, opts.Handlers...)}, {QueueID: workerTestQueue, Handlers: opts.Handlers, ShutdownTimeout: -1}} { + if _, err := NewWorker(c, invalid); err == nil { + t.Fatal("invalid Worker options accepted") + } + } +} + +func TestWorkerShutdownBoundsPendingHandshake(t *testing.T) { + entered := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { close(entered); <-r.Context().Done() })) + defer server.Close() + c, _ := NewClient(server.URL, Options{Token: "test-token"}) + w, err := NewWorker(c, WorkerOptions{QueueID: workerTestQueue, ShutdownTimeout: 50 * time.Millisecond, Handlers: []WorkerHandler{runtimeHandler("test", func(context.Context, WorkerExecution) (json.RawMessage, error) { + t.Error("unexpected handler") + return nil, nil + })}}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("handshake did not start") + } + began := time.Now() + cancel() + select { + case err := <-done: + if !errors.Is(err, ErrWorkerShutdownTimeout) { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("shutdown blocked on pending handshake") + } + if time.Since(began) > 500*time.Millisecond { + t.Fatal("handshake exceeded shutdown bound") + } +} + +func TestWorkerPermanentAuthenticationFailureStops(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "private upstream response", http.StatusUnauthorized) + })) + defer server.Close() + c, _ := NewClient(server.URL, Options{Token: "test-token"}) + w, err := NewWorker(c, WorkerOptions{QueueID: workerTestQueue, Handlers: []WorkerHandler{runtimeHandler("test", func(context.Context, WorkerExecution) (json.RawMessage, error) { + t.Error("unauthorized handler invoked") + return nil, nil + })}}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + err = w.Run(ctx) + var stopped *WorkerRunError + var status *HTTPError + if !errors.As(err, &stopped) || !errors.As(err, &status) || status.StatusCode != 401 || stopped.Unconfirmed != 0 || strings.Contains(err.Error(), "private") { + t.Fatalf("unexpected authentication failure: %v", err) + } +} diff --git a/sdk/go/worker_runtime_types.go b/sdk/go/worker_runtime_types.go new file mode 100644 index 0000000..7720a3b --- /dev/null +++ b/sdk/go/worker_runtime_types.go @@ -0,0 +1,152 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync/atomic" + "time" +) + +var ErrExecutionLeaseLost = errors.New("Worker execution lease expired or lost") +var ErrWorkerShutdownTimeout = errors.New("Worker shutdown grace period expired") +var ErrWorkerAlreadyRun = errors.New("Worker instance may only Run once") + +// WorkerExecution 是交给业务处理器的输入，不包含实例或尝试秘密。 +type WorkerExecution struct { + Task Task // Task 固定任务输入；业务副作用应使用稳定业务身份去重。 + AttemptID string // AttemptID 当前尝试，用于核查与追踪，不代替业务幂等键。 + Epoch uint64 // Epoch 执行代次；接入方资源可用它拒绝旧尝试写入。 +} + +func (e WorkerExecution) String() string { + return fmt.Sprintf("WorkerExecution{TaskID:%s AttemptID:%s Epoch:%d data:redacted}", e.Task.ID, e.AttemptID, e.Epoch) +} +func (e WorkerExecution) GoString() string { return e.String() } + +// WorkerHandler 注册精确版本的业务函数；函数必须遵守 ctx 的取消。 +// 普通 error/panic 表示效果未知；明确的业务失败返回 *WorkerFailure。 +type WorkerHandler struct { + Capability WorkerCapability // Capability 处理器和输入版本，整个 Worker 生命周期固定。 + Handle func(context.Context, WorkerExecution) (json.RawMessage, error) // Handle 返回非 null JSON；不得忽略租约取消继续发起副作用。 +} + +// WorkerFailure 仅用于接入方能够确认的业务失败，不用于网络超时或未知外部效果。 +type WorkerFailure struct { + Code string // Code 1–64 位稳定 ASCII 分类，禁止放原始异常正文。 + Details json.RawMessage // Details 可选非 null JSON，最多 256 KiB；空值使用空对象。 +} + +func (e *WorkerFailure) Error() string { return "Worker handler reported a known failure" } + +// WorkerOptions 配置自动执行器。一个实例绑定一个队列，只能 Run 一次。 +type WorkerOptions struct { + QueueID string // QueueID 凭据有消费权限的队列。 + Capacity uint32 // Capacity 最大并发处理器数，默认 4，范围 1–256。 + Handlers []WorkerHandler // Handlers 1–32 种精确能力，不可重复，函数不得为空。 + ShutdownTimeout time.Duration // ShutdownTimeout 收到 Run context 取消后的收尾总时限，默认 30 秒，最多 5 分钟。 + ReconnectMinDelay time.Duration // ReconnectMinDelay 重连退避起点，默认 200 毫秒，至少 10 毫秒。 + ReconnectMaxDelay time.Duration // ReconnectMaxDelay 重连退避上限，默认 5 秒，最多 30 秒；采用区间内抖动。 +} + +// WorkerEvent 是有界、可能丢弃的本地观察通知，不是持久化审计或结果订阅。 +type WorkerEvent struct { + Type string // Type connected、reconnecting、handler_started、lease_lost、result_applied、result_evidence、draining。 + TaskID string // TaskID 涉及的任务；连接通知为空。 + AttemptID string // AttemptID 涉及的尝试；连接通知为空。 + Code string // Code 稳定错误/结果分类，不包含输入、结果或秘密。 +} + +// WorkerStats 是当前进程的观测值，重启清零，不作为持久化成功率。 +type WorkerStats struct { + Running int64 // Running 尚未返回的处理器数，包含不遵守取消的函数。 + Started uint64 // Started 本进程实际调用处理器的次数。 + RenewalsConfirmed uint64 // RenewalsConfirmed 收到当前续租确认的次数，不代表任务成功。 + ReportsConfirmed uint64 // ReportsConfirmed 收到 APPLIED 的结果报告数，包含 UNKNOWN。 + EvidenceConfirmed uint64 // EvidenceConfirmed 收到 EVIDENCE_ONLY 的报告数。 + Reconnects uint64 // Reconnects 首次连接之后发起重连的次数。 + DroppedEvents uint64 // DroppedEvents 本地通知缓冲满时丢弃的条数。 +} + +// WorkerRunError 表示执行器未能正常收尾；不能据此认为外部效果已撤销。 +type WorkerRunError struct { + Cause error // Cause 连接/权限/协议错误或收尾超时，支持 errors.Is/As。 + Unconfirmed int // Unconfirmed 尚未完成控制核对的已知尝试数。 + Running int64 // Running 返回时仍未退出的处理器数；Go 无法强制杀死 goroutine。 +} + +func (e *WorkerRunError) Error() string { + return fmt.Sprintf("Worker stopped with %d unconfirmed attempts and %d running handlers: %v", e.Unconfirmed, e.Running, e.Cause) +} +func (e *WorkerRunError) Unwrap() error { return e.Cause } + +// Worker 自动管理注册、心跳、续租、重连和有界排空；任务权威状态仍在服务端。 +type Worker struct { + client *Client + options WorkerOptions + registration WorkerRegistration + handlers map[WorkerCapability]WorkerHandler + used atomic.Bool + events chan WorkerEvent + running atomic.Int64 + started atomic.Uint64 + renewals atomic.Uint64 + reports atomic.Uint64 + evidence atomic.Uint64 + reconnects atomic.Uint64 + dropped atomic.Uint64 +} + +func NewWorker(client *Client, options WorkerOptions) (*Worker, error) { + if client == nil || client.token == "" { + return nil, errors.New("authenticated Orbit client is required") + } + if options.Capacity == 0 { + options.Capacity = 4 + } + if options.ShutdownTimeout == 0 { + options.ShutdownTimeout = 30 * time.Second + } + if options.ReconnectMinDelay == 0 { + options.ReconnectMinDelay = 200 * time.Millisecond + } + if options.ReconnectMaxDelay == 0 { + options.ReconnectMaxDelay = 5 * time.Second + } + if options.ShutdownTimeout < time.Millisecond || options.ShutdownTimeout > 5*time.Minute || options.ReconnectMinDelay < 10*time.Millisecond || options.ReconnectMaxDelay < options.ReconnectMinDelay || options.ReconnectMaxDelay > 30*time.Second { + return nil, errors.New("invalid Worker shutdown or reconnect duration") + } + capabilities := make([]WorkerCapability, 0, len(options.Handlers)) + handlers := make(map[WorkerCapability]WorkerHandler, len(options.Handlers)) + for _, h := range options.Handlers { + if h.Handle == nil { + return nil, errors.New("Worker handler function is required") + } + capabilities = append(capabilities, h.Capability) + handlers[h.Capability] = h + } + r, err := NewWorkerRegistration(options.QueueID, options.Capacity, capabilities) + if err != nil { + return nil, err + } + options.Handlers = append([]WorkerHandler(nil), options.Handlers...) + return &Worker{client: client, options: options, registration: r, handlers: handlers, events: make(chan WorkerEvent, 128)}, nil +} + +// Events 返回只读通知流；无需读取也不会阻塞执行，Run 返回时关闭。 +func (w *Worker) Events() <-chan WorkerEvent { return w.events } +func (w *Worker) Stats() WorkerStats { + return WorkerStats{Running: w.running.Load(), Started: w.started.Load(), RenewalsConfirmed: w.renewals.Load(), ReportsConfirmed: w.reports.Load(), EvidenceConfirmed: w.evidence.Load(), Reconnects: w.reconnects.Load(), DroppedEvents: w.dropped.Load()} +} +func (w *Worker) emit(event WorkerEvent) { + select { + case w.events <- event: + default: + w.dropped.Add(1) + } +} +func (w *Worker) String() string { + return fmt.Sprintf("Worker{QueueID:%s Capacity:%d secrets:redacted}", w.options.QueueID, w.options.Capacity) +} +func (w *Worker) GoString() string { return w.String() }