diff --git a/README.md b/README.md index 553df41..7d95f67 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已完成 P0–P2 阶段验收：33 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 、[队列资源控制](docs/api/queues.md) 和 [应用共享配额](docs/api/application-resources.md) 已接入服务端和 SDK，剩余 Pulse 能力与流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已完成 P0–P2 阶段验收：33 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 、[队列资源控制](docs/api/queues.md) 、[应用共享配额](docs/api/application-resources.md) 和 [资源键互斥](docs/api/keyed-concurrency.md) 已接入服务端和 SDK，剩余 Pulse 能力与流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 数据库禁止外键，关联一致性通过服务端事务校验，索引、唯一约束和 CHECK 保留；readiness 同时检查迁移及零外键要求。这不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 阶段验收](docs/reports/phase-2.md)、[安全重试](docs/reports/phase-3-safe-retry.md)、[计划调度](docs/reports/phase-3-schedules.md)、[取消与无外键回归](docs/reports/phase-3-cancellation.md) 及 [队列资源回归](docs/reports/phase-3-queue-resources.md)。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 4741646..49621e3 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -132,3 +132,11 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 ## 12. 应用共享额度 000010_application_resources 新增 ApplicationResources，使用 BINARY(16) 应用主键，12 个持久化字段均有中文注释和 SQL COMMENT，无外键。共享未终结任务数、活跃尝试数、未开始预占数与一秒开始窗口；配置 revision 不随用量变化。停止旧写入节点后回填所有应用，包含跨命名空间的既有在途状态。关联身份行与额度行分开，认证后按应用额度 → 队列 → 任务/尝试 → 许可顺序持锁。后台恢复不因应用停用而停止释放到期尝试。详细字段、权限、SDK 场景及并发边界见 [应用配额接口](api/application-resources.md)。 + +## 13. 显式资源键许可 + +000011_keyed_concurrency 增加 Task.ConcurrencyKey / ConcurrencyKeyDigest、Attempt.ConcurrencyToken、WorkerSession.ConcurrencyKeys，共四个中文注释与 SQL COMMENT 字段。历史值分别为空/空、零和 false，原 BusinessKey 语义不变；33 个实体、273 个直接声明的持久化字段，最终外键数保持零。 + +资源限制复用 ConcurrencyLimit：scope_key=key，key_digest 为域分隔 SHA-256，命名空间内唯一；每键最大许可固定 1。原始 UTF-8 字节精确匹配，不依赖 MySQL 的文本排序规则。一个有键尝试持有队列和资源键两条 Permit，应用活跃尝试只计一次；续租、释放、恢复与状态同事务完成。 + +键行 Revision 在领取和释放时递增，领取后的值保存到 Attempt.ConcurrencyToken；同尝试重连不改变序号，资源行不得删除或重置。达到整数边界时拒绝新领取，禁止回绕。SDK 交付序号，资源端负责原子比较与业务幂等；历史未知效果与执行许可释放分别维护。字段、升级、调用和异常语义见 [资源键契约](api/keyed-concurrency.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index f89a9c3..f7bdca1 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -27,7 +27,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、历史查询、完整 Schema 与真实进程强杀恢复；测试拓扑和剩余生产故障矩阵见 P2 验收报告 | -| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额及 SDK 场景已实现并回归；业务键许可、优先级/公平性和可靠订阅待开发 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级/公平性和可靠订阅待开发 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -46,6 +46,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 运行中取消 | 保存取消意图 → 停止处理器 → 确认结果 → 查询并排空 | P3，已实现完整示例 | | 队列限速与背压 | 暂停 → 满额拒绝 → 原回执核对 → 释放空间 → 限速执行 | P3，已实现完整示例 | | 跨命名空间应用配额 | first 受理 → second 全局满额 → 释放 → 原键成功 → 恢复配置 | P3，已实现完整示例 | +| 跨队列共享资源 | 两队列同键 → 一个有效许可 → 递增序号 → 资源端拒绝旧写入 | P3，已实现完整示例 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | | 审核与外部回执 | 提交流程 → 等待信号 → 授权信号注入 → 继续 | P4 | | 外部作业轮询 | 提交一次 → 有界查询处理中状态 → 回执或最终结果 → 超限核查 | P5 | @@ -84,6 +85,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P3 取消与无外键回归](reports/phase-3-cancellation.md) - [P3 队列资源控制回归](reports/phase-3-queue-resources.md) - [P3 应用共享配额回归](reports/phase-3-application-resources.md) +- [P3 资源键许可回归](reports/phase-3-keyed-concurrency.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 0900231..4f0d854 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本：0.2 - 整理日期：2026-09-17 -- 阶段：P0–P2 已通过阶段验收；P3 安全重试、计划调度、运行中取消、队列资源控制和应用共享配额已通过回归，数据库已改为无外键，P3–P7 继续推进，尚未完成全平台生产验收 +- 阶段：P0–P2 已通过阶段验收；P3 安全重试、计划调度、运行中取消、队列资源控制和应用共享配额已通过回归，资源键许可及真实进程恢复已通过全量回归，数据库已改为无外键，P3–P7 继续推进，尚未完成全平台生产验收 - 适用范围：Orbit 统一平台，包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 diff --git a/docs/api/README.md b/docs/api/README.md index fbcb9cc..8587ae7 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -13,6 +13,7 @@ | `/v1/tasks/{task_id}/attempts/{attempt_id}/evidence` 与证据详情 | P2 已实现 | [执行历史](execution-history.md) | | `/v1/queues` 创建、查询及 `:configure` 控制 | 已实现队列并发、开始速率、积压背压与暂停/恢复 | [队列接口与 SDK 场景](queues.md) | | `/v1/application/resources` 查询及 `:configure` 管理 | P3 已实现跨命名空间应用配额 | [应用配额与 SDK 场景](application-resources.md) | +| 任务 `concurrency_key` 与尝试 `concurrency_token` | P3 已实现跨队列资源键互斥 | [资源键接口与 SDK 场景](keyed-concurrency.md) | | 凭据校验与初始化 CLI | P1 已实现 | [鉴权](authentication.md) | | `/v1/worker/connect` | P2 已实现并通过阶段验收 | [Worker 契约](worker.md)、[Schema](worker-schema.md) | | 任务冻结策略与 Worker `retryable` | P3 安全重试已实现 | [安全重试](retries.md) | diff --git a/docs/api/execution-history.md b/docs/api/execution-history.md index 69d2a83..8a336ed 100644 --- a/docs/api/execution-history.md +++ b/docs/api/execution-history.md @@ -54,6 +54,7 @@ curl "$ORBIT_URL/v1/tasks/$TASK_ID/attempts/$ATTEMPT_ID/evidence/$EVIDENCE_ID" \ | --- | --- | --- | | id、task_id | string / 是 | 尝试与父任务标识 | | epoch | 十进制字符串 / 是 | 单任务执行代次；SDK uint64 | +| concurrency_token | 十进制字符串 / 否 | 有显式资源键时返回跨任务单调许可序号；SDK uint64，无键时为零并省略；历史查询不能授权执行，见 [资源键契约](keyed-concurrency.md) | | state | string / 是 | LEASED、RUNNING、SUCCEEDED、FAILED、CANCELLED、EXPIRED、UNKNOWN；枚举不表示全部状态转换已实现 | | revision | 十进制字符串 / 是 | 记录版本，续租及归属变化可递增；SDK uint64 | | lease_until | UTC RFC 3339 / 是 | 数据库记录的租约期限，**查询不提供执行授权** | diff --git a/docs/api/keyed-concurrency.md b/docs/api/keyed-concurrency.md new file mode 100644 index 0000000..1a6e454 --- /dev/null +++ b/docs/api/keyed-concurrency.md @@ -0,0 +1,117 @@ +# 资源键互斥与单调许可序号 + +状态：P3 已实现。适合多个队列的任务更新同一份对象、文件或其他共享资源。数据库使用 MySQL 事务，不引入外键、Redis 或额外锁服务。 + +## 1. 四种身份各自的作用 + +| 字段 | 作用与范围 | +| --- | --- | +| Idempotency-Key | 一次平台受理命令的身份；原键原内容重报取回原回执 | +| business_key | 业务追踪关联，保持原有语义，不限制并发、不自动去重 | +| concurrency_key | 显式资源键；同一命名空间内跨队列共享一个有效尝试许可 | +| concurrency_token | 此资源键的持久化单调序号；供资源端原子拒绝过时写入，不能代替授权或业务幂等 | + +资源键必须由所有相关提交方按同一约定生成。省略键、使用不同键或处于不同命名空间的任务不会参与同一个互斥组。不要把凭据或敏感业务正文放入键中；任务详情与摘要会返回它。 + +平台控制有效许可，无法撤回已经发往外部的请求。租约到期后旧请求可能仍在运行；资源端需要结合序号与操作幂等保护实际写入。 + +## 2. HTTP 提交与查询 + +沿用 [任务接口](tasks.md) 的鉴权、动作、幂等、状态和错误信封，不增加专门的管理接口。 + +| 位置 / 字段 | 类型 / 默认 | 约束 | +| --- | --- | --- | +| POST /v1/tasks 的 concurrency_key | 可选 string，省略或空字符串表示关闭 | 非空时 1–191 个 Unicode 字符；有效 UTF-8，不含控制字符，首尾不能有 Unicode 空白；null 拒绝 | +| 任务详情、列表摘要、task.assign.task 的 concurrency_key | string，可省略 | 返回受理时冻结值，不能事后换键 | +| task.assign 的 concurrency_token | 非零 uint64 十进制字符串，有资源键时必填 | 无资源键时省略；SDK 为 uint64，不能转换为浮点数 | +| 尝试列表、详情的 concurrency_token | 同上，有资源键时返回 | 历史查询用于核对，不授予执行权限 | +| 计划版本 definition.task 的 concurrency_key | 与提交相同 | 版本固定值，生成任务时继承；无模板插值或动态表达式 | + +键按 UTF-8 字节精确匹配，不做大小写折叠或 Unicode 规范化。例如 `object/A`、`object/a`、组合及分解形式的重音字符属于不同键。若业务认为它们相同，应在提交前统一生成规则。 + +```sh +curl "$ORBIT_URL/v1/tasks" \ + -H "Authorization: Bearer $ORBIT_TOKEN" \ + -H 'Content-Type: application/json' \ + -H 'Idempotency-Key: update-object-42-operation-001' \ + --data '{"queue_id":"11111111111111111111111111111111","handler":"object.update","handler_version":"v1","payload_version":"v1","payload":{"operation_ref":"update-001","object_ref":"objects/42"},"concurrency_key":"object/42"}' +``` + +响应是任务接口的完整信封，首次 201、重报 200；任务对象增加 `"concurrency_key":"object/42"`。原幂等键更换资源键返回 409 IDEMPOTENCY_CONFLICT。非法键返回 400 INVALID_ARGUMENT；权限或队列范围沿用任务接口。资源忙时受理仍可成功，任务保持 WAITING，不消耗尝试次数，但占用未完成任务配额并继续计算总截止时间。 + +## 3. 许可、释放与恢复 + +领取在同一事务中预占应用额度、队列许可和资源键许可，创建尝试并分配序号。有键任务同时持有两条许可记录，应用活跃尝试只计一次。许可覆盖 LEASED 和 RUNNING；业务代码必须等到 Start 确认当前执行权后才运行。 + +| 情况 | 资源键处理 | +| --- | --- | +| Worker 重连、Resume、续租 | 保留同一尝试和序号；续租同步延长许可 | +| 结果重复确认、旧回执重放 | 不重复释放，不影响后续尝试的许可 | +| 未开始取消、明确结束、到期恢复 | 同事务释放；后续尝试取得更大的序号 | +| 安全重试退避 | 释放本次执行许可，等待下次领取；不在退避期间独占资源 | +| UNKNOWN / NEEDS_REVIEW | 释放执行许可，保留未知效果和历史；待核查任务仍占积压配额 | +| 暂停队列、停止派发、断开连接 | 不直接释放在途许可，仍按原租约和结果收尾 | + +因此未知效果不会自动把整个资源键封锁到人工核查结束。若业务要求核查期间禁止其他操作，还需独立的业务冻结状态。当前不提供人工解锁或重置序号接口。 + +相同 `(namespace_id, concurrency_key)` 的新许可序号严格递增，允许间隔；内部领取和释放都推进资源行版本。不得删除或重置这条资源行，否则会破坏单调性。序号空间不足时拒绝领取，不能溢出回绕。不同资源之间不比较序号，单任务的 epoch 也不能替代跨任务资源序号。 + +领取查询在候选 LIMIT 前排除已占用键，避免一批阻塞任务挡住其他键。这不承诺 FIFO、跨队列公平性或最大等待时间。 + +## 4. Worker 与 SDK + +新版 `NewWorker` 自动发送 `hello.concurrency_keys=true`，处理器读取 `e.Task.ConcurrencyKey` 和 `e.ConcurrencyToken`。底层接入者须显式设置 `registration.ConcurrencyKeys=true`，实现资源端校验后再启用。这是整个进程身份的固定注册属性；同进程重连改变它返回 STATE_CONFLICT，升级进程应创建新身份。 + +未声明支持的旧 Worker 继续消费无键任务，不领取带键任务。SDK 拒绝缺少序号、无键却携带序号、未声明支持却收到有键任务等协议错误。重复派发不能静默替换原键或原序号。 + +提交片段（`api`、`ctx`、`queueID` 已初始化）： + +```go +accepted, err := api.SubmitTask(ctx, "update-object-42-operation-001", orbit.SubmitTaskRequest{ + QueueID: queueID, Handler: "object.update", HandlerVersion: "v1", PayloadVersion: "v1", + Payload: json.RawMessage(`{"operation_ref":"update-001","object_ref":"objects/42"}`), + ConcurrencyKey: "object/42", +}) +``` + +资源端需要在**一次原子操作**里完成：核验命名空间、资源键和调用权限；拒绝 `token < highest_token`；提升最高序号；按稳定操作身份查重并写入实际结果。同序号的重复请求仍需要操作幂等。仅先查询最高序号、再分开写入会产生竞争；仅在 Worker 内用 mutex 无法保护多进程和外部系统。 + +最高序号校验只能拒绝资源端已经见过更高序号后的旧写入；更高序号尚未到达时，不能单靠本地最高值证明旧租约仍有效。需要更强授权语义时，资源端还需相应的授权校验或业务版本条件。平台序号也不能替代资源端的身份认证。 + +## 5. 完整可运行场景 + +源码：[keyed-concurrency/main.go](../../sdk/go/examples/keyed-concurrency/main.go)。示例创建两个专用队列，每条队列提交一个相同资源键的任务；两个自动 Worker 各自容量为 4。处理器调用一个受控内存资源模型，查询尝试确认序号增长，最后模拟旧请求验证它被拒绝。 + +从仓库根目录，在专用开发数据库完成迁移并启动服务： + +```sh +go run ./cmd/orbit-migrate up +go run ./cmd/orbit-admin bootstrap --application keyed-example --namespace default \ + --environment test --queue initial --worker-access --output /tmp/orbit-keyed.credentials.json +go run ./cmd/orbit +``` + +上述命令使用预先设置的 `ORBIT_MYSQL_DSN`，服务使用 `ORBIT_HTTP_ADDR=127.0.0.1:8080`。示例凭据需具有全部队列范围的 queues:manage、tasks:submit/read 和 workers:connect/consume。另一个终端执行： + +```sh +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ +ORBIT_CREDENTIALS_FILE=/tmp/orbit-keyed.credentials.json go run ./examples/keyed-concurrency +``` + +正常输出： + +```text +queues=2 completed=2 tokens_increasing=true +resource_writes=2 stale_write_rejected=true worker_drained=true +``` + +示例不访问真实外部资源；内存 mutex 和 map 仅说明原子校验位置，不能复制为生产分布式资源存储。正常结束排空 Worker；每次运行创建独立队列、任务和资源键并保留历史。任务总时限 20 秒，失败非零退出并有界收尾，进程被强杀仍需按历史核对。 + +## 6. 升级与当前边界 + +停止旧服务写入，应用全部迁移到 000011，再启动新服务、升级 Worker SDK，最后提交带资源键的任务。旧节点不维护资源键许可，不能与新节点混跑；新 SDK 的 hello 扩展也不能发给尚未升级的严格旧服务。 + +000011 为历史任务填空键/空摘要、历史尝试填零序号、历史会话填 false，不改变原 business_key。新增四个模型字段均有中文注释和 SQL COMMENT，实体数仍为 33；数据库最终外键数为零。 + +当前基于一个 MySQL 主库及应用额度锁串行化事务，资源键记录随不同键数量持续增长。尚未提供可调的每键并发大于 1、资源管理/清理接口、优先级公平性或生产容量结论。验证证据见 [本阶段报告](../reports/phase-3-keyed-concurrency.md)。 diff --git a/docs/api/tasks.md b/docs/api/tasks.md index 8893f7e..3984907 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -24,6 +24,7 @@ P2 进展：服务端执行内核、WSS 与独立自动 Worker SDK 已可产生 | payload_version | string | 必填 | 输入契约版本，1–32 字符 | | payload | JSON | 必填 | 非 null JSON，最大 256 KiB；平台不解释业务字段 | | business_key | string | 可选 | 最多 191 个 Unicode 字符，无 NUL/换行；用于关联，不提供业务幂等 | +| concurrency_key | string | 可选，空为关闭 | 1–191 个 Unicode 字符，无控制字符和首尾空白；显式资源互斥键，同命名空间跨队列共享，见 [资源键契约](keyed-concurrency.md) | | available_at | time / null | 受理时数据库时间 | 最早可执行时间；必须早于总截止时间，仅允许最多 1 秒的过去时间容差 | | start_deadline_at | time / null | 无独立限制 | 必须晚于当前时间，不早于 available_at，不晚于总截止时间 | | task_timeout_ms | uint64 | 0 或省略为 86400000 | 从受理时计算的总时限，最大 30 天（2592000000 毫秒） | @@ -90,7 +91,7 @@ curl -i "$ORBIT_URL/v1/tasks" \ } ``` -可选响应字段：start_deadline_at、result、cancel_requested_at、completed_at、business_key，未设置时省略。effect_uncertain=true 表示存在尚未核实的授权执行效果，false 时省略；详情、摘要和 Worker 派发均可返回。等待重试时 result/completed_at 省略，历史结果在尝试中保留。终态有 completed_at。状态集合为 WAITING / LEASED / RUNNING / NEEDS_REVIEW / SUCCEEDED / FAILED / CANCELLED；P1 仅主动产生 WAITING 和 CANCELLED。 +可选响应字段：start_deadline_at、result、cancel_requested_at、completed_at、business_key、concurrency_key，未设置时省略。effect_uncertain=true 表示存在尚未核实的授权执行效果，false 时省略；详情、摘要和 Worker 派发均可返回。等待重试时 result/completed_at 省略，历史结果在尝试中保留。终态有 completed_at。状态集合为 WAITING / LEASED / RUNNING / NEEDS_REVIEW / SUCCEEDED / FAILED / CANCELLED；P1 仅主动产生 WAITING 和 CANCELLED。 ### 幂等与失败处理 diff --git a/docs/api/worker-schema.md b/docs/api/worker-schema.md index 38dc2ac..a8bb934 100644 --- a/docs/api/worker-schema.md +++ b/docs/api/worker-schema.md @@ -32,6 +32,7 @@ ACK 的 body 根据原请求选择：worker.capacity → welcome，worker.resume - instance_secret/lease_secret 必须是 32 字节秘密的严格无填充 base64url，长度 43；知道资源 ID 无法替代秘密。 - 结果为非 null JSON；FAILED/UNKNOWN/CANCELLED 必须携带稳定 error_code；SUCCEEDED 省略或使用空 error_code。 - hello.cancellation_notifications 可选且默认 false；Lease 可返回 cancel_requested_at。CANCELLED 需要事务核验已保存意图；通知、取消后执行边界与升级顺序见 [取消契约](cancellation.md)。 +- hello.concurrency_keys 可选且默认 false，未声明时不派发带资源键任务；task.assign.task.concurrency_key 与非零字符串 concurrency_token 必须同时出现。Schema 检查字段配对，SDK 另核验支持声明和键格式，事务维护许可；见 [资源键契约](keyed-concurrency.md)。 - finish.retryable 为可选布尔值，默认 false；true 仅允许 FAILED，Schema 与服务端都校验。派发任务可带 effect_uncertain=true，表示此前授权执行的效果尚未核实。业务重试条件见 [安全重试](retries.md)。 - 返回时间使用 UTC RFC 3339，最多微秒精度。验证器应开启 format 断言以校验实际日历日期。 diff --git a/docs/api/worker.md b/docs/api/worker.md index b683182..60e8260 100644 --- a/docs/api/worker.md +++ b/docs/api/worker.md @@ -66,6 +66,7 @@ go run ./cmd/orbit-admin bootstrap \ | capabilities | 1–32 个 handler / handler_version / payload_version 精确版本组合，不可重复 | | capacity | 总执行槽数量，1–256，不是可用槽的累加值 | | cancellation_notifications | 可选 bool，默认 false；true 接收 task.cancel 通知，自动 Worker 默认开启 | +| concurrency_keys | 可选 bool，默认 false；true 接收资源键和单调序号，自动 Worker 默认开启；同进程重连不可更改 | 重连必须保持凭据、队列、处理器集合和进程秘密，处理器集合的排列顺序不影响身份。新连接递增会话代次并使旧会话失效。知道 instance_id 或 task_id 不能接管执行。 @@ -77,7 +78,7 @@ worker.welcome 返回 session_id、epoch、server_time、expires_at、lease_dura | --- | --- | --- | | worker.resume | AttemptRef，字段见上方 task.start 示例 | 当前 LeaseStatus；成功迁移有效尝试到新会话，不创建尝试或延长期限 | | worker.capacity | sequence 字符串、capacity 0–256、drain bool | 更新后的 welcome 字段 | -| task.assign | 服务端主动发送 | task 完整任务、attempt_id、epoch、lease_secret、lease_until、server_time | +| task.assign | 服务端主动发送 | task 完整任务、attempt_id、epoch、lease_secret、lease_until、server_time；有资源键时带 concurrency_token 字符串 | capacity 使用单调递增序号和覆盖值。同序号同内容可重复，同序号不同内容或旧序号冲突；capacity=0 不释放已有任务的并发许可。drain=true 是进程级持久化排空，重连保留；后续必须保持 drain=true，尝试清除返回 STATE_CONFLICT。新进程使用新实例身份。 @@ -129,7 +130,9 @@ receipt 是原操作的不可变确认；lease 是本次重新核验的执行权 平台在同一事务中预占应用并发与开始额度、分配队列并发许可、创建尝试、递增任务执行代次和次数、保存 LEASED 与派发事件。领取本身消耗一次 max_attempts 预算，包括最终未获开始授权的尝试，预算不会因重连或过期恢复重置。 -运行中的进程总容量跨其所有旧/新会话统计。声明容量减少、断开连接或重复结果，都不能重复释放额度。已实施队列 max_concurrency、rate_per_second 及未终结任务上限；参数含义、窗口突发、暂停及降额语义见 [队列资源控制](queues.md)。[应用级配额](application-resources.md) 已同时生效；业务键许可与公平性继续推进。 +运行中的进程总容量跨其所有旧/新会话统计。声明容量减少、断开连接或重复结果，都不能重复释放额度。已实施队列 max_concurrency、rate_per_second 及未终结任务上限；参数含义、窗口突发、暂停及降额语义见 [队列资源控制](queues.md)。[应用级配额](application-resources.md) 与 [显式资源键许可](keyed-concurrency.md) 同时生效；公平性继续推进。 + +同命名空间的 concurrency_key 跨队列互斥，领取事务同时保存键许可和 concurrency_token；原 business_key 仅追踪。没有声明 concurrency_keys 的 Worker 不领取带键任务。资源端需用单调序号和操作幂等保护实际写入，平台无法撤回外部已发出的请求；未知结果释放执行许可并保留核查状态，详细边界和升级顺序见资源键契约。 领取前核对速率窗口及已有 LEASED 预留，满额时不新建尝试。Start 在同一事务内消费应用与队列的开始预算，原 ACK 重报不重复计数。领取后若配置降额，Start 可能返回 RATE_LIMITED；保持原消息身份在剩余租约内退避核对。续租、结果确认、取消及恢复不消费新开始额度。 diff --git a/docs/examples/README.md b/docs/examples/README.md index 3041749..8faaa11 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -92,6 +92,8 @@ ## 计划场景与验收标准 +已可运行的 [跨队列资源互斥场景](../api/keyed-concurrency.md#5-完整可运行场景) 提供 [完整源码](../../sdk/go/examples/keyed-concurrency/main.go)：两个队列的自动 Worker 操作同一资源，核对递增序号，并模拟资源端拒绝旧写入。输入使用显式 concurrency_key；预期 completed=2、resource_writes=2、stale_write_rejected=true。示例仅使用受控内存资源，无外部副作用；正常排空、保留历史，任务自带总时限。内存 mutex 不是生产分布式存储，生产必须在资源端原子持久化序号、操作身份和实际更新。 + | 场景 | 输入与步骤 | 预期结果 | 必测异常 | 状态 | | --- | --- | --- | --- | --- | | 文件转换 | 对象引用 → 提交任务 → Worker 转换 → 结果引用 | 同一操作获得一次业务结果，历史可查 | 重复提交、Worker 在结果上报前中断 | 受理与执行框架已实现，实际转换示例待补齐 | diff --git a/docs/reports/phase-3-keyed-concurrency.md b/docs/reports/phase-3-keyed-concurrency.md new file mode 100644 index 0000000..ab69182 --- /dev/null +++ b/docs/reports/phase-3-keyed-concurrency.md @@ -0,0 +1,84 @@ +# P3 资源键许可与过时执行隔离回归 + +验证日期：2026-09-17。交付 [资源键互斥接口、SDK 和场景手册](../api/keyed-concurrency.md)。P3 优先级/公平性、可靠结果订阅及 P4–P7 继续推进，持续目标尚未完成。 + +## 1. 已实现内容 + +- 显式 concurrency_key 在命名空间内跨队列共享一个有效尝试许可；原 business_key 保持追踪含义，不改变既有任务并发行为。 +- 领取事务同时维护应用额度、队列许可、键许可、任务和尝试。资源占用时不消耗新的尝试；候选 LIMIT 前过滤忙碌键。 +- concurrency_token 使用持久化键行的单调版本，跨任务递增、允许间隔、拒绝溢出；同尝试重连保留。资源端必须原子比较并按稳定操作身份幂等处理实际写入。 +- 续租、取消、结果、退避及到期恢复同步维护两类许可。UNKNOWN 释放执行许可并保留核查与未知效果，旧结果不能释放后续尝试的许可。 +- Worker 显式协商 concurrency_keys；旧 Worker 不消费有键任务，自动 SDK 默认开启。协议 Schema、SDK 精确 uint64、尝试历史与计划模板同步支持。 +- [000011 迁移](../../internal/storage/mysql/migrations/000011_keyed_concurrency.up.sql) 新增四个中文注释字段，无外键。当前 33 个实体、273 个直接声明的持久化字段均有中文注释，SQL 同步 COMMENT。 +- [完整示例](../../sdk/go/examples/keyed-concurrency/main.go) 创建两个队列、两个 Worker、相同资源键，核对序号并模拟资源端拒绝旧写入。 + +## 2. 环境与命令 + +macOS 本机、Go 1.27.1、Docker MySQL 8.4.10；服务和数据库仅绑定回环，随机私密凭据及临时数据库随测试清理。没有修改既有业务数据库。 + +```sh +make fmt test vet build +make integration +cd sdk/go +GOWORK=off go test -race ./... +``` + +基础竞态测试、vet、三个 CLI 构建及独立 SDK module 验证通过。无集成环境的普通 go test 会跳过真实服务用例，实际集成证据来自 scripts/test-mysql.sh。 + +本次完整集成结果： + +| 分组 | 结果 | +| --- | --- | +| 全部真实 MySQL 集成 | PASS，71.436 秒 | +| 资源键七组数据库场景 | PASS，4.36 秒 | +| v7 无外键升级 | PASS，3.43 秒，70 → 0，325 个既有索引列定义保留 | +| v9 应用额度回填及当前迁移 | PASS，3.81 秒 | +| 历史分页、权限、精确结果和证据 | PASS，0.48 秒 | +| SDK Worker 连接及运行器 | PASS，82.224 秒 | +| 其中资源键、四任务、双队列及 Start ACK 丢失 | PASS，3.32 秒 | +| SDK 计划 / 队列资源 / 应用资源 | PASS，分别 5.634 / 4.697 / 1.372 秒（含包开销） | +| 六项真实进程故障组 | PASS，25.173 秒 | +| 其中资源键跨节点恢复 | PASS，2.27 秒 | +| 十一个完整示例 | 全部通过 | + +示例包括 health、tasks、worker-protocol、worker、history、retry、schedules、cancellation、resources、application-resources 和 keyed-concurrency。上述时长是回归执行时间，不代表吞吐或生产容量。 + +完整套件后进一步移除了空资源键判断对数据库文本排序规则的依赖，并补充仅含组合符和零宽字符的非空键回归。最终定向审计再次通过资源键八组场景、v7 零外键及索引/数据保留、v9 配额迁移和 v4 安全重试升级，总包耗时 21.104 秒。新旧普通任务、计划模板、特殊 UTF-8 键和回滚路径均在这些真实数据库用例中验证。 + +## 3. 数据库与协议证据 + +[数据库用例](../../internal/storage/mysql/keyed_concurrency_integration_test.go) 验证： + +1. 老 Worker 对有键任务不创建尝试；相同 business_key 的无键任务仍可并行；显式启用后获得许可，原进程更改支持声明被拒绝，旧 false 注册摘要仍可重连。 +2. 两个节点、两个队列竞争同一键仅一个领取；队列和键各有一条许可，但应用只计一个尝试。未开始取消后另一个队列拿到更大序号；Start 重报、续租、尝试详情和完成后计数正确。 +3. 前面 20 个同键阻塞任务不遮挡后面的其他资源；阻塞项尝试次数为零。同应用另一个命名空间的相同键独立执行。 +4. 明确失败退避和 UNKNOWN 正确释放执行许可；待核查仍占积压。旧迟到结果仅保存证据，不能释放另一个新任务的许可；未开始过期同样回收。 +5. 领取事件写入失败时，尝试、键行、两类许可和应用计数一起回滚。大小写、组合及分解字符按精确字节区分；仅含组合符或零宽字符的非空键不被文本排序规则折叠为空。 +6. 接近 uint64 上限时仍完整保存序号，最后一次释放可成功；后续授权返回冲突且不消耗尝试，不回绕。 +7. 不可变计划模板传递资源键；连续触发同键任务按许可逐个执行。非法首尾空白、控制字符被拒绝。 + +[协议契约测试](../../internal/transport/httpserver/worker_contract_test.go) 同步检查 hello 扩展、带键派发、超过浮点精度的序号，以及缺键、缺序号、数值型或零序号。 [SDK 线协议测试](../../sdk/go/keyed_concurrency_test.go) 使用真实 TLS WebSocket 检查未协商支持、非法键、字段配对和大序号精度。 + +本阶段修复了尝试详情增加序号后 SELECT/Scan 字段数量不匹配。完整回归还暴露历史查询夹具假设单次全库恢复扫描就能处理指定任务；其他用例留下更早的到期任务后，该假设不成立。用例改为有界分批恢复，仍要求原任务进入 WAITING、第二次尝试成功、分页和证据全部一致，没有降低业务断言。 + +## 4. SDK 与真实故障 + +[SDK 集成用例](../../sdk/go/keyed_concurrency_integration_test.go) 通过真实服务运行四条同键任务，分布在两个队列，各 Worker 容量为 4。处理器同资源活跃数始终为 1；Start 已提交但 ACK 丢失后自动重连，处理器总共执行四次，每条任务只有一次尝试。处理器取得的序号与列表及详情中的序号完全一致。 + +真实进程测试强杀并 Wait 确认节点 A 退出，再启动不同 node_id 的 B。原 Worker 身份 Resume 原 RUNNING 尝试；另一队列不能领取同键任务，尝试数保持零。原任务完成后另一队列得到更大序号；旧 Finish 重放不破坏新尝试，后者仍可续租和正常完成。 + +完整示例实际输出： + +```text +queues=2 completed=2 tokens_increasing=true +resource_writes=2 stale_write_rejected=true worker_drained=true +``` + +## 5. 当前边界 + +- 只保证同一命名空间、显式相同资源键的有效许可互斥。省略键或另一个命名空间不参与；不承诺提交顺序或跨队列公平性。 +- 超时释放不会物理中断旧外部请求；UNKNOWN 不自动冻结整个资源。外部系统须实现原子序号校验、操作幂等和必要的业务冻结。仅有最高已见序号不能拒绝尚未见到更高序号时的所有旧写入。 +- 序号不是认证凭据或执行授权。处理器必须经当前 Start/Resume 授权；历史查询和 task.assign 不能直接授权业务执行。 +- 一个 MySQL 主库与应用额度锁是当前一致性边界。键行不能随任务清理而重置；高基数键增长、锁等待、吞吐和恢复延迟仍需 P6 负载验证。 +- 停止旧写入节点后迁移，再启新服务和新 SDK；混合旧写入节点、数据库复制切换、跨故障域网络分区、备份恢复和实际 RPO/RTO 尚未验收。 +- 示例资源模型使用进程内内存，仅用于清晰展示校验位置，不代表生产资源存储实现。Weave、可靠订阅及整个平台生产验收仍在开发计划中。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 3b13d19..e1ff62e 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -179,4 +179,4 @@ task=<任务ID> state=SUCCEEDED attempts=1 evidence=1 reason=RESULT_CONFLICT dis ## 7. 后续交付 -P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) 与历史查询，并通过 [Schema 及进程故障阶段验收](../reports/phase-2.md)；P3 已实现 [安全重试](../api/retries.md)、[计划调度](../api/schedules.md)、[运行中取消](../api/cancellation.md) 、[队列资源控制及示例](../api/queues.md) 和 [应用共享配额及示例](../api/application-resources.md)，业务键许可、优先级/公平性和结果订阅继续开发；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 +P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) 与历史查询，并通过 [Schema 及进程故障阶段验收](../reports/phase-2.md)；P3 已实现 [安全重试](../api/retries.md)、[计划调度](../api/schedules.md)、[运行中取消](../api/cancellation.md) 、[队列资源控制及示例](../api/queues.md) 和 [应用共享配额及示例](../api/application-resources.md)，[资源键互斥和单调序号示例](../api/keyed-concurrency.md) 已实现，优先级/公平性和结果订阅继续开发；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index 2c57e18..1800376 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -63,9 +63,12 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { finished := retryStart(t, g) retryFinish(t, g, finished, "SUCCEEDED", false) assertApplicationCounts(t, f, 4, 2, 1) - // v10 只新增此表；其余列、索引与数据均是 v9 结构。 + // 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。 f.exec(`DROP TABLE orbit_application_resources`) - f.exec(`DELETE FROM orbit_schema_migrations WHERE version=10`) + f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_concurrency_key, DROP COLUMN concurrency_key, DROP COLUMN concurrency_key_digest`) + f.exec(`ALTER TABLE pulse_attempts DROP COLUMN concurrency_token`) + f.exec(`ALTER TABLE pulse_worker_sessions DROP COLUMN concurrency_keys`) + f.exec(`DELETE FROM orbit_schema_migrations WHERE version>=10`) if err := Migrate(ctx, db); err != nil { t.Fatal(err) } diff --git a/internal/storage/mysql/keyed_concurrency.go b/internal/storage/mysql/keyed_concurrency.go new file mode 100644 index 0000000..dae9251 --- /dev/null +++ b/internal/storage/mysql/keyed_concurrency.go @@ -0,0 +1,71 @@ +package mysql + +import ( + "context" + "crypto/sha256" + "database/sql" + "errors" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" +) + +func concurrencyDigest(key string) *core.Digest { + if key == "" { + return nil + } + digest := core.Digest(sha256.Sum256(append([]byte("orbit-concurrency-key-v1\x00"), []byte(key)...))) + return &digest +} + +// concurrencyKeyValue 在 Go 中判断空字节串，避免 MySQL 文本排序规则将某些字符视为空。 +func concurrencyKeyValue(key string) *string { + if key == "" { + return nil + } + return &key +} + +// keyEligibility 在 LIMIT 前排除已占用资源，避免前 16 个同键任务挡住其他资源。 +// 调用者已锁应用额度与队列；命名空间中的所有领取都遵守此顺序。 +const keyEligibility = `(concurrency_key_digest IS NULL OR NOT EXISTS ( + SELECT 1 FROM pulse_concurrency_limits k WHERE k.namespace_id=pulse_tasks.namespace_id + AND k.scope_key='key' AND k.key_digest=pulse_tasks.concurrency_key_digest AND k.used_permits>=k.max_permits))` + +// reserveConcurrencyKey 为一个显式资源键预占单个许可。零 token 表示未启用或暂不可用。 +// 序号使用资源行单调 revision；释放也递增，序号允许有间隔，禁止删除或重置资源行。 +func reserveConcurrencyKey(ctx context.Context, tx *sql.Tx, ns core.ID, key string) (core.ID, uint64, error) { + if key == "" { + return "", 0, nil + } + id, err := core.NewID() + if err != nil { + return "", 0, err + } + digest := concurrencyDigest(key) + _, err = tx.ExecContext(ctx, `INSERT INTO pulse_concurrency_limits(namespace_id,id,scope_key,key_digest,max_permits) VALUES (?,?,'key',?,1) ON DUPLICATE KEY UPDATE id=pulse_concurrency_limits.id`, ns, id, digest) + if err != nil { + return "", 0, err + } + var used, max uint32 + var revision uint64 + err = tx.QueryRowContext(ctx, `SELECT id,used_permits,max_permits,revision FROM pulse_concurrency_limits WHERE namespace_id=? AND scope_key='key' AND key_digest=? FOR UPDATE`, ns, digest).Scan(&id, &used, &max, &revision) + if err != nil { + return "", 0, err + } + if max != 1 { + return "", 0, errors.New("key permit limit must be one") + } + if used != 0 { + return id, 0, nil + } + // 为当前授权及后续释放各保留一次递增空间，不能溢出后重用旧序号。 + if revision > ^uint64(0)-2 { + return "", 0, fault.Conflict + } + r, err := tx.ExecContext(ctx, `UPDATE pulse_concurrency_limits SET used_permits=1,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=? AND used_permits=0 AND revision=?`, ns, id, revision) + if err := exactlyOne(r, err); err != nil { + return "", 0, err + } + return id, revision + 1, nil +} diff --git a/internal/storage/mysql/keyed_concurrency_integration_test.go b/internal/storage/mysql/keyed_concurrency_integration_test.go new file mode 100644 index 0000000..ba1ea7f --- /dev/null +++ b/internal/storage/mysql/keyed_concurrency_integration_test.go @@ -0,0 +1,333 @@ +package mysql + +import ( + "context" + "encoding/json" + "os" + "sync" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" +) + +func keyedWorker(t *testing.T, f *executionFixture, queue core.ID, node *worker.Service) *executionFixture { + t.Helper() + g := *f + g.resources.QueueID = queue + g.a = node + g.hello.InstanceID = executionID(t) + g.hello.QueueID = queue + g.hello.ConcurrencyKeys = true + var err error + g.binding, _, err = node.Connect(g.ctx, g.token, g.hello) + if err != nil { + t.Fatal(err) + } + return &g +} +func keyedSubmit(t *testing.T, f *executionFixture, key string, max uint32) service.TaskView { + t.Helper() + out, err := f.tasks.Submit(f.ctx, f.token, string(executionID(t)), service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), ConcurrencyKey: key, Retry: pulse.RetryPolicy{MaxAttempts: max, InitialDelayMS: 1000, MaxDelayMS: 30000, AttemptTimeoutMS: 30000}}) + if err != nil { + t.Fatal(err) + } + return out.Task +} +func keyedUsed(t *testing.T, f *executionFixture, key string) uint32 { + t.Helper() + var used uint32 + if err := f.db.QueryRowContext(f.ctx, `SELECT used_permits FROM pulse_concurrency_limits WHERE namespace_id=? AND scope_key='key' AND key_digest=?`, f.resources.NamespaceID, concurrencyDigest(key)).Scan(&used); err != nil { + t.Fatal(err) + } + return used +} +func noKeyClaim(t *testing.T, f *executionFixture) { + t.Helper() + if a, err := f.a.Claim(f.ctx, f.token, f.binding); err != nil || a != nil { + t.Fatal("expected no eligible claim", a, err) + } +} +func TestIntegrationKeyedConcurrency(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, stop := context.WithTimeout(context.Background(), 70*time.Second) + defer stop() + db, err := Open(ctx, dsn, 32, 8) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + fixture := func(t *testing.T) *executionFixture { + f := newExecutionFixture(t, ctx, db) + return keyedWorker(t, f, f.resources.QueueID, f.a) + } + t.Run("legacy_workers_and_tracking_keys_keep_their_contract", func(t *testing.T) { + old := newExecutionFixture(t, ctx, db) + keyed := keyedSubmit(t, old, "resource/one", 1) + noKeyClaim(t, old) + if old.get(keyed.ID).AttemptCount != 0 { + t.Fatal("legacy worker consumed keyed attempt") + } + for i := 0; i < 2; i++ { + if _, err := old.tasks.Submit(ctx, old.token, string(executionID(t)), service.SubmitInput{QueueID: old.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), BusinessKey: "same-tracking-key"}); err != nil { + t.Fatal(err) + } + } + a, _ := old.claim() + b, _ := old.claim() + if a.ConcurrencyToken != 0 || b.ConcurrencyToken != 0 { + t.Fatal("tracking key changed execution semantics") + } + f := keyedWorker(t, old, old.resources.QueueID, old.b) + a, _ = f.claim() + if a.Task.ID != keyed.ID || a.ConcurrencyToken == 0 { + t.Fatal("opted-in worker did not receive key token") + } + hello := f.hello + hello.ConcurrencyKeys = false + _, _, err := f.a.Connect(ctx, f.token, hello) + expectWorkerCode(t, err, "STATE_CONFLICT") + // 旧 flag=false 摘要必须继续允许旧进程重连。 + if _, _, err := old.a.Connect(ctx, old.token, old.hello); err != nil { + t.Fatal("legacy registration digest changed", err) + } + }) + t.Run("cross_queue_race_single_permit_and_monotonic_tokens", func(t *testing.T) { + f := fixture(t) + q, err := f.tasks.CreateQueue(ctx, f.token, "second-queue", service.QueueInput{Name: "second", MaxConcurrency: 4, RatePerSecond: 100}) + if err != nil { + t.Fatal(err) + } + g := keyedWorker(t, f, q.Queue.ID, f.b) + keyedSubmit(t, f, "object/42", 1) + keyedSubmit(t, g, "object/42", 1) + type claim struct { + f *executionFixture + a *worker.Assignment + err error + } + ch := make(chan claim, 2) + var wg sync.WaitGroup + for _, target := range []*executionFixture{f, g} { + wg.Add(1) + go func() { + defer wg.Done() + a, err := target.a.Claim(ctx, target.token, target.binding) + ch <- claim{target, a, err} + }() + } + wg.Wait() + close(ch) + var won claim + for v := range ch { + if v.err != nil { + t.Fatal(v.err) + } + if v.a != nil { + if won.a != nil { + t.Fatal("same key dispatched twice") + } + won = v + } + } + if won.a == nil || won.a.ConcurrencyToken == 0 || keyedUsed(t, f, "object/42") != 1 { + t.Fatal("missing key permit") + } + assertApplicationCounts(t, f, 2, 1, 1) + if f.count("pulse_permits", "attempt_id=? AND released_at IS NULL", won.a.AttemptID) != 2 { + t.Fatal("expected queue and key permits") + } + task := won.a.Task + if _, err := won.f.tasks.CancelTask(ctx, won.f.token, task.ID, "cancel", service.CancelInput{ExpectedRevision: task.Revision, Reason: "release unstarted"}); err != nil { + t.Fatal(err) + } + other := f + if won.f == f { + other = g + } + next, ref := other.claim() + if next.ConcurrencyToken <= won.a.ConcurrencyToken { + t.Fatal("key sequence reset across tasks") + } + message := executionID(t) + if _, err := other.a.Start(ctx, other.token, other.binding, message, ref); err != nil { + t.Fatal(err) + } + if _, err := other.a.Start(ctx, other.token, other.binding, message, ref); err != nil { + t.Fatal(err) + } + if _, err := other.a.Renew(ctx, other.token, other.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + history, err := other.tasks.GetTaskAttempt(ctx, other.token, ref.TaskID, ref.AttemptID) + if err != nil || history.ConcurrencyToken != next.ConcurrencyToken { + t.Fatal("token not preserved in history", err) + } + retryFinish(t, other, ref, "SUCCEEDED", false) + assertApplicationCounts(t, f, 0, 0, 0) + if keyedUsed(t, f, "object/42") != 0 { + t.Fatal("key permit leaked") + } + }) + t.Run("busy_keys_do_not_hide_later_keys_or_other_namespaces", func(t *testing.T) { + f := fixture(t) + keyedSubmit(t, f, "busy", 1) + f.claim() + for i := 0; i < 20; i++ { + keyedSubmit(t, f, "busy", 1) + } + free := keyedSubmit(t, f, "free", 1) + a, _ := f.claim() + if a.Task.ID != free.ID { + t.Fatal("blocked prefix hid free key") + } + if f.count("pulse_tasks", "concurrency_key='busy' AND state='WAITING' AND attempt_count<>0") != 0 { + t.Fatal("blocked keys consumed attempts") + } + g := newExecutionFixture(t, ctx, db, f.resources.ApplicationID) + g = keyedWorker(t, g, g.resources.QueueID, g.a) + keyedSubmit(t, g, "busy", 1) + a, _ = g.claim() + if a.ConcurrencyToken == 0 { + t.Fatal("namespace isolation lost") + } + }) + t.Run("unknown_late_result_and_retry_release_exactly_once", func(t *testing.T) { + f := fixture(t) + task := keyedSubmit(t, f, "resource/retry", 2) + first, ref := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + retryFinish(t, f, ref, "FAILED", true) + if keyedUsed(t, f, "resource/retry") != 0 { + t.Fatal("retry holds key while waiting") + } + makeRetryDue(f, task.ID) + second, ref2 := f.claim() + if second.ConcurrencyToken <= first.ConcurrencyToken { + t.Fatal("retry reused token") + } + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref2); err != nil { + t.Fatal(err) + } + retryFinish(t, f, ref2, "UNKNOWN", false) + assertApplicationCounts(t, f, 1, 0, 0) + if f.get(task.ID).State != pulse.TaskNeedsReview { + t.Fatal("unknown result lost") + } + keyedSubmit(t, f, "resource/retry", 1) + third, ref3 := f.claim() + if third.ConcurrencyToken <= second.ConcurrencyToken { + t.Fatal("key epoch did not advance") + } + late, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref2, Outcome: "SUCCEEDED", Result: json.RawMessage(`{}`)}) + if err != nil || late.Receipt.Disposition != "EVIDENCE_ONLY" { + t.Fatal("late result applied", err) + } + if keyedUsed(t, f, "resource/retry") != 1 { + t.Fatal("late result released another attempt's key") + } + f.expire(ref3) + if _, err := f.storeB.Recover(ctx, 100); err != nil { + t.Fatal(err) + } + if keyedUsed(t, f, "resource/retry") != 0 { + t.Fatal("expiry leaked key") + } + assertApplicationCounts(t, f, 1, 0, 0) + }) + t.Run("claim_rollback_and_exact_utf8_key_identity", func(t *testing.T) { + f := fixture(t) + keyedSubmit(t, f, "asset/A", 1) + trigger := "test_key_event_" + string(executionID(t)) + f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected claim failure'") + out, err := f.a.Claim(ctx, f.token, f.binding) + f.exec("DROP TRIGGER " + trigger) + if err == nil || out != nil { + t.Fatal("failed claim exposed permit") + } + assertApplicationCounts(t, f, 1, 0, 0) + if f.count("pulse_attempts", "1=1") != 0 || f.count("pulse_permits", "1=1") != 0 || f.count("pulse_concurrency_limits", "scope_key='key'") != 0 { + t.Fatal("claim did not fully roll back") + } + f.claim() + for _, key := range []string{"asset/a", "asset/é", "asset/e\u0301"} { + keyedSubmit(t, f, key, 1) + f.claim() + } + if f.count("pulse_concurrency_limits", "scope_key='key' AND used_permits=1") != 4 { + t.Fatal("resource keys normalized unexpectedly") + } + }) + t.Run("sequence_exhaustion_is_not_allowed_to_wrap", func(t *testing.T) { + f := fixture(t) + keyedSubmit(t, f, "exhausted", 1) + ref := retryStart(t, f) + retryFinish(t, f, ref, "SUCCEEDED", false) + f.exec(`UPDATE pulse_concurrency_limits SET revision=18446744073709551613 WHERE namespace_id=? AND scope_key='key'`, f.resources.NamespaceID) + keyedSubmit(t, f, "exhausted", 1) + a, ref := f.claim() + if a.ConcurrencyToken != ^uint64(0)-1 { + t.Fatal("high token truncated") + } + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + retryFinish(t, f, ref, "SUCCEEDED", false) + waiting := keyedSubmit(t, f, "exhausted", 1) + _, err := f.a.Claim(ctx, f.token, f.binding) + expectWorkerCode(t, err, "STATE_CONFLICT") + if f.get(waiting.ID).AttemptCount != 0 { + t.Fatal("exhausted sequence consumed task attempt") + } + }) + t.Run("nonempty_keys_do_not_use_text_collation_for_null", func(t *testing.T) { + f := fixture(t) + for _, key := range []string{"\u0301", "\u200b"} { + task := keyedSubmit(t, f, key, 1) + if task.ConcurrencyKey != key { + t.Fatal("nonempty key collapsed by database collation") + } + a, _ := f.claim() + if a.Task.ID != task.ID || a.Task.ConcurrencyKey != key || a.ConcurrencyToken == 0 { + t.Fatal("exact key did not reach assignment") + } + } + }) + t.Run("schedule_template_and_key_validation", func(t *testing.T) { + f := fixture(t) + sf := newScheduleFixture(t, f) + def := sf.definition() + def.Task.ConcurrencyKey = "schedule/resource" + v := sf.create(def) + sf.due(v, 2*time.Second) + sf.tick() + a, ref := f.claim() + if a.Task.ConcurrencyKey != def.Task.ConcurrencyKey || a.ConcurrencyToken == 0 { + t.Fatal("schedule omitted frozen key") + } + noKeyClaim(t, f) + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + retryFinish(t, f, ref, "SUCCEEDED", false) + b, _ := f.claim() + if b.ConcurrencyToken <= a.ConcurrencyToken { + t.Fatal("schedule reused key token") + } + for _, key := range []string{" leading", "trailing ", "line\nfeed", "\u0000"} { + _, err := f.tasks.Submit(ctx, f.token, string(executionID(t)), service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), ConcurrencyKey: key}) + expectWorkerCode(t, err, "INVALID_ARGUMENT") + } + }) +} diff --git a/internal/storage/mysql/migrations/000011_keyed_concurrency.up.sql b/internal/storage/mysql/migrations/000011_keyed_concurrency.up.sql new file mode 100644 index 0000000..fdeafd5 --- /dev/null +++ b/internal/storage/mysql/migrations/000011_keyed_concurrency.up.sql @@ -0,0 +1,10 @@ +-- 显式资源键互斥：旧业务关联键语义不变；旧任务、会话与尝试默认不启用。 +-- 不使用外键，归属、许可与单调序号在任务事务内维护。 +ALTER TABLE pulse_tasks + ADD COLUMN concurrency_key VARCHAR(191) NULL COMMENT '显式互斥资源键，同命名空间跨队列共享；空表示不启用，不代替业务幂等', + ADD COLUMN concurrency_key_digest BINARY(32) NULL COMMENT '资源键精确 UTF-8 字节的域分隔 SHA-256；与资源键同时为空或有值', + ADD CONSTRAINT ck_task_concurrency_key CHECK ((concurrency_key IS NULL) = (concurrency_key_digest IS NULL)); +ALTER TABLE pulse_attempts + ADD COLUMN concurrency_token BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '当前资源键的单调许可序号，零表示未启用；重连不变，新尝试递增，不可单独证明执行授权'; +ALTER TABLE pulse_worker_sessions + ADD COLUMN concurrency_keys BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否明确支持资源键互斥与序号交付；未声明的旧 Worker 不领取带资源键任务'; diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index 4d5c763..4bb756f 100644 --- a/internal/storage/mysql/no_foreign_keys_integration_test.go +++ b/internal/storage/mysql/no_foreign_keys_integration_test.go @@ -133,12 +133,18 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { if err := rows.Scan(scan...); err != nil { t.Fatal(err) } - result := make([]string, len(cols)) + result := make([]string, 0, len(cols)) for i, value := range values { + if cols[i] == "concurrency_key" || cols[i] == "concurrency_key_digest" { + if value.Valid { + t.Fatal("legacy task unexpectedly enabled keyed concurrency") + } + continue + } if value.Valid { - result[i] = cols[i] + "=" + value.String + result = append(result, cols[i]+"="+value.String) } else { - result[i] = cols[i] + "=" + result = append(result, cols[i]+"=") } } if rows.Next() || rows.Err() != nil { diff --git a/internal/storage/mysql/schedule_tick.go b/internal/storage/mysql/schedule_tick.go index 8430af2..4560da2 100644 --- a/internal/storage/mysql/schedule_tick.go +++ b/internal/storage/mysql/schedule_tick.go @@ -224,7 +224,7 @@ func (s *ScheduleStore) saveOccurrence(ctx context.Context, tx *sql.Tx, v schedu if err != nil { return err } - _, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,owner_kind,owner_id,available_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),'schedule',?,?,?,?,?,?,?)`, v.NamespaceID, value, v.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, v.ID, now, now.Add(time.Duration(in.TaskTimeoutMS)*time.Millisecond), in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now) + _, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,concurrency_key,concurrency_key_digest,owner_kind,owner_id,available_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,'schedule',?,?,?,?,?,?,?)`, v.NamespaceID, value, v.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), v.ID, now, now.Add(time.Duration(in.TaskTimeoutMS)*time.Millisecond), in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now) if err != nil { return err } diff --git a/internal/storage/mysql/task_history.go b/internal/storage/mysql/task_history.go index 5905f0e..cc60b24 100644 --- a/internal/storage/mysql/task_history.go +++ b/internal/storage/mysql/task_history.go @@ -91,11 +91,11 @@ func historyAttemptExists(ctx context.Context, tx *sql.Tx, ns, task, attempt cor return err } -const attemptHistoryColumns = `a.id,a.task_id,a.epoch,a.state,a.revision,a.lease_until,a.authorized_at,a.ended_at,COALESCE(a.error_code,''),a.retryable,a.created_at,a.updated_at` +const attemptHistoryColumns = `a.id,a.task_id,a.epoch,a.state,a.revision,a.lease_until,a.authorized_at,a.ended_at,COALESCE(a.error_code,''),a.retryable,a.created_at,a.updated_at,a.concurrency_token` func scanAttemptSummary(row interface{ Scan(...any) error }) (service.AttemptSummary, error) { var a service.AttemptSummary - err := row.Scan(&a.ID, &a.TaskID, &a.Epoch, &a.State, &a.Revision, &a.LeaseUntil, &a.AuthorizedAt, &a.EndedAt, &a.ErrorCode, &a.Retryable, &a.CreatedAt, &a.UpdatedAt) + err := row.Scan(&a.ID, &a.TaskID, &a.Epoch, &a.State, &a.Revision, &a.LeaseUntil, &a.AuthorizedAt, &a.EndedAt, &a.ErrorCode, &a.Retryable, &a.CreatedAt, &a.UpdatedAt, &a.ConcurrencyToken) return a, err } func (s *TaskStore) ListTaskAttempts(ctx context.Context, t auth.Token, task core.ID, in service.HistoryPageInput) (service.AttemptPage, error) { @@ -148,7 +148,7 @@ func (s *TaskStore) GetTaskAttempt(ctx context.Context, t auth.Token, task, atte return err } var result []byte - err := tx.QueryRowContext(ctx, `SELECT `+attemptHistoryColumns+`,a.session_id,w.worker_instance_id,a.result FROM pulse_attempts a JOIN pulse_worker_sessions w ON w.namespace_id=a.namespace_id AND w.id=a.session_id WHERE a.namespace_id=? AND a.task_id=? AND a.id=?`, t.NamespaceID, task, attempt).Scan(&out.ID, &out.TaskID, &out.Epoch, &out.State, &out.Revision, &out.LeaseUntil, &out.AuthorizedAt, &out.EndedAt, &out.ErrorCode, &out.Retryable, &out.CreatedAt, &out.UpdatedAt, &out.SessionID, &out.WorkerInstanceID, &result) + err := tx.QueryRowContext(ctx, `SELECT `+attemptHistoryColumns+`,a.session_id,w.worker_instance_id,a.result FROM pulse_attempts a JOIN pulse_worker_sessions w ON w.namespace_id=a.namespace_id AND w.id=a.session_id WHERE a.namespace_id=? AND a.task_id=? AND a.id=?`, t.NamespaceID, task, attempt).Scan(&out.ID, &out.TaskID, &out.Epoch, &out.State, &out.Revision, &out.LeaseUntil, &out.AuthorizedAt, &out.EndedAt, &out.ErrorCode, &out.Retryable, &out.CreatedAt, &out.UpdatedAt, &out.ConcurrencyToken, &out.SessionID, &out.WorkerInstanceID, &result) if errors.Is(err, sql.ErrNoRows) { return fault.NotFound } diff --git a/internal/storage/mysql/task_history_integration_test.go b/internal/storage/mysql/task_history_integration_test.go index b20e8a7..e045227 100644 --- a/internal/storage/mysql/task_history_integration_test.go +++ b/internal/storage/mysql/task_history_integration_test.go @@ -43,8 +43,17 @@ func TestIntegrationTaskHistory(t *testing.T) { t.Fatal(err) } f.expire(first) - if _, err := f.storeA.Recover(ctx, 10); err != nil { - t.Fatal(err) + // 扫描按全库到期顺序分批；完整套件可能留下比本任务更早到期的其他夹具。 + for round := 0; ; round++ { + if _, err := f.storeA.Recover(ctx, 10); err != nil { + t.Fatal(err) + } + if f.get(task.ID).State == "WAITING" { + break + } + if round == 19 { + t.Fatal("history fixture was not recovered within bounded batches") + } } f.exec(`UPDATE pulse_tasks SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID) _, second := f.claim() diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index 74af3d9..c77014a 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -195,8 +195,8 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in if err := reserveQueueTask(ctx, tx, token.NamespaceID, in.QueueID); err != nil { return err } - _, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,available_at,start_deadline_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at) - VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,?,?,?,?,?)`, token.NamespaceID, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now) + _, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,concurrency_key,concurrency_key_digest,available_at,start_deadline_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,?,?,?,?,?,?,?)`, token.NamespaceID, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now) if err != nil { return err } @@ -215,14 +215,14 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in return out, err } -const taskColumns = `id,namespace_id,queue_id,state,revision,handler,handler_version,payload_version,payload,result,COALESCE(business_key,''),available_at,start_deadline_at,deadline_at,recovery_safety,policy_snapshot,attempt_count,cancel_requested_at,completed_at,created_at,updated_at,effect_uncertain` +const taskColumns = `id,namespace_id,queue_id,state,revision,handler,handler_version,payload_version,payload,result,COALESCE(business_key,''),COALESCE(concurrency_key,''),available_at,start_deadline_at,deadline_at,recovery_safety,policy_snapshot,attempt_count,cancel_requested_at,completed_at,created_at,updated_at,effect_uncertain` type scanner interface{ Scan(...any) error } func scanTask(row scanner) (service.TaskView, error) { var v service.TaskView var policy, payload, result []byte - err := row.Scan(&v.ID, &v.NamespaceID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.PayloadVersion, &payload, &result, &v.BusinessKey, &v.AvailableAt, &v.StartDeadlineAt, &v.DeadlineAt, &v.RecoverySafety, &policy, &v.AttemptCount, &v.CancelRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain) + err := row.Scan(&v.ID, &v.NamespaceID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.PayloadVersion, &payload, &result, &v.BusinessKey, &v.ConcurrencyKey, &v.AvailableAt, &v.StartDeadlineAt, &v.DeadlineAt, &v.RecoverySafety, &policy, &v.AttemptCount, &v.CancelRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain) if err != nil { return v, err } @@ -272,7 +272,7 @@ func (s *TaskStore) ListTasks(ctx context.Context, token auth.Token, in service. if !p.AllowsQueue(in.QueueID) { return fault.Forbidden } - q := `SELECT id,queue_id,state,revision,handler,handler_version,COALESCE(business_key,''),available_at,deadline_at,attempt_count,created_at,updated_at,effect_uncertain FROM pulse_tasks WHERE namespace_id=? AND queue_id=?` + q := `SELECT id,queue_id,state,revision,handler,handler_version,COALESCE(business_key,''),COALESCE(concurrency_key,''),available_at,deadline_at,attempt_count,created_at,updated_at,effect_uncertain FROM pulse_tasks WHERE namespace_id=? AND queue_id=?` args := []any{token.NamespaceID, in.QueueID} if in.State != "" { q += " AND state=?" @@ -296,7 +296,7 @@ func (s *TaskStore) ListTasks(ctx context.Context, token auth.Token, in service. defer rows.Close() for rows.Next() { var v service.TaskSummary - if err := rows.Scan(&v.ID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.BusinessKey, &v.AvailableAt, &v.DeadlineAt, &v.AttemptCount, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain); err != nil { + if err := rows.Scan(&v.ID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.BusinessKey, &v.ConcurrencyKey, &v.AvailableAt, &v.DeadlineAt, &v.AttemptCount, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain); err != nil { return err } out.Items = append(out.Items, v) diff --git a/internal/storage/mysql/worker_execution.go b/internal/storage/mysql/worker_execution.go index 96d36e0..ca1f368 100644 --- a/internal/storage/mysql/worker_execution.go +++ b/internal/storage/mysql/worker_execution.go @@ -148,7 +148,11 @@ func (s *ExecutionStore) Claim(ctx context.Context, token auth.Token, b worker.B if len(matches) == 0 { return nil } - rows, err := tx.QueryContext(ctx, `SELECT id,gate_id FROM pulse_tasks WHERE namespace_id=? AND queue_id=? AND state='WAITING' AND available_at<=UTC_TIMESTAMP(6) AND deadline_at>UTC_TIMESTAMP(6) AND (start_deadline_at IS NULL OR start_deadline_at>UTC_TIMESTAMP(6)) AND attempt_countUTC_TIMESTAMP(6) AND (start_deadline_at IS NULL OR start_deadline_at>UTC_TIMESTAMP(6)) AND attempt_count 191 || strings.ContainsAny(in.BusinessKey, "\x00\r\n") { return in, fault.Invalid("business_key must be at most 191 characters without control newlines") } + if err := ValidateConcurrencyKey(in.ConcurrencyKey); err != nil { + return in, err + } if in.TaskTimeoutMS == 0 { in.TaskTimeoutMS = 24 * 60 * 60 * 1000 } @@ -170,3 +174,19 @@ func (s *Service) ConfigureQueue(ctx context.Context, token auth.Token, id core. } return s.store.ConfigureQueue(ctx, token, id, key, in) } + +// ValidateConcurrencyKey 不归一化资源键，字节精确相等才共享许可。 +func ValidateConcurrencyKey(key string) error { + if key == "" { + return nil + } + if !utf8.ValidString(key) || utf8.RuneCountInString(key) > 191 || strings.TrimSpace(key) != key { + return fault.Invalid("invalid concurrency_key") + } + for _, c := range key { + if unicode.IsControl(c) { + return fault.Invalid("invalid concurrency_key") + } + } + return nil +} diff --git a/pulse/service/types.go b/pulse/service/types.go index a21ff1b..f7dee74 100644 --- a/pulse/service/types.go +++ b/pulse/service/types.go @@ -25,6 +25,8 @@ type SubmitInput struct { Payload json.RawMessage `json:"payload"` // BusinessKey 可选关联键，不提供业务效果幂等保证。 BusinessKey string `json:"business_key,omitempty"` + // ConcurrencyKey 显式互斥资源键，同命名空间跨队列最多一个有效尝试；不代替业务幂等。 + ConcurrencyKey string `json:"concurrency_key,omitempty"` // AvailableAt 最早执行时间；为空表示受理后即可进入派发范围。 AvailableAt *time.Time `json:"available_at,omitempty"` // StartDeadlineAt 最晚开始时间；为空表示使用任务总截止时间。 @@ -63,6 +65,8 @@ type TaskView struct { EffectUncertain bool `json:"effect_uncertain,omitempty"` // BusinessKey 可选业务关联键，不自动提供业务效果幂等。 BusinessKey string `json:"business_key,omitempty"` + // ConcurrencyKey 显式互斥资源键，同命名空间跨队列最多一个有效尝试；不代替业务幂等。 + ConcurrencyKey string `json:"concurrency_key,omitempty"` // AvailableAt 最早可执行时间，UTC。 AvailableAt time.Time `json:"available_at"` // StartDeadlineAt 最晚开始时间；为空时使用任务总截止时间。 @@ -131,6 +135,8 @@ type TaskSummary struct { EffectUncertain bool `json:"effect_uncertain,omitempty"` // BusinessKey 可选业务关联键，不自动提供业务效果幂等。 BusinessKey string `json:"business_key,omitempty"` + // ConcurrencyKey 显式互斥资源键，同命名空间跨队列最多一个有效尝试；不代替业务幂等。 + ConcurrencyKey string `json:"concurrency_key,omitempty"` // AvailableAt 最早可执行时间，UTC。 AvailableAt time.Time `json:"available_at"` // DeadlineAt 任务总截止时间，UTC。 diff --git a/pulse/worker/types.go b/pulse/worker/types.go index 091bc8e..9f58c2f 100644 --- a/pulse/worker/types.go +++ b/pulse/worker/types.go @@ -21,6 +21,7 @@ type Capability struct { // ConnectInput 在每次连接时提交；同一进程恢复时保留实例和秘密。 type ConnectInput struct { + ConcurrencyKeys bool `json:"concurrency_keys,omitempty"` // ConcurrencyKeys 显式支持互斥资源键和单调序号交付，重连不可变。 CancellationNotifications bool `json:"cancellation_notifications,omitempty"` // CancellationNotifications 显式接受 task.cancel 提示；事实仍从持久化租约核对。 InstanceID core.ID `json:"instance_id"` // InstanceID 进程标识，重启换新值。 InstanceSecret string `json:"instance_secret"` // InstanceSecret 32 字节随机秘密，只保存摘要。 @@ -56,12 +57,13 @@ type CapacityInput struct { // Assignment 仅通过已认证执行连接交付，包含敏感尝试秘密。 type Assignment struct { - Task service.TaskView `json:"task"` // Task 受理快照及业务参数。 - AttemptID core.ID `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 领取时数据库时间。 + ConcurrencyToken uint64 `json:"concurrency_token,omitempty,string"` // ConcurrencyToken 资源键的单调许可序号；无资源键时为零，不能代替当前授权。 + Task service.TaskView `json:"task"` // Task 受理快照及业务参数。 + AttemptID core.ID `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 领取时数据库时间。 } // AttemptRef 证明对一次尝试的控制权；连接代次仍由 Binding 单独验证。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 5865c35..b7fc988 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -59,6 +59,7 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/schedules) (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/cancellation) (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/resources) +(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/keyed-concurrency) # 应用配额场景使用独立应用与两个命名空间，普通 Worker 凭据不扩权。 diff --git a/sdk/go/examples/keyed-concurrency/main.go b/sdk/go/examples/keyed-concurrency/main.go new file mode 100644 index 0000000..3e331e3 --- /dev/null +++ b/sdk/go/examples/keyed-concurrency/main.go @@ -0,0 +1,169 @@ +// keyed-concurrency 演示跨队列资源互斥及资源端单调序号校验。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "sync" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +var errStale = errors.New("stale resource token") + +// resource 仅用于本地示范。生产必须在资源存储事务内持久化最高序号和操作身份， +// 并与实际写入原子提交；Go mutex 无法保护其他进程或外部服务。 +type resource struct { + mu sync.Mutex + namespace, key string + highest uint64 + operations map[string]bool +} + +func (r *resource) apply(namespace, key string, token uint64, operation string) error { + r.mu.Lock() + defer r.mu.Unlock() + if namespace != r.namespace || key != r.key || token == 0 || token < r.highest { + return errStale + } + r.highest = token + // 同一操作的重报不产生第二次模拟写入。 + r.operations[operation] = true + return nil +} +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} +func run() error { + token := os.Getenv("ORBIT_TOKEN") + if path := os.Getenv("ORBIT_CREDENTIALS_FILE"); path != "" && token == "" { + raw, err := os.ReadFile(path) + if err != nil { + return err + } + var c struct { + Token string `json:"token"` + } + if err = json.Unmarshal(raw, &c); err != nil { + return err + } + token = c.Token + } + api, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + op, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + name := "keyed." + op + var queues []string + var tasks []orbit.Task + for i := 0; i < 2; i++ { + q, err := api.CreateQueue(ctx, fmt.Sprintf("queue-%s-%d", op, i), orbit.CreateQueueRequest{Name: fmt.Sprintf("%s.%d", name, i), MaxConcurrency: 4, RatePerSecond: 100}) + if err != nil { + return err + } + queues = append(queues, q.Queue.ID) + in := orbit.SubmitTaskRequest{QueueID: q.Queue.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"object_ref":"objects/shared"}`), ConcurrencyKey: "object/" + op, TaskTimeoutMS: 20000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 10000}} + out, err := api.SubmitTask(ctx, fmt.Sprintf("operation-%s-%d", op, i), in) + if err != nil { + return err + } + tasks = append(tasks, out.Task) + } + r := &resource{namespace: tasks[0].NamespaceID, key: tasks[0].ConcurrencyKey, operations: map[string]bool{}} + handler := orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: name, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + // 本示例每条任务表示一个独立操作，Task.ID 在安全重试期间保持不变。 + if err := r.apply(e.Task.NamespaceID, e.Task.ConcurrencyKey, e.ConcurrencyToken, e.Task.ID); err != nil { + return nil, &orbit.WorkerFailure{Code: "STALE_RESOURCE_TOKEN"} + } + return json.RawMessage(`{"updated":true}`), nil + }} + run, stop := context.WithCancel(ctx) + done := make(chan error, 2) + started := 0 + defer func() { + stop() + for i := 0; i < started; i++ { + <-done + } + }() + for _, queue := range queues { + w, err := orbit.NewWorker(api, orbit.WorkerOptions{QueueID: queue, Capacity: 4, ShutdownTimeout: 5 * time.Second, Handlers: []orbit.WorkerHandler{handler}}) + if err != nil { + return err + } + started++ + go func() { done <- w.Run(run) }() + } + var low, high uint64 + for _, task := range tasks { + for { + current, err := api.GetTask(ctx, task.ID) + if err != nil { + return err + } + if current.State == orbit.TaskSucceeded { + break + } + if current.State == orbit.TaskFailed || current.State == orbit.TaskCancelled || current.State == orbit.TaskNeedsReview { + return fmt.Errorf("unexpected task state: %s", current.State) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(80 * time.Millisecond): + } + } + history, err := api.ListTaskAttempts(ctx, task.ID, orbit.HistoryPageRequest{}) + if err != nil { + return err + } + if len(history.Items) != 1 || history.Items[0].ConcurrencyToken == 0 { + return errors.New("missing resource token") + } + v := history.Items[0].ConcurrencyToken + if low == 0 || v < low { + low = v + } + if v > high { + high = v + } + } + if low == high { + return errors.New("resource token reused") + } + if !errors.Is(r.apply(r.namespace, r.key, low, "late-write"), errStale) { + return errors.New("old resource token accepted") + } + r.mu.Lock() + count := len(r.operations) + r.mu.Unlock() + if count != 2 { + return errors.New("unexpected resource writes") + } + stop() + for started > 0 { + err := <-done + started-- + if err != nil { + return err + } + } + fmt.Printf("queues=2 completed=2 tokens_increasing=true\nresource_writes=%d stale_write_rejected=true worker_drained=true\n", count) + return nil +} diff --git a/sdk/go/history_types.go b/sdk/go/history_types.go index db66527..2800ca7 100644 --- a/sdk/go/history_types.go +++ b/sdk/go/history_types.go @@ -13,18 +13,19 @@ type HistoryPageRequest struct { // AttemptSummary 是执行尝试的查询快照，不提供当前执行授权。 type AttemptSummary struct { - ID string `json:"id"` // ID 尝试标识，不是执行秘密。 - TaskID string `json:"task_id"` // TaskID 所属任务；查询始终限定在该任务下。 - Epoch uint64 `json:"epoch,string"` // Epoch 单任务执行代次，JSON 为十进制字符串。 - State AttemptState `json:"state"` // State 尝试状态，与整个任务状态分开。 - Revision uint64 `json:"revision,string"` // Revision 当前记录版本，可因续租/恢复变化。 - LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 数据库记录的租约期限；查询不能授权业务执行。 - AuthorizedAt *time.Time `json:"authorized_at,omitempty"` // AuthorizedAt 开始批准时间；空表示未批准，不证明副作用已经发生。 - EndedAt *time.Time `json:"ended_at,omitempty"` // EndedAt 尝试结束时间；仍在执行时为空。 - Retryable bool `json:"retryable,omitempty"` // Retryable 该明确失败报告是否声明可安全重试；不代表平台必定安排重试。 - ErrorCode string `json:"error_code,omitempty"` // ErrorCode 稳定错误分类；无错误时为空。 - CreatedAt time.Time `json:"created_at"` // CreatedAt 创建时间，UTC 微秒精度。 - UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最新变更时间，不能代替版本校验。 + ConcurrencyToken uint64 `json:"concurrency_token,omitempty,string"` // ConcurrencyToken 资源键的单调许可序号；无资源键时为零，不能代替当前授权。 + ID string `json:"id"` // ID 尝试标识，不是执行秘密。 + TaskID string `json:"task_id"` // TaskID 所属任务；查询始终限定在该任务下。 + Epoch uint64 `json:"epoch,string"` // Epoch 单任务执行代次，JSON 为十进制字符串。 + State AttemptState `json:"state"` // State 尝试状态，与整个任务状态分开。 + Revision uint64 `json:"revision,string"` // Revision 当前记录版本，可因续租/恢复变化。 + LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 数据库记录的租约期限；查询不能授权业务执行。 + AuthorizedAt *time.Time `json:"authorized_at,omitempty"` // AuthorizedAt 开始批准时间；空表示未批准，不证明副作用已经发生。 + EndedAt *time.Time `json:"ended_at,omitempty"` // EndedAt 尝试结束时间；仍在执行时为空。 + Retryable bool `json:"retryable,omitempty"` // Retryable 该明确失败报告是否声明可安全重试；不代表平台必定安排重试。 + ErrorCode string `json:"error_code,omitempty"` // ErrorCode 稳定错误分类；无错误时为空。 + CreatedAt time.Time `json:"created_at"` // CreatedAt 创建时间，UTC 微秒精度。 + UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最新变更时间，不能代替版本校验。 } // Attempt 增加当前归属与已保存结果，不返回令牌、摘要或原始请求秘密。 diff --git a/sdk/go/keyed_concurrency_integration_test.go b/sdk/go/keyed_concurrency_integration_test.go new file mode 100644 index 0000000..6c195d0 --- /dev/null +++ b/sdk/go/keyed_concurrency_integration_test.go @@ -0,0 +1,217 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestIntegrationWorkerRuntimeKeyedConcurrency(t *testing.T) { + api, _, token := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + name := "keyed." + workerTestID(t) + gateway := newWorkerFaultGateway(t, api.baseURL) + client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()}) + if err != nil { + t.Fatal(err) + } + var active, calls atomic.Int32 + var mu sync.Mutex + tokens := map[string]uint64{} + var last uint64 + handler := runtimeHandler(name, func(ctx context.Context, e WorkerExecution) (json.RawMessage, error) { + calls.Add(1) + if active.Add(1) != 1 { + t.Error("same resource handlers overlapped") + } + defer active.Add(-1) + mu.Lock() + if e.Task.ConcurrencyKey != name || e.ConcurrencyToken <= last { + t.Error("key token was lost, duplicated or reset") + } + last = e.ConcurrencyToken + tokens[e.Task.ID] = e.ConcurrencyToken + mu.Unlock() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(100 * time.Millisecond): + } + return e.Task.Payload, nil + }) + var tasks []Task + var workers []*Worker + for i := 0; i < 2; i++ { + q, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: name + workerTestID(t), MaxConcurrency: 4, RatePerSecond: 100}) + if err != nil { + t.Fatal(err) + } + for j := 0; j < 2; j++ { + out, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: q.Queue.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`), ConcurrencyKey: name}) + if err != nil { + t.Fatal(err) + } + tasks = append(tasks, out.Task) + } + w, err := NewWorker(client, WorkerOptions{QueueID: q.Queue.ID, Capacity: 4, Handlers: []WorkerHandler{handler}}) + if err != nil { + t.Fatal(err) + } + workers = append(workers, w) + } + // 在真实 Start 提交后丢 ACK；恢复必须沿用原 key/token，不能重跑处理器。 + gateway.drop.Store("task.start") + run, stop := context.WithCancel(ctx) + done := make(chan error, 2) + for _, w := range workers { + go func() { done <- w.Run(run) }() + } + defer func() { + stop() + for range workers { + if err := <-done; err != nil { + t.Error(err) + } + } + }() + for _, task := range tasks { + finished := waitRuntimeState(t, ctx, api, task.ID, TaskSucceeded) + if finished.AttemptCount != 1 || finished.ConcurrencyKey != name { + t.Fatal("keyed task lost identity") + } + history, err := api.ListTaskAttempts(ctx, task.ID, HistoryPageRequest{}) + if err != nil || len(history.Items) != 1 { + t.Fatal("missing keyed history", err) + } + mu.Lock() + want := tokens[task.ID] + mu.Unlock() + if want == 0 || history.Items[0].ConcurrencyToken != want { + t.Fatal("history token differs from handler") + } + detail, err := api.GetTaskAttempt(ctx, task.ID, history.Items[0].ID) + if err != nil || detail.ConcurrencyToken != want { + t.Fatal("detail token differs from handler", err) + } + } + gateway.expectDrop(t, "task.start") + if calls.Load() != 4 || active.Load() != 0 { + t.Fatal("incorrect handler count") + } +} + +func TestIntegrationWorkerProcessKeyedConcurrency(t *testing.T) { + a := startOrbitProcess(t) + _, _, token := integrationWorkerClient(t) + api := processClient(t, a, token) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + name := "key-crash." + workerTestID(t) + var registrations []WorkerRegistration + var tasks []Task + for i := 0; i < 2; i++ { + q, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: name + workerTestID(t), MaxConcurrency: 2, RatePerSecond: 100}) + if err != nil { + t.Fatal(err) + } + r, err := NewWorkerRegistration(q.Queue.ID, 1, []WorkerCapability{{Handler: name, HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + t.Fatal(err) + } + r.ConcurrencyKeys = true + registrations = append(registrations, r) + out, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: q.Queue.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), ConcurrencyKey: name}) + if err != nil { + t.Fatal(err) + } + tasks = append(tasks, out.Task) + } + dial := func(api *Client, r WorkerRegistration) *WorkerConnection { + t.Helper() + w, err := api.DialWorker(ctx, r) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { w.Close() }) + return w + } + capacity := func(w *WorkerConnection) { + t.Helper() + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + t.Fatal(err) + } + } + w1 := dial(api, registrations[0]) + capacity(w1) + first, err := w1.NextAssignment(ctx) + if err != nil || first.ConcurrencyToken == 0 { + t.Fatal("first key not granted", err) + } + started, err := w1.Start(ctx, workerTestID(t), first.Attempt()) + if err != nil { + t.Fatal(err) + } + if _, active := started.Lease.ExecutionDeadline(); !active { + t.Fatal("no current execution authorization") + } + a.kill(t) + b := startOrbitProcess(t) + api = processClient(t, b, token) + if a.nodeID == b.nodeID { + t.Fatal("node id reused") + } + w1 = dial(api, registrations[0]) + lease, err := w1.Resume(ctx, workerTestID(t), first.Attempt()) + if err != nil { + t.Fatal(err) + } + if _, active := lease.ExecutionDeadline(); !active { + t.Fatal("original attempt not resumed") + } + w2 := dial(api, registrations[1]) + capacity(w2) + brief, stop := context.WithTimeout(ctx, 300*time.Millisecond) + _, err = w2.NextAssignment(brief) + stop() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatal("restart lost key exclusion", err) + } + waiting, err := api.GetTask(ctx, tasks[1].ID) + if err != nil || waiting.AttemptCount != 0 { + t.Fatal("blocked key consumed attempt", err) + } + history, err := api.GetTaskAttempt(ctx, tasks[0].ID, first.AttemptID) + if err != nil || history.ConcurrencyToken != first.ConcurrencyToken { + t.Fatal("restart changed key token", err) + } + finishID := workerTestID(t) + finish := WorkerFinish{Ref: first.Attempt(), Outcome: WorkerSucceeded, Result: json.RawMessage(`{}`)} + if _, err = w1.Finish(ctx, finishID, finish); err != nil { + t.Fatal(err) + } + second, err := w2.NextAssignment(ctx) + if err != nil || second.ConcurrencyToken <= first.ConcurrencyToken { + t.Fatal("restart reset sequence", err) + } + if _, err = w2.Start(ctx, workerTestID(t), second.Attempt()); err != nil { + t.Fatal(err) + } + // 第二次许可已经开始，旧完成回执的重放不能释放它。 + if out, err := w1.Finish(ctx, finishID, finish); err != nil || !out.Deduplicated { + t.Fatal("duplicate finish changed ownership", err) + } + if _, err = w2.Renew(ctx, workerTestID(t), second.Attempt()); err != nil { + t.Fatal(err) + } + if _, err = w2.Finish(ctx, workerTestID(t), WorkerFinish{Ref: second.Attempt(), Outcome: WorkerSucceeded, Result: json.RawMessage(`{}`)}); err != nil { + t.Fatal(err) + } + for _, task := range tasks { + waitRuntimeState(t, ctx, api, task.ID, TaskSucceeded) + } +} diff --git a/sdk/go/keyed_concurrency_test.go b/sdk/go/keyed_concurrency_test.go new file mode 100644 index 0000000..34b157b --- /dev/null +++ b/sdk/go/keyed_concurrency_test.go @@ -0,0 +1,52 @@ +package orbit + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/coder/websocket" +) + +func TestWorkerConcurrencyKeyWireContract(t *testing.T) { + for _, tc := range []struct { + name string + support bool + key string + token uint64 + valid bool + }{ + {"legacy", false, "", 0, true}, + {"new_unkeyed", true, "", 0, true}, + {"large_counter", true, "object/资源", ^uint64(0) - 1, true}, + {"not_negotiated", false, "object/a", 2, false}, + {"missing_counter", true, "object/a", 0, false}, + {"unexpected_counter", true, "", 2, false}, + {"leading_space", true, " object/a", 2, false}, + {"control", true, "object/\n", 2, false}, + {"too_long", true, strings.Repeat("字", 192), 2, false}, + } { + t.Run(tc.name, func(t *testing.T) { + r := workerTestRegistration(t) + r.ConcurrencyKeys = tc.support + a := workerTestAssignment(r) + a.Task.ConcurrencyKey, a.ConcurrencyToken = tc.key, tc.token + _, w := workerMock(t, r, func(ctx context.Context, c *websocket.Conn) { + workerTestWrite(t, ctx, c, workerFrame{Type: "task.assign", Body: workerTestBody(t, a)}) + <-ctx.Done() + }) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + got, err := w.NextAssignment(ctx) + if tc.valid { + if err != nil || got.ConcurrencyToken != tc.token || got.Task.ConcurrencyKey != tc.key { + t.Fatalf("key/token did not survive wire: %v", err) + } + } else if !errors.Is(err, ErrWorkerProtocol) { + t.Fatalf("invalid assignment accepted: %v", err) + } + }) + } +} diff --git a/sdk/go/types.go b/sdk/go/types.go index 024f1c8..760239d 100644 --- a/sdk/go/types.go +++ b/sdk/go/types.go @@ -20,6 +20,8 @@ type SubmitTaskRequest struct { Payload json.RawMessage `json:"payload"` // BusinessKey 可选关联键，不提供业务效果幂等保证。 BusinessKey string `json:"business_key,omitempty"` + // ConcurrencyKey 显式互斥资源键，同命名空间跨队列最多一个有效尝试；不代替业务幂等。 + ConcurrencyKey string `json:"concurrency_key,omitempty"` // AvailableAt 最早执行时间；为空表示受理后即可进入派发范围。 AvailableAt *time.Time `json:"available_at,omitempty"` // StartDeadlineAt 最晚开始时间；为空表示使用任务总截止时间。 @@ -58,6 +60,8 @@ type Task struct { EffectUncertain bool `json:"effect_uncertain,omitempty"` // BusinessKey 可选业务关联键，不自动提供业务效果幂等。 BusinessKey string `json:"business_key,omitempty"` + // ConcurrencyKey 显式互斥资源键，同命名空间跨队列最多一个有效尝试；不代替业务幂等。 + ConcurrencyKey string `json:"concurrency_key,omitempty"` // AvailableAt 最早可执行时间，UTC。 AvailableAt time.Time `json:"available_at"` // StartDeadlineAt 最晚开始时间；为空时使用任务总截止时间。 @@ -126,6 +130,8 @@ type TaskSummary struct { EffectUncertain bool `json:"effect_uncertain,omitempty"` // BusinessKey 可选业务关联键，不自动提供业务效果幂等。 BusinessKey string `json:"business_key,omitempty"` + // ConcurrencyKey 显式互斥资源键，同命名空间跨队列最多一个有效尝试；不代替业务幂等。 + ConcurrencyKey string `json:"concurrency_key,omitempty"` // AvailableAt 最早可执行时间，UTC。 AvailableAt time.Time `json:"available_at"` // DeadlineAt 任务总截止时间，UTC。 diff --git a/sdk/go/worker_connection.go b/sdk/go/worker_connection.go index 32808b7..484dfb0 100644 --- a/sdk/go/worker_connection.go +++ b/sdk/go/worker_connection.go @@ -399,6 +399,9 @@ func (w *WorkerConnection) command(ctx context.Context, op, id string, ref Worke } func (w *WorkerConnection) validAssignment(a WorkerAssignment) bool { + if (a.Task.ConcurrencyKey == "") != (a.ConcurrencyToken == 0) || (a.Task.ConcurrencyKey != "" && (!w.registration.ConcurrencyKeys || !validConcurrencyKey(a.Task.ConcurrencyKey))) { + return false + } 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 } diff --git a/sdk/go/worker_runtime.go b/sdk/go/worker_runtime.go index de29e81..5585c53 100644 --- a/sdk/go/worker_runtime.go +++ b/sdk/go/worker_runtime.go @@ -202,7 +202,7 @@ func (r *workerRuntime) event(j *runtimeAttempt, kind, code string) { } func (r *workerRuntime) accept(a WorkerAssignment) error { if j := r.jobs[a.AttemptID]; j != nil { - if j.assignment.Attempt() != a.Attempt() { + if j.assignment.Attempt() != a.Attempt() || j.assignment.ConcurrencyToken != a.ConcurrencyToken || j.assignment.Task.ConcurrencyKey != a.Task.ConcurrencyKey { return ErrWorkerProtocol } return nil @@ -553,7 +553,7 @@ func (r *workerRuntime) receive(reply runtimeReply) error { 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} + execution := WorkerExecution{ConcurrencyToken: j.assignment.ConcurrencyToken, 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()) diff --git a/sdk/go/worker_runtime_types.go b/sdk/go/worker_runtime_types.go index 000e866..bb1f6c1 100644 --- a/sdk/go/worker_runtime_types.go +++ b/sdk/go/worker_runtime_types.go @@ -16,9 +16,10 @@ var ErrWorkerAlreadyRun = errors.New("Worker instance may only Run once") // WorkerExecution 是交给业务处理器的输入，不包含实例或尝试秘密。 type WorkerExecution struct { - Task Task // Task 固定任务输入；业务副作用应使用稳定业务身份去重。 - AttemptID string // AttemptID 当前尝试，用于核查与追踪，不代替业务幂等键。 - Epoch uint64 // Epoch 执行代次；接入方资源可用它拒绝旧尝试写入。 + ConcurrencyToken uint64 // ConcurrencyToken 同一命名空间资源键的单调序号，用于资源端拒绝旧执行写入。 + Task Task // Task 固定任务输入；业务副作用应使用稳定业务身份去重。 + AttemptID string // AttemptID 当前尝试，用于核查与追踪，不代替业务幂等键。 + Epoch uint64 // Epoch 单任务执行代次；跨任务资源隔离使用 ConcurrencyToken。 } func (e WorkerExecution) String() string { @@ -143,6 +144,7 @@ func NewWorker(client *Client, options WorkerOptions) (*Worker, error) { return nil, err } r.CancellationNotifications = true + r.ConcurrencyKeys = true options.Handlers = append([]WorkerHandler(nil), options.Handlers...) return &Worker{client: client, options: options, registration: r, handlers: handlers, events: make(chan WorkerEvent, 128)}, nil } diff --git a/sdk/go/worker_types.go b/sdk/go/worker_types.go index 8406502..d44ae29 100644 --- a/sdk/go/worker_types.go +++ b/sdk/go/worker_types.go @@ -8,7 +8,10 @@ import ( "errors" "fmt" "sort" + "strings" "time" + "unicode" + "unicode/utf8" ) // WorkerCapability 精确声明一种处理器与参数版本，不进行静默版本降级。 @@ -21,6 +24,7 @@ type WorkerCapability struct { // WorkerRegistration 是同一进程重连时必须复用的身份；进程重启使用新身份。 // 包含秘密，禁止以 JSON 等形式写入日志；SDK 的 String/GoString 会隐藏秘密。 type WorkerRegistration struct { + ConcurrencyKeys bool `json:"concurrency_keys,omitempty"` // ConcurrencyKeys 显式支持互斥资源键和单调序号交付，重连不可变。 CancellationNotifications bool `json:"cancellation_notifications,omitempty"` // CancellationNotifications 接受 task.cancel 提示；旧服务不支持，须先升级服务端。 InstanceID string `json:"instance_id"` // InstanceID 本进程随机标识，32 位小写十六进制。 InstanceSecret string `json:"instance_secret"` // InstanceSecret 32 字节随机秘密的 base64url 编码，重连保留。 @@ -71,12 +75,13 @@ type WorkerCapacity struct { // 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 领取时数据库时间。 + ConcurrencyToken uint64 `json:"concurrency_token,omitempty,string"` // ConcurrencyToken 资源键的单调许可序号；无资源键时为零，不能代替当前授权。 + 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 返回控制操作所需的证明；不要把证明转交业务处理器。 @@ -215,3 +220,15 @@ func normalizeRegistration(r WorkerRegistration) (WorkerRegistration, error) { func validWorkerAttempt(a WorkerAttempt) bool { return validID(a.TaskID) && validID(a.AttemptID) && a.Epoch > 0 && validWorkerSecret(a.LeaseSecret) } + +func validConcurrencyKey(key string) bool { + if !utf8.ValidString(key) || utf8.RuneCountInString(key) > 191 || strings.TrimSpace(key) != key { + return false + } + for _, c := range key { + if unicode.IsControl(c) { + return false + } + } + return true +}