diff --git a/AGENTS.md b/AGENTS.md index 22c40b7..ed5ed84 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,4 +14,6 @@ - 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 - 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 -- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。P3仍需补齐批量受理、独立任务核查/人工重试、任务查询筛选和事务Outbox接入；不能因订阅完成而跳过原功能范围。 +- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。P3仍需补齐独立任务核查/人工重试、任务查询筛选和事务Outbox接入；不能因订阅完成而跳过原功能范围。 + +- 批量受理最多32项、512 KiB；每项复用单项身份与事务，未知结果必须原键核对。任务提交按认证→应用额度→提交幂等身份→队列持锁，禁止恢复为先抢提交唯一键后等待额度锁；原回执仍须先于配额上限检查返回。 diff --git a/README.md b/README.md index 5b390cd..9605d81 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) 已实现；剩余 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 示例；剩余 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/03-http-websocket-protocol.md b/docs/03-http-websocket-protocol.md index 01c761a..d3dd858 100644 --- a/docs/03-http-websocket-protocol.md +++ b/docs/03-http-websocket-protocol.md @@ -1,6 +1,6 @@ # Orbit HTTP 与 WebSocket 协议草案 -> 版本 0.2 · 2026-09-17。HTTP + WS 已确认。本文的具体路径、消息名、字段及状态码是首版设计建议，业务协议尚未实现，需评审后冻结；健康接口的已实现契约见接口手册。 +> 版本 0.2 · 2026-09-17。HTTP + WS 已确认。本文保留首版设计建议；已实现能力以[正式接口手册](api/README.md)为准，其余路径、消息名、字段和状态码仍待对应阶段实现及验证。 实现更新：HTTP 任务/队列接口已完成 P1；Worker 服务端已进入 P2，实际帧名、字段、回执与边界见 [Worker 契约](api/worker.md)。本文保留早期完整设计，未实现项目不能据此视为已交付。 @@ -18,7 +18,7 @@ | 方法与路径 | 行为 | | --- | --- | | `POST /v1/tasks` | 幂等提交一项任务 | -| `POST /v1/tasks:batchSubmit` | 批量逐项提交，每项独立幂等 | +| `POST /v1/tasks:batch` | 已实现；逐项身份、四类结果与限制见[批量手册](api/batch-tasks.md) | | `GET /v1/tasks/{task_id}` | 查询权威任务状态与结果摘要 | | `GET /v1/tasks/{task_id}/attempts` | 分页查询执行尝试 | | `GET /v1/tasks` | 按有索引的字段分页查询，支持提交幂等键精确查找 | @@ -55,7 +55,7 @@ Weave 的补充接口建议如下，功能范围已确认，具体技术细则 - 首次创建任务、流程或信号记录的建议响应为 `201 Created`，表示资源及幂等记录已可靠提交，不表示业务执行或信号消费完成。 - 相同幂等请求返回 `200` 和原资源 ID，标记 `deduplicated: true`；相同键不同语义请求返回 `409`。 -- 单项提交使用 `Idempotency-Key`；批量每项使用 `submission_key`，由 SDK 统一映射为提交幂等标识。 +- 单项提交使用 `Idempotency-Key`；已实现的批量每项使用 `idempotency_key`，与单项共用受理身份；不接受整批幂等头，见[批量接口](api/batch-tasks.md)。 - 普通任务提交幂等作用域为命名空间内的任务提交操作；流程启动使用独立操作作用域，信号按命名空间、实例与信号 ID 去重。其他管理写操作按命名空间、操作和资源界定作用域，避免同一个键在不同 API 间意外冲突。 - HTTP `request_id` 仅用于追踪，与业务幂等键分开。 - 请求超时后，使用原幂等键查询或重试；不能通过换一个键来猜测原请求是否失败。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index e812d4a..01c46be 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响应丢失及消费者进程故障回归已完成；批量受理、独立任务核查/人工重试、业务键/时间筛选与事务Outbox接入待完成 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景、COMMIT响应丢失及消费者进程故障回归已完成；批量受理、SDK、接口与部分补交示例已完成全量回归；独立任务核查/人工重试、业务键/时间筛选与事务Outbox接入待完成 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -37,11 +37,11 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 ### P3 功能范围核对 -可靠订阅链路完成后，重新对照 [原功能范围](01-functional-spec.md) 和 [事务衔接设计](04-reliability-and-data.md)，以下原定首版能力仍有缺口。不能只按已实现模块反推P3完成；先补齐并回归，再进入完整阶段验收。 +可靠订阅链路完成后，重新对照 [原功能范围](01-functional-spec.md) 和 [事务衔接设计](04-reliability-and-data.md)，以下原定首版能力按核对结果逐项补齐。不能只按已实现模块反推P3完成；全部补齐并回归后，再进入完整阶段验收。 -| 剩余项 | 当前证据与后续交付 | +| 核对项 | 当前证据与后续交付 | | --- | --- | -| 批量任务受理 | [任务API](api/tasks.md) 明确尚未提供；需有界逐项结果、独立幂等、部分失败补交、SDK与运行示例 | +| 批量任务受理 | 已实现[批量API](api/batch-tasks.md)、独立SDK及[补交示例](examples/batch.md)；真实MySQL部分失败与HTTP响应丢失用例已通过，同键并发20轮及全量回归已通过，见[报告](reports/phase-3-batch-tasks.md) | | 独立任务人工核查与重试 | [重试API](api/retries.md) 仍未提供解除不确定性的核查接口；需证据、权限、版本、审计和历史保留，人工重试身份按关联新任务实现并冻结契约 | | 任务查询筛选 | [当前ListTasks](../sdk/go/types.go) 仅队列/状态/分页；需按业务关联键及创建时间范围查询，游标绑定所有筛选，保持授权与有界索引扫描 | | 业务到平台的事务Outbox接入 | [事务Outbox设计](04-reliability-and-data.md#61-业务系统到平台) 尚缺完整接入示例；需业务变更与待发请求同事务、稳定提交身份、响应不明核对和发送恢复，保持核心SDK无ORM/数据库依赖 | @@ -63,6 +63,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 跨队列共享资源 | 两队列同键 → 一个有效许可 → 递增序号 → 资源端拒绝旧写入 | P3，已实现完整示例 | | 优先级与应用隔离 | 近同时到期按优先级 → 旧任务等待补偿 → 应用/队列轮转 | P3，已实现完整示例 | | 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，核心、HTTP/WSS、管理及消费 SDK、独立Inbox示例与真实进程故障回归已完成 | +| 批量文件作业受理 | 逐项身份 → 部分结果 → 同键核对 → 释放额度 → 只补交失败项 | P3，已实现完整示例 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | | 审核与外部回执 | 提交流程 → 等待信号 → 授权信号注入 → 继续 | P4 | | 外部作业轮询 | 提交一次 → 有界查询处理中状态 → 回执或最终结果 → 超限核查 | P5 | @@ -94,6 +95,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P3 订阅通信与管理 SDK 回归](reports/phase-3-result-subscription-transport.md) - [P3 消费 SDK 与节点故障回归](reports/phase-3-event-consumer-sdk.md) - [P3 业务 Inbox 与消费者故障回归](reports/phase-3-result-inbox.md) +- [P3 批量任务受理回归](reports/phase-3-batch-tasks.md) - [接口手册](api/README.md) - [SDK 接入](sdk/README.md) - [通用场景示例](examples/README.md) diff --git a/docs/README.md b/docs/README.md index 338d02d..05772c1 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 579e378..f4ef811 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -9,6 +9,7 @@ | `GET /health/live` | 已实现 | [健康检查](health.md) | | `GET /health/ready` | 已实现，验证数据库、迁移记录及零外键要求 | [健康检查](health.md) | | `/v1/tasks` 提交、查询、分页与待执行取消 | P1 已实现 | [任务接口](tasks.md) | +| `POST /v1/tasks:batch` | P3 已实现逐项受理、独立幂等、部分结果与补交 | [批量任务](batch-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` 创建、查询及 `:configure` 控制 | 已实现队列并发、开始速率、积压背压与暂停/恢复 | [队列接口与 SDK 场景](queues.md) | diff --git a/docs/api/batch-tasks.md b/docs/api/batch-tasks.md new file mode 100644 index 0000000..6f22d9b --- /dev/null +++ b/docs/api/batch-tasks.md @@ -0,0 +1,160 @@ +# 批量任务受理 v1 + +状态：P3 已实现。`POST /v1/tasks:batch` 与 SDK `SubmitTasks` 提交一组独立任务，返回每项的受理结果。它不提供流程顺序、整批原子事务或业务效果恰好一次保证；步骤依赖由后续 Weave 管理。 + +## 1. 请求与权限 + +请求头必须有 `Authorization: Bearer ` 与 `Content-Type: application/json`。**不传 `Idempotency-Key` 请求头**，传入会返回400；每项在正文使用自己的 `idempotency_key`。凭据必须具备 `tasks:submit`。预认证后，每项写入仍在自己的事务内重新校验凭据、动作、目标队列范围和命名空间归属。 + +| 字段 | 类型与约束 | +| --- | --- | +| items | 数组，1至32项，保持输入顺序处理 | +| items[].idempotency_key | 1至128个ASCII可打印非空白字符，同一批内不能重复；调用方发送前持久化 | +| items[].task | 与 [POST /v1/tasks](tasks.md#post-v1tasks) 完全相同的任务对象、默认值及归一化规则 | + +总请求不超过512 KiB，每项payload仍不超过256 KiB。服务端请求上下文最多8秒；调用方可能更早取消。SDK不自动拆批或并发发送，32项是保护边界，不是吞吐承诺。不要通过增加批次大小规避背压。 + +示意请求（将队列ID替换为初始化结果）： + +```json +{ + "items": [ + { + "idempotency_key": "conversion-001", + "task": { + "queue_id": "11111111111111111111111111111111", + "handler": "file.convert", + "handler_version": "v1", + "payload_version": "v1", + "payload": {"source_ref": "objects/input-001.txt", "sequence": 9007199254740993} + } + }, + { + "idempotency_key": "conversion-002", + "task": { + "queue_id": "11111111111111111111111111111111", + "handler": "file.convert", + "handler_version": "v1", + "payload_version": "v1", + "payload": {"source_ref": "objects/input-002.txt"} + } + } + ] +} +``` + +保存为 `batch.json` 后调用： + +```sh +curl -i "$ORBIT_URL/v1/tasks:batch" \ + -H "Authorization: Bearer $ORBIT_TOKEN" \ + -H 'Content-Type: application/json' \ + --data-binary @batch.json +``` + +## 2. 响应与逐项状态 + +整批格式与初始权限检查通过后返回HTTP200，即使其中部分或全部项目被拒绝。必须读取 `data.items`，不能只依据HTTP状态判断受理成功。每个输入恰好对应一条结果，`index` 为**本次请求**内从0开始的索引。只补交子集时索引重新编号，稳定身份仍由原幂等键确定。 + +| status | 字段 | 本次请求的含义与处理 | +| --- | --- | --- | +| ACCEPTED | receipt，无error | 已确认受理事务提交或核对到原回执；任务执行结果另查或订阅 | +| REJECTED | error，无receipt | 本项明确拒绝，不因本次调用新增任务；需按错误原因处理，既有同键任务可能仍存在 | +| UNKNOWN | error.code=UNAVAILABLE | 无法确认提交结果，包括数据库/提交响应异常；必须用原键和原参数核对 | +| NOT_ATTEMPTED | error.code=BATCH_STOPPED | 本请求未尝试该项；之前其他请求可能已受理，仍须沿用原身份 | + +例如第一项受理、第二项队列满额（ID和时间仅作示意）： + +```json +{ + "request_id": "44444444444444444444444444444444", + "data": { + "items": [ + { + "index": 0, + "status": "ACCEPTED", + "receipt": { + "task_id": "22222222222222222222222222222222", + "queue_id": "11111111111111111111111111111111", + "revision": "1", + "created_at": "2026-09-17T00:00:00Z", + "deduplicated": false + } + }, + { + "index": 1, + "status": "REJECTED", + "error": { + "code": "QUEUE_FULL", + "message": "queue unfinished task limit reached; retry later using the original identity", + "retry_after_ms": 1000 + } + } + ] + } +} +``` + +`receipt` 是原始受理回执的精简视图，字段如下： + +| 字段 | 类型与含义 | +| --- | --- | +| task_id / queue_id | 32位小写十六进制标识 | +| revision | 十进制字符串，原受理版本；任务已执行或取消后重报仍返回原版本 | +| created_at | 原任务创建时间，UTC RFC3339微秒精度 | +| deduplicated | bool，本项是否复用已有受理回执 | + +回执不重复携带payload、结果或策略，32项返回大小有界。需要当前完整状态时调用 `GetTask`；取消、核查等后续操作应先查询当前版本。 + +## 3. 整批拒绝与部分失败 + +开始处理任何项目之前，完整读取并严格解析整个请求。以下情况整批返回错误信封，当前请求不写任务： + +- HTTP400：空批/超过32项、无效或重复的项幂等键、传整批幂等头、未知字段/大小写别名/重复JSON键、字段类型或ID/时间编码错误、非法JSON/Unicode、非可空字段为null。 +- HTTP413：总请求超过512 KiB；HTTP415：Content-Type不支持。 +- HTTP401/403：初始凭据无效或无tasks:submit动作权限。 +- HTTP503：预认证数据库不可用；此时还未进入逐项受理。 + +结构正确后，任务语义错误按项返回 `REJECTED/INVALID_ARGUMENT`，例如处理器为空、优先级超过9、有效时间范围不满足要求。队列不存在或跨命名空间为NOT_FOUND，缺少队列范围为PERMISSION_DENIED，同身份不同内容为IDEMPOTENCY_CONFLICT。队列或应用未终结上限分别为QUEUE_FULL、APPLICATION_FULL，按项提供1000毫秒退避提示；HTTP200不附整批Retry-After头。 + +处理遇到未知数据库结果后立即停止余项：当前项UNKNOWN，余项NOT_ATTEMPTED。中途凭据失效时当前项REJECTED/UNAUTHENTICATED，余项停止。上下文取消时，已确认提交的前项保留ACCEPTED，尚未开始的后项NOT_ATTEMPTED；在途项若无法确认提交则UNKNOWN。响应本身也可能丢失，不能依赖一定能收到这些状态。 + +一项队列越权、语义错误、幂等冲突或额度不足，不阻止后续独立项尝试。不在请求之间保存批次游标或整批快照，也不撤销此前已经受理的项目。每项任务、幂等记录、额度、事件与审计继续同事务提交；没有新增数据库表、外键或级联操作。 + +## 4. 幂等与补交 + +1. 在业务自己的事务中保存每项稳定键和完整请求内容，再发给平台。业务变更与平台受理仍有跨系统事务边界，后续事务Outbox示例单独交付。 +2. 批量与单项提交使用相同的 `task.submit` 命名空间身份。批量受理后用同键调用SubmitTask会去重；反方向相同。换批次、重排或只补交子集不会改变身份。 +3. 原回执核对先于当前积压额度检查，队列满额时仍可获取已受理项目的回执；当前凭据与队列授权仍须有效。 +4. UNKNOWN、超时、连接断开或无法验证的响应：保留原键、原参数和原相对时限；不要重算available_at，不生成新键。不确定哪些项成功时可重报整批。 +5. 明确额度不足时，退避并释放/等待配额后，可只补交被拒项目，仍用原身份。参数错误需要修正时先确认该操作从未受理，不能把“换键”当作解决幂等冲突的方法。 +6. 幂等记录当前标记90天过期但尚无在线清理；生产清理窗口按后续保留策略执行，业务不得假定无限去重。 + +队列暂停仍允许受理。暂停、取消和任务执行与批量接口独立，提交顺序不保证执行顺序。业务效果去重仍由Worker/接收方负责。 + +## 5. Go SDK + +```go +// items 来自业务库中的持久化待发记录。 +result, err := client.SubmitTasks(ctx, orbit.SubmitTasksRequest{Items: items}) +if err != nil { + // 网络/响应错误无法证明未提交；保留所有原身份，稍后核对。 + return err +} +for _, item := range result.Items { + switch item.Status { + case orbit.BatchAccepted: + // 将task_id与对应的持久化操作关联；此处不代表业务执行成功。 + fmt.Println(item.Index, item.Receipt.TaskID, item.Receipt.Deduplicated) + case orbit.BatchRejected: + // 根据item.Error.Code决定修正、人工处理或按RetryAfterMS退避。 + fmt.Println(item.Index, item.Error.Code) + case orbit.BatchUnknown, orbit.BatchNotAttempted: + // 使用items[item.Index]的原键和原参数保留待核对状态。 + } +} +``` + +SDK验证输入数量/身份/整体体积及返回数量、顺序、状态和回执归属；不自动重试、拆批或创建新键。`err == nil` 只表示取得可解释的逐项响应。局部错误为 `BatchItemError`，不是顶层APIError；整批HTTP错误仍为APIError，网络错误沿用标准库。独立SDK无数据库依赖。 + +完整运行命令、输入输出和收尾见 [批量场景](../examples/batch.md)，源码见 [SDK示例](../../sdk/go/examples/batch/main.go)，覆盖范围和实际证据见 [回归报告](../reports/phase-3-batch-tasks.md)。旧客户端继续使用单项接口；新SDK调用旧服务该路由会返回404，不会静默退化为其他语义。 diff --git a/docs/api/tasks.md b/docs/api/tasks.md index 97ad2ce..65e9ecb 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -6,7 +6,7 @@ P2 进展：服务端执行内核、WSS 与独立自动 Worker SDK 已可产生 ## 公共约定 -业务接口需要 `Authorization: Bearer `；写入需要 `Content-Type: application/json` 和稳定的 `Idempotency-Key`。凭据决定命名空间，请求体不能覆盖它。详见 [鉴权与初始化](authentication.md)。 +业务接口需要 `Authorization: Bearer `；单项写入需要 `Content-Type: application/json` 和稳定的 `Idempotency-Key`；[批量受理](batch-tasks.md) 在每项正文提供独立幂等键。凭据决定命名空间，请求体不能覆盖它。详见 [鉴权与初始化](authentication.md)。 请求体上限 512 KiB。拒绝未知字段、大小写别名、重复 JSON 键、无效 UTF-8、孤立 Unicode 代理项和超过 32 层的嵌套。除可空时间外，字段不能显式传 null。请求追踪 ID 由服务端生成，与幂等键分开。 @@ -190,4 +190,4 @@ SDK：`Client.CancelTask(ctx, taskID, key, request)`。此操作不承诺回滚 {"request_id":"44444444444444444444444444444444","error":{"code":"IDEMPOTENCY_CONFLICT","message":"idempotency key was used with different content"}} ``` -有限自动重试已实现，具体分类、退避、回执和升级边界见 [安全重试](retries.md)。[运行中取消](cancellation.md) 已实现；批量受理和人工重试尚未实现。[执行历史](execution-history.md) 已提供尝试与证据的列表和详情。完整可运行程序见 [SDK 任务示例](../../sdk/go/examples/tasks/main.go)，已验证异常场景见 [P1 回归报告](../reports/phase-1.md) 及 [取消回归](../reports/phase-3-cancellation.md)。SDK 忽略未知响应字段；更改已有字段语义需升级契约和回归。 +有限自动重试已实现，具体分类、退避、回执和升级边界见 [安全重试](retries.md)。[运行中取消](cancellation.md) 已实现；[批量受理](batch-tasks.md) 已实现逐项结果与原身份补交；独立任务人工重试仍待实现。[执行历史](execution-history.md) 已提供尝试与证据的列表和详情。完整可运行程序见 [SDK 任务示例](../../sdk/go/examples/tasks/main.go)，已验证异常场景见 [P1 回归报告](../reports/phase-1.md) 及 [取消回归](../reports/phase-3-cancellation.md)。SDK 忽略未知响应字段；更改已有字段语义需升级契约和回归。 diff --git a/docs/examples/README.md b/docs/examples/README.md index 48a85d1..2c73822 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -130,3 +130,11 @@ - 输出：inbox_duplicate=true、applied_events=1、ack_confirmed=1、pending=0。 - 覆盖：并发重复、旧版本保护、回滚、真实MySQL COMMIT响应丢失、业务提交后/ACK前消费者SIGKILL及新进程去重。 - 边界：只覆盖接收方本地SQL效果；核心SDK不依赖数据库，外部副作用需要接收方自己的幂等/Outbox设计。 + +## 已可运行：批量受理与部分补交 + +- 源码：[batch](../../sdk/go/examples/batch/main.go)，[完整启动命令和输出](batch.md)。 +- 步骤：独立暂停队列 → 三项批量提交、两项成功一项满额 → 整批核对原回执 → 取消释放额度 → 仅原身份补交第三项 → 取消收尾。 +- 预期：accepted=2、replayed=2、retried_items=1、unfinished=0。 +- 异常：逐项区分REJECTED/UNKNOWN/NOT_ATTEMPTED；真实HTTP响应丢失后恢复无重复，保留当前权限和额度控制。 +- 范围：示例只演示受理，不转换文件、不提供业务Outbox；失败非零退出，每项60秒总时限，保留队列及审计。 diff --git a/docs/examples/batch.md b/docs/examples/batch.md new file mode 100644 index 0000000..e6b8bc2 --- /dev/null +++ b/docs/examples/batch.md @@ -0,0 +1,50 @@ +# 批量文件作业受理与部分补交 + +[完整源码](../../sdk/go/examples/batch/main.go) 使用独立Go SDK，不导入服务端模型或数据库驱动。它演示受理与取消，不启动Worker或转换真实文件。 + +## 前置与运行 + +使用已迁移的MySQL和当前服务。开发凭据需要 `queues:manage/read`、`tasks:submit/read/cancel` 和当前命名空间全部队列范围；标准bootstrap默认具备这些权限。只在专用开发命名空间运行。 + +仓库根目录中，先通过环境配置专用 `ORBIT_MYSQL_DSN`： + +```sh +go run ./cmd/orbit-migrate up +mkdir -p .local +chmod 700 .local +go run ./cmd/orbit-admin bootstrap \ + --application batch-demo --namespace default --environment development \ + --queue files --output .local/batch.credentials.json +go run ./cmd/orbit +``` + +另开终端： + +```sh +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ +ORBIT_CREDENTIALS_FILE=../../.local/batch.credentials.json \ +go run ./examples/batch +``` + +也接受ORBIT_TOKEN，环境值优先。示例自行创建并暂停一个独立队列，未终结上限为2，不改变已有队列或应用配置。前置应用剩余额度至少2。 + +## 调用链与输出 + +1. 三项文件引用与精确整数输入，使用三个稳定键一次提交。 +2. 前两项受理，第三项QUEUE_FULL；整批重报确认前两项ID不变且去重。 +3. 查询并取消第一项释放额度，仅以第三项原键和原参数补交。 +4. 取消剩余两项，核对unfinished=0，队列保持暂停。 + +```text +accepted=2 rejected=1 code=QUEUE_FULL +replayed=2 same_task_ids=true +retried_items=1 same_identity=true accepted=true +queue=<实际32位ID> paused=true unfinished=0 +``` + +每项总时限60秒，示例总调用上下文20秒。所有响应都校验，失败非零退出且不打印token或业务参数。网络异常时示例直接报错，不循环生成新身份；真正生产发送器应从业务库恢复原身份再核对。示例每次运行生成独立身份，**不是持久化业务Outbox实现**。 + +正常完成保留已取消任务、队列和审计，不硬删除。异常中断可能留下任务；恢复扫描器按总时限处理，示例不能保证强杀后的主动清理。对有外部副作用的真实任务仍需核查执行状态。 + +[真实SDK回归](../../sdk/go/batch_integration_test.go) 还通过HTTP代理在服务已返回受理回执后强制断线，再用原身份重报，确认没有重复任务。全部失败分类、部分提交与单项/批量互通见 [接口手册](../api/batch-tasks.md)。 diff --git a/docs/reports/phase-3-batch-tasks.md b/docs/reports/phase-3-batch-tasks.md new file mode 100644 index 0000000..5752ad2 --- /dev/null +++ b/docs/reports/phase-3-batch-tasks.md @@ -0,0 +1,95 @@ +# P3 批量任务受理回归 + +验证日期：2026-09-17。本阶段增加批量受理、SDK及部分补交场景；不代表P3或全平台验收完成。完整集成已通过，下面记录实际覆盖范围与剩余边界。 + +## 1. 实现与证据入口 + +- [服务契约](../../pulse/service/batch.go)：最多32项，逐项独立事务与幂等，ACCEPTED/REJECTED/UNKNOWN/NOT_ATTEMPTED四类结果。未知数据库结果或凭据失效后停止余项，上下文结束保留已确认前项。 +- [HTTP入口](../../internal/transport/httpserver/tasks.go)：`POST /v1/tasks:batch`，全体正文严格解析后才受理，每项正文幂等键，总请求512 KiB。完整响应使用HTTP200，逐项标记结果。 +- [独立SDK](../../sdk/go/batch.go)：SubmitTasks及中文字段说明，不自动生成身份、拆批或重试；校验返回数量、顺序、状态与回执归属。 +- [运行示例](../../sdk/go/examples/batch/main.go)：两项受理、一项满额，原键核对后释放配额，仅补交被拒项并取消收尾。 +- [接口手册](../api/batch-tasks.md)、[场景说明](../examples/batch.md)、[开发计划](../08-development-plan.md) 同步维护。 + +不增加数据库模型或迁移，平台仍为35个实体、289个中文注释持久化字段；不增加外键、级联或SDK数据库依赖。 + +## 2. 验证环境与命令 + +macOS、Go1.27.1、Docker单实例MySQL8.4.10。使用一次性数据库、随机测试凭据及回环端口，结束后自动删除，不修改已有业务库。 + +```sh +make fmt test vet build +make integration +``` + +普通测试没有数据库环境时会跳过Integration；数据库/HTTP/进程证据来自真实集成入口。并发修复另在一次性数据库重复执行： + +```sh +go test -race -count=20 \ + -run '^TestIntegrationTaskBatch/concurrent_replay_and_shared_application_backpressure$' \ + -v ./internal/storage/mysql +``` + +该定向命令要求 `ORBIT_TEST_MYSQL_DSN` 指向一次性数据库；正式完整脚本会自动准备隔离环境。 + +| 验证组 | 实际结果 | +| --- | --- | +| fmt / race / vet / build | 已通过 | +| 定向批量MySQL六组场景 | PASS，5.559秒，含包开销 | +| 真实HTTP丢失批量响应及SDK补交 | PASS，1.458秒，含包开销 | +| 完整批量示例 | 已通过，unfinished=0 | +| 同键并发与共享配额20次重复 | PASS，10.110秒；每轮8个调用者 | +| 全量MySQL集成 | PASS，94.874秒 | +| SDK Worker完整组 | PASS，88.099秒 | +| SDK调度 / 队列资源 / 批量受理 | PASS，5.640 / 4.474 / 1.361秒 | +| SDK订阅管理 / 事件消费 | PASS，33.508 / 51.852秒 | +| Inbox六组真实数据库/消费者故障 | PASS，35.783秒 | +| SDK应用配额 | PASS，1.426秒 | +| 九项节点/Worker进程故障 | PASS，69.057秒；连同Inbox消费者共十项真实进程故障 | +| 全部运行示例 | 15个核心SDK示例 + 1个独立Inbox示例通过 | + +上述完整脚本正常退出，状态码0；没有以尚在运行的日志推断完成。文档链接与JSON代码块核对通过，核心SDK依赖检查未包含服务端模型或MySQL驱动。以上是单次回归耗时，不是吞吐或性能指标。 + +## 3. 事务、权限与配额 + +[数据库回归](../../internal/storage/mysql/task_batch_integration_test.go) 实际验证： + +1. 有效/参数无效/不存在队列/队列满额混合批次，每项结果独立；数据库仅有两份任务、受理幂等、事件及审计。 +2. 单项与批量共用原始回执；取消后重报仍返回原受理版本，满额也不阻止已提交身份核对。释放配额后原失败身份受理成功，同键不同内容仍冲突。 +3. 跨命名空间队列与队列权限拒绝；即使已有回执也不能绕过当前队列范围。 +4. 第一项提交后撤销凭据，第二项重新认证拒绝，第三项未尝试；只有第一项任务存在。 +5. 第一项提交后取消请求上下文，后缀未写入；恢复后整批原键重报，首项去重、后缀受理。 +6. 真实审计触发器故障使新项事务回滚，任务/幂等/事件/配额不残留；既有已受理项保留。接口保守返回UNKNOWN，修复故障后原身份补交成功。 +7. 八个调用者同时重报四个相同身份，应用上限3；前三个稳定任务ID、仅三次新受理，第四项APPLICATION_FULL，另一命名空间也受共享额度限制。 + +单元及HTTP边界还覆盖整批非法键/重复键/超量、全项参数无效时仍认证、未知字段/大小写别名/重复JSON键/null、超大正文、禁止整批幂等头、精确整数、不完整或矛盾的SDK回执，以及上下文和内部错误不泄露细节。 + +## 4. 并发回归发现与修复 + +最初同键并发满额场景实际出现MySQL 1213死锁，API按既有保守规则报告UNKNOWN。测试没有将UNKNOWN改判成功，也没有通过放宽断言隐藏问题。 + +任务提交原来先抢占幂等唯一键，再等待应用额度锁。在满额回滚和同键INSERT竞争中触发锁冲突。本轮调整为：认证 → 应用额度锁 → 提交幂等身份 → 队列 → 任务/事件/审计。取得应用锁时不检查或扣减额度；读取已有回执仍在配额检查之前。 + +提取只在应用锁已持有时使用的队列锁辅助函数，避免重复查询应用锁。每项事务结束后释放全部锁，整个批次不持有跨项锁。没有新增外部锁服务、数据库外键或盲目重放事务。当前场景通过不表示全部MySQL负载不再可能死锁；其他未知数据库错误仍必须原身份核对。 + +## 5. 真实HTTP故障与示例 + +[SDK集成](../../sdk/go/batch_integration_test.go) 启动真实HTTP代理：先读完平台已经生成的受理响应，再强制关闭客户端连接、丢弃响应。SDK返回网络错误，未自动补发。再次用原键/原参数请求，两项返回同一任务且deduplicated=true，第三项保持QUEUE_FULL；查询仅存在两行任务。取消一项后只补交原第三项，受理成功。 + +完整示例输出： + +```text +accepted=2 rejected=1 code=QUEUE_FULL +replayed=2 same_task_ids=true +retried_items=1 same_identity=true accepted=true +queue=<实际ID> paused=true unfinished=0 +``` + +这验证受理与HTTP回执恢复，不等同于Worker业务效果恰好一次，也不是业务事务Outbox。任务、队列和审计保留，无硬删除。 + +## 6. 兼容与尚未验收范围 + +- 无数据库迁移变更；批量与单项受理的身份、策略、事件和审计相同。新SDK调用旧节点返回404，不静默降级。 +- 单项与批量均使用调整后的锁顺序。部署应统一升级写入节点；未做新旧二进制混合压力或滚动发布验收。 +- 批次逐项串行，8秒HTTP预算不保证每次32项全部尝试；大参数回执不重复携带原输入。调用方必须处理部分提交和响应全丢。 +- 本轮为本机单MySQL正确性验证，未验证数据库故障切换、跨区域网络、百万租户容量或生产RPO/RTO。 +- P3仍缺独立任务人工核查/重试、业务键及时间筛选、事务Outbox示例；P4–P7仍按原目标推进。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 983a651..dc5977a 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -1,6 +1,6 @@ # Go SDK 接入说明 -已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询、[自动 Worker](worker-runtime.md)、[Worker 底层协议 SDK](worker.md) 和 [执行历史查询](../api/execution-history.md)。[安全重试](../api/retries.md) 已提供有限策略、WorkerFailure.Retryable、历史查询及完整例子。[计划调度](../api/schedules.md) 已支持创建、查询、版本发布、暂停/恢复、触发历史与完整示例。[订阅管理](../api/subscriptions.md) 已支持创建、查询、暂停恢复、交付历史与人工重试；[事件消费 SDK](event-consumer.md) 已支持连接、自动心跳、重连与确认。流程构建等后续接口按 [开发计划](../08-development-plan.md) 继续实现。 +已实现健康查询、任务单项与[批量提交](../api/batch-tasks.md)/查询/分页/待执行取消、队列创建/查询、[自动 Worker](worker-runtime.md)、[Worker 底层协议 SDK](worker.md) 和 [执行历史查询](../api/execution-history.md)。[安全重试](../api/retries.md) 已提供有限策略、WorkerFailure.Retryable、历史查询及完整例子。[计划调度](../api/schedules.md) 已支持创建、查询、版本发布、暂停/恢复、触发历史与完整示例。[订阅管理](../api/subscriptions.md) 已支持创建、查询、暂停恢复、交付历史与人工重试；[事件消费 SDK](event-consumer.md) 已支持连接、自动心跳、重连与确认。流程构建等后续接口按 [开发计划](../08-development-plan.md) 继续实现。 module：`git.yuntuo.hk/k2safe/orbit/sdk/go`。SDK 使用独立 go.mod，依赖 Go 标准库与 `github.com/coder/websocket v1.8.15`，不引用服务端模型或数据库实现。 @@ -105,6 +105,7 @@ fmt.Println(cancelled.Task.State) | 方法 | 请求 / 返回 | 文档 | | --- | --- | --- | +| SubmitTasks | SubmitTasksRequest / SubmitTasksResult，32项独立幂等与结果 | [批量受理](../api/batch-tasks.md) / [运行场景](../examples/batch.md) | | CreateSubscription / GetSubscription / ListSubscriptions | 订阅受理、当前配置与队列分页 | [订阅](../api/subscriptions.md) | | ConfigureSubscription | 显式暂停/继续与版本、原因 | [订阅](../api/subscriptions.md) | | ListEventDeliveries / GetEventDelivery / RetryEventDelivery | 交付历史、冻结结果和耗尽后追加预算 | [订阅](../api/subscriptions.md) | diff --git a/internal/storage/mysql/task_batch_integration_test.go b/internal/storage/mysql/task_batch_integration_test.go new file mode 100644 index 0000000..1d5cc5b --- /dev/null +++ b/internal/storage/mysql/task_batch_integration_test.go @@ -0,0 +1,230 @@ +package mysql + +import ( + "context" + "encoding/json" + "fmt" + "os" + "sync" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func taskBatch(f *executionFixture, n int) service.BatchSubmitInput { + in := service.BatchSubmitInput{} + for i := 0; i < n; i++ { + in.Items = append(in.Items, service.BatchSubmitItem{IdempotencyKey: fmt.Sprintf("batch-%d", i), Task: service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}}) + } + return in +} +func requireBatchStatuses(t *testing.T, out service.BatchSubmitResult, err error, wants ...service.BatchItemStatus) { + t.Helper() + if err != nil || len(out.Items) != len(wants) { + t.Fatal(out, err) + } + for i, want := range wants { + if out.Items[i].Index != i || out.Items[i].Status != want { + t.Fatalf("item %d=%+v want=%s", i, out.Items[i], want) + } + } +} + +type batchAfterCommitStore struct { + service.Store + after func() +} + +func (s *batchAfterCommitStore) Submit(ctx context.Context, token auth.Token, key string, in service.SubmitInput) (service.SubmitResult, error) { + out, err := s.Store.Submit(ctx, token, key, in) + if err == nil && s.after != nil { + fn := s.after + s.after = nil + fn() + } + return out, err +} + +func TestIntegrationTaskBatch(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + 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("partial_rejections_replay_and_quota", func(t *testing.T) { + f := fixture(t) + configureResources(t, f, 10, 4, 2, true) + in := taskBatch(f, 5) + in.Items[1].Task.Handler = "" + in.Items[2].Task.QueueID = executionID(t) + out, err := f.tasks.SubmitBatch(ctx, f.token, in) + requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchRejected, service.BatchRejected, service.BatchAccepted, service.BatchRejected) + if out.Items[1].Error.Code != "INVALID_ARGUMENT" || out.Items[2].Error.Code != "NOT_FOUND" || out.Items[4].Error.Code != "QUEUE_FULL" || out.Items[4].Error.RetryAfterMS != 1000 { + t.Fatal(out) + } + assertQueueCount(t, f, 2) + assertApplicationCounts(t, f, 2, 0, 0) + for table, filter := range map[string]string{"orbit_idempotency_keys": "operation='task.submit'", "orbit_audit_events": "action='task.submit'", "orbit_events": "type='task.accepted'"} { + if f.count(table, filter) != 2 { + t.Fatal("partial batch leaked writes", table) + } + } + if string(f.get(out.Items[0].Receipt.TaskID).Payload) != `{"n":9007199254740993}` { + t.Fatal("payload precision lost") + } + one, err := f.tasks.Submit(ctx, f.token, in.Items[0].IdempotencyKey, in.Items[0].Task) + if err != nil || !one.Deduplicated || one.Task.ID != out.Items[0].Receipt.TaskID { + t.Fatal("single batch identity diverged", err) + } + first := out.Items[0].Receipt + if _, err := f.tasks.CancelTask(ctx, f.token, first.TaskID, "cancel", service.CancelInput{ExpectedRevision: first.Revision, Reason: "free batch slot"}); err != nil { + t.Fatal(err) + } + in.Items = []service.BatchSubmitItem{in.Items[0], in.Items[4]} + again, err := f.tasks.SubmitBatch(ctx, f.token, in) + requireBatchStatuses(t, again, err, service.BatchAccepted, service.BatchAccepted) + if !again.Items[0].Receipt.Deduplicated || again.Items[0].Receipt.Revision != 1 || again.Items[1].Receipt.Deduplicated { + t.Fatal("receipt/rejected identity not reusable") + } + in.Items[0].Task.Payload = json.RawMessage(`{"changed":true}`) + conflict, err := f.tasks.SubmitBatch(ctx, f.token, in) + requireBatchStatuses(t, conflict, err, service.BatchRejected, service.BatchAccepted) + if conflict.Items[0].Error.Code != "IDEMPOTENCY_CONFLICT" { + t.Fatal(conflict) + } + assertQueueCount(t, f, 2) + assertApplicationCounts(t, f, 2, 0, 0) + }) + t.Run("cross_namespace_and_queue_scope", func(t *testing.T) { + f, g := fixture(t), fixture(t) + in := taskBatch(f, 2) + in.Items[0].Task.QueueID = g.resources.QueueID + out, err := f.tasks.SubmitBatch(ctx, f.token, in) + requireBatchStatuses(t, out, err, service.BatchRejected, service.BatchAccepted) + if out.Items[0].Error.Code != "NOT_FOUND" { + t.Fatal(out) + } + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, `{"actions":["tasks:submit"],"queues":["`+string(g.resources.QueueID)+`"]}`, f.resources.NamespaceID, f.resources.CredentialID) + out, err = f.tasks.SubmitBatch(ctx, f.token, service.BatchSubmitInput{Items: in.Items[1:]}) + requireBatchStatuses(t, out, err, service.BatchRejected) + if out.Items[0].Error.Code != "PERMISSION_DENIED" || out.Items[0].Receipt != nil { + t.Fatal("cached receipt bypassed current queue scope") + } + if g.count("pulse_tasks", "1=1") != 0 { + t.Fatal("cross namespace write") + } + }) + t.Run("revoke_between_items_is_rechecked", func(t *testing.T) { + f := fixture(t) + s := &batchAfterCommitStore{Store: NewTaskStore(db), after: func() { + f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + }} + out, err := service.New(s).SubmitBatch(ctx, f.token, taskBatch(f, 3)) + requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchRejected, service.BatchNotAttempted) + if out.Items[1].Error.Code != "UNAUTHENTICATED" || f.count("pulse_tasks", "1=1") != 1 { + t.Fatal(out) + } + }) + t.Run("cancel_after_first_commit_leaves_recoverable_prefix", func(t *testing.T) { + f := fixture(t) + run, stop := context.WithCancel(ctx) + defer stop() + s := &batchAfterCommitStore{Store: NewTaskStore(db), after: stop} + in := taskBatch(f, 3) + out, err := service.New(s).SubmitBatch(run, f.token, in) + requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchNotAttempted, service.BatchNotAttempted) + if f.count("pulse_tasks", "1=1") != 1 { + t.Fatal("cancelled suffix wrote tasks") + } + again, err := f.tasks.SubmitBatch(ctx, f.token, in) + requireBatchStatuses(t, again, err, service.BatchAccepted, service.BatchAccepted, service.BatchAccepted) + if !again.Items[0].Receipt.Deduplicated || again.Items[0].Receipt.TaskID != out.Items[0].Receipt.TaskID { + t.Fatal("lost committed prefix") + } + assertApplicationCounts(t, f, 3, 0, 0) + }) + t.Run("audit_failure_rolls_back_item_and_preserves_previous_commit", func(t *testing.T) { + f := fixture(t) + in := taskBatch(f, 3) + if _, err := f.tasks.Submit(ctx, f.token, in.Items[0].IdempotencyKey, in.Items[0].Task); err != nil { + t.Fatal(err) + } + trigger := "batch_audit_" + string(executionID(t)) + f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='private injected audit fault'") + out, err := f.tasks.SubmitBatch(ctx, f.token, in) + f.exec("DROP TRIGGER " + trigger) + requireBatchStatuses(t, out, err, service.BatchAccepted, service.BatchUnknown, service.BatchNotAttempted) + if out.Items[1].Error.Code != "UNAVAILABLE" { + t.Fatal(out) + } + assertQueueCount(t, f, 1) + assertApplicationCounts(t, f, 1, 0, 0) + for table, filter := range map[string]string{"orbit_idempotency_keys": "operation='task.submit'", "orbit_audit_events": "action='task.submit'", "orbit_events": "type='task.accepted'"} { + if f.count(table, filter) != 1 { + t.Fatal("atomic item rollback failed", table) + } + } + again, err := f.tasks.SubmitBatch(ctx, f.token, in) + requireBatchStatuses(t, again, err, service.BatchAccepted, service.BatchAccepted, service.BatchAccepted) + assertQueueCount(t, f, 3) + }) + t.Run("concurrent_replay_and_shared_application_backpressure", func(t *testing.T) { + f := fixture(t) + grantApplicationResources(f) + configureApplication(t, f, 100, 10, 3) + in := taskBatch(f, 4) + type result struct { + out service.BatchSubmitResult + err error + } + ch := make(chan result, 8) + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { defer wg.Done(); out, err := f.tasks.SubmitBatch(ctx, f.token, in); ch <- result{out, err} }() + } + wg.Wait() + close(ch) + ids := map[int]core.ID{} + fresh := 0 + for r := range ch { + requireBatchStatuses(t, r.out, r.err, service.BatchAccepted, service.BatchAccepted, service.BatchAccepted, service.BatchRejected) + for i := 0; i < 3; i++ { + receipt := r.out.Items[i].Receipt + if previous := ids[i]; previous != "" && previous != receipt.TaskID { + t.Fatal("duplicate identity") + } + ids[i] = receipt.TaskID + if !receipt.Deduplicated { + fresh++ + } + } + if r.out.Items[3].Error.Code != "APPLICATION_FULL" { + t.Fatal(r.out) + } + } + if fresh != 3 || f.count("pulse_tasks", "1=1") != 3 { + t.Fatal("duplicate creation", fresh) + } + assertApplicationCounts(t, f, 3, 0, 0) + g := newExecutionFixture(t, ctx, db, f.resources.ApplicationID) + out, err := g.tasks.SubmitBatch(ctx, g.token, taskBatch(g, 1)) + requireBatchStatuses(t, out, err, service.BatchRejected) + if out.Items[0].Error.Code != "APPLICATION_FULL" { + t.Fatal(out) + } + }) +} diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index 244e2be..5b3b201 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -28,6 +28,11 @@ func NewTaskStore(db *sql.DB) *TaskStore { return &TaskStore{db: db} } var _ service.Store = (*TaskStore)(nil) +// AuthorizeSubmit 用于批量请求的只读预认证；每项事务仍独立校验凭据、动作与队列范围。 +func (s *TaskStore) AuthorizeSubmit(ctx context.Context, token auth.Token) error { + return s.transaction(ctx, token, "tasks:submit", func(*sql.Tx, auth.Permissions) error { return nil }) +} + func (s *TaskStore) transaction(ctx context.Context, token auth.Token, action string, fn func(*sql.Tx, auth.Permissions) error) error { return WithTx(ctx, s.db, func(tx *sql.Tx) error { p, err := authenticate(ctx, tx, token) @@ -159,6 +164,12 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in if !p.AllowsQueue(in.QueueID) { return fault.Forbidden } + // 先按既有应用额度锁串行进入受理，再占用幂等唯一键。 + // 满额回滚时，同键并发INSERT的重复检查锁不能与应用额度锁互相等待。 + // 此处只取得锁，不检查/扣减额度；已有回执在满额时仍可返回。 + if _, err := lockApplicationResources(ctx, tx, token.NamespaceID, false); err != nil { + return err + } taskID, err := core.NewID() if err != nil { return err @@ -174,7 +185,7 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in out.Deduplicated = true return nil } - if _, _, err := lockQueue(ctx, tx, token.NamespaceID, in.QueueID); err != nil { + if _, _, err := lockQueueAfterApplication(ctx, tx, token.NamespaceID, in.QueueID); err != nil { return err } now, err := dbNow(ctx, tx) diff --git a/internal/storage/mysql/worker_execution.go b/internal/storage/mysql/worker_execution.go index 0f11ee2..8afef23 100644 --- a/internal/storage/mysql/worker_execution.go +++ b/internal/storage/mysql/worker_execution.go @@ -49,6 +49,11 @@ func lockQueue(ctx context.Context, tx *sql.Tx, ns, id core.ID) (paused bool, ma if _, err = lockApplicationResources(ctx, tx, ns, false); err != nil { return } + return lockQueueAfterApplication(ctx, tx, ns, id) +} + +// lockQueueAfterApplication 仅在同一事务已持有应用额度排他锁时调用。 +func lockQueueAfterApplication(ctx context.Context, tx *sql.Tx, ns, id core.ID) (paused bool, max uint32, err error) { err = tx.QueryRowContext(ctx, "SELECT paused,max_concurrency FROM pulse_queues WHERE namespace_id=? AND id=? FOR UPDATE", ns, id).Scan(&paused, &max) if errors.Is(err, sql.ErrNoRows) { err = fault.NotFound diff --git a/internal/transport/httpserver/batch_test.go b/internal/transport/httpserver/batch_test.go new file mode 100644 index 0000000..4020714 --- /dev/null +++ b/internal/transport/httpserver/batch_test.go @@ -0,0 +1,79 @@ +package httpserver + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "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" +) + +type batchAuthStore struct { + service.Store + authCalls int +} + +func batchHTTPToken(t *testing.T) string { + t.Helper() + token, _, err := auth.Issue(core.ID(strings.Repeat("1", 32)), core.ID(strings.Repeat("2", 32))) + if err != nil { + t.Fatal(err) + } + return "Bearer " + token +} + +func (s *batchAuthStore) AuthorizeSubmit(context.Context, auth.Token) error { + s.authCalls++ + return fault.Unauthorized +} + +func TestBatchHTTPRejectsWholeMalformedEnvelopeBeforeWrites(t *testing.T) { + for _, test := range []struct{ body, key, code string }{ + {`{"items":[{"idempotency_key":"a","task":{}},{"idempotency_key":"b","task":{"unexpected":1}}]}`, "", "INVALID_ARGUMENT"}, + {`{"items":[{"idempotency_key":"a","task":{"retry":{"MaxAttempts":1}}}]}`, "", "INVALID_ARGUMENT"}, + {`{"items":[{"idempotency_key":"a","task":{"payload":{"x":1,"x":2}}}]}`, "", "INVALID_ARGUMENT"}, + {`{"items":[null]}`, "", "INVALID_ARGUMENT"}, + {`{"items":[{"idempotency_key":"a","task":null}]}`, "", "INVALID_ARGUMENT"}, + {`{"items":[{"idempotency_key":"a","task":{}},{"idempotency_key":"a","task":{}}]}`, "", "INVALID_ARGUMENT"}, + {`{"items":[]}`, "", "INVALID_ARGUMENT"}, + {`{"items":[{"idempotency_key":"a","task":{}}]}`, "whole-batch", "INVALID_ARGUMENT"}, + {`{"items":[{"idempotency_key":"a","task":{"payload":"` + strings.Repeat("x", 512<<10) + `"}}]}`, "", "BODY_TOO_LARGE"}, + } { + s := &batchAuthStore{} + h := New("", nil, service.New(s)).Handler + r := httptest.NewRequest("POST", "/v1/tasks:batch", strings.NewReader(test.body)) + r.Header.Set("Content-Type", "application/json") + r.Header.Set("Authorization", batchHTTPToken(t)) + if test.key != "" { + r.Header.Set("Idempotency-Key", test.key) + } + w := httptest.NewRecorder() + h.ServeHTTP(w, r) + var got struct { + Error struct { + Code string `json:"code"` + } `json:"error"` + } + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || got.Error.Code != test.code || s.authCalls != 0 { + t.Fatalf("got=%s auth=%d want=%s", w.Body.String(), s.authCalls, test.code) + } + } +} + +func TestBatchHTTPAuthenticatesAllInvalidItems(t *testing.T) { + s := &batchAuthStore{} + r := httptest.NewRequest("POST", "/v1/tasks:batch", strings.NewReader(`{"items":[{"idempotency_key":"a","task":{}}]}`)) + r.Header.Set("Content-Type", "application/json") + r.Header.Set("Authorization", batchHTTPToken(t)) + w := httptest.NewRecorder() + New("", nil, service.New(s)).Handler.ServeHTTP(w, r) + if w.Code != http.StatusUnauthorized || s.authCalls != 1 { + t.Fatal(w.Body.String(), s.authCalls) + } +} diff --git a/internal/transport/httpserver/tasks.go b/internal/transport/httpserver/tasks.go index df73a38..8b7cc9e 100644 --- a/internal/transport/httpserver/tasks.go +++ b/internal/transport/httpserver/tasks.go @@ -27,6 +27,17 @@ import ( type endpoint func(*http.Request, auth.Token) (any, int, error) func registerTasks(mux *http.ServeMux, s *service.Service) { + mux.HandleFunc("POST /v1/tasks:batch", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + if len(r.Header.Values("Idempotency-Key")) != 0 { + return nil, 0, fault.Invalid("batch requires per-item idempotency_key, not an Idempotency-Key header") + } + var in service.BatchSubmitInput + if err := decodeBody(r, &in); err != nil { + return nil, 0, err + } + out, err := s.SubmitBatch(r.Context(), t, in) + return out, http.StatusOK, err + })) mux.HandleFunc("POST /v1/tasks", wrap(func(r *http.Request, t auth.Token) (any, int, error) { var in service.SubmitInput if err := decodeBody(r, &in); err != nil { diff --git a/pulse/service/batch.go b/pulse/service/batch.go new file mode 100644 index 0000000..c2fe3d4 --- /dev/null +++ b/pulse/service/batch.go @@ -0,0 +1,127 @@ +package service + +import ( + "context" + "errors" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +// MaxBatchTasks 限制一次请求的独立任务数；HTTP 总请求仍限制为512 KiB。 +const MaxBatchTasks = 32 + +type BatchSubmitInput struct { + // Items 按输入顺序分别受理，不保证整批原子性。 + Items []BatchSubmitItem `json:"items"` +} + +type BatchSubmitItem struct { + // IdempotencyKey 必须由调用方持久化，与单项提交共用命名空间内的幂等身份。 + IdempotencyKey string `json:"idempotency_key"` + // Task 与单项受理完全相同的输入，重报时保留原值。 + Task SubmitInput `json:"task"` +} + +type BatchItemStatus string + +const ( + BatchAccepted BatchItemStatus = "ACCEPTED" + BatchRejected BatchItemStatus = "REJECTED" + BatchUnknown BatchItemStatus = "UNKNOWN" + BatchNotAttempted BatchItemStatus = "NOT_ATTEMPTED" +) + +// BatchTaskReceipt 是原始受理回执的精简视图，避免重复返回大参数。 +type BatchTaskReceipt struct { + // TaskID 已持久化的任务标识，业务结果需另行查询或订阅。 + TaskID core.ID `json:"task_id"` + // QueueID 原始回执所属的队列标识。 + QueueID core.ID `json:"queue_id"` + // Revision 原始受理版本，十进制字符串，不代表当前任务版本。 + Revision uint64 `json:"revision,string"` + // CreatedAt 原始受理时间，UTC微秒精度。 + CreatedAt time.Time `json:"created_at"` + // Deduplicated 本项是否返回已有幂等回执。 + Deduplicated bool `json:"deduplicated"` +} + +type BatchItemError struct { + // Code 稳定错误码；UNKNOWN不能被视为确认回滚。 + Code string `json:"code"` + // Message 对外说明，不包含数据库错误或任务输入。 + Message string `json:"message"` + // RetryAfterMS 建议最小退避毫秒数，省略为未指定，不代表可更换幂等键。 + RetryAfterMS uint32 `json:"retry_after_ms,omitempty"` +} + +type BatchSubmitItemResult struct { + // Index 本次请求内从0开始的索引，结果始终与输入同序且数量相同。 + Index int `json:"index"` + // Status 区分已受理、明确拒绝、结果未知及本次未尝试。 + Status BatchItemStatus `json:"status"` + // Receipt 仅ACCEPTED包含原始受理回执。 + Receipt *BatchTaskReceipt `json:"receipt,omitempty"` + // Error 仅非ACCEPTED包含稳定错误及退避提示。 + Error *BatchItemError `json:"error,omitempty"` +} + +type BatchSubmitResult struct { + // Items 每项独立结果，HTTP成功不能替代逐项检查。 + Items []BatchSubmitItemResult `json:"items"` +} + +// SubmitBatch 逐项重用单项事务；不建立整批幂等身份或跨项锁。 +func (s *Service) SubmitBatch(ctx context.Context, token auth.Token, in BatchSubmitInput) (BatchSubmitResult, error) { + if len(in.Items) < 1 || len(in.Items) > MaxBatchTasks { + return BatchSubmitResult{}, fault.Invalid("items must contain 1 to 32 tasks") + } + keys := make(map[string]struct{}, len(in.Items)) + for _, item := range in.Items { + if err := ValidateKey(item.IdempotencyKey); err != nil { + return BatchSubmitResult{}, err + } + if _, exists := keys[item.IdempotencyKey]; exists { + return BatchSubmitResult{}, fault.Invalid("item idempotency keys must be distinct within a batch") + } + keys[item.IdempotencyKey] = struct{}{} + } + // 即使全部任务参数无效，也必须先验证凭据和动作权限；每项写入会再次认证。 + if err := s.store.AuthorizeSubmit(ctx, token); err != nil { + return BatchSubmitResult{}, err + } + out := BatchSubmitResult{Items: make([]BatchSubmitItemResult, len(in.Items))} + stopped := false + for i, item := range in.Items { + result := BatchSubmitItemResult{Index: i} + if stopped || ctx.Err() != nil { + result.Status = BatchNotAttempted + result.Error = &BatchItemError{Code: "BATCH_STOPPED", Message: "this item was not attempted in this request; retry using the original identity", RetryAfterMS: 1000} + } else { + accepted, err := s.Submit(ctx, token, item.IdempotencyKey, item.Task) + if err == nil { + result.Status = BatchAccepted + result.Receipt = &BatchTaskReceipt{TaskID: accepted.Task.ID, QueueID: accepted.Task.QueueID, Revision: accepted.Task.Revision, CreatedAt: accepted.Task.CreatedAt, Deduplicated: accepted.Deduplicated} + } else { + var public *fault.Error + if !errors.As(err, &public) || public.Code == "UNAVAILABLE" { + // 驱动超时、COMMIT响应丢失等不能证明未提交，停止继续占用数据库。 + public = fault.Unavailable + result.Status, stopped = BatchUnknown, true + } else { + result.Status = BatchRejected + stopped = public.Code == "UNAUTHENTICATED" + } + result.Error = &BatchItemError{Code: public.Code, Message: public.Message} + switch public.Code { + case "QUEUE_FULL", "APPLICATION_FULL", "UNAVAILABLE": + result.Error.RetryAfterMS = 1000 + } + } + } + out.Items[i] = result + } + return out, nil +} diff --git a/pulse/service/batch_test.go b/pulse/service/batch_test.go new file mode 100644 index 0000000..14b13b1 --- /dev/null +++ b/pulse/service/batch_test.go @@ -0,0 +1,128 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "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 batchStore struct { + Store + authErr error + authCalls, calls int + submit func(int, string, SubmitInput) (SubmitResult, error) +} + +func (s *batchStore) AuthorizeSubmit(context.Context, auth.Token) error { + s.authCalls++ + return s.authErr +} +func (s *batchStore) Submit(_ context.Context, _ auth.Token, key string, in SubmitInput) (SubmitResult, error) { + s.calls++ + return s.submit(s.calls, key, in) +} +func batchInput(n int) BatchSubmitInput { + in := BatchSubmitInput{} + for i := 0; i < n; i++ { + in.Items = append(in.Items, BatchSubmitItem{IdempotencyKey: string(rune('a' + i)), Task: SubmitInput{QueueID: core.ID(strings.Repeat("1", 32)), Handler: "file.convert", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}}) + } + return in +} +func acceptedBatchTask(in SubmitInput) SubmitResult { + return SubmitResult{Task: TaskView{ID: core.ID(strings.Repeat("2", 32)), QueueID: in.QueueID, Revision: 1, CreatedAt: time.Now().UTC()}} +} + +func TestBatchEnvelopeBeforeWritesAndAuthorization(t *testing.T) { + for _, n := range []int{0, 33} { + s := &batchStore{} + if _, err := New(s).SubmitBatch(context.Background(), auth.Token{}, batchInput(n)); fault.Code(err) != "INVALID_ARGUMENT" || s.authCalls != 0 || s.calls != 0 { + t.Fatal(n, err) + } + } + for _, key := range []string{"", "a b", strings.Repeat("a", 129), "a"} { + in := batchInput(2) + in.Items[1].IdempotencyKey = key + s := &batchStore{} + if _, err := New(s).SubmitBatch(context.Background(), auth.Token{}, in); fault.Code(err) != "INVALID_ARGUMENT" || s.calls != 0 { + t.Fatal(key, err) + } + } + for _, denied := range []error{fault.Unauthorized, fault.Forbidden, errors.New("database private detail")} { + s := &batchStore{authErr: denied} + in := batchInput(1) + in.Items[0].Task.Handler = "" + if _, err := New(s).SubmitBatch(context.Background(), auth.Token{}, in); err != denied || s.calls != 0 || s.authCalls != 1 { + t.Fatal(err) + } + } +} + +func TestBatchPartialResultsAndIndependentNormalization(t *testing.T) { + in := batchInput(5) + in.Items[1].Task.Priority = 10 + s := &batchStore{submit: func(n int, key string, in SubmitInput) (SubmitResult, error) { + if in.TaskTimeoutMS == 0 || in.Retry.MaxAttempts != 1 || string(in.Payload) != `{"n":9007199254740993}` { + t.Fatal("normalization or precision lost") + } + switch key { + case "c": + return SubmitResult{}, fault.QueueFull + case "d": + return SubmitResult{}, fault.IdempotencyConflict + } + out := acceptedBatchTask(in) + out.Deduplicated = key == "e" + return out, nil + }} + out, err := New(s).SubmitBatch(context.Background(), auth.Token{}, in) + if err != nil || len(out.Items) != 5 || s.calls != 4 { + t.Fatal(out, err) + } + for i, want := range []BatchItemStatus{BatchAccepted, BatchRejected, BatchRejected, BatchRejected, BatchAccepted} { + if out.Items[i].Status != want || out.Items[i].Index != i { + t.Fatal(out) + } + } + if out.Items[1].Error.Code != "INVALID_ARGUMENT" || out.Items[2].Error.RetryAfterMS != 1000 || !out.Items[4].Receipt.Deduplicated { + t.Fatal(out) + } +} + +func TestBatchUnknownAndCancellationNeverClaimRollback(t *testing.T) { + for _, fail := range []error{errors.New("secret database error"), context.DeadlineExceeded, fault.Unavailable, fault.Unauthorized} { + s := &batchStore{submit: func(n int, _ string, in SubmitInput) (SubmitResult, error) { + if n == 1 { + return acceptedBatchTask(in), nil + } + return SubmitResult{}, fail + }} + out, err := New(s).SubmitBatch(context.Background(), auth.Token{}, batchInput(4)) + if err != nil || s.calls != 2 || out.Items[0].Status != BatchAccepted || out.Items[2].Status != BatchNotAttempted || out.Items[3].Status != BatchNotAttempted { + t.Fatal(out, err) + } + want := BatchUnknown + if fail == fault.Unauthorized { + want = BatchRejected + } + if out.Items[1].Status != want || strings.Contains(out.Items[1].Error.Message, "secret") { + t.Fatal(out) + } + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s := &batchStore{submit: func(_ int, _ string, in SubmitInput) (SubmitResult, error) { + cancel() + return acceptedBatchTask(in), nil + }} + out, err := New(s).SubmitBatch(ctx, auth.Token{}, batchInput(3)) + if err != nil || s.calls != 1 || out.Items[0].Status != BatchAccepted || out.Items[1].Status != BatchNotAttempted { + t.Fatal(out, err) + } +} diff --git a/pulse/service/types.go b/pulse/service/types.go index d4b0a61..bc4702b 100644 --- a/pulse/service/types.go +++ b/pulse/service/types.go @@ -216,6 +216,7 @@ type QueueResult struct { // Store 实现命名空间隔离、事务授权与持久化，不在事务外产生业务副作用。 type Store interface { + AuthorizeSubmit(context.Context, auth.Token) error Submit(context.Context, auth.Token, string, SubmitInput) (SubmitResult, error) GetTask(context.Context, auth.Token, core.ID) (TaskView, error) ListTasks(context.Context, auth.Token, ListInput) (TaskPage, error) diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 9e490d7..e57dd03 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -52,6 +52,7 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - (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 '^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) @@ -61,6 +62,7 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/schedules) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/cancellation) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/resources) +(cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/batch) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/keyed-concurrency) (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/priority) (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/subscriptions) diff --git a/sdk/go/batch.go b/sdk/go/batch.go new file mode 100644 index 0000000..bf4b798 --- /dev/null +++ b/sdk/go/batch.go @@ -0,0 +1,121 @@ +package orbit + +import ( + "context" + "errors" + "time" +) + +// MaxBatchTasks 是单次批量受理的项数上限，总请求还受512 KiB限制。 +const MaxBatchTasks = 32 + +type SubmitTasksRequest struct { + // Items 1至32项独立任务，总请求不超过512 KiB，不保证整批原子性。 + Items []SubmitTasksItem `json:"items"` +} + +type SubmitTasksItem struct { + // IdempotencyKey 调用方持久化的稳定键，与SubmitTask共用幂等身份；批内不得重复。 + IdempotencyKey string `json:"idempotency_key"` + // Task 原始任务参数，重报必须保留原值。 + Task SubmitTaskRequest `json:"task"` +} + +type BatchItemStatus string + +const ( + BatchAccepted BatchItemStatus = "ACCEPTED" + BatchRejected BatchItemStatus = "REJECTED" + BatchUnknown BatchItemStatus = "UNKNOWN" + BatchNotAttempted BatchItemStatus = "NOT_ATTEMPTED" +) + +type BatchTaskReceipt struct { + // TaskID 已受理任务标识，实际结果需要查询或订阅。 + TaskID string `json:"task_id"` + // QueueID 任务所属队列。 + QueueID string `json:"queue_id"` + // Revision 原始受理版本，JSON为十进制字符串，不代表当前版本。 + Revision uint64 `json:"revision,string"` + // CreatedAt 原始受理时间，UTC微秒精度。 + CreatedAt time.Time `json:"created_at"` + // Deduplicated 是否复用相同身份的已有回执。 + Deduplicated bool `json:"deduplicated"` +} + +type BatchItemError struct { + // Code 稳定错误码；UNKNOWN结果使用UNAVAILABLE，不证明回滚。 + Code string `json:"code"` + // Message 不包含数据库错误或输入的对外说明。 + Message string `json:"message"` + // RetryAfterMS 建议最小退避毫秒数，零为未指定，SDK不自动重试。 + RetryAfterMS uint32 `json:"retry_after_ms,omitempty"` +} + +type SubmitTasksItemResult struct { + // Index 本次请求内从0开始的索引，与Items输入同序。 + Index int `json:"index"` + // Status 每项独立状态；HTTP成功不表示全部任务已受理。 + Status BatchItemStatus `json:"status"` + // Receipt 仅ACCEPTED存在，省略大参数以保证回执大小有界。 + Receipt *BatchTaskReceipt `json:"receipt,omitempty"` + // Error 仅非ACCEPTED存在，调用方需分别处理明确拒绝、未知及未尝试。 + Error *BatchItemError `json:"error,omitempty"` +} + +type SubmitTasksResult struct { + // Items 与输入数量和顺序一致的逐项结果。 + Items []SubmitTasksItemResult `json:"items"` +} + +// SubmitTasks 逐项提交独立任务。返回error时整批可能已部分或全部受理，必须保留原键及参数核对。 +// error为nil时仍需检查每项Status。此方法不会自动重试、拆批、生成新键或启动并发请求。 +func (c *Client) SubmitTasks(ctx context.Context, in SubmitTasksRequest) (SubmitTasksResult, error) { + if len(in.Items) < 1 || len(in.Items) > MaxBatchTasks { + return SubmitTasksResult{}, errors.New("items must contain 1 to 32 tasks") + } + keys := make(map[string]struct{}, len(in.Items)) + for _, item := range in.Items { + if err := validateIdempotencyKey(item.IdempotencyKey); err != nil { + return SubmitTasksResult{}, err + } + if _, exists := keys[item.IdempotencyKey]; exists { + return SubmitTasksResult{}, errors.New("item idempotency keys must be distinct within a batch") + } + keys[item.IdempotencyKey] = struct{}{} + } + var out SubmitTasksResult + if err := c.callJSON(ctx, "POST", "/v1/tasks:batch", "", in, &out); err != nil { + return SubmitTasksResult{}, err + } + // 拒绝不完整或错误关联的回执，避免调用方将截断的成功响应误认为全部已受理。 + if len(out.Items) != len(in.Items) { + return SubmitTasksResult{}, errors.New("invalid Orbit batch result count") + } + for i, item := range out.Items { + valid := item.Index == i + switch item.Status { + case BatchAccepted: + r := item.Receipt + valid = valid && item.Error == nil && r != nil && validID(r.TaskID) && r.QueueID == in.Items[i].Task.QueueID && validID(r.QueueID) && r.Revision > 0 && !r.CreatedAt.IsZero() + case BatchRejected, BatchUnknown, BatchNotAttempted: + e := item.Error + valid = valid && item.Receipt == nil && e != nil && len(e.Code) > 0 && len(e.Code) <= 64 && len(e.Message) > 0 && len(e.Message) <= 1024 && e.RetryAfterMS <= 3600000 + if valid && item.Status == BatchUnknown { + valid = e.Code == "UNAVAILABLE" + } + if valid && item.Status == BatchNotAttempted { + valid = e.Code == "BATCH_STOPPED" + } + if valid && item.Status == BatchRejected { + valid = e.Code != "UNAVAILABLE" && e.Code != "BATCH_STOPPED" + } + default: + valid = false + } + if !valid { + return SubmitTasksResult{}, errors.New("invalid Orbit batch item result; retry using original identities") + } + } + return out, nil +} diff --git a/sdk/go/batch_integration_test.go b/sdk/go/batch_integration_test.go new file mode 100644 index 0000000..379f0b1 --- /dev/null +++ b/sdk/go/batch_integration_test.go @@ -0,0 +1,107 @@ +package orbit + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestIntegrationTaskBatchSDK(t *testing.T) { + api, _, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + created, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "batch." + workerTestID(t), MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2}) + if err != nil { + t.Fatal(err) + } + q := created.Queue + if _, err := api.ConfigureQueue(ctx, q.ID, workerTestID(t), ConfigureQueueRequest{ExpectedRevision: q.Revision, Reason: "batch acceptance regression", Paused: true, MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2}); err != nil { + t.Fatal(err) + } + input := SubmitTaskRequest{QueueID: q.ID, Handler: "file.convert", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)} + in := SubmitTasksRequest{Items: []SubmitTasksItem{{IdempotencyKey: workerTestID(t), Task: input}, {IdempotencyKey: workerTestID(t), Task: input}, {IdempotencyKey: workerTestID(t), Task: input}}} + // 真实HTTP代理先完整读到数据库受理回执，再强制断开客户端连接，丢弃整批响应。 + target, _ := url.Parse(api.baseURL) + proxy := httputil.NewSingleHostReverseProxy(target) + var dropped atomic.Bool + var calls atomic.Int32 + proxy.ModifyResponse = func(r *http.Response) error { + if r.Request.URL.Path != "/v1/tasks:batch" { + return nil + } + calls.Add(1) + if dropped.CompareAndSwap(false, true) { + raw, err := io.ReadAll(r.Body) + r.Body.Close() + if err != nil { + return err + } + if r.StatusCode != 200 || !strings.Contains(string(raw), `"ACCEPTED"`) || !strings.Contains(string(raw), `"QUEUE_FULL"`) { + t.Error("unexpected pre-drop server result", string(raw)) + } + return io.ErrUnexpectedEOF + } + return nil + } + proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) { + conn, _, hijackErr := w.(http.Hijacker).Hijack() + if hijackErr == nil { + conn.Close() + } + } + server := httptest.NewServer(proxy) + defer server.Close() + client, err := NewClient(server.URL, Options{Token: api.token}) + if err != nil { + t.Fatal(err) + } + if _, err := client.SubmitTasks(ctx, in); err == nil || !dropped.Load() || calls.Load() != 1 { + t.Fatal("response loss not surfaced without silent retry", err) + } + // 原连接已关闭；用原键和原内容核对，前两项去重，满额项保持明确拒绝。 + out, err := client.SubmitTasks(ctx, in) + if err != nil || len(out.Items) != 3 { + t.Fatal(out, err) + } + for i := 0; i < 2; i++ { + if out.Items[i].Status != BatchAccepted || !out.Items[i].Receipt.Deduplicated { + t.Fatal("lost response duplicated task", out) + } + } + if out.Items[2].Status != BatchRejected || out.Items[2].Error.Code != "QUEUE_FULL" || out.Items[2].Error.RetryAfterMS != 1000 { + t.Fatal(out) + } + page, err := api.ListTasks(ctx, ListTasksRequest{QueueID: q.ID}) + if err != nil || len(page.Items) != 2 { + t.Fatal("duplicate durable rows", page, err) + } + single, err := api.SubmitTask(ctx, in.Items[0].IdempotencyKey, input) + if err != nil || !single.Deduplicated || single.Task.ID != out.Items[0].Receipt.TaskID || string(single.Task.Payload) != `{"n":9007199254740993}` { + t.Fatal("batch/single mismatch", err) + } + if _, err := api.CancelTask(ctx, single.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: single.Task.Revision, Reason: "free slot for partial retry"}); err != nil { + t.Fatal(err) + } + suffix, err := api.SubmitTasks(ctx, SubmitTasksRequest{Items: in.Items[2:]}) + if err != nil || suffix.Items[0].Status != BatchAccepted || suffix.Items[0].Receipt.Deduplicated || suffix.Items[0].Index != 0 { + t.Fatal(suffix, err) + } + for _, id := range []string{out.Items[1].Receipt.TaskID, suffix.Items[0].Receipt.TaskID} { + current, err := api.GetTask(ctx, id) + if err != nil { + t.Fatal(err) + } + if _, err := api.CancelTask(ctx, id, workerTestID(t), CancelTaskRequest{ExpectedRevision: current.Revision, Reason: "finish batch regression"}); err != nil { + t.Fatal(err) + } + } + t.Log("lost real HTTP response: two committed tasks recovered by identity, no duplicates; backpressured suffix independently accepted after capacity released") +} diff --git a/sdk/go/batch_test.go b/sdk/go/batch_test.go new file mode 100644 index 0000000..285adc1 --- /dev/null +++ b/sdk/go/batch_test.go @@ -0,0 +1,75 @@ +package orbit + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" +) + +func TestBatchSDKWireValidationAndNoAutomaticRetry(t *testing.T) { + id := strings.Repeat("1", 32) + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.URL.Path != "/v1/tasks:batch" || r.Method != "POST" || r.Header.Get("Authorization") != "Bearer test-token" || r.Header.Get("Idempotency-Key") != "" { + t.Error("invalid batch transport") + } + raw, _ := io.ReadAll(r.Body) + if !strings.Contains(string(raw), `"idempotency_key":"one"`) || !strings.Contains(string(raw), `9007199254740993`) { + t.Error("identity/precision lost") + } + if calls.Load() == 1 { + w.WriteHeader(503) + fmt.Fprintf(w, `{"request_id":%q,"error":{"code":"UNAVAILABLE","message":"retry original identities"}}`, id) + return + } + fmt.Fprintf(w, `{"request_id":%q,"data":{"items":[{"index":0,"status":"ACCEPTED","receipt":{"task_id":%q,"queue_id":%q,"revision":"9007199254740993","created_at":"2026-09-17T00:00:00Z","deduplicated":true}}]}}`, id, id, id) + })) + defer srv.Close() + c, _ := NewClient(srv.URL, Options{Token: "test-token"}) + in := SubmitTasksRequest{Items: []SubmitTasksItem{{IdempotencyKey: "one", Task: SubmitTaskRequest{QueueID: id, Payload: json.RawMessage(`{"n":9007199254740993}`)}}}} + if _, err := c.SubmitTasks(context.Background(), in); err == nil || calls.Load() != 1 { + t.Fatal("request silently retried", err) + } + out, err := c.SubmitTasks(context.Background(), in) + if err != nil || out.Items[0].Receipt.Revision != 9007199254740993 || !out.Items[0].Receipt.Deduplicated { + t.Fatal(out, err) + } + for _, bad := range []SubmitTasksRequest{{}, {Items: make([]SubmitTasksItem, 33)}, {Items: []SubmitTasksItem{in.Items[0], in.Items[0]}}, {Items: []SubmitTasksItem{{IdempotencyKey: " "}}}} { + if _, err := c.SubmitTasks(context.Background(), bad); err == nil { + t.Fatal("accepted invalid batch") + } + } + in.Items[0].Task.Payload = json.RawMessage(`"` + strings.Repeat("x", 512<<10) + `"`) + if _, err := c.SubmitTasks(context.Background(), in); err == nil || calls.Load() != 2 { + t.Fatal("unbounded batch reached network") + } +} + +func TestBatchSDKRejectsIncompleteAndContradictoryReceipts(t *testing.T) { + id := strings.Repeat("1", 32) + for _, data := range []string{ + `{"items":[]}`, + `{"items":[{"index":1,"status":"REJECTED","error":{"code":"INVALID_ARGUMENT","message":"bad"}}]}`, + `{"items":[{"index":0,"status":"UNKNOWN","error":{"code":"INVALID_ARGUMENT","message":"bad"}}]}`, + `{"items":[{"index":0,"status":"REJECTED","error":{"code":"UNAVAILABLE","message":"unknown"}}]}`, + `{"items":[{"index":0,"status":"NOT_ATTEMPTED","error":{"code":"UNAVAILABLE","message":"bad"}}]}`, + `{"items":[{"index":0,"status":"ACCEPTED"}]}`, + `{"items":[{"index":0,"status":"REJECTED","receipt":{},"error":{"code":"QUEUE_FULL","message":"full"}}]}`, + `{"items":[{"index":0,"status":"NEW_STATUS"}]}`, + } { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, `{"request_id":%q,"data":%s}`, id, data) })) + c, _ := NewClient(srv.URL, Options{Token: "test-token"}) + out, err := c.SubmitTasks(context.Background(), SubmitTasksRequest{Items: []SubmitTasksItem{{IdempotencyKey: "one", Task: SubmitTaskRequest{QueueID: id}}}}) + srv.Close() + if err == nil || len(out.Items) != 0 { + t.Fatal("malformed response exposed as success", out, data) + } + } +} diff --git a/sdk/go/doc.go b/sdk/go/doc.go index f10d435..1262e07 100644 --- a/sdk/go/doc.go +++ b/sdk/go/doc.go @@ -1,6 +1,7 @@ // Package orbit 是 Orbit 的独立 Go SDK，提供健康、任务、队列、计划及结果订阅。 // WorkerConnection 是底层接口；NewWorker/Run 自动管理处理器、续租、恢复与排空。 // EventConnection 是结果交付协议；EventConsumer 自动管理心跳、重连与确认，业务端负责事务去重。 +// SubmitTasks 提供有界批量受理，每项沿用独立稳定身份并返回可核对的受理状态。 // 流程 SDK 等后续能力按阶段继续开发。 // SDK 不导入服务端模型、迁移或数据库依赖。 package orbit diff --git a/sdk/go/examples/batch/main.go b/sdk/go/examples/batch/main.go new file mode 100644 index 0000000..ac7f5d4 --- /dev/null +++ b/sdk/go/examples/batch/main.go @@ -0,0 +1,122 @@ +// batch 演示批量部分受理、稳定身份核对和释放配额后仅补交失败项。 +package main + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "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 token == "" && os.Getenv("ORBIT_CREDENTIALS_FILE") != "" { + raw, err := os.ReadFile(os.Getenv("ORBIT_CREDENTIALS_FILE")) + if err != nil { + return fmt.Errorf("read credentials: %w", err) + } + var c struct { + Token string `json:"token"` + } + if json.Unmarshal(raw, &c) != nil { + return fmt.Errorf("invalid credentials file") + } + 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(), 20*time.Second) + defer cancel() + var random [16]byte + if _, err := rand.Read(random[:]); err != nil { + return err + } + key := "batch-example-" + hex.EncodeToString(random[:]) + q, err := api.CreateQueue(ctx, key+"-queue", orbit.CreateQueueRequest{Name: key, MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2}) + if err != nil { + return err + } + if _, err := api.ConfigureQueue(ctx, q.Queue.ID, key+"-pause", orbit.ConfigureQueueRequest{ExpectedRevision: q.Queue.Revision, Reason: "isolated batch acceptance example", Paused: true, MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2}); err != nil { + return err + } + request := orbit.SubmitTasksRequest{} + // 演示运行生成独立身份；生产必须先将每项键与完整参数持久化到业务库，再发送。 + for i := 0; i < 3; i++ { + payload := json.RawMessage(fmt.Sprintf(`{"source_ref":"objects/input-%d.txt","sequence":9007199254740993}`, i)) + request.Items = append(request.Items, orbit.SubmitTasksItem{IdempotencyKey: fmt.Sprintf("%s-%d", key, i), Task: orbit.SubmitTaskRequest{QueueID: q.Queue.ID, Handler: "file.convert", HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, TaskTimeoutMS: 60000}}) + } + first, err := api.SubmitTasks(ctx, request) + if err != nil { + return err + } + if first.Items[0].Status != orbit.BatchAccepted || first.Items[1].Status != orbit.BatchAccepted || first.Items[2].Status != orbit.BatchRejected || first.Items[2].Error.Code != "QUEUE_FULL" { + return fmt.Errorf("unexpected partial acceptance") + } + fmt.Println("accepted=2 rejected=1 code=QUEUE_FULL") + replay, err := api.SubmitTasks(ctx, request) + if err != nil { + return err + } + for i := 0; i < 2; i++ { + if replay.Items[i].Status != orbit.BatchAccepted || !replay.Items[i].Receipt.Deduplicated || replay.Items[i].Receipt.TaskID != first.Items[i].Receipt.TaskID { + return fmt.Errorf("batch replay changed identity") + } + } + if replay.Items[2].Status != orbit.BatchRejected || replay.Items[2].Error.Code != "QUEUE_FULL" { + return fmt.Errorf("backpressure changed unexpectedly") + } + fmt.Println("replayed=2 same_task_ids=true") + cancelTask := func(id string) error { + current, err := api.GetTask(ctx, id) + if err != nil { + return err + } + out, err := api.CancelTask(ctx, id, key+"-cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: current.Revision, Reason: "finish batch example"}) + if err != nil { + return err + } + if out.Task.State != orbit.TaskCancelled { + return fmt.Errorf("expected confirmed cancellation") + } + return nil + } + if err := cancelTask(first.Items[0].Receipt.TaskID); err != nil { + return err + } + // 子批次索引重新从0开始；键和参数仍来自最初被拒项，不重新生成。 + retry, err := api.SubmitTasks(ctx, orbit.SubmitTasksRequest{Items: request.Items[2:]}) + if err != nil { + return err + } + if retry.Items[0].Status != orbit.BatchAccepted || retry.Items[0].Receipt.Deduplicated { + return fmt.Errorf("failed item was not independently accepted") + } + fmt.Println("retried_items=1 same_identity=true accepted=true") + for _, id := range []string{first.Items[1].Receipt.TaskID, retry.Items[0].Receipt.TaskID} { + if err := cancelTask(id); err != nil { + return err + } + } + final, err := api.GetQueue(ctx, q.Queue.ID) + if err != nil { + return err + } + if final.UnfinishedTasks != 0 { + return fmt.Errorf("unfinished tasks remain") + } + fmt.Printf("queue=%s paused=%t unfinished=%d\n", final.ID, final.Paused, final.UnfinishedTasks) + return nil +} diff --git a/sdk/go/tasks.go b/sdk/go/tasks.go index 454bec1..f62bb63 100644 --- a/sdk/go/tasks.go +++ b/sdk/go/tasks.go @@ -93,19 +93,31 @@ func validID(id string) bool { return true } func (c *Client) call(ctx context.Context, method, path, key string, in, out any) error { + if method != "GET" { + if err := validateIdempotencyKey(key); err != nil { + return err + } + } + return c.callJSON(ctx, method, path, key, in, out) +} + +func validateIdempotencyKey(key string) error { + if len(key) < 1 || len(key) > 128 { + return errors.New("a stable Idempotency-Key of 1 to 128 ASCII characters is required") + } + for _, ch := range key { + if ch < 33 || ch > 126 { + return errors.New("Idempotency-Key must not contain spaces or control characters") + } + } + return nil +} + +// callJSON 仅执行有界HTTP交换；写入调用者负责校验整项或批量内的独立幂等键。 +func (c *Client) callJSON(ctx context.Context, method, path, key string, in, out any) error { if c.token == "" { return errors.New("Orbit token is required for business APIs") } - if method != "GET" { - if len(key) < 1 || len(key) > 128 { - return errors.New("a stable Idempotency-Key of 1 to 128 ASCII characters is required") - } - for _, ch := range key { - if ch < 33 || ch > 126 { - return errors.New("Idempotency-Key must not contain spaces or control characters") - } - } - } var body []byte var err error if in != nil {