From 5c8825e4bc40fb26289d802f849b6ae575188b43 Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 17:25:56 +0800 Subject: [PATCH] feat: enforce durable queue resource limits and backpressure --- README.md | 4 +- docs/07-data-model.md | 4 +- docs/08-development-plan.md | 4 +- docs/README.md | 2 +- docs/api/README.md | 2 +- docs/api/authentication.md | 2 +- docs/api/queues.md | 100 ++++- docs/api/schedules.md | 2 + docs/api/tasks.md | 1 + docs/api/worker.md | 4 +- docs/examples/README.md | 9 + docs/reports/phase-3-queue-resources.md | 75 ++++ docs/sdk/README.md | 2 +- docs/sdk/worker-runtime.md | 2 + docs/sdk/worker.md | 1 + internal/fault/fault.go | 1 + .../migrations/000009_queue_resources.up.sql | 15 + internal/storage/mysql/queue_resources.go | 131 ++++++ .../mysql/queue_resources_integration_test.go | 376 ++++++++++++++++++ .../mysql/retry_upgrade_integration_test.go | 6 + internal/storage/mysql/schedule_tick.go | 14 +- internal/storage/mysql/task_cancel.go | 5 + internal/storage/mysql/task_store.go | 13 +- internal/storage/mysql/worker_execution.go | 20 +- internal/storage/mysql/worker_recovery.go | 10 + internal/storage/mysql/worker_results.go | 5 + internal/transport/httpserver/tasks.go | 26 ++ pulse/model/queue.go | 11 +- pulse/service/service.go | 18 +- pulse/service/types.go | 32 +- pulse/worker/service.go | 1 + scripts/test-mysql.sh | 2 + sdk/go/examples/resources/main.go | 150 +++++++ sdk/go/queue_resources_integration_test.go | 115 ++++++ sdk/go/tasks.go | 21 +- sdk/go/types.go | 30 +- sdk/go/worker_process_integration_test.go | 43 ++ sdk/go/worker_runtime.go | 2 +- 38 files changed, 1231 insertions(+), 30 deletions(-) create mode 100644 docs/reports/phase-3-queue-resources.md create mode 100644 internal/storage/mysql/migrations/000009_queue_resources.up.sql create mode 100644 internal/storage/mysql/queue_resources.go create mode 100644 internal/storage/mysql/queue_resources_integration_test.go create mode 100644 sdk/go/examples/resources/main.go create mode 100644 sdk/go/queue_resources_integration_test.go diff --git a/README.md b/README.md index 851ad8f..619de25 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,9 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已完成 P0–P2 阶段验收：32 个数据模型、版本化迁移、凭据及范围校验、队列与任务 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) 已接入服务端和 SDK，剩余 Pulse 能力与流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已完成 P0–P2 阶段验收：32 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 和 [队列资源控制](docs/api/queues.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)。 +数据库禁止外键，关联一致性通过服务端事务校验，索引、唯一约束和 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 da9666f..c2574d8 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -1,6 +1,6 @@ # 数据模型与实现基线 -更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消与计划调度；剩余资源控制、订阅及流程编排继续开发。 +更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度和队列限速/背压；应用配额、公平性、订阅及流程编排继续开发。 ## 1. 模块与依赖 @@ -124,4 +124,6 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 000008_no_foreign_keys 移除历史 70 个外键，保留索引、唯一约束、CHECK、触发器和数据。旧迁移已在回归库应用，按不可变校验和约定保留原文件；新装按顺序运行至当前版本，最终结构同样为零外键。后续迁移禁止再引入外键。升级旧库时先停用旧服务，应用完整迁移再启动新版；迁移需要 DDL 锁，应在目标部署环境演练维护窗口。 +000009_queue_resources 增加 Queue.MaxUnfinishedTasks、UnfinishedTasks、StartWindowUntil、StartsInWindow，Go 字段与 SQL COMMENT 均有中文语义，实体仍为 32。未终结任务按 WAITING/LEASED/RUNNING/NEEDS_REVIEW 回填；窗口初始为空、计数为零。运行计数与任务/开始授权同事务维护，不改变队列配置 revision，配置变更不能重置它们。降额允许已有使用量暂时超过新上限，不删除任务或撤销已有许可；详见 [队列契约](api/queues.md) 和 [资源回归](reports/phase-3-queue-resources.md)。 + 命令与验证记录见 [开发计划](08-development-plan.md)、[基础阶段验证报告](reports/phase-0.md) 与 [取消及无外键回归](reports/phase-3-cancellation.md)。历史阶段报告描述当时结构，当前约束以本页为准。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 8312840..2d2d599 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 按真实拓扑验证 | 待开发 | @@ -44,6 +44,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 有限安全重试 | 明确失败 → 持久化退避 → 新尝试 → 查询完整历史 | P3，已实现完整示例 | | 定期数据同步 | 发布计划 → 触发任务 → 暂停/恢复 → 查看触发记录 | P3，已实现通用快照示例 | | 运行中取消 | 保存取消意图 → 停止处理器 → 确认结果 → 查询并排空 | P3，已实现完整示例 | +| 队列限速与背压 | 暂停 → 满额拒绝 → 原回执核对 → 释放空间 → 限速执行 | P3，已实现完整示例 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | | 审核与外部回执 | 提交流程 → 等待信号 → 授权信号注入 → 继续 | P4 | | 外部作业轮询 | 提交一次 → 有界查询处理中状态 → 回执或最终结果 → 超限核查 | P5 | @@ -80,6 +81,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P3 安全重试阶段记录](reports/phase-3-safe-retry.md) - [P3 计划调度阶段记录](reports/phase-3-schedules.md) - [P3 取消与无外键回归](reports/phase-3-cancellation.md) +- [P3 队列资源控制回归](reports/phase-3-queue-resources.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 4419f6f..fcdc1db 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 97082a1..1271572 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -11,7 +11,7 @@ | `/v1/tasks` 提交、查询、分页与待执行取消 | P1 已实现 | [任务接口](tasks.md) | | `/v1/tasks/{task_id}/attempts` 与尝试详情 | P2 已实现 | [执行历史](execution-history.md) | | `/v1/tasks/{task_id}/attempts/{attempt_id}/evidence` 与证据详情 | P2 已实现 | [执行历史](execution-history.md) | -| `/v1/queues` 创建与查询 | P1 已实现 | [队列接口](queues.md) | +| `/v1/queues` 创建、查询及 `:configure` 控制 | 已实现队列并发、开始速率、积压背压与暂停/恢复 | [队列接口与 SDK 场景](queues.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/authentication.md b/docs/api/authentication.md index 56bd4e2..fb50a0b 100644 --- a/docs/api/authentication.md +++ b/docs/api/authentication.md @@ -19,7 +19,7 @@ go run ./cmd/orbit-admin bootstrap \ ORBIT_MYSQL_DSN 通过环境注入，命令不读取 .env。所有名称必填，以字母/数字开头，后续允许字母、数字、`_ . : -`；environment 最长 32，其余最长 128。输出文件必须不存在，以 0600 创建；原始凭据不会打印到 stdout/stderr。`.local/` 和 `*.credentials.json` 已被 Git 忽略。 -输出 JSON 包含 application_id / namespace_id / queue_id / credential_id / token / expires_at。初始凭据有效期为 30 天，权限包括当前命名空间内的五个 P1 动作和全部队列。初始队列 max_concurrency=4、rate_per_second=10；这是开发初始配置，执行限流随 P2–P3 落地。 +输出 JSON 包含 application_id / namespace_id / queue_id / credential_id / token / expires_at。初始凭据有效期为 30 天，权限包括当前命名空间内的五个 P1 动作和全部队列。初始队列 max_concurrency=4、rate_per_second=10、max_unfinished_tasks=100000；并发、开始限速与积压背压已生效，语义及管理方式见 [队列接口](queues.md)。这些默认值是开发基线，不是容量承诺。 资源、凭据摘要和初始化审计在一个事务中提交。交付文件先写入并同步到磁盘，成功后再提交事务。如果文件写入失败则回滚并移除文件；如果已经交付后提交无法确认，保留文件并报错，应使用其中凭据核对结果，不能直接覆盖或重复创建。 diff --git a/docs/api/queues.md b/docs/api/queues.md index e1ba13a..e813d1e 100644 --- a/docs/api/queues.md +++ b/docs/api/queues.md @@ -1,6 +1,6 @@ # 队列接口 v1 -状态：P1 已实现创建与查询。更新、暂停、实际派发并发与速率控制随 P2–P3 实现；当前参数是后续执行所需的持久化配置。 +状态：已实现创建、查询、配置替换、暂停/恢复、共享并发、开始速率与未终结任务背压。MySQL 是跨节点额度的权威存储；没有额外缓存或消息队列依赖。 ## POST /v1/queues @@ -11,6 +11,7 @@ | name | string | 1–128 字符，以 ASCII 字母/数字开头，其后允许字母、数字、`_ . : -`；命名空间内唯一 | | max_concurrency | uint32 | 必填，1–100000 | | rate_per_second | uint32 | 必填，1–1000000 | +| max_unfinished_tasks | uint32 | 可选，省略/零为 100000；有效范围 1–1000000，包含待核查任务 | ```sh curl "$ORBIT_URL/v1/queues" \ @@ -30,7 +31,8 @@ curl "$ORBIT_URL/v1/queues" \ "queue":{ "id":"11111111111111111111111111111111", "namespace_id":"33333333333333333333333333333333", - "name":"images","paused":false,"max_concurrency":8,"rate_per_second":100,"revision":"1" + "name":"images","paused":false,"max_concurrency":8,"rate_per_second":100, + "max_unfinished_tasks":100000,"unfinished_tasks":"0","revision":"1" } } } @@ -49,3 +51,97 @@ curl "$ORBIT_URL/v1/queues/$ORBIT_QUEUE_ID" -H "Authorization: Bearer $ORBIT_TOK ``` SDK：`Client.GetQueue(ctx, queueID)`。范围限制、重复创建和同名冲突已纳入 [P1 回归](../reports/phase-1.md)。这些配置上限不是生产容量承诺。 + +unfinished_tasks 为 uint64 的十进制字符串，表示当前 WAITING、LEASED、RUNNING、NEEDS_REVIEW 的总数。它是运行计数，变化不增加队列配置 revision。创建/配置回执保存操作时的快照，获取实时积压需要再次 GetQueue。 + +## POST /v1/queues/{queue_id}:configure + +需要 queues:manage 及对应队列范围。完整替换控制参数，不改变名称、队列身份、已受理任务的冻结策略或历史结果。 + +| 字段 | 类型 | 要求 | +| --- | --- | --- | +| expected_revision | string | 必填，非零 uint64 的十进制字符串 | +| reason | string | 必填，非空白，最多 1024 个 Unicode 字符，写入审计 | +| paused | bool | 必须显式提供；省略/null 都拒绝 | +| max_concurrency | uint32 | 必填，1–100000 | +| rate_per_second | uint32 | 必填，1–1000000 | +| max_unfinished_tasks | uint32 | 必填，1–1000000；配置命令不使用零值默认 | + +```sh +curl "$ORBIT_URL/v1/queues/$ORBIT_QUEUE_ID:configure" \ + -H "Authorization: Bearer $ORBIT_TOKEN" \ + -H 'Content-Type: application/json' \ + -H 'Idempotency-Key: queue-control-001' \ + --data '{"expected_revision":"1","reason":"limit background traffic","paused":false,"max_concurrency":2,"rate_per_second":1,"max_unfinished_tasks":100}' +``` + +成功 200 的 data 形状与创建一致，队列配置 revision 加一。配置、并发限制同步、审计、事件和幂等回执在一个事务提交。原键、原 expected_revision 和原内容重报返回原快照，deduplicated=true；同键不同内容为 409 IDEMPOTENCY_CONFLICT，新键使用旧版本为 409 STATE_CONFLICT。权限每次重新检查。 + +SDK：`Client.ConfigureQueue(ctx, queueID, key, orbit.ConfigureQueueRequest{...})` 返回 CreateQueueResult。先 GetQueue 读取版本，网络结果不明时保留原命令核对；只有确认版本冲突并重新评估后，才建立新的配置操作。 + +## 并发、限速与背压的区别 + +| 控制 | 实际行为 | +| --- | --- | +| paused=true | 阻止新的领取和 Start；继续受理任务、计划生成、续租、结果确认、取消及恢复 | +| max_concurrency | 限制队列有效执行许可，包括 LEASED 和 RUNNING；许可在结果/取消/恢复事务中释放 | +| rate_per_second | 每个数据库一秒计数窗口内最多提交的 Start 授权数；续租和原回执重报不计数 | +| max_unfinished_tasks | 限制新任务进入队列；等待、退避、预占、运行、待核查都占一个名额 | + +速率窗口由第一次成功 Start 建立，保存到 MySQL；所有节点共享。窗口结束后的下一次 Start 建立新窗口。一次窗口内可突发至设定值，窗口交界附近可能出现接近两倍的短时突发；这不是“任意滑动一秒”的承诺，也不限制业务处理器内部发出请求的速率。开始授权成功不证明外部业务效果已经发生。 + +领取还预留尚未开始尝试所需的窗口空间，额度不足时保持 WAITING，不创建额外尝试。若领取后管理员降低速率，Start 可返回 WS RATE_LIMITED；不批准开始、不写回执、不增加尝试次数。自动 SDK 在同一租约内按原消息身份退避核对，底层客户端也必须遵守剩余租约，不能绕过开始授权执行。限流不延长总截止、开始截止或尝试预算。 + +降并发不撤销已有许可，超出新上限时等待在途工作收尾再领取；降速率保留当前窗口已用次数；降积压上限保留已有任务，禁止新受理直到低于新上限。调整参数、暂停/恢复、断开连接和节点重启都不能清零使用量。 + +任务第一次进入 SUCCEEDED、FAILED 或 CANCELLED 时，积压额度同事务减一；重试和待核查不释放、不重复增加。运行中取消只保存意图，仍占名额。迟到证据及重复确认不改变计数。人工核查解除入口仍待 P5，不能为了释放额度把未知结果改成成功/取消。 + +### HTTP 429 QUEUE_FULL + +满额时新的 Submit 返回 429，`Retry-After: 1`，错误信封为： + +```json +{"request_id":"44444444444444444444444444444444","error":{"code":"QUEUE_FULL","message":"queue unfinished task limit reached; retry later using the original identity"}} +``` + +该请求没有受理任务，新增占位幂等记录也回滚。已受理请求的同身份重报优先返回原回执，不被当前满额拒绝。释放空间后使用被拒请求的原键、原内容有限重试；建议至少等待 Retry-After 并加入抖动/总截止。若绝对 available_at/start_deadline_at 已过期，原请求会被参数规则拒绝，不能无限重试。 + +SDK 返回 `*orbit.APIError`，Code=QUEUE_FULL、StatusCode=429、RetryAfter=time.Second。RetryAfter 解析服务端整数秒建议，缺失/无效/超过一小时为零；SDK 不自动重报业务写请求。取消、查询、结果上报和管理配置不受受理满额限制。 + +计划触发也使用相同名额。部分补跑已生成的任务正常提交；满额的原触发点保留为下一游标，不创建假任务或跳过记录。下一批继续使用冻结的漏跑策略，因此等待期间超过补偿窗口的点仍可能按策略记录 SKIPPED。队列暂停仅控制执行，要停止计划生成应另外暂停计划。 + +## 完整 SDK 场景 + +源码：[resources/main.go](../../sdk/go/examples/resources/main.go)。先完成当前迁移并启动服务，初始化带 Worker 权限且 queues:manage 范围为全部队列的专用开发凭据： + +```sh +mkdir -p .local +chmod 700 .local +go run ./cmd/orbit-admin bootstrap \ + --application resources-demo --namespace default --environment development \ + --queue bootstrap --worker-access --output .local/resources.credentials.json +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/resources.credentials.json \ + go run ./examples/resources +``` + +示例创建独立队列（积压 2、并发 2、每秒 1 次开始），暂停后提交两个任务；第三个被拒，重复第一个仍得到原回执；取消第一个释放空间，用第三个原身份重新提交，然后恢复并执行两个纯数据回显任务。 + +```text +backpressure=QUEUE_FULL http=429 +duplicate=true rejected_request_retried=true +task=<第二个任务ID> state=SUCCEEDED attempts=1 +task=<第三个任务ID> state=SUCCEEDED attempts=1 +unfinished=0 rate_per_second=1 worker_drained=true +``` + +每次运行创建独立队列和任务，历史保留，不硬删除。无外部业务副作用；任务自带 20 秒总截止，失败非零退出并有界收尾。示例遇到未知网络结果会退出，不自动换键重建资源。 + +## 升级与未完成范围 + +000009 增加四个带中文说明的队列字段，回填所有既有未终结任务；旧配置版本及任务内容保持不变。先停止旧服务和写入，运行迁移，再启动新版。只读检查不能代替升级维护窗口；大表回填及混合版本升级待生产环境验证。 + +创建命令省略新字段时，幂等摘要保留旧字段集合。旧创建回执不含新字段，SDK 解码为零，只表示历史快照缺少信息，实时配置另行 GetQueue。配置接口是新能力；自动 SDK 已能处理 RATE_LIMITED，进行降额管理前应先升级 Worker。 + +本阶段验证见 [资源控制报告](../reports/phase-3-queue-resources.md)。仍需应用级跨命名空间配额、同业务键许可、可配置优先级及应用间公平性、可靠事件积压限制和容量验收；不能用单队列控制替代这些要求。 diff --git a/docs/api/schedules.md b/docs/api/schedules.md index f0fb032..15f5a25 100644 --- a/docs/api/schedules.md +++ b/docs/api/schedules.md @@ -56,6 +56,8 @@ Cron 段顺序为分钟、小时、月日、月、星期；支持 `*`、列表 所有任务策略在 task 内冻结，字段同 SubmitTaskRequest；payload 最大 256 KiB，数字文本保持精确。task 中禁止绝对 available_at / start_deadline_at，每次生成时立即可受理，task_timeout_ms 从本次实际创建时间计算。计划结束不取消已经生成的任务，也不缩短其任务时限。 +生成任务受 [队列积压上限](queues.md) 约束。满额时保留未受理触发点，本批较早已生成部分仍提交；释放空间后按冻结的漏跑策略接续，超过补偿窗口的点仍按策略跳过。队列 paused 仅停止领取/开始，停止计划生成需使用计划暂停接口。 + 创建请求示例，ID 和未来时间替换为实际值： ```json diff --git a/docs/api/tasks.md b/docs/api/tasks.md index 74bb8e2..10d7319 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -180,6 +180,7 @@ SDK：`Client.CancelTask(ctx, taskID, key, request)`。此操作不承诺回滚 | 409 STATE_CONFLICT | 当前状态或版本不允许操作，先查询 | | 413 BODY_TOO_LARGE | 减小请求或改为传递业务对象引用 | | 415 UNSUPPORTED_MEDIA_TYPE | 使用 application/json | +| 429 QUEUE_FULL | 本次未受理，按 Retry-After 有限退避并保留原身份；已受理的原回执不受当前满额限制 | | 503 UNAVAILABLE | 结果可能未知；保留原幂等身份，有限退避后核对 | ```json diff --git a/docs/api/worker.md b/docs/api/worker.md index 2cfe533..ac28a26 100644 --- a/docs/api/worker.md +++ b/docs/api/worker.md @@ -129,7 +129,9 @@ receipt 是原操作的不可变确认；lease 是本次重新核验的执行权 平台在同一事务中分配队列并发许可、创建尝试、递增任务执行代次和次数、保存 LEASED 与派发事件。领取本身消耗一次 max_attempts 预算，包括最终未获开始授权的尝试，预算不会因重连或过期恢复重置。 -运行中的进程总容量跨其所有旧/新会话统计。声明容量减少、断开连接或重复结果，都不能重复释放额度。当前已实施队列 max_concurrency；rate_per_second、其他配额和调度策略在 P3 完成。 +运行中的进程总容量跨其所有旧/新会话统计。声明容量减少、断开连接或重复结果，都不能重复释放额度。已实施队列 max_concurrency、rate_per_second 及未终结任务上限；参数含义、窗口突发、暂停及降额语义见 [队列资源控制](queues.md)。应用级配额、业务键许可与公平性仍在 P3 推进。 + +领取前核对速率窗口及已有 LEASED 预留，满额时不新建尝试。Start 在同一队列事务内消费开始预算，原 ACK 重报不重复计数。领取后若配置降额，Start 可能返回 RATE_LIMITED；保持原消息身份在剩余租约内退避核对。续租、结果确认、取消及恢复不消费新开始额度。 每个节点每 2 秒发起一批恢复扫描，每批最多 100 项： diff --git a/docs/examples/README.md b/docs/examples/README.md index 78f080a..d3993d5 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -72,6 +72,15 @@ - 异常：普通 context 错误、通知丢失和未配合停止不能自动判定取消成功；未知效果进入核查。CAS 冲突先查询，网络结果未知保留原身份核对。 - 清理：无外部业务副作用；历史保留，每次运行独立身份，失败非零退出并有界收尾。 +## 已可运行：队列限速与积压背压 + +- 源码：[资源控制示例](../../sdk/go/examples/resources/main.go)，[初始化、字段与完整输出](../api/queues.md)。 +- 前置：当前迁移与服务，凭据具备全部队列的 queues:manage、任务及 Worker 权限。 +- 步骤：创建并暂停积压上限为 2 的队列 → 第三个任务被 429 拒绝 → 满额下核对第一条原回执 → 取消释放空间 → 用被拒原身份再提交 → 恢复并按每秒 1 次开始执行。 +- 预期：QUEUE_FULL、duplicate=true、两个任务 SUCCEEDED/attempts=1、unfinished=0。 +- 异常：网络结果未知保留原身份，降额不清零用量，待核查任务仍占名额；跨节点及重启证据见 [资源回归](../reports/phase-3-queue-resources.md)。 +- 清理：每次新建独立队列，任务 20 秒总截止；无外部副作用，历史保留，失败非零退出并有界收尾。 + ## 计划场景与验收标准 | 场景 | 输入与步骤 | 预期结果 | 必测异常 | 状态 | diff --git a/docs/reports/phase-3-queue-resources.md b/docs/reports/phase-3-queue-resources.md new file mode 100644 index 0000000..d75e768 --- /dev/null +++ b/docs/reports/phase-3-queue-resources.md @@ -0,0 +1,75 @@ +# P3 队列资源控制回归记录 + +验证日期：2026-09-17。本阶段完成队列配置、并发降额、开始限速及未终结任务背压。P3 仍需应用级配额、业务键许可、优先级/公平性和可靠订阅；持续目标保持进行中。 + +## 1. 已实现范围 + +- `POST /v1/queues/{id}:configure` 完整替换暂停、并发、开始速率和未终结任务上限，带授权、CAS、稳定身份、事件和管理审计。 +- 基于数据库时间的一秒开始计数窗口，多节点共享；Start 事务消费额度，旧回执重报不重复计数，节点更换不清零。 +- 领取预留尚未开始的窗口空间，满额时保持 WAITING。领取后降额可使 Start 返回 RATE_LIMITED，自动 SDK 保留原消息身份退避核对。 +- 未终结任务额度涵盖等待、预占、运行和核查；首次受理增加、首次终态减少，重试和迟到证据不重复变更。 +- 满额受理返回 HTTP 429 / QUEUE_FULL / Retry-After: 1；原已受理请求仍可补取回执。SDK 提供类型化错误及 RetryAfter，不自动重报写请求。 +- 计划补跑受相同积压上限限制，保留未受理的原触发点；释放空间后按冻结漏跑策略接续。 +- 迁移 000009 增加四个队列字段并回填既有未终结任务，Go/SQL 均有中文说明，无数据库外键。 +- 源码核对 257 个直接声明的持久化字段均有中文注释；实体仍为 32 个。 +- [队列接口与完整 SDK 场景](../api/queues.md)、[resources 示例](../../sdk/go/examples/resources/main.go) 同步交付。 + +## 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 ./... +GOWORK=off go vet ./... +``` + +全部通过。完整集成后补充了并发开始竞争、Start 回滚、配置作用域及旧库额度回填断言，对 QueueResources 与 SafeRetryUpgrade 重新运行真实 MySQL 竞态回归并通过；修改的测试包再次通过 vet。没有用跳过数据库的普通单元结果代替真实集成证据。 + +| 测试组 | 实际结果 | +| --- | --- | +| 完整 MySQL 集成 | PASS，64.830 秒 | +| SDK Worker 全部连接/运行器回归 | PASS，80.819 秒 | +| SDK 计划生命周期 | PASS，含包开销 4.579 秒 | +| 新增 SDK 资源生命周期 | PASS，3.19 秒，含包开销 4.414 秒 | +| 真实服务/Worker 进程故障组 | PASS，21.425 秒，包含新增积压额度强杀恢复 | +| 补充资源与升级检查 | PASS，16.228 秒；资源用例 11.00 秒，升级用例 3.85 秒 | +| 九个完整示例 | health / tasks / worker-protocol / worker / history / retry / schedules / cancellation / resources 全部通过 | + +上述为测试用时，不是吞吐或延迟指标。 + +## 3. 数据库及竞争证据 + +[资源集成用例](../../internal/storage/mysql/queue_resources_integration_test.go) 验证： + +- 积压上限为 3 时，16 个并发提交恰好 3 个受理，其余为 QUEUE_FULL；拒绝请求不残留幂等占位，已受理原身份在满额下仍可查询回执。取消后可再次受理。 +- 两个节点、两个独立 Worker 实例先各领取任务，随后将开始上限从 4 降到 2，四个并发 Start 恰好两个通过，其余限流；任务/许可保留。 +- 强制 Start 事件写入失败后，任务仍为 LEASED，开始计数为零，原消息可在故障解除后重试。 +- 跨命名空间及队列范围受限的配置修改被拒，拒绝操作不残留配置身份。 +- 暂停禁止领取和 Start，但仍可提交与取消；配置同身份返回原快照，旧版本新操作冲突，成功变更各有一条审计。 +- rate=1 时未开始预留阻止额外领取；Start ACK 重报不重复计数。切换服务节点并 Resume 后窗口仍有效，两次实际 authorized_at 至少相隔一秒。 +- 从两张在途许可降到并发 1、积压 1，不撤销在途执行；超过上限时不再领取。已领取 Start 限流后，用原消息在下一窗口成功，不增加尝试次数。 +- 强制任务事件写入失败时回滚积压增加；明确失败后重试保持一个名额，UNKNOWN 耗尽后核查继续占名额；WAITING 截止及重复恢复只释放一次。 +- 计划在积压 1 下只生成一条任务，游标停在下一原触发点；取消释放空间后下一批准确接续该触发点。 + +既有取消、结果 ACK 丢失、许可恢复及计划事务回归同步通过。额外的计数核对直接比较队列 counter 与任务真实状态集合，避免只检查返回 DTO。 + +[既有 v4 升级用例](../../internal/storage/mysql/retry_upgrade_integration_test.go) 继续走正式迁移到当前版本，新增验证 WAITING 和 NEEDS_REVIEW 合计回填为 2、默认上限 100000、原队列 revision=1 不变。旧任务各字段、策略、状态及重复迁移检查仍通过。[无外键升级用例](../../internal/storage/mysql/no_foreign_keys_integration_test.go) 同步验证当前外键为零及既有索引保留。 + +## 4. HTTP、SDK 和进程恢复 + +[SDK 用例](../../sdk/go/queue_resources_integration_test.go) 对真实 HTTP/WS 服务创建独立队列，验证省略 paused 被拒、429/QUEUE_FULL/RetryAfter、满额下重复受理回执、取消释放、原被拒身份重试、暂停期间不消耗尝试、恢复后自动 Worker 完成两个任务及精确整数保留。读取实际开始历史验证 rate=1 的授权间隔，最终 unfinished_tasks=0，运行计数不改变配置 revision。 + +[进程用例](../../sdk/go/worker_process_integration_test.go) 的 TestIntegrationWorkerProcessQueueBackpressure 在任务受理后强杀节点 A，Wait 确认退出，再启动不同身份的 B：新提交仍被满额拒绝，原身份仍返回原任务；取消释放后，之前被拒的新任务可用原身份受理。该过程没有修改额度或任务状态来模拟恢复。 + +完整资源示例输出 QUEUE_FULL/429、duplicate=true，最终两个任务 SUCCEEDED/attempts=1、unfinished=0、worker_drained=true。业务处理仅回显通用输入，无外部副作用。原有服务强杀、Worker 强杀转核查及两节点调度补跑回归也全部通过。 + +## 5. 边界与后续 + +一秒计数窗口允许窗口内突发，窗口交界可能出现短时聚集；不保证任意滑动一秒上限。限速针对平台开始授权，不限制业务处理器的内部请求。降低配置不回滚已有外部效果。 + +队列积压计数与任务写入在同一事务，直接绕过服务修改数据库可能造成漂移。尚无自动修复计数或硬删除入口；迁移回填要求停止旧服务及写入，未做大表在线回填验收。历史回执新增字段的兼容规则见接口手册。 + +当前验证同机进程、多节点身份及单 MySQL，尚未覆盖数据库复制切换、跨故障域、混合版本滚动升级、热点队列吞吐及百万租户容量。应用级配额、同业务键并发许可、优先级/公平性、可靠交付积压限制与保留清理仍需继续实现和回归；本阶段不等于 P3 或平台最终验收完成。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index f948e7b..bbab2e9 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)，资源配额和结果订阅继续开发；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)，应用级配额、业务键许可、优先级/公平性和结果订阅继续开发；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 diff --git a/docs/sdk/worker-runtime.md b/docs/sdk/worker-runtime.md index 9d61043..4e934f6 100644 --- a/docs/sdk/worker-runtime.md +++ b/docs/sdk/worker-runtime.md @@ -81,6 +81,8 @@ SDK 的 Run 每个实例只能调用一次，包括已经正常退出的实例 Run 的重连恢复同一次控制操作。平台的 [安全重试](../api/retries.md) 使用同一 Task.ID 派发新的 AttemptID/Epoch，运行器按新的有效授权调用处理器，预算与退避由服务端持久化管理。 +Start 返回 EXECUTION_PAUSED 或 RATE_LIMITED 时，运行器保留原 start 消息身份退避核对，不调用处理器；有限授权到期后按服务端恢复规则处理。新领取已受 [共享队列额度](../api/queues.md) 约束，正常等待限流不消耗新的尝试预算。配置降额不意味着已在执行的 goroutine 已停止。 + ## 5. 结果与错误分类 | 处理器结果 | SDK 报告 / 服务端行为 | diff --git a/docs/sdk/worker.md b/docs/sdk/worker.md index 2eba32a..2a8ff49 100644 --- a/docs/sdk/worker.md +++ b/docs/sdk/worker.md @@ -117,6 +117,7 @@ Result 是不超过 256 KiB 的非 null JSON，可传对象引用；原始异常 ## 6. 错误与有界行为 - 握手 HTTP 拒绝返回 `*HTTPError`；业务控制拒绝返回 `*WorkerError`，通过 Code 分支处理。Error() 只显示分类，避免未知错误正文进入日志。 +- Start 的 RATE_LIMITED 表示当前窗口不允许开始；保留原消息身份在剩余租约内退避，不能调用业务处理器。队列控制、降额与窗口定义见 [队列接口](../api/queues.md)。 - 传输中断返回 `ErrWorkerDisconnected`；消息格式/版本不兼容返回 `ErrWorkerProtocol`；主动关闭为 `ErrWorkerClosed`；超时/取消支持标准 context 错误。 - 最多 32 个在途控制请求，达到上限返回 `ErrWorkerBusy`，请求未发送。同一消息 ID 已在途时也拒绝第二个并发等待者。 - 派发缓冲最多 8 条；读取不及时会以 `ErrWorkerBackpressure` 关闭连接。持久化预占由服务端恢复，不无限积压内存。 diff --git a/internal/fault/fault.go b/internal/fault/fault.go index 478988b..f13dec9 100644 --- a/internal/fault/fault.go +++ b/internal/fault/fault.go @@ -25,6 +25,7 @@ var ( Conflict = New("STATE_CONFLICT", "resource state does not allow this operation") IdempotencyConflict = New("IDEMPOTENCY_CONFLICT", "idempotency key was used with different content") Unavailable = New("UNAVAILABLE", "service temporarily unavailable; retry using the original operation identity") + QueueFull = New("QUEUE_FULL", "queue unfinished task limit reached; retry later using the original identity") ) func Invalid(message string) error { return New("INVALID_ARGUMENT", message) } diff --git a/internal/storage/mysql/migrations/000009_queue_resources.up.sql b/internal/storage/mysql/migrations/000009_queue_resources.up.sql new file mode 100644 index 0000000..34e0a80 --- /dev/null +++ b/internal/storage/mysql/migrations/000009_queue_resources.up.sql @@ -0,0 +1,15 @@ +-- 停止旧服务后迁移；未终结任务包含等待、预占、运行和待核查，不能因重试重复计数。 +-- 无数据库外键。额度与任务状态由服务端在同一队列事务内维护。 +ALTER TABLE pulse_queues + ADD COLUMN max_unfinished_tasks INT UNSIGNED NOT NULL DEFAULT 100000 COMMENT '队列允许积压的未终结任务上限；降额不删除已有任务', + ADD COLUMN unfinished_tasks BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '未终结任务数，包含待核查；受理增加、首次进入终态减少，重试不重复计数', + ADD COLUMN start_window_until DATETIME(6) NULL COMMENT '数据库时间的一秒开始限流窗口结束时刻；空表示尚未开始计数', + ADD COLUMN starts_in_window INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '当前窗口已提交的开始授权次数，回执重报不重复计数，降额不清零', + ADD CONSTRAINT ck_queue_unfinished_limit CHECK (max_unfinished_tasks > 0), + ADD CONSTRAINT ck_queue_start_window CHECK (start_window_until IS NOT NULL OR starts_in_window=0); + +UPDATE pulse_queues q LEFT JOIN ( + SELECT namespace_id,queue_id,COUNT(*) AS unfinished FROM pulse_tasks + WHERE state IN ('WAITING','LEASED','RUNNING','NEEDS_REVIEW') GROUP BY namespace_id,queue_id +) t ON t.namespace_id=q.namespace_id AND t.queue_id=q.id +SET q.unfinished_tasks=COALESCE(t.unfinished,0); diff --git a/internal/storage/mysql/queue_resources.go b/internal/storage/mysql/queue_resources.go new file mode 100644 index 0000000..398852e --- /dev/null +++ b/internal/storage/mysql/queue_resources.go @@ -0,0 +1,131 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" +) + +// 以下额度操作要求调用者已取得队列排他锁，且在同一事务中写入任务/尝试。 +func reserveQueueTask(ctx context.Context, tx *sql.Tx, ns, queue core.ID) error { + r, err := tx.ExecContext(ctx, `UPDATE pulse_queues SET unfinished_tasks=unfinished_tasks+1 WHERE namespace_id=? AND id=? AND unfinished_tasks0`, ns, queue) + return exactlyOne(r, err) +} + +// queueStartBudget 只读取当前窗口，不把空轮询写成新的窗口。 +// 窗口由第一次成功 Start 建立；以数据库时间计算，节点重启不清零。 +func queueStartBudget(ctx context.Context, tx *sql.Tx, ns, queue core.ID, now time.Time) (uint32, uint32, *time.Time, error) { + var rate, used uint32 + var until *time.Time + err := tx.QueryRowContext(ctx, `SELECT rate_per_second,starts_in_window,start_window_until FROM pulse_queues WHERE namespace_id=? AND id=?`, ns, queue).Scan(&rate, &used, &until) + if err != nil { + return 0, 0, nil, err + } + if until == nil || !until.After(now) { + used, until = 0, nil + } + return rate, used, until, nil +} + +// queueCanClaim 把尚未 Start 的预占也算入当前窗口可用空间，避免限流时空耗尝试预算。 +func queueCanClaim(ctx context.Context, tx *sql.Tx, ns, queue core.ID, now time.Time) (bool, error) { + rate, used, _, err := queueStartBudget(ctx, tx, ns, queue, now) + if err != nil || used >= rate { + return false, err + } + var reserved uint64 + err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=? AND queue_id=? AND state='LEASED'`, ns, queue).Scan(&reserved) + return err == nil && reserved < uint64(rate-used), err +} + +func consumeQueueStart(ctx context.Context, tx *sql.Tx, ns, queue core.ID, now time.Time) error { + rate, used, until, err := queueStartBudget(ctx, tx, ns, queue, now) + if err != nil { + return err + } + if used >= rate { + return worker.RateLimited + } + if until == nil { + end := now.Add(time.Second) + until = &end + } + r, err := tx.ExecContext(ctx, `UPDATE pulse_queues SET start_window_until=?,starts_in_window=? WHERE namespace_id=? AND id=?`, until, used+1, ns, queue) + return exactlyOne(r, err) +} + +// ConfigureQueue 完整替换控制参数，运行计数不参与配置版本，也不因配置变更重置。 +func (s *TaskStore) ConfigureQueue(ctx context.Context, token auth.Token, id core.ID, key string, in service.QueueConfigInput) (service.QueueResult, error) { + var out service.QueueResult + err := s.transaction(ctx, token, "queues:manage", func(tx *sql.Tx, p auth.Permissions) error { + if !p.AllowsQueue(id) { + return fault.NotFound + } + identity, cached, err := reserveIdentity(ctx, tx, token.NamespaceID, "queue.configure", string(id), key, "queue", id, in) + if err != nil { + return err + } + if cached != nil { + if err := json.Unmarshal(cached, &out); err != nil { + return err + } + out.Deduplicated = true + return nil + } + if _, _, err := lockQueue(ctx, tx, token.NamespaceID, id); err != nil { + return err + } + before, err := readQueue(ctx, tx, token.NamespaceID, id) + if err != nil { + return err + } + if before.Revision != in.ExpectedRevision { + return fault.Conflict + } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + r, err := tx.ExecContext(ctx, `UPDATE pulse_queues SET paused=?,max_concurrency=?,rate_per_second=?,max_unfinished_tasks=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, in.Paused, in.MaxConcurrency, in.RatePerSecond, in.MaxUnfinishedTasks, now, token.NamespaceID, id, in.ExpectedRevision) + if err := exactlyOne(r, err); err != nil { + return err + } + if _, _, err := queueLimit(ctx, tx, token.NamespaceID, id, in.MaxConcurrency); err != nil { + return err + } + out.Queue, err = readQueue(ctx, tx, token.NamespaceID, id) + if err != nil { + return err + } + if err := appendAudit(ctx, tx, token, identity, id, "queue", "queue.configure", in.Reason, before, out.Queue); err != nil { + return err + } + if err := emitEvent(ctx, tx, token.NamespaceID, id, out.Queue.Revision, "queue", "queue.configured", out.Queue); err != nil { + return err + } + return saveIdentity(ctx, tx, token.NamespaceID, identity, out) + }) + return out, err +} diff --git a/internal/storage/mysql/queue_resources_integration_test.go b/internal/storage/mysql/queue_resources_integration_test.go new file mode 100644 index 0000000..5b0e6c4 --- /dev/null +++ b/internal/storage/mysql/queue_resources_integration_test.go @@ -0,0 +1,376 @@ +package mysql + +import ( + "context" + "encoding/json" + "os" + "sync" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" +) + +func configureResources(t *testing.T, f *executionFixture, rate, concurrency, unfinished uint32, paused bool) service.QueueView { + t.Helper() + q, err := f.tasks.GetQueue(f.ctx, f.token, f.resources.QueueID) + if err != nil { + t.Fatal(err) + } + out, err := f.tasks.ConfigureQueue(f.ctx, f.token, q.ID, string(executionID(t)), service.QueueConfigInput{ExpectedRevision: q.Revision, Reason: "controlled resource regression", Paused: paused, MaxConcurrency: concurrency, RatePerSecond: rate, MaxUnfinishedTasks: unfinished}) + if err != nil { + t.Fatal(err) + } + return out.Queue +} +func assertQueueCount(t *testing.T, f *executionFixture, expected uint64) { + t.Helper() + q, err := f.tasks.GetQueue(f.ctx, f.token, f.resources.QueueID) + if err != nil || q.UnfinishedTasks != expected { + t.Fatalf("queue count=%d want=%d err=%v", q.UnfinishedTasks, expected, err) + } + actual := f.count("pulse_tasks", "queue_id=? AND state IN ('WAITING','LEASED','RUNNING','NEEDS_REVIEW')", q.ID) + if uint64(actual) != expected { + t.Fatalf("counter differs from actual states: %d", actual) + } +} +func waitStartWindow(t *testing.T, f *executionFixture) { + t.Helper() + var until, now time.Time + if err := f.db.QueryRowContext(f.ctx, `SELECT start_window_until,UTC_TIMESTAMP(6) FROM pulse_queues WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.QueueID).Scan(&until, &now); err != nil { + t.Fatal(err) + } + timer := time.NewTimer(max(time.Millisecond, until.Sub(now)+10*time.Millisecond)) + defer timer.Stop() + select { + case <-timer.C: + case <-f.ctx.Done(): + t.Fatal(f.ctx.Err()) + } +} +func TestIntegrationQueueResources(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, stop := context.WithTimeout(context.Background(), 60*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 { return newExecutionFixture(t, ctx, db) } + t.Run("concurrent_nodes_share_start_budget", func(t *testing.T) { + f := fixture(t) + configureResources(t, f, 4, 4, 10, false) + hello := f.hello + hello.InstanceID = executionID(t) + b, _, err := f.b.Connect(ctx, f.token, hello) + if err != nil { + t.Fatal(err) + } + type item struct { + api *worker.Service + binding worker.Binding + ref worker.AttemptRef + } + var items []item + for i := 0; i < 4; i++ { + f.submit(1) + api, binding := f.a, f.binding + if i%2 == 1 { + api, binding = f.b, b + } + a, err := api.Claim(ctx, f.token, binding) + if err != nil || a == nil { + t.Fatal("claim reservation failed", err) + } + items = append(items, item{api, binding, worker.AttemptRef{TaskID: a.Task.ID, AttemptID: a.AttemptID, Epoch: a.Epoch, LeaseSecret: a.LeaseSecret}}) + } + configureResources(t, f, 2, 4, 10, false) + start := make(chan struct{}) + results := make(chan error, 4) + var wg sync.WaitGroup + for _, item := range items { + message := executionID(t) + wg.Add(1) + go func() { + defer wg.Done() + <-start + _, err := item.api.Start(ctx, f.token, item.binding, message, item.ref) + results <- err + }() + } + close(start) + wg.Wait() + close(results) + allowed := 0 + for err := range results { + if err == nil { + allowed++ + } else if fault.Code(err) != "RATE_LIMITED" { + t.Fatal(err) + } + } + if allowed != 2 { + t.Fatalf("concurrent starts=%d, want 2", allowed) + } + assertQueueCount(t, f, 4) + }) + t.Run("start_budget_rollback_and_configuration_scope", func(t *testing.T) { + f := fixture(t) + q := configureResources(t, f, 1, 4, 10, false) + f.submit(1) + _, ref := f.claim() + message := executionID(t) + f.exec(`CREATE TRIGGER orbit_start_budget_fail BEFORE INSERT ON orbit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled rollback'`) + _, err := f.a.Start(ctx, f.token, f.binding, message, ref) + f.exec(`DROP TRIGGER orbit_start_budget_fail`) + if err == nil || f.get(ref.TaskID).State != pulse.TaskLeased { + t.Fatal("failed start committed") + } + var used uint32 + if err := db.QueryRowContext(ctx, `SELECT starts_in_window FROM pulse_queues WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, q.ID).Scan(&used); err != nil || used != 0 { + t.Fatal("rolled back start spent budget") + } + if _, err := f.a.Start(ctx, f.token, f.binding, message, ref); err != nil { + t.Fatal(err) + } + other := fixture(t) + in := service.QueueConfigInput{ExpectedRevision: q.Revision, Reason: "scope", MaxConcurrency: 4, RatePerSecond: 1, MaxUnfinishedTasks: 10} + _, err = other.tasks.ConfigureQueue(ctx, other.token, q.ID, "cross-namespace", in) + expectWorkerCode(t, err, "NOT_FOUND") + raw, _ := json.Marshal(auth.Permissions{Actions: []string{"queues:manage"}, Queues: []string{string(other.resources.QueueID)}}) + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.resources.NamespaceID, f.resources.CredentialID) + _, err = f.tasks.ConfigureQueue(ctx, f.token, q.ID, "denied-queue", in) + expectWorkerCode(t, err, "NOT_FOUND") + if f.count("orbit_idempotency_keys", "operation='queue.configure' AND scope_key=?", string(q.ID)) != 1 { + t.Fatal("denied config left identity") + } + }) + t.Run("concurrent_admission_duplicate_and_release", func(t *testing.T) { + f := fixture(t) + configureResources(t, f, 10, 4, 3, false) + input := service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)} + type result struct { + key string + out service.SubmitResult + err error + } + results := make(chan result, 16) + var wg sync.WaitGroup + for i := 0; i < 16; i++ { + key := string(executionID(t)) + wg.Add(1) + go func() { + defer wg.Done() + out, err := f.tasks.Submit(ctx, f.token, key, input) + results <- result{key, out, err} + }() + } + wg.Wait() + close(results) + var accepted []result + for r := range results { + if r.err == nil { + accepted = append(accepted, r) + } else { + expectWorkerCode(t, r.err, "QUEUE_FULL") + } + } + if len(accepted) != 3 || f.count("orbit_idempotency_keys", "operation='task.submit'") != 3 { + t.Fatal("admission exceeded limit or left rejected identities") + } + assertQueueCount(t, f, 3) + again, err := f.tasks.Submit(ctx, f.token, accepted[0].key, input) + if err != nil || !again.Deduplicated || again.Task.ID != accepted[0].out.Task.ID { + t.Fatal("full queue rejected original receipt") + } + _, err = f.tasks.CancelTask(ctx, f.token, again.Task.ID, "release", service.CancelInput{ExpectedRevision: again.Task.Revision, Reason: "release slot"}) + if err != nil { + t.Fatal(err) + } + assertQueueCount(t, f, 2) + if _, err := f.tasks.Submit(ctx, f.token, "new-after-release", input); err != nil { + t.Fatal(err) + } + assertQueueCount(t, f, 3) + }) + t.Run("configuration_cas_audit_and_pause", func(t *testing.T) { + f := fixture(t) + q := configureResources(t, f, 10, 4, 10, true) + task := f.submit(1) + if a, err := f.a.Claim(ctx, f.token, f.binding); err != nil || a != nil { + t.Fatal("paused queue dispatched") + } + in := service.QueueConfigInput{ExpectedRevision: q.Revision, Reason: "resume", MaxConcurrency: 4, RatePerSecond: 10, MaxUnfinishedTasks: 10} + first, err := f.tasks.ConfigureQueue(ctx, f.token, q.ID, "resume", in) + if err != nil { + t.Fatal(err) + } + again, err := f.tasks.ConfigureQueue(ctx, f.token, q.ID, "resume", in) + if err != nil || !again.Deduplicated || again.Queue.Revision != first.Queue.Revision { + t.Fatal("config receipt changed") + } + _, err = f.tasks.ConfigureQueue(ctx, f.token, q.ID, "stale", in) + expectWorkerCode(t, err, "STATE_CONFLICT") + _, ref := f.claim() + configureResources(t, f, 10, 4, 10, true) + _, err = f.a.Start(ctx, f.token, f.binding, executionID(t), ref) + expectWorkerCode(t, err, "EXECUTION_PAUSED") + _, err = f.tasks.CancelTask(ctx, f.token, task.ID, "cancel-paused", service.CancelInput{ExpectedRevision: f.get(task.ID).Revision, Reason: "cancel while paused"}) + if err != nil { + t.Fatal(err) + } + assertQueueCount(t, f, 0) + if f.count("orbit_audit_events", "resource_id=? AND action='queue.configure'", q.ID) != 3 { + t.Fatal("configuration audit count") + } + }) + t.Run("rate_window_survives_reconnect_and_receipt_replay", func(t *testing.T) { + f := fixture(t) + configureResources(t, f, 1, 4, 10, false) + f.submit(1) + f.submit(1) + _, first := f.claim() + if a, err := f.a.Claim(ctx, f.token, f.binding); err != nil || a != nil { + t.Fatal("unstarted reservation did not throttle claims") + } + message := executionID(t) + if _, err := f.a.Start(ctx, f.token, f.binding, message, first); err != nil { + t.Fatal(err) + } + if out, err := f.a.Start(ctx, f.token, f.binding, message, first); err != nil || !out.Deduplicated { + t.Fatal("start replay rejected by quota") + } + binding, _, err := f.b.Connect(ctx, f.token, f.hello) + if err != nil { + t.Fatal(err) + } + if _, err := f.b.Resume(ctx, f.token, binding, first); err != nil { + t.Fatal(err) + } + f.a, f.binding = f.b, binding + if a, err := f.a.Claim(ctx, f.token, f.binding); err != nil || a != nil { + t.Fatal("node handoff reset rate window") + } + waitStartWindow(t, f) + _, second := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), second); err != nil { + t.Fatal(err) + } + var gap int64 + if err := db.QueryRowContext(ctx, `SELECT TIMESTAMPDIFF(MICROSECOND,a.authorized_at,b.authorized_at) FROM pulse_attempts a JOIN pulse_attempts b ON b.namespace_id=a.namespace_id WHERE a.namespace_id=? AND a.id=? AND b.id=?`, f.resources.NamespaceID, first.AttemptID, second.AttemptID).Scan(&gap); err != nil || gap < 1000000 { + t.Fatalf("rate gap=%d err=%v", gap, err) + } + retryFinish(t, f, first, "SUCCEEDED", false) + retryFinish(t, f, second, "SUCCEEDED", false) + assertQueueCount(t, f, 0) + }) + t.Run("lower_limits_preserve_inflight_and_throttled_start", func(t *testing.T) { + f := fixture(t) + configureResources(t, f, 2, 4, 10, false) + f.submit(1) + f.submit(1) + f.submit(1) + _, a := f.claim() + _, b := f.claim() + configureResources(t, f, 1, 1, 1, false) + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), a); err != nil { + t.Fatal(err) + } + message := executionID(t) + _, err := f.a.Start(ctx, f.token, f.binding, message, b) + expectWorkerCode(t, err, "RATE_LIMITED") + if f.get(b.TaskID).State != pulse.TaskLeased || f.get(b.TaskID).AttemptCount != 1 || f.used() != 2 { + t.Fatal("rate rejection consumed or revoked work") + } + retryFinish(t, f, a, "SUCCEEDED", false) + if got, err := f.a.Claim(ctx, f.token, f.binding); err != nil || got != nil { + t.Fatal("lowered concurrency admitted excess") + } + waitStartWindow(t, f) + if _, err := f.a.Start(ctx, f.token, f.binding, message, b); err != nil { + t.Fatal(err) + } + retryFinish(t, f, b, "SUCCEEDED", false) + assertQueueCount(t, f, 1) + waitStartWindow(t, f) + f.claim() + }) + t.Run("rollback_retry_unknown_and_expiry_accounting", func(t *testing.T) { + f := fixture(t) + configureResources(t, f, 10, 4, 2, false) + f.exec(`CREATE TRIGGER orbit_resource_test_fail BEFORE INSERT ON orbit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled rollback'`) + _, err := f.tasks.Submit(ctx, f.token, "rollback", service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}) + f.exec(`DROP TRIGGER orbit_resource_test_fail`) + if err == nil { + t.Fatal("injected failure ignored") + } + assertQueueCount(t, f, 0) + task := retryTask(t, f, 2, pulse.RecoveryIdempotent) + ref := retryStart(t, f) + retryFinish(t, f, ref, "FAILED", true) + assertQueueCount(t, f, 1) + makeRetryDue(f, task.ID) + ref = retryStart(t, f) + retryFinish(t, f, ref, "UNKNOWN", false) + assertQueueCount(t, f, 1) + waiting := f.submit(1) + f.exec(`UPDATE pulse_tasks SET available_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 2 SECOND),deadline_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, waiting.ID) + if _, err := f.storeA.Recover(ctx, 100); err != nil { + t.Fatal(err) + } + assertQueueCount(t, f, 1) + if _, err := f.storeA.Recover(ctx, 100); err != nil { + t.Fatal(err) + } + assertQueueCount(t, f, 1) + }) + t.Run("schedule_backpressure_preserves_unaccepted_cursor", func(t *testing.T) { + f := fixture(t) + configureResources(t, f, 10, 4, 1, false) + sf := newScheduleFixture(t, f) + v := sf.create(sf.definition()) + sf.due(v, 2*time.Second) + sf.tick() + assertQueueCount(t, f, 1) + items := sf.occurrences(v.ID) + if len(items) != 1 || items[0].TaskID == nil { + t.Fatal("schedule exceeded backlog limit") + } + next := sf.current(v.ID).NextFireAt + if next == nil || !next.Equal(items[0].ScheduledAt.Add(time.Second)) { + t.Fatal("backpressure skipped unaccepted point") + } + id := *items[0].TaskID + _, err := f.tasks.CancelTask(ctx, f.token, id, "free-schedule", service.CancelInput{ExpectedRevision: 1, Reason: "free slot"}) + if err != nil { + t.Fatal(err) + } + sf.due(v, 0) + sf.tick() + items = sf.occurrences(v.ID) + if len(items) != 2 { + t.Fatal("schedule did not resume after release") + } + found := false + for _, item := range items { + if item.ScheduledAt.Equal(*next) { + found = true + } + } + if !found { + t.Fatal("resumed wrong original time") + } + assertQueueCount(t, f, 1) + }) +} diff --git a/internal/storage/mysql/retry_upgrade_integration_test.go b/internal/storage/mysql/retry_upgrade_integration_test.go index cf287f4..49b96b9 100644 --- a/internal/storage/mysql/retry_upgrade_integration_test.go +++ b/internal/storage/mysql/retry_upgrade_integration_test.go @@ -94,6 +94,12 @@ func TestIntegrationSafeRetryUpgrade(t *testing.T) { if err := CheckSchema(ctx, db); err != nil { t.Fatal(err) } + // 000009 必须回填 WAITING 和 NEEDS_REVIEW，并保留既有队列配置版本。 + var unfinished uint64 + var maxUnfinished, queueRevision uint64 + if err := db.QueryRowContext(ctx, `SELECT unfinished_tasks,max_unfinished_tasks,revision FROM pulse_queues WHERE namespace_id=? AND id=?`, ns, queue).Scan(&unfinished, &maxUnfinished, &queueRevision); err != nil || unfinished != 2 || maxUnfinished != 100000 || queueRevision != 1 { + t.Fatalf("resource upgrade changed queue or missed states: count=%d max=%d revision=%d err=%v", unfinished, maxUnfinished, queueRevision, err) + } for state, id := range ids { var gotState, gotPolicy string var uncertain bool diff --git a/internal/storage/mysql/schedule_tick.go b/internal/storage/mysql/schedule_tick.go index 4a95d5c..acb8028 100644 --- a/internal/storage/mysql/schedule_tick.go +++ b/internal/storage/mysql/schedule_tick.go @@ -153,8 +153,15 @@ func (s *ScheduleStore) tickOne(ctx context.Context, ns, id, queue core.ID) (int if err != nil { return err } + backpressured := false for _, decision := range batch.Decisions { if err := s.saveOccurrence(ctx, tx, v, version, decision, now); err != nil { + if errors.Is(err, fault.QueueFull) { + // 已提交本批较早的决定仍保留；满额的原触发点留待下次按冻结策略处理。 + t := decision.ScheduledAt + next, backpressured = &t, true + break + } return err } if decision.Kind == "SKIPPED" { @@ -163,7 +170,9 @@ func (s *ScheduleStore) tickOne(ctx context.Context, ns, id, queue core.ID) (int fired++ } } - next = batch.Next + if !backpressured { + next = batch.Next + } if next == nil && cursor < v.CurrentVersion { cursor++ needsInitialization = true @@ -205,6 +214,9 @@ func (s *ScheduleStore) saveOccurrence(ctx context.Context, tx *sql.Tx, v schedu if in.QueueID != v.QueueID { return errors.New("schedule template queue mismatch") } + if err := reserveQueueTask(ctx, tx, v.NamespaceID, v.QueueID); err != nil { + return err + } policy, err := json.Marshal(in.Retry) if err != nil { return err diff --git a/internal/storage/mysql/task_cancel.go b/internal/storage/mysql/task_cancel.go index 40544ac..1d68d0d 100644 --- a/internal/storage/mysql/task_cancel.go +++ b/internal/storage/mysql/task_cancel.go @@ -101,6 +101,11 @@ func (s *TaskStore) CancelTask(ctx context.Context, token auth.Token, id core.ID if err := exactlyOne(result, err); err != nil { return err } + if next.Terminal() { + if err := releaseQueueTask(ctx, tx, token.NamespaceID, view.QueueID); err != nil { + return err + } + } out.Task, err = readTask(ctx, tx, token.NamespaceID, id, false) if err != nil { return err diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index aec921f..8d365c0 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -171,7 +171,7 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in return nil } var queue core.ID - if err := tx.QueryRowContext(ctx, "SELECT id FROM pulse_queues WHERE namespace_id=? AND id=? FOR SHARE", token.NamespaceID, in.QueueID).Scan(&queue); errors.Is(err, sql.ErrNoRows) { + if err := tx.QueryRowContext(ctx, "SELECT id FROM pulse_queues WHERE namespace_id=? AND id=? FOR UPDATE", token.NamespaceID, in.QueueID).Scan(&queue); errors.Is(err, sql.ErrNoRows) { return fault.NotFound } else if err != nil { return err @@ -195,6 +195,9 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in if err != nil { return err } + 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) if err != nil { @@ -338,7 +341,11 @@ func (s *TaskStore) CreateQueue(ctx context.Context, token auth.Token, key strin out.Deduplicated = true return nil } - _, err = tx.ExecContext(ctx, "INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,policy) VALUES (?,?,?,?,?,'{}')", token.NamespaceID, id, in.Name, in.MaxConcurrency, in.RatePerSecond) + maxUnfinished := in.MaxUnfinishedTasks + if maxUnfinished == 0 { + maxUnfinished = 100000 + } + _, err = tx.ExecContext(ctx, "INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,max_unfinished_tasks,policy) VALUES (?,?,?,?,?,?,'{}')", token.NamespaceID, id, in.Name, in.MaxConcurrency, in.RatePerSecond, maxUnfinished) var dbErr *driver.MySQLError if errors.As(err, &dbErr) && dbErr.Number == 1062 { return fault.Conflict @@ -359,7 +366,7 @@ func (s *TaskStore) CreateQueue(ctx context.Context, token auth.Token, key strin } func readQueue(ctx context.Context, tx *sql.Tx, ns, id core.ID) (service.QueueView, error) { var q service.QueueView - err := tx.QueryRowContext(ctx, "SELECT id,namespace_id,name,paused,max_concurrency,rate_per_second,revision FROM pulse_queues WHERE namespace_id=? AND id=?", ns, id).Scan(&q.ID, &q.NamespaceID, &q.Name, &q.Paused, &q.MaxConcurrency, &q.RatePerSecond, &q.Revision) + err := tx.QueryRowContext(ctx, "SELECT id,namespace_id,name,paused,max_concurrency,rate_per_second,max_unfinished_tasks,unfinished_tasks,revision FROM pulse_queues WHERE namespace_id=? AND id=?", ns, id).Scan(&q.ID, &q.NamespaceID, &q.Name, &q.Paused, &q.MaxConcurrency, &q.RatePerSecond, &q.MaxUnfinishedTasks, &q.UnfinishedTasks, &q.Revision) if errors.Is(err, sql.ErrNoRows) { err = fault.NotFound } diff --git a/internal/storage/mysql/worker_execution.go b/internal/storage/mysql/worker_execution.go index cd8036f..738b977 100644 --- a/internal/storage/mysql/worker_execution.go +++ b/internal/storage/mysql/worker_execution.go @@ -81,8 +81,13 @@ func queueLimit(ctx context.Context, tx *sql.Tx, ns, queue core.ID, max uint32) var used uint32 var configured uint32 err = tx.QueryRowContext(ctx, `SELECT id,used_permits,max_permits FROM pulse_concurrency_limits WHERE namespace_id=? AND scope_key='queue' AND key_digest=? FOR UPDATE`, ns, digest).Scan(&id, &used, &configured) - if err == nil && configured != max { - err = errors.New("queue limit configuration requires coordinated update") + // 降额时保留已有许可；队列上限控制新的领取，存储上限至少容纳在途数。 + target := max + if used > target { + target = used + } + if err == nil && configured != target { + _, err = tx.ExecContext(ctx, `UPDATE pulse_concurrency_limits SET max_permits=?,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, target, ns, id) } return id, used, err } @@ -123,6 +128,14 @@ func (s *ExecutionStore) Claim(ctx context.Context, token auth.Token, b worker.B if used >= max { return nil } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + canClaim, err := queueCanClaim(ctx, tx, token.NamespaceID, state.queue, now) + if err != nil || !canClaim { + return err + } args := []any{token.NamespaceID, state.queue} matches := make([]string, 0, len(state.capabilities)) for _, c := range state.capabilities { @@ -465,6 +478,9 @@ func (s *ExecutionStore) executionCommand(ctx context.Context, token auth.Token, if task.view.StartDeadlineAt != nil && !task.view.StartDeadlineAt.After(now) { return worker.LeaseLost } + if err := consumeQueueStart(ctx, tx, token.NamespaceID, task.view.QueueID, now); err != nil { + return err + } authorized = &now } else if a.State != pulse.AttemptRunning || authorized == nil { return fault.Conflict diff --git a/internal/storage/mysql/worker_recovery.go b/internal/storage/mysql/worker_recovery.go index b37b383..d91e23c 100644 --- a/internal/storage/mysql/worker_recovery.go +++ b/internal/storage/mysql/worker_recovery.go @@ -98,6 +98,11 @@ func (s *ExecutionStore) Recover(ctx context.Context, limit int) (worker.Recover if err := exactlyOne(result, err); err != nil { return err } + if changed.Terminal() { + if err := releaseQueueTask(ctx, tx, c.ns, c.queue); err != nil { + return err + } + } return emitEvent(ctx, tx, c.ns, c.id, task.view.Revision+1, "task", resultEvent(changed), map[string]any{"reason": "START_DEADLINE_EXCEEDED", "node_id": s.node, "effect_uncertain": task.view.EffectUncertain}) } if task.view.State != pulse.TaskLeased && task.view.State != pulse.TaskRunning { @@ -155,6 +160,11 @@ func (s *ExecutionStore) Recover(ctx context.Context, limit int) (worker.Recover if err := releasePermits(ctx, tx, c.ns, attempt.ID, now); err != nil { return err } + if next.Terminal() { + if err := releaseQueueTask(ctx, tx, c.ns, c.queue); err != nil { + return err + } + } changed = next return emitEvent(ctx, tx, c.ns, c.id, task.view.Revision+1, "task", resultEvent(next), map[string]any{"attempt_id": attempt.ID, "epoch": fmt.Sprint(attempt.Epoch), "reason": reason, "node_id": s.node, "effect_uncertain": decision.Uncertain, "decision": decision.Reason, "next_available_at": decision.NextAt, "delay_ms": decision.DelayMS, "attempt_count": task.view.AttemptCount, "max_attempts": task.maxAttempts}) }) diff --git a/internal/storage/mysql/worker_results.go b/internal/storage/mysql/worker_results.go index 4bc3d55..14654fd 100644 --- a/internal/storage/mysql/worker_results.go +++ b/internal/storage/mysql/worker_results.go @@ -112,6 +112,11 @@ func (s *ExecutionStore) Finish(ctx context.Context, token auth.Token, b worker. if err := releasePermits(ctx, tx, token.NamespaceID, a.ID, now); err != nil { return err } + if taskState.Terminal() { + if err := releaseQueueTask(ctx, tx, token.NamespaceID, task.view.QueueID); err != nil { + return err + } + } event := map[string]any{"attempt_id": a.ID, "epoch": fmt.Sprint(a.Epoch), "state": taskState, "outcome": in.Outcome, "error_code": in.ErrorCode, "retryable": in.Retryable, "effect_uncertain": decision.Uncertain, "decision": decision.Reason, "next_available_at": decision.NextAt, "delay_ms": decision.DelayMS, "attempt_count": task.view.AttemptCount, "max_attempts": task.maxAttempts} if err := emitEvent(ctx, tx, token.NamespaceID, in.Ref.TaskID, task.view.Revision+1, "task", resultEvent(taskState), event); err != nil { return err diff --git a/internal/transport/httpserver/tasks.go b/internal/transport/httpserver/tasks.go index 92533c3..6dc0fed 100644 --- a/internal/transport/httpserver/tasks.go +++ b/internal/transport/httpserver/tasks.go @@ -100,6 +100,23 @@ func registerTasks(mux *http.ServeMux, s *service.Service) { out, err := s.GetQueue(r.Context(), t, core.ID(r.PathValue("id"))) return out, 200, err })) + mux.HandleFunc("POST /v1/queues/{command}", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + command := r.PathValue("command") + if !strings.HasSuffix(command, ":configure") { + return nil, 0, fault.NotFound + } + var in service.QueueConfigInput + if err := decodeBody(r, &in); err != nil { + return nil, 0, err + } + // 完整替换命令要求 paused 显式给出，避免遗漏被解释为解除暂停。 + key, err := idempotencyKey(r) + if err != nil { + return nil, 0, err + } + out, err := s.ConfigureQueue(r.Context(), t, core.ID(strings.TrimSuffix(command, ":configure")), key, in) + return out, 200, err + })) } func createdStatus(duplicate bool) int { if duplicate { @@ -172,6 +189,10 @@ func writeError(w http.ResponseWriter, id string, err error) { if status == 0 { status = 503 } + if known.Code == "QUEUE_FULL" { + status = http.StatusTooManyRequests + w.Header().Set("Retry-After", "1") + } if status == 401 { w.Header().Set("WWW-Authenticate", `Bearer realm="orbit"`) } @@ -243,6 +264,11 @@ func exactFields(raw []byte, t reflect.Type) error { name := strings.Split(field.Tag.Get("json"), ",")[0] if name != "" && name != "-" { names[name] = field.Type + if field.Tag.Get("required") == "true" { + if _, ok := object[name]; !ok { + return errors.New("missing required field") + } + } } } for name, value := range object { diff --git a/pulse/model/queue.go b/pulse/model/queue.go index ce4087f..54b6b2d 100644 --- a/pulse/model/queue.go +++ b/pulse/model/queue.go @@ -2,6 +2,7 @@ package model import ( core "git.yuntuo.hk/k2safe/orbit/model" + "time" ) // Queue 表示任务路由与资源控制队列，不要求每个租户单独创建队列。 @@ -15,8 +16,16 @@ type Queue struct { Paused bool `db:"paused"` // MaxConcurrency 允许同时持有的有效执行许可上限，必须大于零。 MaxConcurrency uint32 `db:"max_concurrency"` - // RatePerSecond 每秒新执行的速率配置；实际限流由共享协调逻辑落实。 + // RatePerSecond 每个数据库一秒窗口最多提交的开始授权数，由所有节点共享。 RatePerSecond uint32 `db:"rate_per_second"` + // MaxUnfinishedTasks 未终结任务数上限；降额不删除或终止已受理任务。 + MaxUnfinishedTasks uint32 `db:"max_unfinished_tasks"` + // UnfinishedTasks 等待、预占、运行及待核查的任务总数，首次终结时同事务减一。 + UnfinishedTasks uint64 `db:"unfinished_tasks"` + // StartWindowUntil 当前一秒计数窗口结束时刻，空表示尚无开始授权。 + StartWindowUntil *time.Time `db:"start_window_until"` + // StartsInWindow 当前窗口的开始授权次数，重复确认不计数，调整配置不清零。 + StartsInWindow uint32 `db:"starts_in_window"` // Policy 队列资源与执行配置的 JSON 文档。 Policy core.JSON `db:"policy"` } diff --git a/pulse/service/service.go b/pulse/service/service.go index eb60c0c..b674d17 100644 --- a/pulse/service/service.go +++ b/pulse/service/service.go @@ -143,7 +143,7 @@ func (s *Service) CreateQueue(ctx context.Context, token auth.Token, key string, if err := ValidateKey(key); err != nil { return QueueResult{}, err } - if !ValidName(in.Name, 128) || in.MaxConcurrency == 0 || in.MaxConcurrency > 100000 || in.RatePerSecond == 0 || in.RatePerSecond > 1000000 { + if !ValidName(in.Name, 128) || in.MaxConcurrency == 0 || in.MaxConcurrency > 100000 || in.RatePerSecond == 0 || in.RatePerSecond > 1000000 || in.MaxUnfinishedTasks > 1000000 { return QueueResult{}, fault.Invalid("queue name, concurrency or rate is invalid") } return s.store.CreateQueue(ctx, token, key, in) @@ -154,3 +154,19 @@ func (s *Service) GetQueue(ctx context.Context, token auth.Token, id core.ID) (Q } return s.store.GetQueue(ctx, token, id) } + +func (s *Service) ConfigureQueue(ctx context.Context, token auth.Token, id core.ID, key string, in QueueConfigInput) (QueueResult, error) { + if id.Validate() != nil { + return QueueResult{}, fault.Invalid("queue id is invalid") + } + if err := ValidateKey(key); err != nil { + return QueueResult{}, err + } + if in.ExpectedRevision == 0 || strings.TrimSpace(in.Reason) == "" || !utf8.ValidString(in.Reason) || utf8.RuneCountInString(in.Reason) > 1024 { + return QueueResult{}, fault.Invalid("reason and nonzero expected_revision are required") + } + if in.MaxConcurrency == 0 || in.MaxConcurrency > 100000 || in.RatePerSecond == 0 || in.RatePerSecond > 1000000 || in.MaxUnfinishedTasks == 0 || in.MaxUnfinishedTasks > 1000000 { + return QueueResult{}, fault.Invalid("queue limits are invalid") + } + return s.store.ConfigureQueue(ctx, token, id, key, in) +} diff --git a/pulse/service/types.go b/pulse/service/types.go index 31c61f9..a21ff1b 100644 --- a/pulse/service/types.go +++ b/pulse/service/types.go @@ -149,14 +149,20 @@ type TaskPage struct { NextCursor string `json:"next_cursor,omitempty"` } type QueueInput struct { + // MaxUnfinishedTasks 未终结任务上限；省略或零使用 100000，最高 1000000。 + MaxUnfinishedTasks uint32 `json:"max_unfinished_tasks,omitempty"` // Name 队列名称，同一命名空间内唯一。 Name string `json:"name"` - // MaxConcurrency 最大同时执行数量，供执行阶段使用。 + // MaxConcurrency 允许占用的有效执行许可上限。 MaxConcurrency uint32 `json:"max_concurrency"` - // RatePerSecond 每秒开始数量上限，供执行阶段使用。 + // RatePerSecond 数据库一秒窗口最多提交的开始授权数。 RatePerSecond uint32 `json:"rate_per_second"` } type QueueView struct { + // MaxUnfinishedTasks 未终结任务上限，包含待核查；只限制新的受理。 + MaxUnfinishedTasks uint32 `json:"max_unfinished_tasks"` + // UnfinishedTasks 当前未终结任务数，JSON 十进制字符串；运行计数不改变配置版本。 + UnfinishedTasks uint64 `json:"unfinished_tasks,string"` // ID 资源标识，32 位小写十六进制字符串。 ID core.ID `json:"id"` // NamespaceID 命名空间标识，由凭据确定隔离范围。 @@ -165,13 +171,30 @@ type QueueView struct { Name string `json:"name"` // Paused 是否停止新执行；不禁止受理新任务。 Paused bool `json:"paused"` - // MaxConcurrency 最大同时执行数量，供执行阶段使用。 + // MaxConcurrency 允许占用的有效执行许可上限。 MaxConcurrency uint32 `json:"max_concurrency"` - // RatePerSecond 每秒开始数量上限，供执行阶段使用。 + // RatePerSecond 数据库一秒窗口最多提交的开始授权数。 RatePerSecond uint32 `json:"rate_per_second"` // Revision 并发版本，JSON 使用十进制字符串避免跨语言精度丢失。 Revision uint64 `json:"revision,string"` } + +// QueueConfigInput 完整替换控制配置，必须保留同一操作的原版本及原因。 +type QueueConfigInput struct { + // ExpectedRevision 非零的队列配置版本。 + ExpectedRevision uint64 `json:"expected_revision,string"` + // Reason 管理原因，写入不可变审计。 + Reason string `json:"reason"` + // Paused 停止新的领取和开始；在途续租、结果与受理继续。 + Paused bool `json:"paused" required:"true"` + // MaxConcurrency 有效并发上限，降低时已有许可自然收尾。 + MaxConcurrency uint32 `json:"max_concurrency"` + // RatePerSecond 数据库一秒窗口最多开始授权数，降低时保留原计数。 + RatePerSecond uint32 `json:"rate_per_second"` + // MaxUnfinishedTasks 未终结任务上限，降低时不删除历史或在途任务。 + MaxUnfinishedTasks uint32 `json:"max_unfinished_tasks"` +} + type QueueResult struct { // Queue 本次操作返回的队列快照。 Queue QueueView `json:"queue"` @@ -186,6 +209,7 @@ type Store interface { ListTasks(context.Context, auth.Token, ListInput) (TaskPage, error) CancelTask(context.Context, auth.Token, core.ID, string, CancelInput) (CancelResult, error) CreateQueue(context.Context, auth.Token, string, QueueInput) (QueueResult, error) + ConfigureQueue(context.Context, auth.Token, core.ID, string, QueueConfigInput) (QueueResult, error) GetQueue(context.Context, auth.Token, core.ID) (QueueView, error) ListTaskAttempts(context.Context, auth.Token, core.ID, HistoryPageInput) (AttemptPage, error) GetTaskAttempt(context.Context, auth.Token, core.ID, core.ID) (AttemptView, error) diff --git a/pulse/worker/service.go b/pulse/worker/service.go index b6eada6..467bcd3 100644 --- a/pulse/worker/service.go +++ b/pulse/worker/service.go @@ -18,6 +18,7 @@ import ( var StaleSession = fault.New("STALE_SESSION", "connection is expired or superseded") var LeaseLost = fault.New("LEASE_LOST", "attempt no longer grants execution authority") var GateClosed = fault.New("EXECUTION_PAUSED", "queue or execution gate is paused") +var RateLimited = fault.New("RATE_LIMITED", "queue start limit reached; retry the original start within its lease") type Service struct{ store Store } diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 51d2956..3a3f851 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -51,12 +51,14 @@ if [ -z "$TASK_HTTP_PORT" ]; then printf '%s\n' 'Bootstrap HTTP server did not s 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 --schedule-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 'IntegrationWorker(Connection|Runtime)' -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 '^TestIntegrationSchedule' -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 '^TestIntegrationQueueResourcesSDK$' -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) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/history) (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/retry) (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) # 停止原服务后再做进程故障演练，避免其他恢复扫描器掩盖接管验证。 kill "$TASK_SERVER_PID" diff --git a/sdk/go/examples/resources/main.go b/sdk/go/examples/resources/main.go new file mode 100644 index 0000000..cf60497 --- /dev/null +++ b/sdk/go/examples/resources/main.go @@ -0,0 +1,150 @@ +// resources 演示队列暂停、积压背压、幂等核对、释放空间与限速执行。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "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 := 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 := "resources." + op + created, err := api.CreateQueue(ctx, "queue-"+op, orbit.CreateQueueRequest{Name: name, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2}) + if err != nil { + return err + } + q := created.Queue + configure := func(paused bool) error { + out, err := api.ConfigureQueue(ctx, q.ID, fmt.Sprintf("configure-%s-%t", op, paused), orbit.ConfigureQueueRequest{ExpectedRevision: q.Revision, Reason: "resource control example", Paused: paused, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2}) + if err == nil { + q = out.Queue + } + return err + } + if err := configure(true); err != nil { + return err + } + in := orbit.SubmitTaskRequest{QueueID: q.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"object_ref":"objects/example","n":9007199254740993}`), TaskTimeoutMS: 20000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 10000}} + first, err := api.SubmitTask(ctx, "first-"+op, in) + if err != nil { + return err + } + second, err := api.SubmitTask(ctx, "second-"+op, in) + if err != nil { + return err + } + _, err = api.SubmitTask(ctx, "third-"+op, in) + var rejected *orbit.APIError + if !errors.As(err, &rejected) || rejected.Code != "QUEUE_FULL" { + return fmt.Errorf("expected QUEUE_FULL, got %v", err) + } + fmt.Printf("backpressure=%s http=%d\n", rejected.Code, rejected.StatusCode) + duplicate, err := api.SubmitTask(ctx, "first-"+op, in) + if err != nil { + return err + } + if !duplicate.Deduplicated || duplicate.Task.ID != first.Task.ID { + return errors.New("original receipt was not preserved") + } + if _, err := api.CancelTask(ctx, first.Task.ID, "cancel-"+op, orbit.CancelTaskRequest{ExpectedRevision: first.Task.Revision, Reason: "free one unfinished slot"}); err != nil { + return err + } + // 明确 429 表示未受理；释放空间后仍使用第三个任务的原幂等键与原内容。 + third, err := api.SubmitTask(ctx, "third-"+op, in) + if err != nil { + return err + } + fmt.Println("duplicate=true rejected_request_retried=true") + w, err := orbit.NewWorker(api, orbit.WorkerOptions{QueueID: q.ID, Capacity: 2, ShutdownTimeout: 5 * time.Second, Handlers: []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 + } + return e.Task.Payload, nil + }}}}) + if err != nil { + return err + } + run, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + waited := false + defer func() { + stop() + if !waited { + <-done + } + }() + if err := configure(false); err != nil { + return err + } + for _, id := range []string{second.Task.ID, third.Task.ID} { + for { + task, err := api.GetTask(ctx, id) + if err != nil { + return err + } + if task.State == orbit.TaskSucceeded { + fmt.Printf("task=%s state=%s attempts=%d\n", id, task.State, task.AttemptCount) + break + } + if task.State == orbit.TaskFailed || task.State == orbit.TaskNeedsReview || task.State == orbit.TaskCancelled { + return fmt.Errorf("unexpected task state: %s", task.State) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(80 * time.Millisecond): + } + } + } + q, err = api.GetQueue(ctx, q.ID) + if err != nil { + return err + } + if q.UnfinishedTasks != 0 { + return errors.New("queue still has unfinished work") + } + stop() + err = <-done + waited = true + if err != nil { + return err + } + fmt.Printf("unfinished=%d rate_per_second=%d worker_drained=true\n", q.UnfinishedTasks, q.RatePerSecond) + return nil +} diff --git a/sdk/go/queue_resources_integration_test.go b/sdk/go/queue_resources_integration_test.go new file mode 100644 index 0000000..f67f892 --- /dev/null +++ b/sdk/go/queue_resources_integration_test.go @@ -0,0 +1,115 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" +) + +func TestIntegrationQueueResourcesSDK(t *testing.T) { + api, _, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + name := "resources." + workerTestID(t) + created, err := api.CreateQueue(ctx, "create-"+name, CreateQueueRequest{Name: name, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2}) + if err != nil { + t.Fatal(err) + } + q := created.Queue + configure := func(paused bool) { + t.Helper() + out, err := api.ConfigureQueue(ctx, q.ID, workerTestID(t), ConfigureQueueRequest{ExpectedRevision: q.Revision, Reason: "SDK resource regression", Paused: paused, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2}) + if err != nil { + t.Fatal(err) + } + q = out.Queue + } + configure(true) + // 丢失 paused 不能被当成解除暂停；这是原始 HTTP 对象，SDK 本身总会发送 bool。 + var ignored CreateQueueResult + err = api.call(ctx, "POST", "/v1/queues/"+q.ID+":configure", workerTestID(t), map[string]any{"expected_revision": "2", "reason": "missing pause", "max_concurrency": 2, "rate_per_second": 1, "max_unfinished_tasks": 2}, &ignored) + var apiErr *APIError + if !errors.As(err, &apiErr) || apiErr.StatusCode != 400 { + t.Fatalf("missing paused accepted: %v", err) + } + input := SubmitTaskRequest{QueueID: q.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)} + first, err := api.SubmitTask(ctx, name+"-one", input) + if err != nil { + t.Fatal(err) + } + second, err := api.SubmitTask(ctx, name+"-two", input) + if err != nil { + t.Fatal(err) + } + _, err = api.SubmitTask(ctx, name+"-three", input) + if !errors.As(err, &apiErr) || apiErr.StatusCode != 429 || apiErr.Code != "QUEUE_FULL" || apiErr.RetryAfter != time.Second { + t.Fatalf("queue did not backpressure: %v", err) + } + if duplicate, err := api.SubmitTask(ctx, name+"-one", input); err != nil || !duplicate.Deduplicated || duplicate.Task.ID != first.Task.ID { + t.Fatal("full queue lost original identity") + } + if _, err := api.CancelTask(ctx, first.Task.ID, name+"-cancel", CancelTaskRequest{ExpectedRevision: first.Task.Revision, Reason: "free backlog slot"}); err != nil { + t.Fatal(err) + } + third, err := api.SubmitTask(ctx, name+"-three", input) + if err != nil { + t.Fatal(err) + } + w, err := NewWorker(api, WorkerOptions{QueueID: q.ID, Capacity: 2, Handlers: []WorkerHandler{runtimeHandler(name, func(ctx context.Context, e WorkerExecution) (json.RawMessage, error) { return e.Task.Payload, nil })}}) + if err != nil { + t.Fatal(err) + } + run, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(run) }() + waited := false + defer func() { + stop() + if !waited { + <-done + } + }() + select { + case <-time.After(350 * time.Millisecond): + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + current, err := api.GetTask(ctx, second.Task.ID) + if err != nil || current.AttemptCount != 0 { + t.Fatal("paused queue consumed attempt") + } + configure(false) + for _, id := range []string{second.Task.ID, third.Task.ID} { + finished := waitRuntimeState(t, ctx, api, id, TaskSucceeded) + if finished.AttemptCount != 1 || string(finished.Result) != `{"n":9007199254740993}` { + t.Fatal("resource controls altered execution") + } + } + var authorized []time.Time + for _, id := range []string{second.Task.ID, third.Task.ID} { + attempts, err := api.ListTaskAttempts(ctx, id, HistoryPageRequest{}) + if err != nil || len(attempts.Items) != 1 || attempts.Items[0].AuthorizedAt == nil { + t.Fatal("missing start history") + } + authorized = append(authorized, *attempts.Items[0].AuthorizedAt) + } + gap := authorized[1].Sub(authorized[0]) + if gap < 0 { + gap = -gap + } + if gap < time.Second { + t.Fatalf("start rate exceeded: %s", gap) + } + q, err = api.GetQueue(ctx, q.ID) + if err != nil || q.UnfinishedTasks != 0 || q.Revision != 3 { + t.Fatal("quota count or configuration revision drifted") + } + stop() + err = <-done + waited = true + if err != nil { + t.Fatal(err) + } +} diff --git a/sdk/go/tasks.go b/sdk/go/tasks.go index 7e62b9a..454bec1 100644 --- a/sdk/go/tasks.go +++ b/sdk/go/tasks.go @@ -10,6 +10,7 @@ import ( "net/http" "net/url" "strconv" + "time" ) // APIError 提供可分支处理的错误码和请求追踪标识，不包含内部数据库错误。 @@ -18,6 +19,8 @@ type APIError struct { Code string Message string RequestID string + // RetryAfter 服务端建议的最小退避时间；缺失、无效或超过一小时返回零，SDK 不自动重试。 + RetryAfter time.Duration } func (e *APIError) Error() string { @@ -56,7 +59,7 @@ func (c *Client) ListTasks(ctx context.Context, in ListTasksRequest) (TaskPage, return out, err } -// CancelTask 当前只取消 WAITING 状态且不归属流程的任务；重复请求使用原键与原版本。 +// CancelTask 保存取消意图；运行中等待结果确认，重复请求保留原键与原版本。 func (c *Client) CancelTask(ctx context.Context, id, key string, in CancelTaskRequest) (CancelTaskResult, error) { var out CancelTaskResult if !validID(id) { @@ -150,7 +153,11 @@ func (c *Client) call(ctx context.Context, method, path, key string, in, out any decoded := json.Unmarshal(raw, &envelope) if response.StatusCode < 200 || response.StatusCode >= 300 { if decoded == nil && envelope.Error != nil && len(envelope.Error.Code) <= 64 && len(envelope.Error.Message) <= 1024 && validID(envelope.RequestID) { - return &APIError{response.StatusCode, envelope.Error.Code, envelope.Error.Message, envelope.RequestID} + var retryAfter time.Duration + if seconds, err := strconv.ParseUint(response.Header.Get("Retry-After"), 10, 32); err == nil && seconds > 0 && seconds <= 3600 { + retryAfter = time.Duration(seconds) * time.Second + } + return &APIError{StatusCode: response.StatusCode, Code: envelope.Error.Code, Message: envelope.Error.Message, RequestID: envelope.RequestID, RetryAfter: retryAfter} } return &HTTPError{StatusCode: response.StatusCode} } @@ -159,3 +166,13 @@ func (c *Client) call(ctx context.Context, method, path, key string, in, out any } return json.Unmarshal(envelope.Data, out) } + +// ConfigureQueue 完整替换队列控制参数；重复请求保留原版本及幂等键。 +func (c *Client) ConfigureQueue(ctx context.Context, id, key string, in ConfigureQueueRequest) (CreateQueueResult, error) { + var out CreateQueueResult + if !validID(id) { + return out, errors.New("invalid queue id") + } + err := c.call(ctx, "POST", "/v1/queues/"+id+":configure", key, in, &out) + return out, err +} diff --git a/sdk/go/types.go b/sdk/go/types.go index 00fdca4..024f1c8 100644 --- a/sdk/go/types.go +++ b/sdk/go/types.go @@ -144,14 +144,20 @@ type TaskPage struct { NextCursor string `json:"next_cursor,omitempty"` } type CreateQueueRequest struct { + // MaxUnfinishedTasks 未终结任务上限，包含待核查；省略或零使用 100000。 + MaxUnfinishedTasks uint32 `json:"max_unfinished_tasks,omitempty"` // Name 队列名称，同一命名空间内唯一。 Name string `json:"name"` - // MaxConcurrency 最大同时执行数量，供执行阶段使用。 + // MaxConcurrency 允许占用的有效执行许可上限。 MaxConcurrency uint32 `json:"max_concurrency"` - // RatePerSecond 每秒开始数量上限，供执行阶段使用。 + // RatePerSecond 数据库一秒窗口最多提交的开始授权数。 RatePerSecond uint32 `json:"rate_per_second"` } type Queue struct { + // MaxUnfinishedTasks 未终结任务上限，只限制新的受理。 + MaxUnfinishedTasks uint32 `json:"max_unfinished_tasks"` + // UnfinishedTasks 未终结任务数，线上为十进制字符串；变化不增加配置版本。 + UnfinishedTasks uint64 `json:"unfinished_tasks,string"` // ID 资源标识，32 位小写十六进制字符串。 ID string `json:"id"` // NamespaceID 命名空间标识，由凭据确定隔离范围。 @@ -160,9 +166,9 @@ type Queue struct { Name string `json:"name"` // Paused 是否停止新执行；不禁止受理新任务。 Paused bool `json:"paused"` - // MaxConcurrency 最大同时执行数量，供执行阶段使用。 + // MaxConcurrency 允许占用的有效执行许可上限。 MaxConcurrency uint32 `json:"max_concurrency"` - // RatePerSecond 每秒开始数量上限，供执行阶段使用。 + // RatePerSecond 数据库一秒窗口最多提交的开始授权数。 RatePerSecond uint32 `json:"rate_per_second"` // Revision 并发版本，JSON 使用十进制字符串避免跨语言精度丢失。 Revision uint64 `json:"revision,string"` @@ -202,3 +208,19 @@ type RetryPolicy struct { MaxDelayMS uint64 `json:"max_delay_ms"` // MaxDelayMS 间隔上限。 AttemptTimeoutMS uint64 `json:"attempt_timeout_ms"` // AttemptTimeoutMS 单次尝试时限。 } + +// ConfigureQueueRequest 完整替换控制配置，必须保留同一操作的原版本及原因。 +type ConfigureQueueRequest struct { + // ExpectedRevision 非零的队列配置版本。 + ExpectedRevision uint64 `json:"expected_revision,string"` + // Reason 管理原因，写入不可变审计。 + Reason string `json:"reason"` + // Paused 停止新的领取和开始；在途续租、结果与受理继续。 + Paused bool `json:"paused"` + // MaxConcurrency 有效并发上限，降低时已有许可自然收尾。 + MaxConcurrency uint32 `json:"max_concurrency"` + // RatePerSecond 数据库一秒窗口最多开始授权数，降低时保留原计数。 + RatePerSecond uint32 `json:"rate_per_second"` + // MaxUnfinishedTasks 未终结任务上限，降低时不删除历史或在途任务。 + MaxUnfinishedTasks uint32 `json:"max_unfinished_tasks"` +} diff --git a/sdk/go/worker_process_integration_test.go b/sdk/go/worker_process_integration_test.go index b8fc957..232f077 100644 --- a/sdk/go/worker_process_integration_test.go +++ b/sdk/go/worker_process_integration_test.go @@ -147,6 +147,49 @@ func awaitProcessDrop(t *testing.T, ctx context.Context, g *workerFaultGateway, } } +func TestIntegrationWorkerProcessQueueBackpressure(t *testing.T) { + a := startOrbitProcess(t) + _, _, token := integrationWorkerClient(t) + api := processClient(t, a, token) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + name := "quota-crash-" + workerTestID(t) + q, err := api.CreateQueue(ctx, name, CreateQueueRequest{Name: name, MaxConcurrency: 1, RatePerSecond: 1, MaxUnfinishedTasks: 1}) + if err != nil { + t.Fatal(err) + } + in := SubmitTaskRequest{QueueID: q.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)} + first, err := api.SubmitTask(ctx, name+"-first", in) + if err != nil { + t.Fatal(err) + } + a.kill(t) + b := startOrbitProcess(t) + api = processClient(t, b, token) + if a.nodeID == b.nodeID { + t.Fatal("node identity reused") + } + _, err = api.SubmitTask(ctx, name+"-second", in) + var rejected *APIError + if !errors.As(err, &rejected) || rejected.Code != "QUEUE_FULL" { + t.Fatalf("restart lost quota: %v", err) + } + dupe, err := api.SubmitTask(ctx, name+"-first", in) + if err != nil || !dupe.Deduplicated || dupe.Task.ID != first.Task.ID { + t.Fatal("restart lost original receipt") + } + if _, err := api.CancelTask(ctx, first.Task.ID, name+"-release", CancelTaskRequest{ExpectedRevision: first.Task.Revision, Reason: "free recovered queue"}); err != nil { + t.Fatal(err) + } + if _, err := api.SubmitTask(ctx, name+"-second", in); err != nil { + t.Fatal(err) + } + current, err := api.GetQueue(ctx, q.Queue.ID) + if err != nil || current.UnfinishedTasks != 1 { + t.Fatal("recovered quota counter mismatch") + } +} + // 两次在真实数据库提交后丢 ACK 并强杀节点，由另外的真实进程核对原操作。 func TestIntegrationWorkerProcessACKRecovery(t *testing.T) { a := startOrbitProcess(t) diff --git a/sdk/go/worker_runtime.go b/sdk/go/worker_runtime.go index 40ca29f..de29e81 100644 --- a/sdk/go/worker_runtime.go +++ b/sdk/go/worker_runtime.go @@ -497,7 +497,7 @@ func (r *workerRuntime) receive(reply runtimeReply) error { r.remove(j) } return nil - case "EXECUTION_PAUSED", "UNAVAILABLE": + case "EXECUTION_PAUSED", "RATE_LIMITED", "UNAVAILABLE": if j.retryDelay == 0 { j.retryDelay = 200 * time.Millisecond } else {