diff --git a/AGENTS.md b/AGENTS.md index 9696920..07df215 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,9 +14,11 @@ - 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 - 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 -- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，P3仍需补齐任务查询筛选和事务Outbox接入；不能因订阅完成而跳过原功能范围。 +- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，任务查询已支持业务键及创建时间，P3仍需补齐任务详情来源追踪和事务Outbox接入；不能因订阅完成而跳过原功能范围。 - 批量受理最多32项、512 KiB；每项复用单项身份与事务，未知结果必须原键核对。任务提交按认证→应用额度→提交幂等身份→队列持锁，禁止恢复为先抢提交唯一键后等待额度锁；原回执仍须先于配额上限检查返回。 - 独立任务核查必须显式声明所有历史执行已结束，并提交原因、证据和版本；NEEDS_REVIEW只转SUCCEEDED/FAILED。核查不改写旧尝试或迟到证据。人工重试仅无所有者且无未知效果的FAILED/CANCELLED任务，创建retry_of关联新任务，每个原任务最多一个直接后继；继续重试沿后继推进。 - tasks:review、tasks:retry、tasks:audit:read是显式管理权限，普通提交/Worker凭据不自动获得。核查结果流锁最后获取，审计/回执/额度与状态同事务；恢复不得绕过Weave/计划所有权、门控及现有额度限制。 + +- 任务列表队列必填，business_key精确匹配（包含尾部空格），created_from包含/created_before不包含。v2游标绑定所有筛选，不能因网络或游标错误删筛选降级；旧游标仅兼容旧查询。四种固定索引覆盖状态/业务键组合，禁止全量取回后内存过滤。业务键仅Go空字符串映射NULL。 diff --git a/README.md b/README.md index 8df345b..c6d56a9 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已完成 P0–P2 阶段验收：35 个数据模型、版本化迁移、凭据及范围校验、队列与任务 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) 和 [优先级/节点公平派发](docs/api/priority-and-fairness.md) 已接入服务端和 SDK，[可靠结果订阅](docs/api/subscriptions.md) 已实现持久化核心、HTTP/WSS、管理 SDK 与示例，[消费 SDK](docs/sdk/event-consumer.md) 和 [独立 MySQL Inbox 示例](docs/examples/result-inbox.md) 已实现；[批量任务受理](docs/api/batch-tasks.md) 已提供逐项幂等、失败分类与 SDK 示例；[人工核查与重试](docs/api/task-recovery.md) 已提供带证据的管理接口、关联新任务、审计查询及运行示例；剩余 Pulse 能力与流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已完成 P0–P2 阶段验收：35 个数据模型、版本化迁移、凭据及范围校验、队列与任务 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) 和 [优先级/节点公平派发](docs/api/priority-and-fairness.md) 已接入服务端和 SDK，[可靠结果订阅](docs/api/subscriptions.md) 已实现持久化核心、HTTP/WSS、管理 SDK 与示例，[消费 SDK](docs/sdk/event-consumer.md) 和 [独立 MySQL Inbox 示例](docs/examples/result-inbox.md) 已实现；[批量任务受理](docs/api/batch-tasks.md) 已提供逐项幂等、失败分类与 SDK 示例；[人工核查与重试](docs/api/task-recovery.md) 已提供带证据的管理接口、关联新任务、审计查询及运行示例；[业务键与时间范围查询](docs/api/task-queries.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/01-functional-spec.md b/docs/01-functional-spec.md index e6b80eb..3bb81a1 100644 --- a/docs/01-functional-spec.md +++ b/docs/01-functional-spec.md @@ -1,6 +1,6 @@ # Orbit 功能说明：Pulse 与公共能力 -> 版本 0.2 · 2026-09-17。Pulse 与 Weave 首版功能范围已确认；标注“建议”的细节待评审。Weave 具体范围另见流程设计。基础、任务执行、调度、资源控制和可靠订阅已实现；批量受理及独立任务核查/人工重试已实现，查询筛选等继续开发，实现状态见持续开发计划。 +> 版本 0.2 · 2026-09-17。Pulse 与 Weave 首版功能范围已确认；标注“建议”的细节待评审。Weave 具体范围另见流程设计。基础、任务执行、调度、资源控制和可靠订阅已实现；批量受理、独立任务核查/人工重试及业务键/创建时间查询已实现，事务Outbox接入等继续开发，实现状态见持续开发计划。 ## 1. 目标与责任边界 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 5e654db..7c78bed 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -168,3 +168,9 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 核查和重试按认证→应用额度→操作幂等→队列→任务/尝试持锁，事务内校验状态、父任务/计划归属和无活动执行。核查的状态/result、额度释放、审计、回执及结果日志一起提交，结果流锁最后取得。人工重试创建新任务与受理事件，审计关联保存在原任务上；旧终态、旧尝试和迟到证据不覆盖。详见[接口契约](api/task-recovery.md)。 真实v7升级回归检查70个外键全部删除、原任务保留、323个未替换索引组件不变，并单独核验新的重试唯一索引覆盖原两列。已有非法重复重试关联或自关联不会被静默清理；迁移失败需核查实际数据。后续仍禁止外键及级联。 + +## 000016 任务业务键及创建时间查询 + +BusinessKey沿用既有可空VARCHAR(191)，Go/SQL中文注释明确精确匹配、区分大小写和尾部空格；排序规则显式utf8mb4_0900_bin，历史文本原样保留，不增加模型字段。新增(namespace_id,queue_id,business_key,created_at,id)及增加state的组合索引，配合原两种创建时间索引覆盖有无状态/业务键的四种查询。 + +HTTP/SDK创建时间范围使用包含下界、不包含上界；v2游标以固定长度摘要绑定全部规范化筛选，当前权限逐页校验。单项提交、计划生成和人工重试仅把Go空字符串映射为NULL，不用连接排序规则折叠空格。仍为35个实体、289个中文注释字段，无外键/级联；具体接口与升级边界见[查询手册](api/task-queries.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 97a5b26..e36a5fd 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/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景、COMMIT响应丢失及消费者进程故障回归已完成；批量受理、SDK、接口与部分补交示例已完成全量回归；独立任务核查/人工重试、审计查询、两次真实HTTP回执丢失及完整示例已通过全量回归；业务键/时间筛选与事务Outbox接入待完成 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景、COMMIT响应丢失及消费者进程故障回归已完成；批量受理、SDK、接口与部分补交示例已完成全量回归；独立任务核查/人工重试、审计查询、两次真实HTTP回执丢失及完整示例已通过全量回归；业务键/创建时间筛选、SDK与示例已通过全量回归；任务详情来源追踪与事务Outbox接入待完成 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -43,7 +43,8 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | --- | --- | | 批量任务受理 | 已实现[批量API](api/batch-tasks.md)、独立SDK及[补交示例](examples/batch.md)；真实MySQL部分失败与HTTP响应丢失用例已通过，同键并发20轮及全量回归已通过，见[报告](reports/phase-3-batch-tasks.md) | | 独立任务人工核查与重试 | 已实现[管理API及SDK](api/task-recovery.md)、专门权限、显式安全声明、证据/版本/审计和关联新任务；[完整场景](examples/recovery.md)及[回归报告](reports/phase-3-task-recovery.md)同步交付 | -| 任务查询筛选 | [当前ListTasks](../sdk/go/types.go) 仅队列/状态/分页；需按业务关联键及创建时间范围查询，游标绑定所有筛选，保持授权与有界索引扫描 | +| 任务查询筛选 | 已实现[业务键及创建时间查询](api/task-queries.md)，v2游标绑定全部筛选、旧游标受限兼容、四种索引路径与[完整场景](examples/task-query.md)；[回归报告](reports/phase-3-task-queries.md)同步记录 | +| 任务详情来源追踪 | 原功能要求能查看关联计划；当前TaskView未返回来源计划/触发记录。需补齐任务到计划及触发版本的查询，验证当前权限、同命名空间归属与来源缺失处理；流程来源随Weave接入 | | 业务到平台的事务Outbox接入 | [事务Outbox设计](04-reliability-and-data.md#61-业务系统到平台) 尚缺完整接入示例；需业务变更与待发请求同事务、稳定提交身份、响应不明核对和发送恢复，保持核心SDK无ORM/数据库依赖 | 凭据安全重绑、历史清理、全量监控/容量与MySQL故障切换仍按P6执行；Weave的通用执行门控/内部事件接续随P4–P5落实，不将外部结果订阅当作内部流程引擎。 @@ -64,6 +65,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 优先级与应用隔离 | 近同时到期按优先级 → 旧任务等待补偿 → 应用/队列轮转 | P3，已实现完整示例 | | 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，核心、HTTP/WSS、管理及消费 SDK、独立Inbox示例与真实进程故障回归已完成 | | 批量文件作业受理 | 逐项身份 → 部分结果 → 同键核对 → 释放额度 → 只补交失败项 | P3，已实现完整示例 | +| 业务操作定位 | 精确业务键 → 创建区间 → 游标分页 → 查询详情 | P3，已实现完整示例 | | 独立任务人工恢复 | 未知结果 → 证据核查 → 保留原任务 → 关联新任务 → 审计查询 | P3，已实现完整示例 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | | 审核与外部回执 | 提交流程 → 等待信号 → 授权信号注入 → 继续 | P4 | @@ -98,6 +100,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P3 业务 Inbox 与消费者故障回归](reports/phase-3-result-inbox.md) - [P3 批量任务受理回归](reports/phase-3-batch-tasks.md) - [P3 人工核查与重试回归](reports/phase-3-task-recovery.md) +- [P3 业务键与时间查询回归](reports/phase-3-task-queries.md) - [接口手册](api/README.md) - [SDK 接入](sdk/README.md) - [通用场景示例](examples/README.md) diff --git a/docs/README.md b/docs/README.md index 8dbaf38..b7e949a 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 5a560e3..7620bcc 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -8,7 +8,7 @@ | --- | --- | --- | | `GET /health/live` | 已实现 | [健康检查](health.md) | | `GET /health/ready` | 已实现，验证数据库、迁移记录及零外键要求 | [健康检查](health.md) | -| `/v1/tasks` 提交、查询、分页与待执行取消 | P1 已实现 | [任务接口](tasks.md) | +| `/v1/tasks` 提交、查询、分页与待执行取消 | 已实现，P3扩展业务键与创建时间筛选 | [任务接口](tasks.md) / [查询契约](task-queries.md) | | `POST /v1/tasks:batch` | P3 已实现逐项受理、独立幂等、部分结果与补交 | [批量任务](batch-tasks.md) | | `/v1/tasks/{task_id}:review` / `:retry` 与 `/operations` | P3 已实现带证据核查、关联新任务与操作审计 | [人工恢复](task-recovery.md) | | `/v1/tasks/{task_id}/attempts` 与尝试详情 | P2 已实现 | [执行历史](execution-history.md) | diff --git a/docs/api/task-queries.md b/docs/api/task-queries.md new file mode 100644 index 0000000..5d9e54b --- /dev/null +++ b/docs/api/task-queries.md @@ -0,0 +1,117 @@ +# 按业务键与创建时间查询任务 + +状态：P3 已实现HTTP、Go SDK及[运行示例](../examples/task-query.md)。这是 `GET /v1/tasks` 的扩展，返回任务摘要，查询不执行或重试任务。验证记录见[本阶段报告](../reports/phase-3-task-queries.md)。 + +## 1. 权限与字段 + +请求使用 `Authorization: Bearer `，需要当前命名空间的 `tasks:read` 及指定队列范围，不携带幂等键。命名空间由凭据确定，不接受调用方指定。队列不在权限范围返回403；有权限但不存在该队列/无匹配任务时返回空列表，不披露其他命名空间资源。 + +| 参数 | 类型与默认 | 要求 | +| --- | --- | --- | +| queue_id | string，必填 | 32位小写十六进制队列标识 | +| state | string，可选 | WAITING、LEASED、RUNNING、NEEDS_REVIEW、SUCCEEDED、FAILED、CANCELLED之一；省略表示所有状态 | +| business_key | string，可选 | 1–191个Unicode字符，无NUL/回车/换行；精确关联匹配，显式空值拒绝 | +| created_from | RFC3339时间，可选 | 创建时间下界，包含该时刻 | +| created_before | RFC3339时间，可选 | 创建时间上界，不包含该时刻；与下界同时给出时必须更晚 | +| limit | integer，默认20 | 1–100；可以在翻页时更改 | +| cursor | string，可选 | 上页next_cursor，最长1024字符，必须保持其他筛选不变 | + +参数只能出现一次，未知名称、非法百分号编码、空时间、日期字符串、非规范状态或非法范围返回400。时间允许RFC3339时区偏移并规范化UTC；精度最多到微秒，UTC年份2000–9998。上下界可以仅给一个，不设置默认时间窗口。JSON的null不是URL查询值，省略参数才表示不设该条件。 + +business_key区分大小写、重音字符、Unicode编码形式以及尾部空格。例如 `object:1`、`object:1 `、`Object:1` 是不同键；`é` 与 `e` 加组合重音也不同。不修剪/规范化输入，不使用LIKE，`%` 和 `_` 是普通字符；关联键不承担业务幂等。纯空格键若已提交也按原值匹配，空字符串提交仍表示没有关联键；本接口不提供“只查无关联键”筛选。 + +## 2. 完整请求和响应 + +使用URL编码，尤其是键中的 `+`、`%`、空格和时间中的时区加号： + +```sh +curl --get "$ORBIT_URL/v1/tasks" \ + -H "Authorization: Bearer $ORBIT_TOKEN" \ + --data-urlencode "queue_id=$ORBIT_QUEUE_ID" \ + --data-urlencode 'business_key=object:operation-001' \ + --data-urlencode 'state=FAILED' \ + --data-urlencode 'created_from=2026-09-17T00:00:00Z' \ + --data-urlencode 'created_before=2026-09-18T00:00:00Z' \ + --data-urlencode 'limit=20' +``` + +HTTP200完整示意响应： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{ + "items":[{ + "id":"22222222222222222222222222222222", + "queue_id":"11111111111111111111111111111111", + "state":"FAILED", + "revision":"4", + "handler":"object.prepare", + "handler_version":"v1", + "business_key":"object:operation-001", + "available_at":"2026-09-17T00:00:00Z", + "deadline_at":"2026-09-17T00:01:00Z", + "attempt_count":1, + "created_at":"2026-09-17T00:00:00Z", + "updated_at":"2026-09-17T00:00:01Z" + }] + } +} +``` + +有下一页时data还包含next_cursor字符串；没有匹配时items为 `[]`。摘要不携带payload/result/冻结策略/retry_of，按需调用 `GetTask` 获取详情。字段完整说明见[任务接口](tasks.md)。不返回全量总数或大offset接口。 + +## 3. 分页与并发语义 + +固定按 `(created_at DESC,id DESC)` 排序。相同创建时间以任务ID打破平局；取消、执行和核查改变状态/updated_at，不改变创建排序键。每页取至多limit+1行以判断下一页。 + +v2游标同时绑定命名空间、队列、状态、精确业务键和规范化UTC时间界限；长业务键通过固定长度摘要绑定，191字符的合法键不会撑破游标上限。修改/删除任何筛选、混用其他命名空间或传入非法游标返回400，不能通过更换参数扩大原页查询。游标是排序位置，不是授权证明，每页都重新校验当前凭据和队列范围。 + +跨页不是数据库快照：遍历期间新增任务、任务状态变化可能影响后续页面，不能将动态状态查询当作完整审计导出。新任务排在已读位置之前时，从首页刷新才会看到；状态筛选的任务离开该状态时可能从后续页面消失。需要可靠处理结果使用[结果订阅](subscriptions.md)，核对历史使用[执行历史](execution-history.md)与[操作审计](task-recovery.md)。 + +旧版未带版本的游标仅兼容原队列/状态查询，不能附加业务键或时间界限。SDK把游标当不透明字符串，不应解析、改写或自行拼接。 + +## 4. SDK调用 + +`Client.ListTasks(ctx, orbit.ListTasksRequest)` 返回 `TaskPage`。新增字段BusinessKey、CreatedFrom、CreatedBefore，其余字段保持不变。BusinessKey空字符串在SDK中表示省略，时间指针nil表示无该界限。 + +```go +query := orbit.ListTasksRequest{ + QueueID: queueID, + State: orbit.TaskFailed, + BusinessKey: "object:operation-001", + CreatedFrom: &from, + CreatedBefore: &before, + Limit: 20, +} +for { + page, err := client.ListTasks(ctx, query) + if err != nil { return err } + for _, task := range page.Items { + fmt.Println(task.ID, task.State, task.CreatedAt) + } + if page.NextCursor == "" { break } + query.Cursor = page.NextCursor +} +``` + +片段假设已有客户端及time.Time类型from/before；完整可运行源码见[示例](../../sdk/go/examples/task-query/main.go)。SDK校验ID、范围、状态、键长度/UTF-8及时间精度，并用url.Values编码，时间转UTC，不修改调用方时间对象。超时只影响读取，可保持条件重新请求；没有自动重试或自动翻页。 + +## 5. 存储、升级与边界 + +迁移000016明确business_key使用 `utf8mb4_0900_bin`，保证尾部空格有意义；现有文本不重写、不修剪，不增加模型字段或数据库外键。单项受理、计划触发和人工重试按Go字符串是否为空决定NULL，不依赖连接排序规则的NULLIF比较。 + +查询使用四个固定索引之一，全部将等值过滤放在创建时间及ID之前： + +| 筛选组合（均含命名空间与队列） | 索引 | +| --- | --- | +| 无状态/业务键 | ix_task_created | +| 状态 | ix_task_state_created | +| 业务键 | ix_task_queue_business_created，新增 | +| 业务键与状态 | ix_task_queue_business_state_created，新增 | + +上下界与游标在索引时间范围内继续扫描，不按全部任务取回后内存过滤。SQL索引名来自固定服务端分支，查询值全部参数化。实际MySQL执行计划验证四组走对应range/ref索引且无filesort；这是查询路径验证，不是生产吞吐或百万租户容量证明。 + +升级需先停止旧写入节点、应用迁移，再统一启用新版节点；列排序规则和索引调整会有DDL成本，应演练目标规模维护窗口。新SDK查询旧节点时，新参数会被拒绝，不能删除筛选后自动降级。新旧节点混合分页与生产滚动升级尚未验收。 + +通用400/401/403/503错误见[任务手册](tasks.md)。业务键与时间查询不新增权限，不授予执行、取消或人工核查权限。 diff --git a/docs/api/tasks.md b/docs/api/tasks.md index 98beec2..e22dd69 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -124,8 +124,12 @@ SDK：`Client.GetTask(ctx, taskID)`。 | --- | --- | | queue_id | 必填，有权限的队列 ID | | state | 可选，取上述状态之一 | +| business_key | 可选，精确匹配1–191个Unicode字符，区分尾部空格；显式空值拒绝 | +| created_from / created_before | 可选，RFC3339微秒时间；创建时间包含起点、不包含终点，两者同时设置时起点必须更早 | | limit | 可选，默认 20；1–100 | -| cursor | 可选，上一页返回值；必须保持 queue_id/state 不变，最长 1024 字符 | +| cursor | 可选，上一页返回值；必须保持队列、状态、业务键和时间范围不变，最长1024字符 | + +完整筛选、时间边界、索引与兼容说明见[业务键与时间查询](task-queries.md)，并提供[完整运行示例](../examples/task-query.md)。 按 (created_at DESC, id DESC) 游标分页，不使用大 offset。不承诺跨页快照：读取过程中状态可能变化。列表省略 payload/result/retry 等大字段，返回摘要。 diff --git a/docs/examples/README.md b/docs/examples/README.md index 7f372cb..ccb36b4 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -146,3 +146,10 @@ - 预期：原任务FAILED、新任务SUCCEEDED，原尝试和审计保留；管理凭据显式--recovery-access。 - 异常：真实HTTP在核查/重试提交后丢响应，原身份恢复同一审计和新任务；迟到成功只进入证据。 - 范围：本例不操作外部系统，不能替代真实证据核查、业务幂等或持久化管理操作Outbox。 + +## 已可运行：业务键和创建时间查询 + +- 源码：[task-query](../../sdk/go/examples/task-query/main.go)，[完整启动与异常说明](task-query.md)。 +- 步骤：提交三个模拟任务 → 精确业务键及时间范围过滤 → 每页一条遍历 → 验证上界排除 → 取消收尾。 +- 预期：matched=2、pages=2、before_excluded=true、cancelled=3。 +- 范围：普通提交/读取/取消权限；不调用外部系统、不启动Worker，所有历史保留，状态分页不是数据库快照。 diff --git a/docs/examples/task-query.md b/docs/examples/task-query.md new file mode 100644 index 0000000..8ee8e7d --- /dev/null +++ b/docs/examples/task-query.md @@ -0,0 +1,36 @@ +# 业务键与时间范围查询示例 + +源码：[task-query/main.go](../../sdk/go/examples/task-query/main.go)。本例提交三个模拟对象任务，不启动Worker，演示精确业务键、创建时间区间和任务摘要分页。 + +## 准备与运行 + +按[SDK初始化步骤](../sdk/README.md)迁移数据库、初始化专用开发队列及普通提交/读取/取消凭据，启动当前Orbit服务。从仓库根目录执行： + +```sh +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/development.credentials.json \ + go run ./examples/task-query +``` + +也可使用ORBIT_TOKEN与ORBIT_QUEUE_ID，环境值优先。无需额外管理权限，凭据秘密不输出。队列应有至少三个未终结任务空位；每次运行的处理器与业务键带独立标识，避免已有任务干扰。 + +## 实际调用与结果 + +1. 提交两个完全相同业务键的任务，以及一个只相差尾部空格的任务。业务键含中文、加号、百分号及重音字符。 +2. 用服务端回执中的CreatedAt设置包含起点、不包含终点的时间范围，避免依赖客户端时钟同步。 +3. ListTasks按原键、WAITING状态和创建区间查询，每页1条，沿NextCursor遍历；仅匹配前两个任务。 +4. 将created_before设为第二个任务的精确创建时间，确认它被排除，只剩第一个任务。 +5. 分别取消本例三个任务，保留任务、受理与取消审计。 + +预期输出： + +```text +matched=2 pages=2 before_excluded=true cancelled=3 +``` + +重复运行创建新的独立样本。代码没有外部业务副作用，也不删除数据库历史。失败非零退出，可能留下已受理任务；用回执、业务键和时间范围查询后核对，各任务总时限60秒，当前示例没有Worker会进入未开始超时处理。 + +完整参数、游标兼容和异常说明见[查询契约](../api/task-queries.md)。翻页时保持所有筛选，发生INVALID_ARGUMENT不要丢弃过滤条件“兜底查全部”。状态变化的跨页结果不是快照，不能当作完整审计导出。 + +`make integration` 会在一次性MySQL中运行此示例及实际HTTP SDK回归，验证原键/尾部空格、时间边界、分页、变更条件拒绝，以及取消后创建时间不改变。 diff --git a/docs/reports/phase-3-task-queries.md b/docs/reports/phase-3-task-queries.md new file mode 100644 index 0000000..f0812ad --- /dev/null +++ b/docs/reports/phase-3-task-queries.md @@ -0,0 +1,75 @@ +# P3 业务键与创建时间查询回归 + +验证日期：2026-09-17。当前已实现服务端、SDK、迁移、接口及可运行场景；全量回归已正常退出，状态码0。P3的任务详情来源追踪和业务事务Outbox接入仍待交付，不能据此宣布Pulse或全平台完成。 + +## 1. 实现证据 + +- [服务输入](../../pulse/service/service.go)：精确业务键、UTC创建时间下界/上界、有限页大小，非法值不进入存储。 +- [查询存储](../../internal/storage/mysql/task_query.go)：固定四种索引分支、参数化SQL、创建时间/ID游标分页，v2摘要绑定全部筛选，旧游标只兼容原队列/状态查询。 +- [HTTP入口](../../internal/transport/httpserver/tasks.go)与[独立SDK](../../sdk/go/task_query.go)：URL编码、UTC规范化、显式空参数与未知参数拒绝，SDK无数据库依赖。 +- [迁移000016](../../internal/storage/mysql/migrations/000016_task_query_filters.up.sql)：business_key使用精确NO PAD二进制排序，增加两种队列业务键索引，中文字段含义同步。无新增模型字段、无外键/级联，仍为35个模型、289个持久化字段。 +- [接口手册](../api/task-queries.md)、[完整示例说明](../examples/task-query.md)及[源码](../../sdk/go/examples/task-query/main.go)。 + +## 2. 真实数据库验证 + +[TaskQuery集成](../../internal/storage/mysql/task_query_integration_test.go)使用真实受理任务，仅将测试创建时间调为精确边界来建立确定性数据： + +1. 相同业务键不同任务、相同创建时间不同ID，按时间/ID倒序逐页遍历，无重复或遗漏。 +2. 精确区分尾部空格、大小写、预组合/组合Unicode及无关联键，创建下界包含、上界排除。 +3. 同命名空间其他队列、其他命名空间同键任务不进入本查询；191字符键仍生成有界游标并能完整翻页，纯空格键可原样受理与检索。 +4. 更改/删除业务键、更改状态/时间/队列/命名空间时，旧游标拒绝；等价UTC时刻及修改页大小仍可继续。 +5. 已取得游标后撤销队列范围，继续查询仍被当前权限拒绝。 +6. 原版无版本游标仅用于旧查询；重复字段、null、大小写别名、未知字段、未知版本和纳秒时间游标拒绝。 +7. 实际HTTP验证Unicode/保留字符编码、显式空键、重复参数、错误日期、非法UTF-8/编码、非法时间范围和未知名称。 +8. EXPLAIN FORMAT=JSON对四种筛选验证对应索引、range/ref访问、无filesort。验证的是查询计划，不是生产吞吐或容量。 + +旧v7升级测试同时保存包含重音字符和尾部空格的已有业务键，升级后原任务快照不变，原323个未替换索引组件保留，列排序规则正确，外键70→0。应用额度v9升级夹具同步还原新增索引与排序规则再实际重放迁移。 + +单项受理、计划生成、人工重试只在Go字符串为空时写NULL，避免不同连接排序规则下NULLIF比较把纯空格折叠为空。旧数据不推测补回丢失的值。 + +## 3. SDK与示例证据 + +[SDK单元测试](../../sdk/go/task_query_test.go)验证URL编码、时间规范化、精确revision、非法参数本地拒绝；[服务测试](../../pulse/service/task_query_test.go)验证时间/业务键边界与存储隔离。 + +[真实HTTP SDK集成](../../sdk/go/task_query_integration_test.go)提交同键/尾部空格不同键任务，分页查询、变更条件拒绝、精确时间边界和取消后原创建时间排序均通过。完整示例输出： + +```text +matched=2 pages=2 before_excluded=true cancelled=3 +``` + +示例使用普通提交/读取/取消凭据，不增加管理权限，不启动Worker，不调用外部业务系统，保留任务及审计。 + +## 4. 环境与复现 + +macOS、Go1.27.1、Docker单实例MySQL8.4.10，隔离数据库、随机测试凭据与回环端口，结束自动清理。 + +```sh +make fmt test vet build +make integration +``` + +定向数据库回归已通过（16.148秒，含升级与包开销）；新增的真实受理和最长业务键场景随后进入全量回归。普通未配置数据库的go test跳过Integration，不能替代真实数据库验证。 + +| 验证组 | 本轮实际结果 | +| --- | --- | +| fmt / race / vet / build | PASS | +| 全量MySQL集成 / 任务查询用例 | PASS，101.225 / 0.38秒 | +| SDK Worker完整组 | PASS，90.485秒 | +| SDK计划 / 队列 / 批量 | PASS，5.704 / 4.461 / 1.368秒 | +| 新查询SDK真实HTTP | PASS，1.324秒 | +| SDK订阅管理 / 事件消费 | PASS，33.394 / 51.884秒 | +| SDK人工核查与重试 | PASS，3.405秒 | +| Inbox真实数据库与消费者故障 | PASS，34.643秒 | +| SDK应用配额 | PASS，1.369秒 | +| 九项节点/Worker进程故障 | PASS，60.625秒；连同Inbox消费者共十项进程故障 | +| 全部运行示例 | 17个核心SDK示例 + 1个独立Inbox示例通过 | + +完整脚本的终态退出码为0，未以中途日志推断完成。以上耗时是本次正确性回归记录，不是吞吐或容量指标。文档链接、JSON代码块及核心SDK无数据库/服务端依赖检查均通过。 + +## 5. 已知边界与后续 + +- 查询需要队列，时间界限可选。结果不含总数、大载荷或跨页快照，不提供前缀/模糊搜索、未设置业务键专用筛选或跨队列聚合。 +- 时间相同依赖ID稳定排序，状态可能跨页变化；可靠结果消费使用订阅，历史审计使用已有历史接口。 +- v2游标绑定筛选但不是权限证明；旧节点拒绝新参数，不允许删条件降级。新旧二进制混合查询和生产滚动升级尚未验收。 +- 新索引增加写入和存储成本；迁移需DDL维护窗口演练，生产规模容量、数据库切换及RPO/RTO仍在P6验证。 +- 任务详情尚未返回来源计划/触发记录，已对照原功能清单纳入P3补齐；业务事务Outbox也是剩余接入项；Weave执行与恢复仍按P4–P5推进，持续目标保持原范围。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index a20bd1c..01211f4 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -119,7 +119,7 @@ fmt.Println(cancelled.Task.State) | Liveness / Readiness | Health | [健康](../api/health.md) | | SubmitTask | SubmitTaskRequest / SubmitTaskResult | [任务](../api/tasks.md) | | GetTask | taskID / Task | [任务](../api/tasks.md) | -| ListTasks | ListTasksRequest / TaskPage | [任务](../api/tasks.md) | +| ListTasks | ListTasksRequest / TaskPage，业务键及创建时间范围 | [查询](../api/task-queries.md) | | CancelTask | taskID、key、CancelTaskRequest / CancelTaskResult | [任务](../api/tasks.md) | | CreateQueue | key、CreateQueueRequest / CreateQueueResult | [队列](../api/queues.md) | | GetQueue | queueID / Queue | [队列](../api/queues.md) | @@ -128,7 +128,7 @@ fmt.Println(cancelled.Task.State) | ListAttemptEvidence | taskID、attemptID、HistoryPageRequest / AttemptEvidencePage | [执行历史](../api/execution-history.md) | | GetAttemptEvidence | taskID、attemptID、evidenceID / AttemptEvidence | [执行历史](../api/execution-history.md) | -所有方法第一个参数为 context.Context，返回值之外都返回 error。计划方法的具体参数见对应手册。命令方法要求调用方提供稳定 key，不自动生成或重试。ListTasksRequest 接受 QueueID、State、Limit、Cursor。 +所有方法第一个参数为context.Context，返回值之外都返回error。计划方法的具体参数见对应手册。命令方法要求调用方提供稳定key，不自动生成或重试。ListTasksRequest接受QueueID、State、BusinessKey、CreatedFrom、CreatedBefore、Limit、Cursor；[筛选与游标语义](../api/task-queries.md)和[运行示例](../examples/task-query.md)已提供。 Options.HTTPClient 可注入专用 HTTP 客户端。SDK 复制配置，不修改调用方实例；默认超时 10 秒，可用 context 缩短。禁止自动跟随重定向，避免凭据被转交其他地址。生产配置 HTTPS；SDK 不承担部署 TLS 终止。 diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index e74a51f..0e87df0 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -64,6 +64,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { retryFinish(t, g, finished, "SUCCEEDED", false) assertApplicationCounts(t, f, 4, 2, 1) // 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。 + f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_queue_business_created, DROP INDEX ix_task_queue_business_state_created, MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '业务侧关联键，用于检索追踪，不自动提供业务幂等'`) f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_retry_not_self, DROP INDEX uq_task_retry_parent, ADD INDEX fk_pulse_tasks_3(namespace_id,retry_of)`) f.exec(`DROP TABLE orbit_application_resources`) f.exec(`DROP TABLE orbit_result_events,orbit_event_streams`) diff --git a/internal/storage/mysql/migrations/000016_task_query_filters.up.sql b/internal/storage/mysql/migrations/000016_task_query_filters.up.sql new file mode 100644 index 0000000..413dc2d --- /dev/null +++ b/internal/storage/mysql/migrations/000016_task_query_filters.up.sql @@ -0,0 +1,6 @@ +-- 业务关联键按精确值查询；NO PAD二进制排序区分尾部空格，不规范化Unicode。 +-- 两种新索引覆盖已授权队列内的关联键及可选状态过滤，不增加外键。 +ALTER TABLE pulse_tasks + MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NULL COMMENT '可空业务关联键，精确匹配并区分大小写及尾部空格，不自动提供业务幂等', + ADD KEY ix_task_queue_business_created(namespace_id,queue_id,business_key,created_at,id) COMMENT '队列内按业务键及创建时间分页', + ADD KEY ix_task_queue_business_state_created(namespace_id,queue_id,business_key,state,created_at,id) COMMENT '队列内按业务键和状态及创建时间分页'; diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index 51f0e3a..aa95c81 100644 --- a/internal/storage/mysql/no_foreign_keys_integration_test.go +++ b/internal/storage/mysql/no_foreign_keys_integration_test.go @@ -89,7 +89,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { indexes := func() []string { t.Helper() // v15 将原 retry_of 非唯一索引替换为相同列的唯一索引，下方单独验证其定义。 - rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) + rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) if err != nil { t.Fatal(err) } @@ -113,6 +113,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { exec(`INSERT INTO orbit_namespaces(id,application_id,name,environment) VALUES (?,?,?,'test')`, ns, app, string(ns)) exec(`INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,policy) VALUES (?,?,?,4,10,'{}')`, ns, queue, string(queue)) exec(`INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,available_at,deadline_at,attempt_timeout_ms,max_attempts) VALUES (?,?,?,'object.prepare','v1','v1','{"n":9007199254740993}','{}',UTC_TIMESTAMP(6),DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR),5000,3)`, ns, task, queue) + exec(`UPDATE pulse_tasks SET business_key=? WHERE namespace_id=? AND id=?`, "关联键é尾部 ", ns, task) taskSnapshot := func() []string { t.Helper() rows, err := db.QueryContext(ctx, `SELECT * FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, task) @@ -172,6 +173,10 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { if err := db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(CONCAT(COLUMN_NAME,':',NON_UNIQUE) ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND INDEX_NAME='uq_task_retry_parent'`).Scan(&retryIndex); err != nil || retryIndex != "namespace_id:0,retry_of:0" { t.Fatalf("retry uniqueness replacement=%q err=%v", retryIndex, err) } + var collation string + if err := db.QueryRowContext(ctx, `SELECT COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND COLUMN_NAME='business_key'`).Scan(&collation); err != nil || collation != "utf8mb4_0900_bin" { + t.Fatalf("business key collation=%q err=%v", collation, err) + } if afterIndexes := indexes(); !reflect.DeepEqual(beforeIndexes, afterIndexes) { before, after := map[string]bool{}, map[string]bool{} for _, index := range beforeIndexes { diff --git a/internal/storage/mysql/schedule_tick.go b/internal/storage/mysql/schedule_tick.go index fc90417..98a06d9 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,concurrency_key,concurrency_key_digest,priority,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), in.Priority, 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,priority,owner_kind,owner_id,available_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,'schedule',?,?,?,?,?,?,?)`, v.NamespaceID, value, v.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, businessKeyValue(in.BusinessKey), concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, 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_query.go b/internal/storage/mysql/task_query.go new file mode 100644 index 0000000..91c79ae --- /dev/null +++ b/internal/storage/mysql/task_query.go @@ -0,0 +1,180 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/base64" + "encoding/hex" + "encoding/json" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +// taskCursor 是已授权查询的排序位置，不能代替凭据；v2摘要绑定全部筛选。 +type taskCursor struct { + Version int `json:"v,omitempty"` + Namespace core.ID `json:"n"` + Queue core.ID `json:"q"` + State pulse.TaskState `json:"s"` + Created time.Time `json:"t"` + ID core.ID `json:"i"` + Filter string `json:"f,omitempty"` +} + +func taskFilterDigest(in service.ListInput) (string, error) { + digest, err := requestDigest(struct { + Domain string `json:"domain"` + Queue core.ID `json:"queue"` + State pulse.TaskState `json:"state"` + BusinessKey string `json:"business_key"` + From *time.Time `json:"from"` + Before *time.Time `json:"before"` + }{"orbit.task-query.v2", in.QueueID, in.State, in.BusinessKey, in.CreatedFrom, in.CreatedBefore}) + return hex.EncodeToString(digest[:]), err +} + +func decodeTaskCursor(value string, ns core.ID, in service.ListInput, filter string) (taskCursor, error) { + var c taskCursor + invalid := fault.Invalid("cursor does not match this task query") + raw, err := base64.RawURLEncoding.Strict().DecodeString(value) + if err != nil || len(raw) > 768 { + return c, invalid + } + if _, err = jsonvalue.Canonical(raw); err != nil { + return c, invalid + } + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil || fields == nil { + return c, invalid + } + for k, v := range fields { + switch k { + case "v", "n", "q", "s", "t", "i", "f": + default: + return c, invalid + } + if string(v) == "null" { + return c, invalid + } + } + for _, k := range []string{"n", "q", "s", "t", "i"} { + if _, ok := fields[k]; !ok { + return c, invalid + } + } + if json.Unmarshal(raw, &c) != nil || c.Namespace != ns || c.Queue != in.QueueID || c.State != in.State || c.ID.Validate() != nil || c.Created.Year() < 2000 || c.Created.Year() > 9998 || c.Created.Nanosecond()%1000 != 0 { + return c, invalid + } + if c.Version == 0 { + // 旧版未带版本的游标仅兼容原队列/状态查询，不能附加新筛选。 + if _, ok := fields["v"]; ok { + return c, invalid + } + if _, ok := fields["f"]; ok { + return c, invalid + } + if in.BusinessKey != "" || in.CreatedFrom != nil || in.CreatedBefore != nil { + return c, invalid + } + } else if c.Version != 2 || c.Filter != filter { + return c, invalid + } + if in.CreatedFrom != nil && c.Created.Before(*in.CreatedFrom) || in.CreatedBefore != nil && !c.Created.Before(*in.CreatedBefore) { + return c, invalid + } + return c, nil +} + +const taskSummaryColumns = `id,queue_id,state,revision,handler,handler_version,COALESCE(business_key,''),COALESCE(concurrency_key,''),priority,available_at,deadline_at,attempt_count,created_at,updated_at,effect_uncertain` + +// taskListSQL 选择固定索引名，等值筛选全部位于创建时间前；游标不依赖状态更新时间。 +func taskListSQL(ns core.ID, in service.ListInput, filter string) (string, []any, error) { + index := "ix_task_created" + if in.State != "" { + index = "ix_task_state_created" + } + if in.BusinessKey != "" { + index = "ix_task_queue_business_created" + if in.State != "" { + index = "ix_task_queue_business_state_created" + } + } + q := `SELECT ` + taskSummaryColumns + ` FROM pulse_tasks FORCE INDEX (` + index + `) WHERE namespace_id=? AND queue_id=?` + args := []any{ns, in.QueueID} + if in.BusinessKey != "" { + q += " AND business_key=?" + args = append(args, in.BusinessKey) + } + if in.State != "" { + q += " AND state=?" + args = append(args, in.State) + } + if in.CreatedFrom != nil { + q += " AND created_at>=?" + args = append(args, *in.CreatedFrom) + } + if in.CreatedBefore != nil { + q += " AND created_at in.Limit { + out.Items = out.Items[:in.Limit] + last := out.Items[len(out.Items)-1] + raw, err := json.Marshal(taskCursor{Version: 2, Namespace: token.NamespaceID, Queue: in.QueueID, State: in.State, Created: last.CreatedAt, ID: last.ID, Filter: filter}) + if err != nil { + return err + } + out.NextCursor = base64.RawURLEncoding.EncodeToString(raw) + } + return nil + }) + return out, err +} diff --git a/internal/storage/mysql/task_query_integration_test.go b/internal/storage/mysql/task_query_integration_test.go new file mode 100644 index 0000000..147034e --- /dev/null +++ b/internal/storage/mysql/task_query_integration_test.go @@ -0,0 +1,270 @@ +package mysql + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "sort" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver" + core "git.yuntuo.hk/k2safe/orbit/model" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func TestIntegrationTaskQuery(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, stop := context.WithTimeout(context.Background(), 90*time.Second) + defer stop() + db, err := Open(ctx, dsn, 12, 4) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + f := newExecutionFixture(t, ctx, db) + g := newExecutionFixture(t, ctx, db) + from := time.Date(2026, 9, 17, 0, 0, 0, 123456000, time.UTC) + before := from.Add(time.Hour) + makeTask := func(owner *executionFixture, key string, at time.Time, state pulse.TaskState) core.ID { + t.Helper() + out, err := owner.tasks.Submit(ctx, owner.token, string(executionID(t)), service.SubmitInput{QueueID: owner.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), BusinessKey: key}) + if err != nil || out.Task.BusinessKey != key { + t.Fatal("business key changed on admission", err) + } + v := out.Task + owner.exec(`UPDATE pulse_tasks SET created_at=? WHERE namespace_id=? AND id=?`, at, owner.resources.NamespaceID, v.ID) + if state == pulse.TaskCancelled { + if _, err := owner.tasks.CancelTask(ctx, owner.token, v.ID, string(executionID(t)), service.CancelInput{ExpectedRevision: v.Revision, Reason: "query regression"}); err != nil { + t.Fatal(err) + } + } + return v.ID + } + key := "任务 /+%? é " + ids := []core.ID{makeTask(f, key, from, pulse.TaskWaiting), makeTask(f, key, from, pulse.TaskWaiting), makeTask(f, key, from.Add(time.Minute), pulse.TaskWaiting)} + sort.Slice(ids[:2], func(i, j int) bool { return ids[i] > ids[j] }) + expected := []core.ID{ids[2], ids[0], ids[1]} + makeTask(f, key, from.Add(-time.Microsecond), pulse.TaskWaiting) + makeTask(f, key, before, pulse.TaskWaiting) + makeTask(f, key, from, pulse.TaskCancelled) + for _, other := range []string{strings.TrimSuffix(key, " "), strings.TrimSuffix(key, " ") + " ", strings.Replace(key, "é", "É", 1), strings.Replace(key, "é", "e\u0301", 1), ""} { + makeTask(f, other, from, pulse.TaskWaiting) + } + makeTask(g, key, from, pulse.TaskWaiting) + spaceID := makeTask(f, " ", from, pulse.TaskWaiting) + spaces, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: " "}) + if err != nil || len(spaces.Items) != 1 || spaces.Items[0].ID != spaceID { + t.Fatal("space key not preserved", spaces, err) + } + + otherQueue, err := f.tasks.CreateQueue(ctx, f.token, string(executionID(t)), service.QueueInput{Name: "other-query", MaxConcurrency: 1, RatePerSecond: 10}) + if err != nil { + t.Fatal(err) + } + otherTask, err := f.tasks.Submit(ctx, f.token, string(executionID(t)), service.SubmitInput{QueueID: otherQueue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), BusinessKey: key}) + if err != nil { + t.Fatal(err) + } + f.exec(`UPDATE pulse_tasks SET created_at=? WHERE namespace_id=? AND id=?`, from, f.resources.NamespaceID, otherTask.Task.ID) + longKey := strings.Repeat("界", 191) + makeTask(f, longKey, from, pulse.TaskWaiting) + makeTask(f, longKey, from, pulse.TaskWaiting) + longPage, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: longKey, Limit: 1}) + if err != nil || longPage.NextCursor == "" || len(longPage.NextCursor) > 1024 { + t.Fatal("maximum key overflowed cursor", err) + } + lastLong, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: longKey, Limit: 1, Cursor: longPage.NextCursor}) + if err != nil || len(lastLong.Items) != 1 || lastLong.Items[0].ID == longPage.Items[0].ID || lastLong.NextCursor != "" { + t.Fatal("maximum key pagination", lastLong, err) + } + base := service.ListInput{QueueID: f.resources.QueueID, BusinessKey: key, State: pulse.TaskWaiting, CreatedFrom: &from, CreatedBefore: &before, Limit: 1} + var firstCursor string + t.Run("precise_key_time_and_same_timestamp_pages", func(t *testing.T) { + in := base + var got []core.ID + for { + page, err := f.tasks.ListTasks(ctx, f.token, in) + if err != nil { + t.Fatal(err) + } + for _, v := range page.Items { + got = append(got, v.ID) + if v.BusinessKey != key || v.State != pulse.TaskWaiting { + t.Fatal("wrong query result") + } + } + if firstCursor == "" { + firstCursor = page.NextCursor + } + if page.NextCursor == "" { + break + } + in.Cursor = page.NextCursor + } + if fmt.Sprint(got) != fmt.Sprint(expected) { + t.Fatal("page boundaries", got, expected) + } + in = base + in.State = "" + in.Limit = 100 + page, err := f.tasks.ListTasks(ctx, f.token, in) + if err != nil || len(page.Items) != 4 { + t.Fatal(page, err) + } + in = base + in.CreatedFrom = nil + page, err = f.tasks.ListTasks(ctx, f.token, in) + if err != nil || len(page.Items) != 1 || page.NextCursor == "" { + t.Fatal(page, err) + } + }) + t.Run("cursor_binds_filters_and_current_authority", func(t *testing.T) { + for name, change := range map[string]func(*service.ListInput){"key": func(v *service.ListInput) { v.BusinessKey = strings.TrimSuffix(key, " ") }, "remove_key": func(v *service.ListInput) { v.BusinessKey = "" }, "state": func(v *service.ListInput) { v.State = "" }, "lower": func(v *service.ListInput) { z := from.Add(-time.Second); v.CreatedFrom = &z }, "upper": func(v *service.ListInput) { z := before.Add(time.Second); v.CreatedBefore = &z }, "queue": func(v *service.ListInput) { v.QueueID = g.resources.QueueID }} { + t.Run(name, func(t *testing.T) { + in := base + in.Cursor = firstCursor + change(&in) + _, err := f.tasks.ListTasks(ctx, f.token, in) + expectWorkerCode(t, err, "INVALID_ARGUMENT") + }) + } + in := base + in.Cursor = firstCursor + _, err := g.tasks.ListTasks(ctx, g.token, in) + expectWorkerCode(t, err, "INVALID_ARGUMENT") + zone := time.FixedZone("equivalent", 8*3600) + lo, hi := from.In(zone), before.In(zone) + in.CreatedFrom = &lo + in.CreatedBefore = &hi + in.Limit = 2 + page, err := f.tasks.ListTasks(ctx, f.token, in) + if err != nil || len(page.Items) != 2 { + t.Fatal("equivalent UTC or page size not supported", page, err) + } + f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(g.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID) + _, err = f.tasks.ListTasks(ctx, f.token, in) + expectWorkerCode(t, err, "PERMISSION_DENIED") + f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY('*')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + }) + t.Run("legacy_cursor_and_strict_cursor_validation", func(t *testing.T) { + legacy := taskCursor{Namespace: f.resources.NamespaceID, Queue: f.resources.QueueID, State: pulse.TaskWaiting, Created: from.Add(time.Minute), ID: ids[2]} + raw, _ := json.Marshal(legacy) + encoded := base64.RawURLEncoding.EncodeToString(raw) + _, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, State: pulse.TaskWaiting, Cursor: encoded}) + if err != nil { + t.Fatal("old cursor rejected", err) + } + in := base + in.Cursor = encoded + _, err = f.tasks.ListTasks(ctx, f.token, in) + expectWorkerCode(t, err, "INVALID_ARGUMENT") + valid, _ := base64.RawURLEncoding.DecodeString(firstCursor) + for _, raw := range []string{strings.TrimSuffix(string(valid), "}") + `,"v":2}`, strings.Replace(string(valid), `"v":2`, `"v":3`, 1), strings.Replace(string(valid), `"v":2`, `"v":null`, 1), strings.Replace(string(valid), `"v":2`, `"V":2`, 1), strings.TrimSuffix(string(valid), "}") + `,"secret":1}`, strings.Replace(string(valid), ".123456Z", ".123456001Z", 1)} { + in.Cursor = base64.RawURLEncoding.EncodeToString([]byte(raw)) + _, err = f.tasks.ListTasks(ctx, f.token, in) + expectWorkerCode(t, err, "INVALID_ARGUMENT") + } + }) + t.Run("http_validation_and_encoded_unicode", func(t *testing.T) { + server := httptest.NewServer(httpserver.New("", func(context.Context) error { return nil }, f.tasks).Handler) + defer server.Close() + call := func(query string, status int) service.TaskPage { + t.Helper() + r, _ := http.NewRequestWithContext(ctx, "GET", server.URL+"/v1/tasks?queue_id="+string(f.resources.QueueID)+"&"+query, nil) + r.Header.Set("Authorization", "Bearer "+f.resources.Token) + resp, err := server.Client().Do(r) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode != status { + t.Fatalf("query status=%d want=%d body=%s", resp.StatusCode, status, raw) + } + var out struct { + Data service.TaskPage `json:"data"` + } + if err = json.Unmarshal(raw, &out); err != nil { + t.Fatal(err) + } + return out.Data + } + q := url.Values{"business_key": {key}, "state": {"WAITING"}, "created_from": {from.Format(time.RFC3339Nano)}, "created_before": {before.Format(time.RFC3339Nano)}} + if page := call(q.Encode(), 200); len(page.Items) != 3 { + t.Fatal(page) + } + for _, q := range []string{"business_key=", "business_key=a&business_key=b", "business_key=%ff", "business_key=%00", "created_from=", "created_from=2026-02-30T00:00:00Z", "created_from=2026-09-17T00:00:00.000000001Z", "created_from=2026-09-17", "created_before=1999-01-01T00:00:00Z", "created_from=2026-09-17T00:00:00Z&created_before=2026-09-17T00:00:00Z", "CreatedFrom=1", "created_from=%zz", "created_before=a&created_before=b"} { + call(q, 400) + } + }) + t.Run("all_filter_shapes_use_ordered_composite_indexes", func(t *testing.T) { + cases := []struct { + state pulse.TaskState + key, index string + }{{"", "", "ix_task_created"}, {pulse.TaskWaiting, "", "ix_task_state_created"}, {"", key, "ix_task_queue_business_created"}, {pulse.TaskWaiting, key, "ix_task_queue_business_state_created"}} + for _, tc := range cases { + in := base + in.State = tc.state + in.BusinessKey = tc.key + digest, err := taskFilterDigest(in) + if err != nil { + t.Fatal(err) + } + q, args, err := taskListSQL(f.resources.NamespaceID, in, digest) + if err != nil { + t.Fatal(err) + } + var raw string + if err = db.QueryRowContext(ctx, "EXPLAIN FORMAT=JSON "+q, args...).Scan(&raw); err != nil { + t.Fatal(err) + } + var plan map[string]any + if err = json.Unmarshal([]byte(raw), &plan); err != nil { + t.Fatal(err) + } + found := false + var walk func(any) + walk = func(v any) { + switch x := v.(type) { + case map[string]any: + if x["using_filesort"] == true { + t.Fatal("query uses filesort", raw) + } + if x["table_name"] == "pulse_tasks" { + found = true + if x["key"] != tc.index || (x["access_type"] != "range" && x["access_type"] != "ref") { + t.Fatal("unbounded query access", raw) + } + } + for _, v := range x { + walk(v) + } + case []any: + for _, v := range x { + walk(v) + } + } + } + walk(plan) + if !found { + t.Fatal("task query plan missing", raw) + } + t.Logf("state=%q business_key=%t index=%s no filesort", tc.state, tc.key != "", tc.index) + } + }) +} diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index 0b7f42a..2fb662a 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -5,7 +5,6 @@ import ( "crypto/sha256" "crypto/subtle" "database/sql" - "encoding/base64" "encoding/json" "errors" "time" @@ -208,6 +207,14 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in return out, err } +// businessKeyValue 仅把空字符串映射为NULL，不能由连接排序规则把空格折叠为空值。 +func businessKeyValue(key string) *string { + if key == "" { + return nil + } + return &key +} + // insertAcceptedTask 要求调用方已持有应用额度和队列锁，并将身份、事件及审计同事务提交。 func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID) (service.TaskView, error) { now, err := dbNow(ctx, tx) @@ -233,7 +240,7 @@ func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in return service.TaskView{}, 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,concurrency_key,concurrency_key_digest,priority,available_at,start_deadline_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at,retry_of) - VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,?,?,?,?,?,?,?,?,?)`, ns, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now, retryOf) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, ns, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, businessKeyValue(in.BusinessKey), concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now, retryOf) if err != nil { return service.TaskView{}, err } @@ -283,65 +290,6 @@ func (s *TaskStore) GetTask(ctx context.Context, token auth.Token, id core.ID) ( return out, err } -type taskCursor struct { - Namespace core.ID `json:"n"` - Queue core.ID `json:"q"` - State pulse.TaskState `json:"s"` - Created time.Time `json:"t"` - ID core.ID `json:"i"` -} - -func (s *TaskStore) ListTasks(ctx context.Context, token auth.Token, in service.ListInput) (service.TaskPage, error) { - out := service.TaskPage{Items: []service.TaskSummary{}} - err := s.transaction(ctx, token, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error { - if !p.AllowsQueue(in.QueueID) { - return fault.Forbidden - } - q := `SELECT id,queue_id,state,revision,handler,handler_version,COALESCE(business_key,''),COALESCE(concurrency_key,''),priority,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=?" - args = append(args, in.State) - } - if in.Cursor != "" { - raw, err := base64.RawURLEncoding.DecodeString(in.Cursor) - var c taskCursor - if err != nil || json.Unmarshal(raw, &c) != nil || c.Namespace != token.NamespaceID || c.Queue != in.QueueID || c.State != in.State || c.Created.IsZero() || c.ID.Validate() != nil { - return fault.Invalid("cursor does not match this query") - } - q += " AND (created_at < ? OR (created_at=? AND id in.Limit { - out.Items = out.Items[:in.Limit] - last := out.Items[len(out.Items)-1] - raw, err := json.Marshal(taskCursor{token.NamespaceID, in.QueueID, in.State, last.CreatedAt, last.ID}) - if err != nil { - return err - } - out.NextCursor = base64.RawURLEncoding.EncodeToString(raw) - } - return nil - }) - return out, err -} func (s *TaskStore) CreateQueue(ctx context.Context, token auth.Token, key string, in service.QueueInput) (service.QueueResult, error) { var out service.QueueResult err := s.transaction(ctx, token, "queues:manage", func(tx *sql.Tx, p auth.Permissions) error { diff --git a/internal/transport/httpserver/tasks.go b/internal/transport/httpserver/tasks.go index f07de65..27d5142 100644 --- a/internal/transport/httpserver/tasks.go +++ b/internal/transport/httpserver/tasks.go @@ -60,7 +60,7 @@ func registerTasks(mux *http.ServeMux, s *service.Service) { return nil, 0, fault.Invalid("invalid query encoding") } for k, v := range q { - if len(v) != 1 || (k != "queue_id" && k != "state" && k != "limit" && k != "cursor") { + if len(v) != 1 || (k != "queue_id" && k != "state" && k != "limit" && k != "cursor" && k != "business_key" && k != "created_from" && k != "created_before") { return nil, 0, fault.Invalid("invalid query parameter") } } @@ -72,7 +72,20 @@ func registerTasks(mux *http.ServeMux, s *service.Service) { return nil, 0, fault.Invalid("invalid limit") } } - out, err := s.ListTasks(r.Context(), t, service.ListInput{QueueID: core.ID(q.Get("queue_id")), State: pulse.TaskState(q.Get("state")), Limit: limit, Cursor: q.Get("cursor")}) + in := service.ListInput{QueueID: core.ID(q.Get("queue_id")), State: pulse.TaskState(q.Get("state")), Limit: limit, Cursor: q.Get("cursor"), BusinessKey: q.Get("business_key")} + if q.Has("business_key") && in.BusinessKey == "" { + return nil, 0, fault.Invalid("business_key filter must not be empty") + } + for name, field := range map[string]**time.Time{"created_from": &in.CreatedFrom, "created_before": &in.CreatedBefore} { + if q.Has(name) { + v, err := time.Parse(time.RFC3339Nano, q.Get(name)) + if err != nil { + return nil, 0, fault.Invalid("invalid creation time bound") + } + *field = &v + } + } + out, err := s.ListTasks(r.Context(), t, in) return out, 200, err })) mux.HandleFunc("POST /v1/tasks/{command}", wrap(func(r *http.Request, t auth.Token) (any, int, error) { diff --git a/pulse/model/task.go b/pulse/model/task.go index 8d05a1c..a415dea 100644 --- a/pulse/model/task.go +++ b/pulse/model/task.go @@ -33,7 +33,7 @@ type Task struct { PolicySnapshot core.JSON `db:"policy_snapshot"` // RecoverySafety 恢复安全等级；未知值按需要核查处理，不能默认自动重跑。 RecoverySafety RecoverySafety `db:"recovery_safety"` - // BusinessKey 业务侧关联键，用于检索追踪，不自动提供业务幂等；可为空，表示尚未设置该关联或值。 + // BusinessKey 可空业务关联键，按精确Unicode值检索，区分大小写及尾部空格；不自动提供业务幂等。 BusinessKey *string `db:"business_key"` // ConcurrencyKey 显式互斥资源键，命名空间内跨队列共享；空表示不启用。 ConcurrencyKey *string `db:"concurrency_key"` diff --git a/pulse/service/service.go b/pulse/service/service.go index a374bfc..67d756f 100644 --- a/pulse/service/service.go +++ b/pulse/service/service.go @@ -132,6 +132,21 @@ func (s *Service) ListTasks(ctx context.Context, token auth.Token, in ListInput) default: return TaskPage{}, fault.Invalid("invalid task state") } + if !utf8.ValidString(in.BusinessKey) || utf8.RuneCountInString(in.BusinessKey) > 191 || strings.ContainsAny(in.BusinessKey, "\x00\r\n") { + return TaskPage{}, fault.Invalid("business_key must be at most 191 characters without control newlines") + } + for _, field := range []**time.Time{&in.CreatedFrom, &in.CreatedBefore} { + if *field != nil { + v := (**field).UTC() + if v.Year() < 2000 || v.Year() > 9998 || v.Nanosecond()%1000 != 0 { + return TaskPage{}, fault.Invalid("creation bounds must use supported years and microsecond precision") + } + *field = &v + } + } + if in.CreatedFrom != nil && in.CreatedBefore != nil && !in.CreatedFrom.Before(*in.CreatedBefore) { + return TaskPage{}, fault.Invalid("created_from must be before created_before") + } return s.store.ListTasks(ctx, token, in) } func (s *Service) CancelTask(ctx context.Context, token auth.Token, id core.ID, key string, in CancelInput) (CancelResult, error) { diff --git a/pulse/service/task_query_test.go b/pulse/service/task_query_test.go new file mode 100644 index 0000000..efcf1c3 --- /dev/null +++ b/pulse/service/task_query_test.go @@ -0,0 +1,56 @@ +package service + +import ( + "context" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +type queryValidationStore struct { + Store + calls int + input ListInput +} + +func (s *queryValidationStore) ListTasks(_ context.Context, _ auth.Token, in ListInput) (TaskPage, error) { + s.calls++ + s.input = in + return TaskPage{}, nil +} +func TestTaskQueryBoundsAndNormalization(t *testing.T) { + store := &queryValidationStore{} + api := New(store) + from := time.Date(2026, 9, 17, 12, 0, 0, 123456000, time.FixedZone("example", 8*3600)) + before := from.Add(time.Hour) + base := ListInput{QueueID: core.ID(strings.Repeat("1", 32)), BusinessKey: "订单 /+% é ", CreatedFrom: &from, CreatedBefore: &before} + for name, change := range map[string]func(*ListInput){ + "bad_utf8": func(v *ListInput) { v.BusinessKey = "\xff" }, "long_key": func(v *ListInput) { v.BusinessKey = strings.Repeat("字", 192) }, "newline": func(v *ListInput) { v.BusinessKey = "x\n" }, "zero_time": func(v *ListInput) { z := time.Time{}; v.CreatedFrom = &z }, "nanosecond": func(v *ListInput) { z := from.Add(time.Nanosecond); v.CreatedFrom = &z }, "equal_bounds": func(v *ListInput) { v.CreatedBefore = &from }, "reverse_bounds": func(v *ListInput) { v.CreatedFrom = &before; v.CreatedBefore = &from }, + } { + t.Run(name, func(t *testing.T) { + in := base + change(&in) + if _, err := api.ListTasks(context.Background(), auth.Token{}, in); fault.Code(err) != "INVALID_ARGUMENT" { + t.Fatal(err) + } + }) + } + if store.calls != 0 { + t.Fatal("invalid query reached persistence") + } + if _, err := api.ListTasks(context.Background(), auth.Token{}, base); err != nil { + t.Fatal(err) + } + if store.input.BusinessKey != base.BusinessKey || store.input.CreatedFrom.Location() != time.UTC || !store.input.CreatedFrom.Equal(from) || store.input.Limit != 20 { + t.Fatal("query normalization lost meaning") + } + for _, in := range []ListInput{{QueueID: base.QueueID, CreatedFrom: &from}, {QueueID: base.QueueID, CreatedBefore: &before}} { + if _, err := api.ListTasks(context.Background(), auth.Token{}, in); err != nil { + t.Fatal(err) + } + } +} diff --git a/pulse/service/types.go b/pulse/service/types.go index 05f969f..419e4e0 100644 --- a/pulse/service/types.go +++ b/pulse/service/types.go @@ -113,6 +113,12 @@ type CancelResult struct { Deduplicated bool `json:"deduplicated"` } type ListInput struct { + // BusinessKey 可选精确业务关联键；空表示不筛选，区分大小写、Unicode形式及尾部空格。 + BusinessKey string + // CreatedFrom 可选创建时间下界（包含），UTC微秒精度。 + CreatedFrom *time.Time + // CreatedBefore 可选创建时间上界（不包含），与下界同时给出时必须更晚。 + CreatedBefore *time.Time // QueueID 队列标识，必须属于当前命名空间。 QueueID core.ID // State 当前任务状态；终态不会被重复请求重开。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index ceb9096..5cd0d66 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -48,11 +48,13 @@ done if [ -z "$TASK_HTTP_PORT" ]; then printf '%s\n' 'Bootstrap HTTP server did not start.' >&2; exit 1; fi (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" "$TASK_GO" run ./examples/health) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/sdk.credentials.json" "$TASK_GO" run ./examples/tasks) +(cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/sdk.credentials.json" "$TASK_GO" run ./examples/task-query) 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 --event-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_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 '^TestIntegrationTaskBatchSDK$' -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 '^TestIntegrationTaskQuerySDK$' -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 '^TestIntegrationSubscriptionManagementSDK$' -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 '^TestIntegrationEvent(Connection|Consumer)' -v .) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/worker-protocol) diff --git a/sdk/go/examples/task-query/main.go b/sdk/go/examples/task-query/main.go new file mode 100644 index 0000000..5b7150d --- /dev/null +++ b/sdk/go/examples/task-query/main.go @@ -0,0 +1,104 @@ +// task-query 演示业务键精确匹配、创建时间边界和任务摘要分页，不启动Worker。 +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, queue := os.Getenv("ORBIT_TOKEN"), os.Getenv("ORBIT_QUEUE_ID") + if path := os.Getenv("ORBIT_CREDENTIALS_FILE"); path != "" { + raw, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read private credentials: %w", err) + } + var credentials struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if err := json.Unmarshal(raw, &credentials); err != nil { + return fmt.Errorf("invalid credentials file") + } + if token == "" { + token = credentials.Token + } + if queue == "" { + queue = credentials.QueueID + } + } + client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + if err != nil { + return err + } + if queue == "" { + return fmt.Errorf("ORBIT_QUEUE_ID is required") + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + op, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + businessKey := "object /+% é " + op + " " + var tasks []orbit.Task + for i, key := range []string{businessKey, businessKey, businessKey[:len(businessKey)-1]} { + out, err := client.SubmitTask(ctx, fmt.Sprintf("query-%s-%d", op, i), orbit.SubmitTaskRequest{QueueID: queue, Handler: "query." + op, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"object_ref":"examples/input"}`), BusinessKey: key, TaskTimeoutMS: 60000}) + if err != nil { + return err + } + tasks = append(tasks, out.Task) + } + from, before := tasks[0].CreatedAt, tasks[2].CreatedAt.Add(time.Microsecond) + query := orbit.ListTasksRequest{QueueID: queue, State: orbit.TaskWaiting, BusinessKey: businessKey, CreatedFrom: &from, CreatedBefore: &before, Limit: 1} + matched, pages := 0, 0 + for { + page, err := client.ListTasks(ctx, query) + if err != nil { + return err + } + pages++ + for _, task := range page.Items { + if task.BusinessKey != businessKey { + return errors.New("business key was not exact") + } + matched++ + } + if page.NextCursor == "" { + break + } + query.Cursor = page.NextCursor + } + if matched != 2 || pages != 2 { + return errors.New("unexpected filtered pages") + } + query.Cursor = "" + end := tasks[1].CreatedAt + query.CreatedBefore = &end + bounded, err := client.ListTasks(ctx, query) + if err != nil { + return err + } + if len(bounded.Items) != 1 || bounded.Items[0].ID != tasks[0].ID { + return errors.New("exclusive time boundary failed") + } + for i, task := range tasks { + if _, err := client.CancelTask(ctx, task.ID, fmt.Sprintf("query-cancel-%s-%d", op, i), orbit.CancelTaskRequest{ExpectedRevision: task.Revision, Reason: "complete query example"}); err != nil { + return err + } + } + fmt.Printf("matched=%d pages=%d before_excluded=true cancelled=%d\n", matched, pages, len(tasks)) + return nil +} diff --git a/sdk/go/task_query.go b/sdk/go/task_query.go new file mode 100644 index 0000000..e658f9a --- /dev/null +++ b/sdk/go/task_query.go @@ -0,0 +1,56 @@ +package orbit + +import ( + "context" + "errors" + "net/url" + "strconv" + "strings" + "time" + "unicode/utf8" +) + +// ListTasks 查询同一队列内的任务摘要，筛选固定后沿NextCursor分页；不承诺跨页快照。 +func (c *Client) ListTasks(ctx context.Context, in ListTasksRequest) (TaskPage, error) { + var out TaskPage + if !validID(in.QueueID) || in.Limit < 0 || in.Limit > 100 || len(in.Cursor) > 1024 { + return out, errors.New("invalid task query queue or page") + } + switch in.State { + case "", TaskWaiting, TaskLeased, TaskRunning, TaskNeedsReview, TaskSucceeded, TaskFailed, TaskCancelled: + default: + return out, errors.New("invalid task state") + } + if !utf8.ValidString(in.BusinessKey) || utf8.RuneCountInString(in.BusinessKey) > 191 || strings.ContainsAny(in.BusinessKey, "\x00\r\n") { + return out, errors.New("invalid business key filter") + } + q := url.Values{} + q.Set("queue_id", in.QueueID) + if in.State != "" { + q.Set("state", string(in.State)) + } + if in.BusinessKey != "" { + q.Set("business_key", in.BusinessKey) + } + if in.Limit != 0 { + q.Set("limit", strconv.Itoa(in.Limit)) + } + if in.Cursor != "" { + q.Set("cursor", in.Cursor) + } + for name, field := range map[string]*time.Time{"created_from": in.CreatedFrom, "created_before": in.CreatedBefore} { + if field == nil { + continue + } + v := field.UTC() + if v.Year() < 2000 || v.Year() > 9998 || v.Nanosecond()%1000 != 0 { + return out, errors.New("invalid task creation bound") + } + q.Set(name, v.Format(time.RFC3339Nano)) + } + if in.CreatedFrom != nil && in.CreatedBefore != nil && !in.CreatedFrom.Before(*in.CreatedBefore) { + return out, errors.New("created_from must be before created_before") + } + err := c.call(ctx, "GET", "/v1/tasks?"+q.Encode(), "", nil, &out) + return out, err +} diff --git a/sdk/go/task_query_integration_test.go b/sdk/go/task_query_integration_test.go new file mode 100644 index 0000000..afab925 --- /dev/null +++ b/sdk/go/task_query_integration_test.go @@ -0,0 +1,64 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" +) + +func TestIntegrationTaskQuerySDK(t *testing.T) { + api, queue, _ := integrationWorkerClient(t) + ctx, stop := context.WithTimeout(context.Background(), 20*time.Second) + defer stop() + op := workerTestID(t) + key := "query /+% é " + op + " " + var tasks []Task + for _, k := range []string{key, key, key[:len(key)-1]} { + v, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: queue, Handler: "query." + op, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"sequence":9007199254740993}`), BusinessKey: k, TaskTimeoutMS: 60000}) + if err != nil || v.Task.BusinessKey != k { + t.Fatal(v, err) + } + tasks = append(tasks, v.Task) + } + from := tasks[0].CreatedAt + before := tasks[2].CreatedAt.Add(time.Microsecond) + query := ListTasksRequest{QueueID: queue, State: TaskWaiting, BusinessKey: key, CreatedFrom: &from, CreatedBefore: &before, Limit: 1} + page, err := api.ListTasks(ctx, query) + if err != nil || len(page.Items) != 1 || page.Items[0].ID != tasks[1].ID || page.NextCursor == "" { + t.Fatal(page, err) + } + query.Cursor = page.NextCursor + last, err := api.ListTasks(ctx, query) + if err != nil || len(last.Items) != 1 || last.Items[0].ID != tasks[0].ID || last.NextCursor != "" { + t.Fatal(last, err) + } + changed := query + changed.BusinessKey = key[:len(key)-1] + _, err = api.ListTasks(ctx, changed) + var apiErr *APIError + if !errors.As(err, &apiErr) || apiErr.Code != "INVALID_ARGUMENT" { + t.Fatal("cursor accepted changed filter", err) + } + query.Cursor = "" + end := tasks[1].CreatedAt + query.CreatedBefore = &end + bounded, err := api.ListTasks(ctx, query) + if err != nil || len(bounded.Items) != 1 || bounded.Items[0].ID != tasks[0].ID || bounded.NextCursor != "" { + t.Fatal("exclusive upper boundary", bounded, err) + } + for _, task := range tasks { + if _, err := api.CancelTask(ctx, task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: task.Revision, Reason: "complete task query regression"}); err != nil { + t.Fatal(err) + } + } + query.State = TaskCancelled + query.CreatedBefore = &before + query.Limit = 20 + cancelled, err := api.ListTasks(ctx, query) + if err != nil || len(cancelled.Items) != 2 { + t.Fatal("created time changed after cancellation", cancelled, err) + } + t.Log("exact Unicode/trailing-space filter, inclusive lower/exclusive upper time, cursor binding and stable creation order verified over real HTTP") +} diff --git a/sdk/go/task_query_test.go b/sdk/go/task_query_test.go new file mode 100644 index 0000000..4b8e470 --- /dev/null +++ b/sdk/go/task_query_test.go @@ -0,0 +1,47 @@ +package orbit + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestTaskQueryEncodingAndBounds(t *testing.T) { + const id = "11111111111111111111111111111111" + from := time.Date(2026, 9, 17, 8, 0, 0, 123456000, time.FixedZone("sample", 8*3600)) + before := from.Add(time.Hour) + key := "业务 /+% é " + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + q := r.URL.Query() + if r.Method != "GET" || r.URL.Path != "/v1/tasks" || r.Header.Get("Idempotency-Key") != "" || r.Header.Get("Authorization") != "Bearer example-token" || q.Get("business_key") != key || q.Get("created_from") != "2026-09-17T00:00:00.123456Z" || q.Get("created_before") != "2026-09-17T01:00:00.123456Z" || q.Get("cursor") != "opaque+cursor/=" { + t.Error("query lost meaning", r.URL.String()) + } + json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": TaskPage{Items: []TaskSummary{{ID: id, Revision: 9007199254740993, BusinessKey: key}}}}) + })) + defer server.Close() + c, _ := NewClient(server.URL, Options{Token: "example-token"}) + base := ListTasksRequest{QueueID: id, State: TaskWaiting, BusinessKey: key, CreatedFrom: &from, CreatedBefore: &before, Limit: 2, Cursor: "opaque+cursor/="} + page, err := c.ListTasks(context.Background(), base) + if err != nil || len(page.Items) != 1 || page.Items[0].BusinessKey != key || page.Items[0].Revision != 9007199254740993 { + t.Fatal(page, err) + } + for name, change := range map[string]func(*ListTasksRequest){"queue": func(v *ListTasksRequest) { v.QueueID = "../queue" }, "state": func(v *ListTasksRequest) { v.State = "unknown" }, "limit": func(v *ListTasksRequest) { v.Limit = 101 }, "cursor": func(v *ListTasksRequest) { v.Cursor = strings.Repeat("a", 1025) }, "utf8": func(v *ListTasksRequest) { v.BusinessKey = "\xff" }, "key": func(v *ListTasksRequest) { v.BusinessKey = strings.Repeat("字", 192) }, "newline": func(v *ListTasksRequest) { v.BusinessKey = "x\n" }, "equal": func(v *ListTasksRequest) { v.CreatedBefore = &from }, "nanosecond": func(v *ListTasksRequest) { z := from.Add(time.Nanosecond); v.CreatedFrom = &z }, "year": func(v *ListTasksRequest) { z := time.Time{}; v.CreatedFrom = &z }} { + t.Run(name, func(t *testing.T) { + in := base + change(&in) + if _, err := c.ListTasks(context.Background(), in); err == nil { + t.Fatal("invalid query accepted") + } + }) + } + if calls.Load() != 1 { + t.Fatal("invalid query reached HTTP") + } +} diff --git a/sdk/go/tasks.go b/sdk/go/tasks.go index f62bb63..ef9605f 100644 --- a/sdk/go/tasks.go +++ b/sdk/go/tasks.go @@ -8,7 +8,6 @@ import ( "fmt" "io" "net/http" - "net/url" "strconv" "time" ) @@ -42,22 +41,6 @@ func (c *Client) GetTask(ctx context.Context, id string) (Task, error) { err := c.call(ctx, "GET", "/v1/tasks/"+id, "", nil, &out) return out, err } -func (c *Client) ListTasks(ctx context.Context, in ListTasksRequest) (TaskPage, error) { - var out TaskPage - q := url.Values{} - q.Set("queue_id", in.QueueID) - if in.State != "" { - q.Set("state", string(in.State)) - } - if in.Limit != 0 { - q.Set("limit", strconv.Itoa(in.Limit)) - } - if in.Cursor != "" { - q.Set("cursor", in.Cursor) - } - err := c.call(ctx, "GET", "/v1/tasks?"+q.Encode(), "", nil, &out) - return out, err -} // CancelTask 保存取消意图；运行中等待结果确认，重复请求保留原键与原版本。 func (c *Client) CancelTask(ctx context.Context, id, key string, in CancelTaskRequest) (CancelTaskResult, error) { diff --git a/sdk/go/types.go b/sdk/go/types.go index 9fd8e10..08458c1 100644 --- a/sdk/go/types.go +++ b/sdk/go/types.go @@ -108,6 +108,12 @@ type CancelTaskResult struct { Deduplicated bool `json:"deduplicated"` } type ListTasksRequest struct { + // BusinessKey 可选精确业务关联键；空表示不筛选，区分大小写、Unicode形式及尾部空格。 + BusinessKey string + // CreatedFrom 可选创建时间下界（包含），UTC微秒精度。 + CreatedFrom *time.Time + // CreatedBefore 可选创建时间上界（不包含），与下界同时给出时必须更晚。 + CreatedBefore *time.Time // QueueID 队列标识，必须属于当前命名空间。 QueueID string // State 当前任务状态；终态不会被重复请求重开。