From 2b8213a4f707c7a09ddce30f47aaa53494e122b2 Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 22:40:42 +0800 Subject: [PATCH] feat: add scoped task source tracing and SDK examples --- AGENTS.md | 4 +- README.md | 2 +- docs/01-functional-spec.md | 2 +- docs/07-data-model.md | 4 + docs/08-development-plan.md | 17 +- docs/api/README.md | 1 + docs/api/schedules.md | 4 + docs/api/task-source.md | 131 ++++++++++++ docs/api/tasks.md | 2 + docs/examples/README.md | 9 +- docs/examples/recovery.md | 5 +- docs/reports/phase-3-task-source.md | 76 +++++++ docs/sdk/README.md | 4 +- ...tion_resources_upgrade_integration_test.go | 2 +- internal/storage/mysql/task_source.go | 90 ++++++++ .../mysql/task_source_integration_test.go | 200 ++++++++++++++++++ internal/transport/httpserver/history.go | 7 + pulse/service/source.go | 43 ++++ pulse/service/types.go | 1 + sdk/go/examples/recovery/main.go | 8 + sdk/go/examples/schedules/main.go | 8 + sdk/go/examples/tasks/main.go | 8 + sdk/go/recovery_integration_test.go | 8 + sdk/go/schedule_integration_test.go | 4 + sdk/go/task_source.go | 68 ++++++ sdk/go/task_source_test.go | 61 ++++++ 26 files changed, 760 insertions(+), 9 deletions(-) create mode 100644 docs/api/task-source.md create mode 100644 docs/reports/phase-3-task-source.md create mode 100644 internal/storage/mysql/task_source.go create mode 100644 internal/storage/mysql/task_source_integration_test.go create mode 100644 pulse/service/source.go create mode 100644 sdk/go/task_source.go create mode 100644 sdk/go/task_source_test.go diff --git a/AGENTS.md b/AGENTS.md index 07df215..7094503 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,7 +14,7 @@ - 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 - 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 -- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，任务查询已支持业务键及创建时间，P3仍需补齐任务详情来源追踪和事务Outbox接入；不能因订阅完成而跳过原功能范围。 +- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，任务查询已支持业务键、创建时间及直接来源（普通/人工重试/计划固定版本），P3仍需补齐事务Outbox接入；不能因订阅完成而跳过原功能范围。 - 批量受理最多32项、512 KiB；每项复用单项身份与事务，未知结果必须原键核对。任务提交按认证→应用额度→提交幂等身份→队列持锁，禁止恢复为先抢提交唯一键后等待额度锁；原回执仍须先于配额上限检查返回。 @@ -22,3 +22,5 @@ - tasks:review、tasks:retry、tasks:audit:read是显式管理权限，普通提交/Worker凭据不自动获得。核查结果流锁最后获取，审计/回执/额度与状态同事务；恢复不得绕过Weave/计划所有权、门控及现有额度限制。 - 任务列表队列必填，business_key精确匹配（包含尾部空格），created_from包含/created_before不包含。v2游标绑定所有筛选，不能因网络或游标错误删筛选降级；旧游标仅兼容旧查询。四种固定索引覆盖状态/业务键组合，禁止全量取回后内存过滤。业务键仅Go空字符串映射NULL。 + +- 任务来源通过独立只读路由查询，tasks:read仅授予当前任务的有限来源引用；计划定义仍需schedules:read。显式核对命名空间、队列、计划、触发与固定版本，悬空/错误/未实现所有者不得默认为DIRECT；不依赖外键，也不向派发路径加入来源查询。 diff --git a/README.md b/README.md index c6d56a9..82dc3c6 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) 已提供带证据的管理接口、关联新任务、审计查询及运行示例；[业务键与时间范围查询](docs/api/task-queries.md) 已提供过滤、分页与SDK场景；剩余 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场景；[任务来源查询](docs/api/task-source.md) 已提供人工重试父任务、计划触发记录与固定版本；剩余 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 3bb81a1..83d256c 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 具体范围另见流程设计。基础、任务执行、调度、资源控制和可靠订阅已实现；批量受理、独立任务核查/人工重试及业务键/创建时间查询已实现，事务Outbox接入等继续开发，实现状态见持续开发计划。 +> 版本 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 7c78bed..fcdb787 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -174,3 +174,7 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 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)。 + +### 任务直接来源视图 + +[来源查询](api/task-source.md)复用Task.OwnerKind/OwnerID/RetryOf、Schedule、ScheduleVersion与ScheduleOccurrence，沿已有(namespace_id,task_id)唯一索引及计划版本键定位，不增加表、持久化字段、索引或迁移。只读响应模型的每个字段也有中文注释。任务、计划、触发版本与队列归属在认证事务中显式验证，缺失关联不被默认为普通任务；流程来源随Weave实现。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index e36a5fd..bf67cdd 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回执丢失及完整示例已通过全量回归；业务键/创建时间筛选、SDK与示例已通过全量回归；任务详情来源追踪与事务Outbox接入待完成 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景、COMMIT响应丢失及消费者进程故障回归已完成；批量受理、SDK、接口与部分补交示例已完成全量回归；独立任务核查/人工重试、审计查询、两次真实HTTP回执丢失及完整示例已通过全量回归；业务键/创建时间筛选、SDK与示例已通过全量回归；任务直接来源（普通/人工重试/计划固定版本）及SDK示例已通过全量回归；事务Outbox接入待完成 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -44,11 +44,23 @@ 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)同步交付 | | 任务查询筛选 | 已实现[业务键及创建时间查询](api/task-queries.md)，v2游标绑定全部筛选、旧游标受限兼容、四种索引路径与[完整场景](examples/task-query.md)；[回归报告](reports/phase-3-task-queries.md)同步记录 | -| 任务详情来源追踪 | 原功能要求能查看关联计划；当前TaskView未返回来源计划/触发记录。需补齐任务到计划及触发版本的查询，验证当前权限、同命名空间归属与来源缺失处理；流程来源随Weave接入 | +| 任务详情来源追踪 | 已实现[独立来源接口](api/task-source.md)、SDK和普通/计划/人工恢复场景，返回计划及触发固定版本；同命名空间归属、当前权限和来源缺失拒绝已通过[全量回归](reports/phase-3-task-source.md)，流程来源随Weave接入 | | 业务到平台的事务Outbox接入 | [事务Outbox设计](04-reliability-and-data.md#61-业务系统到平台) 尚缺完整接入示例；需业务变更与待发请求同事务、稳定提交身份、响应不明核对和发送恢复，保持核心SDK无ORM/数据库依赖 | 凭据安全重绑、历史清理、全量监控/容量与MySQL故障切换仍按P6执行；Weave的通用执行门控/内部事件接续随P4–P5落实，不将外部结果订阅当作内部流程引擎。 +### P3 事务Outbox接入的剩余验收 + +以下为下一项实现与测试要求，目前尚未交付： + +1. 独立业务示例module，以业务记录和待发请求同一个MySQL事务提交；核心SDK保持无ORM/数据库依赖，业务表示例也禁止外键，字段提供中文注释。 +2. 同时冻结目标命名空间/队列、原幂等键、完整请求及内容摘要。相同业务操作重报不新建待发项，内容不同必须冲突；凭据原文不能进入Outbox或日志。 +3. 提交、领取与确认分别使用短事务；HTTP在事务外执行。多发送进程通过数据库租约及递增代次协调，旧代次不能覆盖新结果；重启后继续读取持久化待发项。 +4. 平台已受理但响应丢失、业务COMMIT回执未知、发送进程退出，都必须用原身份核对。只确认平台受理，不将受理标记当作业务执行成功。 +5. 退避、发送次数及总发送期限均有上限，补发窗口不超过平台幂等保留期；权限、内容冲突、预算耗尽和不能可靠确认的响应需保留可核查记录，不生成新键绕过。 +6. 真实数据库回归覆盖业务回滚、并发重复、发送竞争、过期租约、迟到确认和原始数字精度；真实HTTP丢响应、MySQL COMMIT响应丢失和发送进程SIGKILL后恢复一并验收。 +7. 提供完整初始化、提交、运行/恢复、状态核对及失败处理命令；接入全量回归，记录实际测试证据后再完成P3范围核对。 + ## 4. SDK 与示例交付清单 @@ -101,6 +113,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P3 批量任务受理回归](reports/phase-3-batch-tasks.md) - [P3 人工核查与重试回归](reports/phase-3-task-recovery.md) - [P3 业务键与时间查询回归](reports/phase-3-task-queries.md) +- [P3 任务直接来源回归](reports/phase-3-task-source.md) - [接口手册](api/README.md) - [SDK 接入](sdk/README.md) - [通用场景示例](examples/README.md) diff --git a/docs/api/README.md b/docs/api/README.md index 7620bcc..d7a7e71 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` 提交、查询、分页与待执行取消 | 已实现，P3扩展业务键与创建时间筛选 | [任务接口](tasks.md) / [查询契约](task-queries.md) | +| `GET /v1/tasks/{task_id}/source` | P3已实现普通、人工重试与计划固定版本来源 | [来源查询](task-source.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/schedules.md b/docs/api/schedules.md index 15f5a25..cf77fde 100644 --- a/docs/api/schedules.md +++ b/docs/api/schedules.md @@ -165,6 +165,8 @@ ListSchedules 需要 queue_id；计划列表和 occurrences 均可用 limit（ FIRED 表示已创建单点任务；COALESCED 表示范围已合并到 task_id；SKIPPED 没有 task_id。它们都不是业务执行结果，继续 GetTask 查询，再通过 [尝试与证据接口](execution-history.md) 核对执行。 +从任务反查计划使用[GetTaskSource](task-source.md)，返回来源计划、触发记录、当时固定版本和合并边界。该只读引用需要tasks:read，完整计划/版本内容仍需schedules:read；发布新版本不改变老任务来源。 + ## 6. SDK 与完整示例 所有方法首参数为 context.Context，写操作需要稳定 key。ListSchedules 的参数为 `(ctx, queueID, SchedulePageRequest)`；ListScheduleOccurrences 为 `(ctx, scheduleID, SchedulePageRequest)`。 @@ -185,7 +187,9 @@ ORBIT_URL=http://127.0.0.1:8080 \ 前置数据库与目录步骤见 [SDK 接入](../sdk/README.md)。源码：[完整调度示例](../../sdk/go/examples/schedules/main.go)。示例创建暂停的每秒计划，恢复后让自动 Worker 返回对象输入快照，查询两次完成，再暂停并排空。计划本身有十秒结束边界，任务与历史保留；重复运行生成独立身份，不调用外部业务系统。 ```text +source_kind=SCHEDULE fixed_version=1 schedule=<计划ID> occurrence=<触发ID> task=<任务ID> version=1 state=SUCCEEDED +source_kind=SCHEDULE fixed_version=1 schedule=<计划ID> occurrence=<触发ID> task=<任务ID> version=1 state=SUCCEEDED paused=true completed_tasks=2 current_version=1 worker_drained=true running=0 diff --git a/docs/api/task-source.md b/docs/api/task-source.md new file mode 100644 index 0000000..adec3d5 --- /dev/null +++ b/docs/api/task-source.md @@ -0,0 +1,131 @@ +# 任务直接来源查询 + +状态：P3已实现服务端、独立SDK及普通任务、计划、人工恢复示例。接口只读，使用已有模型和索引，不增加数据库外键或迁移。验证证据见[回归报告](../reports/phase-3-task-source.md)。 + +## 1. 请求与权限 + +`GET /v1/tasks/{task_id}/source` + +请求头：`Authorization: Bearer `。不需要请求体、Content-Type或Idempotency-Key，不接受任何查询参数。task_id为32位小写十六进制ID；认证决定命名空间，URL不能覆盖它。 + +权限为`tasks:read`及任务所属队列范围。每次请求验证当前凭据；不存在、跨命名空间或不在队列范围的任务均返回404。只获得来源引用，不因此获得计划定义、模板、管理或执行权限；读取完整计划及固定版本仍需`schedules:read`，见[计划手册](schedules.md)。 + +```sh +curl "$ORBIT_URL/v1/tasks/$TASK_ID/source" \ + -H "Authorization: Bearer $ORBIT_TOKEN" +``` + +## 2. 响应字段 + +HTTP 200，使用`request_id`和`data`信封；X-Request-ID与正文一致。下表字段均位于data。 + +| 字段 | 类型 / 必填 | 含义 | +| --- | --- | --- | +| task_id | string / 是 | 已授权查询的任务ID | +| kind | string / 是 | DIRECT、MANUAL_RETRY或SCHEDULE | +| retry_of | string / 仅MANUAL_RETRY | 同命名空间、同队列的直接原任务ID | +| schedule | object / 仅SCHEDULE | 触发该任务的固定关联，字段如下 | +| schedule.schedule_id | string / 是 | 来源计划ID | +| schedule.schedule_version | 十进制字符串 / 是 | 实际触发时的非零uint64版本，发布新版本不改写它 | +| schedule.occurrence_id | string / 是 | 唯一关联该任务的触发记录ID | +| schedule.scheduled_at | UTC RFC3339 / 是 | 原计划时间，最多微秒精度，与实际受理/执行时间分开 | +| schedule.decision | string / 是 | FIRED单点触发或COALESCED合并触发 | +| schedule.covered_until | UTC RFC3339 / 仅COALESCED | 合并范围的排除上界，严格晚于scheduled_at | + +DIRECT表示普通任务受理，包括单项和批量受理；任务内部的自动安全重试仍属于同一个任务，不改变来源。MANUAL_RETRY表示[人工重试](task-recovery.md)创建的新任务，只返回直接父任务，不递归遍历整条重试链。父任务继续保留原历史。 + +SCHEDULE中的版本来自触发记录，不能用计划最新版本替换。COALESCED对应区间`[scheduled_at, covered_until)`；合并范围内由计划定义确定原始时间点，不承诺每个自然时间单位都有一次触发。SKIPPED未创建任务，因此不会成为任务来源。 + +普通受理的完整响应： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{ + "task_id":"11111111111111111111111111111111", + "kind":"DIRECT" + } +} +``` + +人工重试的完整响应： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{ + "task_id":"11111111111111111111111111111111", + "kind":"MANUAL_RETRY", + "retry_of":"22222222222222222222222222222222" + } +} +``` + +计划合并触发的完整响应： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{ + "task_id":"11111111111111111111111111111111", + "kind":"SCHEDULE", + "schedule":{ + "schedule_id":"22222222222222222222222222222222", + "schedule_version":"1", + "occurrence_id":"33333333333333333333333333333333", + "scheduled_at":"2026-09-17T00:00:00Z", + "decision":"COALESCED", + "covered_until":"2026-09-17T00:03:00.000001Z" + } + } +} +``` + +单点触发返回decision=FIRED并省略covered_until。所有不适用的可选字段省略，不用null占位。来源不包含任务payload/result、计划模板、策略、凭据或执行租约秘密。任务当前状态另用GetTask读取。 + +## 3. 关联校验与错误 + +平台采用无外键设计，本接口在认证事务内显式核对命名空间、任务、队列、计划、不可变版本与触发记录。人工重试父任务须为同队列中已失败/取消的独立任务。无所有者但存在触发记录、悬空父任务、缺失版本、错误队列、未知所有者或当前未支持的门控均拒绝，不能把关联损坏降级成DIRECT。 + +| HTTP / code | 处理 | +| --- | --- | +| 400 INVALID_ARGUMENT | 检查task_id；删除不受支持、重复或格式错误的查询参数 | +| 401 UNAUTHENTICATED | 检查凭据、过期、撤销及应用/命名空间状态 | +| 403 PERMISSION_DENIED | 需要tasks:read动作权限 | +| 404 NOT_FOUND | 任务不存在或不在当前命名空间/队列范围；旧版本节点也可能没有本路由 | +| 503 UNAVAILABLE | 数据库或来源关联不能可靠确认；不返回部分来源，不推测DIRECT | + +读取不改变任务状态、版本、额度、尝试或审计。网络错误可以在有限预算内重新读取；SDK不自动重试。持续503应排查数据关联和服务日志，不能通过删除来源字段“修复”，也不能根据来源查询失败重新提交任务。 + +## 4. SDK调用与场景 + +```go +func (c *Client) GetTaskSource(ctx context.Context, taskID string) (TaskSource, error) +``` + +下方为已有客户端与taskID的调用片段；完整初始化、错误处理和可运行程序见后面的示例。 + +```go +source, err := client.GetTaskSource(ctx, taskID) +if err != nil { return err } +switch source.Kind { +case "DIRECT": + fmt.Println("普通受理", source.TaskID) +case "MANUAL_RETRY": + fmt.Println("直接原任务", *source.RetryOf) +case "SCHEDULE": + fmt.Println("来源计划", source.Schedule.ScheduleID, + "固定版本", source.Schedule.ScheduleVersion, + "触发记录", source.Schedule.OccurrenceID) +} +``` + +SDK的ScheduleVersion为uint64，JSON仍为字符串；时间使用time.Time。SDK验证响应task_id一致性、来源种类与字段组合、ID、版本及时间范围；未知kind或矛盾数据返回错误，不返回可误用的部分来源。协议新增来源种类时，接入方需升级SDK再使用。 + +| 场景 | 完整源码 | 运行说明与新增输出 | +| --- | --- | --- | +| 单项受理、取消后读取来源 | [tasks](../../sdk/go/examples/tasks/main.go) | [SDK初始化](../sdk/README.md)，source_kind=DIRECT | +| 按计划找到触发任务，再反查来源 | [schedules](../../sdk/go/examples/schedules/main.go) | [计划场景](schedules.md#6-sdk-与完整示例)，source_kind=SCHEDULE、fixed_version=1 | +| 人工核查后创建关联新任务 | [recovery](../../sdk/go/examples/recovery/main.go) | [恢复场景](../examples/recovery.md)，source_kind=MANUAL_RETRY、parent=原任务ID | + +上述示例均运行在隔离回归中；新发布计划版本后老任务仍指向旧版本，由真实数据库与SDK集成测试验证。Weave流程来源随P4–P5实现，目前没有WORKFLOW响应。此接口为新增只读路由，不更改Task/TaskSummary或Worker消息结构，不增加权限配置；新客户端调用旧节点的404不能理解为DIRECT。 diff --git a/docs/api/tasks.md b/docs/api/tasks.md index e22dd69..7f29c60 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -116,6 +116,8 @@ curl "$ORBIT_URL/v1/tasks/$TASK_ID" -H "Authorization: Bearer $ORBIT_TOKEN" SDK：`Client.GetTask(ctx, taskID)`。 +任务直接来源独立通过`GET /v1/tasks/{task_id}/source`读取，SDK为`GetTaskSource(ctx, taskID)`；可定位普通受理、人工重试父任务、来源计划/触发记录/固定版本。使用相同的tasks:read和队列范围，详情见[来源查询](task-source.md)。来源未嵌入Task或列表摘要，读取完整计划仍需计划权限。 + ## GET /v1/tasks 权限：tasks:read。查询参数均只能出现一次： diff --git a/docs/examples/README.md b/docs/examples/README.md index ccb36b4..debe034 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -15,7 +15,7 @@ - 源码：[SDK 任务受理示例](../../sdk/go/examples/tasks/main.go)。 - 前置：完成迁移、初始化凭据、启动服务，在没有 Worker 消费的专用队列运行此受理/取消示例，详见 [SDK 接入](../sdk/README.md)。 - 输入：对象引用、格式与精确整数；提交两次同一身份，再查询、分页并带版本取消。 -- 预期：重复请求返回同一任务，最终 CANCELLED/revision=2。当前不启动 Worker，不实际生成文件。 +- 预期：重复请求返回同一任务，最终 CANCELLED/revision=2，source_kind=DIRECT。当前不启动 Worker，不实际生成文件。 - 异常：凭据失效、权限不足、内容冲突、版本竞争和未知网络结果，见 [任务接口](../api/tasks.md)。 - 清理：示例以取消结束；任务和审计历史保留。重复运行会生成独立示例任务。 @@ -153,3 +153,10 @@ - 步骤：提交三个模拟任务 → 精确业务键及时间范围过滤 → 每页一条遍历 → 验证上界排除 → 取消收尾。 - 预期：matched=2、pages=2、before_excluded=true、cancelled=3。 - 范围：普通提交/读取/取消权限；不调用外部系统、不启动Worker，所有历史保留，状态分页不是数据库快照。 + +## 已可运行：从任务定位来源 + +- [来源接口与SDK](../api/task-source.md)随现有tasks、schedules、recovery三个示例运行，无额外服务或数据库表。 +- 步骤：取得任务ID → GetTaskSource → 区分普通提交/人工重试/计划触发 → 按权限读取父任务或固定计划版本。 +- 预期：source_kind分别为DIRECT、MANUAL_RETRY、SCHEDULE；人工重试关联原任务，计划任务保留实际触发版本。 +- 异常：来源缺失或归属不匹配返回503，不能降级为DIRECT；新计划版本不覆盖历史版本，引用查询不授予计划定义读取权限。 diff --git a/docs/examples/recovery.md b/docs/examples/recovery.md index bf46e6c..50999cf 100644 --- a/docs/examples/recovery.md +++ b/docs/examples/recovery.md @@ -35,16 +35,19 @@ ORBIT_URL=http://127.0.0.1:8080 \ 3. 原键重报核查，确认同audit_id和deduplicated=true。 4. RetryTask声明可安全重新执行并指定新总预算60秒，取得retry_of指向原任务的新任务。 5. Worker开始新任务、报告SUCCEEDED；查询原任务仍FAILED、新任务SUCCEEDED。 -6. 查询原任务操作列表及重试审计详情，排空并关闭Worker连接。 +6. 查询原任务操作列表及重试审计详情，通过GetTaskSource核对新任务的直接原任务，排空并关闭Worker连接。 预期输出，标识由实际执行生成： ```text +source_kind=MANUAL_RETRY parent=<原任务ID> source=<原任务ID> state=FAILED retry=<新任务ID> state=SUCCEEDED audit=<审计ID> operations=4 ``` 操作数包含本例受理、原结果、核查和人工重试；以本次实际查询为准。示例不删除任务、审计或尝试。重复运行创建独立演示任务，每次保留原历史。 +[来源接口](../api/task-source.md)仅返回直接父任务，重复人工重试沿后继继续，不递归加载整条链；查询不产生新的审计或改写历史。 + ## 接入片段 下方在已有客户端和已核实的current任务上下文中调用；完整imports/main及错误处理见源码。 diff --git a/docs/reports/phase-3-task-source.md b/docs/reports/phase-3-task-source.md new file mode 100644 index 0000000..661889b --- /dev/null +++ b/docs/reports/phase-3-task-source.md @@ -0,0 +1,76 @@ +# P3 任务直接来源回归 + +验证日期：2026-09-17。服务端、SDK和三类现有场景已接入直接来源查询；定向MySQL、基础检查及全量隔离回归均已通过，完整脚本正常退出，状态码0。业务事务Outbox仍是P3待交付项，Weave按P4–P5继续。 + +## 1. 实现范围 + +- [服务模型](../../pulse/service/source.go)与[存储查询](../../internal/storage/mysql/task_source.go)：DIRECT、MANUAL_RETRY、SCHEDULE三类来源；计划ID、触发记录、冻结版本及合并范围；人工重试仅一跳父任务。 +- [HTTP路由](../../internal/transport/httpserver/history.go)提供GET /v1/tasks/{task_id}/source，沿用tasks:read和队列范围，拒绝查询参数。 +- [独立SDK](../../sdk/go/task_source.go)验证响应ID、类型组合、版本和时间，保留uint64精度；没有数据库依赖或自动重试。 +- [接口手册](../api/task-source.md)、[普通提交](../../sdk/go/examples/tasks/main.go)、[计划](../../sdk/go/examples/schedules/main.go)及[人工恢复](../../sdk/go/examples/recovery/main.go)同步提供调用与输出。 + +来源独立读取，不改变Task/TaskSummary及Worker协议，不在派发路径增加关联查询。复用现有(namespace_id,task_id)触发唯一索引及计划版本键，不新增迁移、索引或持久化字段；平台仍为35个实体、289个中文注释字段。新只读结构的字段也均有中文说明。 + +## 2. 真实MySQL与HTTP + +[TaskSource集成](../../internal/storage/mysql/task_source_integration_test.go)已验证： + +1. 普通受理和人工派生返回正确类型及直接父任务，读取前后任务快照不变。 +2. 跨命名空间、未授权队列返回NOT_FOUND，撤销凭据后拒绝继续查询；缺失或跨命名空间父任务返回UNAVAILABLE。 +3. 计划发布v2后，已生成任务仍指向v1和原触发记录，任务内容不变。 +4. 只有tasks:read的凭据能查询有限来源引用，但读取完整计划仍被PERMISSION_DENIED拒绝。 +5. 实际HTTP拒绝非法ID、参数与编码，隐藏无权限任务；响应不包含模板、payload或租约秘密。 +6. COALESCED来源保留原触发时间和排除上界。 +7. 模拟无外键数据库中的悬空所有者、缺失版本、错误计划队列、反向关联残留、未知所有者及门控，均拒绝且返回零值而非部分来源；恢复正确关联后查询成功。 + +定向回归总计24.058秒（含包及升级开销），TaskSource四组9.46秒通过。v7旧库升级验证外键70→0、323个未替换索引组件保留、已有任务内容不变；应用资源升级夹具还原历史真实utf8mb4_0900_as_cs后重放后续迁移。该调整仅修正测试夹具，生产迁移000016的精确utf8mb4_0900_bin保持不变。 + +## 3. SDK与完整场景 + +[SDK单元](../../sdk/go/task_source_test.go)覆盖三类有效响应、FIRED/COALESCED、9007199254740993版本精度、错误task_id、未知种类、字段矛盾、非法版本/时间/ID以及本地拒绝不安全路径。 + +[计划真实HTTP集成](../../sdk/go/schedule_integration_test.go)在发布新版本后核对旧任务固定来源；[恢复真实HTTP集成](../../sdk/go/recovery_integration_test.go)在回执丢失后以原身份恢复，再查询直接来源与人工重试关联。 + +已有三个可运行程序新增输出，原输出继续保留： + +```text +source_kind=DIRECT +source_kind=SCHEDULE fixed_version=1 +source_kind=MANUAL_RETRY parent=<原任务ID> +``` + +这些行来自不同程序，不是同一条任务同时具有三种来源。完整启动命令分别见[SDK入门](../sdk/README.md)、[计划场景](../api/schedules.md#6-sdk-与完整示例)及[恢复场景](../examples/recovery.md)。 + +## 4. 复现与结果 + +macOS、Go1.27.1、Docker单实例MySQL8.4.10，随机测试凭据、回环端口和一次性数据库。脚本完成后清理隔离资源。 + +```sh +make fmt test vet build +make integration +``` + +普通未配置数据库的go test会跳过Integration，不能替代真实数据库回归。本轮全量脚本完成后的实际结果： + +| 验证组 | 本轮结果 | +| --- | --- | +| fmt / race / vet / build | PASS | +| 全量MySQL / TaskSource四组 | PASS，108.438 / 9.98秒 | +| SDK Worker完整组 | PASS，87.120秒 | +| SDK计划 / 队列 / 批量 / 查询 | PASS，5.815 / 4.419 / 1.391 / 1.461秒 | +| SDK订阅管理 / 事件消费 | PASS，33.366 / 52.813秒 | +| SDK人工恢复与来源 | PASS，3.398秒 | +| Inbox数据库与消费者故障 | PASS，34.591秒 | +| SDK应用额度 | PASS，1.370秒 | +| 九项节点/Worker进程故障 | PASS，60.842秒；加上Inbox消费者，共十项进程故障 | +| 可运行程序 | 17个核心SDK示例 + 1个独立Inbox示例通过，三个新增来源输出符合预期 | + +零外键和历史迁移校验继续通过；文档链接、JSON块及核心SDK无服务端/数据库依赖检查通过。上述耗时为这次正确性回归记录，不是吞吐或容量结论。没有使用中途PASS日志推断脚本完成。 + +## 5. 交付边界 + +- 此接口只读直接来源，没有链式递归、来源全文或逆向全量任务搜索；状态和执行历史使用对应查询接口。 +- tasks:read只授予当前任务的有限来源引用，完整计划/版本仍独立鉴权。来源查询不产生执行权或恢复权限。 +- 关联损坏返回503，需要排查；不自动猜测或修复归属。所有者与来源由写入事务固定，读取不依赖数据库外键。 +- Weave流程来源随流程引擎实现，当前未知所有者明确拒绝；旧节点缺少新路由时404不能作为DIRECT使用。 +- 本轮不新增执行/持久化协议，也不提供生产吞吐、数据清理或数据库故障切换保证；这些仍按P6验证。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 01211f4..2a11ebf 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -32,13 +32,14 @@ ORBIT_URL=http://127.0.0.1:8080 \ ```text liveness=alive readiness=ready scope=database_and_schema +source_kind=DIRECT task=<32位任务ID> duplicate=true listed=1 final_state=CANCELLED revision=2 ``` 重复运行示例会新建一个独立任务并取消它，listed 数量可能增加，最多显示当前页 10 条。数据和审计保留，不执行硬删除。示例只演示受理闭环，尚不执行文件转换。 - [健康示例源码](../../sdk/go/examples/health/main.go)：只读，无业务资源写入。 -- [任务示例源码](../../sdk/go/examples/tasks/main.go)：提交两次同一身份、获取当前状态、列表、带版本取消。 +- [任务示例源码](../../sdk/go/examples/tasks/main.go)：提交两次同一身份、获取当前状态、列表、带版本取消，再核对直接来源。 - 任务示例也接受 ORBIT_TOKEN 和 ORBIT_QUEUE_ID，环境值优先于凭据文件。 - 示例发生错误会打印错误并非零退出；打印内容不包含原始 token。 - 初始凭据 30 天有效。初始化行为和错误核对方式见 [凭据文档](../api/authentication.md)。 @@ -119,6 +120,7 @@ fmt.Println(cancelled.Task.State) | Liveness / Readiness | Health | [健康](../api/health.md) | | SubmitTask | SubmitTaskRequest / SubmitTaskResult | [任务](../api/tasks.md) | | GetTask | taskID / Task | [任务](../api/tasks.md) | +| GetTaskSource | taskID / TaskSource，直接来源及计划固定版本 | [来源查询](../api/task-source.md) | | ListTasks | ListTasksRequest / TaskPage，业务键及创建时间范围 | [查询](../api/task-queries.md) | | CancelTask | taskID、key、CancelTaskRequest / CancelTaskResult | [任务](../api/tasks.md) | | CreateQueue | key、CreateQueueRequest / CreateQueueResult | [队列](../api/queues.md) | diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index 0e87df0..5a45a10 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -64,7 +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 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_0900_as_cs 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/task_source.go b/internal/storage/mysql/task_source.go new file mode 100644 index 0000000..0793eb9 --- /dev/null +++ b/internal/storage/mysql/task_source.go @@ -0,0 +1,90 @@ +package mysql + +import ( + "context" + "database/sql" + "errors" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func (s *TaskStore) GetTaskSource(ctx context.Context, t auth.Token, id core.ID) (service.TaskSourceView, error) { + out := service.TaskSourceView{TaskID: id} + err := s.transaction(ctx, t, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error { + var queue core.ID + var owner sql.NullString + var ownerID, gate, retry *core.ID + err := tx.QueryRowContext(ctx, `SELECT queue_id,owner_kind,owner_id,gate_id,retry_of FROM pulse_tasks WHERE namespace_id=? AND id=?`, t.NamespaceID, id).Scan(&queue, &owner, &ownerID, &gate, &retry) + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + if err != nil { + return err + } + if !p.AllowsQueue(queue) { + return fault.NotFound + } + // 来源字段由创建事务固定；当前未实现的门控/所有者不能被误判为普通受理。 + if gate != nil || owner.Valid != (ownerID != nil) || owner.Valid && retry != nil { + return fault.Unavailable + } + if owner.Valid { + if owner.String != "schedule" { + return fault.Unavailable + } + var source service.TaskScheduleSource + // 每条关联均显式限定命名空间、计划、版本与任务队列，不依赖数据库外键。 + err := tx.QueryRowContext(ctx, `SELECT o.schedule_id,o.schedule_version,o.id,o.scheduled_at,o.decision,o.covered_until + FROM pulse_schedule_occurrences o + JOIN pulse_schedules s ON s.namespace_id=o.namespace_id AND s.id=o.schedule_id AND s.queue_id=? + JOIN pulse_schedule_versions v ON v.namespace_id=s.namespace_id AND v.schedule_id=s.id AND v.version=o.schedule_version AND v.version<=s.current_version + WHERE o.namespace_id=? AND o.task_id=? AND o.schedule_id=? AND JSON_UNQUOTE(JSON_EXTRACT(v.task_template,'$.queue_id'))=?`, queue, t.NamespaceID, id, *ownerID, string(queue)).Scan(&source.ScheduleID, &source.ScheduleVersion, &source.OccurrenceID, &source.ScheduledAt, &source.Decision, &source.CoveredUntil) + if errors.Is(err, sql.ErrNoRows) { + return fault.Unavailable + } + if err != nil { + return err + } + if source.ScheduleVersion == 0 || !sourceTimeValid(source.ScheduledAt) || source.CoveredUntil != nil && !sourceTimeValid(*source.CoveredUntil) || source.Decision == "FIRED" && source.CoveredUntil != nil || source.Decision == "COALESCED" && (source.CoveredUntil == nil || !source.CoveredUntil.After(source.ScheduledAt)) || source.Decision != "FIRED" && source.Decision != "COALESCED" { + return fault.Unavailable + } + out.Kind = "SCHEDULE" + out.Schedule = &source + return nil + } + var occurrences int + if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_schedule_occurrences WHERE namespace_id=? AND task_id=?`, t.NamespaceID, id).Scan(&occurrences); err != nil { + return err + } + if occurrences != 0 { + return fault.Unavailable + } + if retry == nil { + out.Kind = "DIRECT" + return nil + } + var parent core.ID + err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_tasks WHERE namespace_id=? AND id=? AND id<>? AND queue_id=? AND owner_kind IS NULL AND owner_id IS NULL AND gate_id IS NULL AND state IN ('FAILED','CANCELLED')`, t.NamespaceID, *retry, id, queue).Scan(&parent) + if errors.Is(err, sql.ErrNoRows) { + return fault.Unavailable + } + if err != nil { + return err + } + out.Kind = "MANUAL_RETRY" + out.RetryOf = &parent + return nil + }) + if err != nil { + return service.TaskSourceView{}, err + } + return out, nil +} + +func sourceTimeValid(t time.Time) bool { + return t.Year() >= 2000 && t.Year() <= 9998 && t.Nanosecond()%1000 == 0 +} diff --git a/internal/storage/mysql/task_source_integration_test.go b/internal/storage/mysql/task_source_integration_test.go new file mode 100644 index 0000000..ba844ba --- /dev/null +++ b/internal/storage/mysql/task_source_integration_test.go @@ -0,0 +1,200 @@ +package mysql + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "reflect" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/pulse/schedule" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func TestIntegrationTaskSource(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, 16, 4) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + t.Run("direct_and_manual_retry_are_read_only_and_scoped", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantTaskRecovery(f) + original := f.submit(1) + source, err := f.tasks.GetTaskSource(ctx, f.token, original.ID) + if err != nil || source.TaskID != original.ID || source.Kind != "DIRECT" || source.Schedule != nil || source.RetryOf != nil { + t.Fatal(source, err) + } + cancelled, err := f.tasks.CancelTask(ctx, f.token, original.ID, "cancel", service.CancelInput{ExpectedRevision: original.Revision, Reason: "retry source regression"}) + if err != nil { + t.Fatal(err) + } + child, err := f.tasks.RetryTask(ctx, f.token, original.ID, "retry", manualRetryInput(cancelled.Task)) + if err != nil { + t.Fatal(err) + } + source, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID) + if err != nil || source.Kind != "MANUAL_RETRY" || source.Schedule != nil || source.RetryOf == nil || *source.RetryOf != original.ID { + t.Fatal(source, err) + } + if !reflect.DeepEqual(f.get(original.ID), cancelled.Task) || !reflect.DeepEqual(f.get(child.Task.ID), child.Task) { + t.Fatal("source read changed task") + } + other := newExecutionFixture(t, ctx, db) + _, err = other.tasks.GetTaskSource(ctx, other.token, child.Task.ID) + expectWorkerCode(t, err, "NOT_FOUND") + f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(other.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID) + _, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID) + expectWorkerCode(t, err, "NOT_FOUND") + f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY('*')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + // 模拟无外键库的缺失与跨命名空间原任务，不能返回未验证的标识。 + for _, parent := range []core.ID{executionID(t), other.submit(1).ID} { + f.exec(`UPDATE pulse_tasks SET retry_of=? WHERE namespace_id=? AND id=?`, parent, f.resources.NamespaceID, child.Task.ID) + _, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID) + expectWorkerCode(t, err, "UNAVAILABLE") + } + f.exec(`UPDATE pulse_tasks SET retry_of=? WHERE namespace_id=? AND id=?`, original.ID, f.resources.NamespaceID, child.Task.ID) + f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + _, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID) + expectWorkerCode(t, err, "UNAUTHENTICATED") + }) + t.Run("schedule_source_preserves_published_version_and_permission_boundary", func(t *testing.T) { + f := newScheduleFixture(t, newExecutionFixture(t, ctx, db)) + in := f.definition() + plan := f.create(in) + f.due(plan, 2*time.Second) + f.tick() + o := f.occurrences(plan.ID)[0] + if o.TaskID == nil { + t.Fatal("missing task") + } + before := f.get(*o.TaskID) + source, err := f.tasks.GetTaskSource(ctx, f.token, *o.TaskID) + if err != nil || source.Kind != "SCHEDULE" || source.Schedule == nil || source.RetryOf != nil { + t.Fatal(source, err) + } + link := source.Schedule + if link.ScheduleID != plan.ID || link.OccurrenceID != o.ID || link.ScheduleVersion != o.ScheduleVersion || !link.ScheduledAt.Equal(o.ScheduledAt) || link.Decision != o.Decision { + t.Fatal("incorrect schedule lineage", link, o) + } + in.Task.Payload = json.RawMessage(`{"version":2}`) + now := f.current(plan.ID) + published, err := f.api.Publish(ctx, f.token, plan.ID, "publish", schedule.PublishInput{ExpectedRevision: now.Revision, Reason: "source stays on original version", Definition: in}) + if err != nil || published.Schedule.CurrentVersion != 2 { + t.Fatal(published, err) + } + after, err := f.tasks.GetTaskSource(ctx, f.token, *o.TaskID) + if err != nil || !reflect.DeepEqual(source, after) || !reflect.DeepEqual(before, f.get(*o.TaskID)) { + t.Fatal("latest publication changed source", after, err) + } + // tasks:read仅授予此任务的来源引用，不因此授予完整计划/模板查询权限。 + f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY('tasks:read')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + if _, err = f.tasks.GetTaskSource(ctx, f.token, *o.TaskID); err != nil { + t.Fatal(err) + } + _, err = f.api.Get(ctx, f.token, plan.ID) + expectWorkerCode(t, err, "PERMISSION_DENIED") + server := httptest.NewServer(httpserver.New("", func(context.Context) error { return nil }, f.tasks).Handler) + defer server.Close() + call := func(path string, status int) []byte { + t.Helper() + r, _ := http.NewRequestWithContext(ctx, "GET", server.URL+path, 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("status=%d want=%d body=%s", resp.StatusCode, status, raw) + } + return raw + } + path := "/v1/tasks/" + string(*o.TaskID) + "/source" + raw := call(path, 200) + for _, secret := range []string{"task_template", "payload", "lease_secret", "secret_digest", "token"} { + if strings.Contains(string(raw), secret) { + t.Fatal("source leaked unrelated data") + } + } + for _, suffix := range []string{"?limit=1", "?namespace_id=x", "?unknown=%zz"} { + call(path+suffix, 400) + } + call("/v1/tasks/not-an-id/source", 400) + call("/v1/tasks/"+string(executionID(t))+"/source", 404) + }) + t.Run("coalesced_source_retains_exclusive_range", func(t *testing.T) { + f := newScheduleFixture(t, newExecutionFixture(t, ctx, db)) + in := f.definition() + in.Policy.Misfire = "COALESCE" + plan := f.create(in) + f.due(plan, 3*time.Second) + f.tick() + found := false + for _, o := range f.occurrences(plan.ID) { + if o.Decision != "COALESCED" { + continue + } + found = true + source, err := f.tasks.GetTaskSource(ctx, f.token, *o.TaskID) + if err != nil || source.Schedule == nil || source.Schedule.CoveredUntil == nil || !source.Schedule.CoveredUntil.Equal(*o.CoveredUntil) || !source.Schedule.ScheduledAt.Equal(o.ScheduledAt) { + t.Fatal(source, err) + } + } + if !found { + t.Fatal("coalesced trigger missing") + } + }) + t.Run("dangling_mismatched_and_unknown_owners_fail_closed", func(t *testing.T) { + f := newScheduleFixture(t, newExecutionFixture(t, ctx, db)) + plan := f.create(f.definition()) + f.due(plan, time.Second) + f.tick() + o := f.occurrences(plan.ID)[0] + id := *o.TaskID + broken := func() { + t.Helper() + out, err := f.tasks.GetTaskSource(ctx, f.token, id) + expectWorkerCode(t, err, "UNAVAILABLE") + if out.TaskID != "" || out.Schedule != nil { + t.Fatal("partial invalid source returned") + } + } + f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, id) + broken() + f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, plan.ID, f.resources.NamespaceID, id) + f.exec(`UPDATE pulse_schedule_occurrences SET schedule_version=999 WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, o.ID) + broken() + f.exec(`UPDATE pulse_schedule_occurrences SET schedule_version=? WHERE namespace_id=? AND id=?`, o.ScheduleVersion, f.resources.NamespaceID, o.ID) + f.exec(`UPDATE pulse_schedules SET queue_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, plan.ID) + broken() + f.exec(`UPDATE pulse_schedules SET queue_id=? WHERE namespace_id=? AND id=?`, f.resources.QueueID, f.resources.NamespaceID, plan.ID) + f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, id) + broken() + f.exec(`UPDATE pulse_tasks SET owner_kind='workflow',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, id) + broken() + f.exec(`UPDATE pulse_tasks SET owner_kind='schedule',owner_id=?,gate_id=? WHERE namespace_id=? AND id=?`, plan.ID, executionID(t), f.resources.NamespaceID, id) + broken() + f.exec(`UPDATE pulse_tasks SET gate_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, id) + if _, err = f.tasks.GetTaskSource(ctx, f.token, id); err != nil { + t.Fatal("restored valid source still refused", err) + } + }) +} diff --git a/internal/transport/httpserver/history.go b/internal/transport/httpserver/history.go index da21060..cb3f030 100644 --- a/internal/transport/httpserver/history.go +++ b/internal/transport/httpserver/history.go @@ -32,6 +32,13 @@ func historyQuery(r *http.Request, paged bool) (service.HistoryPageInput, error) return in, nil } func registerHistory(mux *http.ServeMux, s *service.Service) { + mux.HandleFunc("GET /v1/tasks/{task}/source", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + if _, err := historyQuery(r, false); err != nil { + return nil, 0, err + } + out, err := s.GetTaskSource(r.Context(), t, core.ID(r.PathValue("task"))) + return out, 200, err + })) mux.HandleFunc("GET /v1/tasks/{task}/operations", wrap(func(r *http.Request, t auth.Token) (any, int, error) { in, err := historyQuery(r, true) if err != nil { diff --git a/pulse/service/source.go b/pulse/service/source.go new file mode 100644 index 0000000..d0b430a --- /dev/null +++ b/pulse/service/source.go @@ -0,0 +1,43 @@ +package service + +import ( + "context" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "time" +) + +// TaskSourceView 只读的直接来源，不遍历整条重试链，也不授予来源资源管理权限。 +type TaskSourceView struct { + // TaskID 本次已授权查询的任务标识。 + TaskID core.ID `json:"task_id"` + // Kind 当前支持DIRECT、MANUAL_RETRY、SCHEDULE；分别是普通受理、人工派生、计划触发。 + Kind string `json:"kind"` + // RetryOf 人工重试的直接原任务，限同一命名空间及队列；其他种类省略。 + RetryOf *core.ID `json:"retry_of,omitempty"` + // Schedule 计划触发的固定来源，仅SCHEDULE返回，不包含任务模板或计划配置。 + Schedule *TaskScheduleSource `json:"schedule,omitempty"` +} + +// TaskScheduleSource 关联已生成任务的触发记录及当时的不可变计划版本。 +type TaskScheduleSource struct { + // ScheduleID 原计划标识，不因发布新版本而变化。 + ScheduleID core.ID `json:"schedule_id"` + // ScheduleVersion 触发时使用的固定版本，JSON字符串避免精度丢失。 + ScheduleVersion uint64 `json:"schedule_version,string"` + // OccurrenceID 唯一关联该任务的触发记录标识。 + OccurrenceID core.ID `json:"occurrence_id"` + // ScheduledAt 原计划时间，UTC微秒精度，不等于实际受理/执行时间。 + ScheduledAt time.Time `json:"scheduled_at"` + // Decision FIRED为单点触发，COALESCED为合并触发；跳过不会产生任务。 + Decision string `json:"decision"` + // CoveredUntil 合并范围的排除边界，仅COALESCED返回。 + CoveredUntil *time.Time `json:"covered_until,omitempty"` +} + +func (s *Service) GetTaskSource(ctx context.Context, t auth.Token, id core.ID) (TaskSourceView, error) { + if err := validateHistoryIDs(id); err != nil { + return TaskSourceView{}, err + } + return s.store.GetTaskSource(ctx, t, id) +} diff --git a/pulse/service/types.go b/pulse/service/types.go index 419e4e0..380ae23 100644 --- a/pulse/service/types.go +++ b/pulse/service/types.go @@ -224,6 +224,7 @@ type QueueResult struct { // Store 实现命名空间隔离、事务授权与持久化，不在事务外产生业务副作用。 type Store interface { + GetTaskSource(context.Context, auth.Token, core.ID) (TaskSourceView, error) ReviewTask(context.Context, auth.Token, core.ID, string, ReviewInput) (ReviewResult, error) RetryTask(context.Context, auth.Token, core.ID, string, ManualRetryInput) (ManualRetryResult, error) ListTaskOperations(context.Context, auth.Token, core.ID, HistoryPageInput) (TaskOperationPage, error) diff --git a/sdk/go/examples/recovery/main.go b/sdk/go/examples/recovery/main.go index f0661b0..0139694 100644 --- a/sdk/go/examples/recovery/main.go +++ b/sdk/go/examples/recovery/main.go @@ -177,6 +177,14 @@ func run() error { if _, err := w.Capacity(ctx, id, orbit.WorkerCapacity{Sequence: 2, Drain: true}); err != nil { return err } + lineage, err := client.GetTaskSource(ctx, child.ID) + if err != nil { + return err + } + if lineage.Kind != "MANUAL_RETRY" || lineage.RetryOf == nil || *lineage.RetryOf != original.ID { + return errors.New("manual retry source mismatch") + } + fmt.Printf("source_kind=%s parent=%s\n", lineage.Kind, *lineage.RetryOf) fmt.Printf("source=%s state=%s retry=%s state=%s audit=%s operations=%d\n", original.ID, original.State, child.ID, child.State, audit.ID, len(page.Items)) return nil } diff --git a/sdk/go/examples/schedules/main.go b/sdk/go/examples/schedules/main.go index f9fb5c4..67754e2 100644 --- a/sdk/go/examples/schedules/main.go +++ b/sdk/go/examples/schedules/main.go @@ -99,6 +99,14 @@ func run() error { return err } if task.State == orbit.TaskSucceeded { + origin, err := client.GetTaskSource(ctx, task.ID) + if err != nil { + return err + } + if origin.Kind != "SCHEDULE" || origin.Schedule == nil || origin.Schedule.ScheduleID != created.Schedule.ID || origin.Schedule.OccurrenceID != item.ID || origin.Schedule.ScheduleVersion != item.ScheduleVersion { + return errors.New("task source did not match frozen occurrence") + } + fmt.Printf("source_kind=%s fixed_version=%d\n", origin.Kind, origin.Schedule.ScheduleVersion) completed[task.ID] = true fmt.Printf("schedule=%s occurrence=%s task=%s version=%d state=%s\n", created.Schedule.ID, item.ID, task.ID, item.ScheduleVersion, task.State) } diff --git a/sdk/go/examples/tasks/main.go b/sdk/go/examples/tasks/main.go index 1d695d7..8748927 100644 --- a/sdk/go/examples/tasks/main.go +++ b/sdk/go/examples/tasks/main.go @@ -79,6 +79,14 @@ func run() error { if err != nil { return err } + origin, err := client.GetTaskSource(ctx, current.ID) + if err != nil { + return err + } + if origin.Kind != "DIRECT" { + return fmt.Errorf("expected direct task source") + } + fmt.Printf("source_kind=%s\n", origin.Kind) fmt.Printf("task=%s duplicate=%t listed=%d final_state=%s revision=%d\n", current.ID, again.Deduplicated, len(page.Items), cancelled.Task.State, cancelled.Task.Revision) if cancelled.Task.State != orbit.TaskCancelled { return fmt.Errorf("expected CANCELLED") diff --git a/sdk/go/recovery_integration_test.go b/sdk/go/recovery_integration_test.go index 57cb955..7247168 100644 --- a/sdk/go/recovery_integration_test.go +++ b/sdk/go/recovery_integration_test.go @@ -126,6 +126,14 @@ func TestIntegrationTaskRecoverySDK(t *testing.T) { if err != nil || !child.Deduplicated || child.Task.ID != retryEnvelope.Data.Task.ID || child.AuditID != retryEnvelope.Data.AuditID || child.Task.RetryOf == nil || *child.Task.RetryOf != unknown.ID { t.Fatal(child, err) } + lineage, err := api.GetTaskSource(ctx, child.Task.ID) + if err != nil || lineage.Kind != "MANUAL_RETRY" || lineage.RetryOf == nil || *lineage.RetryOf != unknown.ID || lineage.Schedule != nil { + t.Fatal("manual retry source", lineage, err) + } + originalSource, err := api.GetTaskSource(ctx, unknown.ID) + if err != nil || originalSource.Kind != "DIRECT" || originalSource.RetryOf != nil { + t.Fatal("direct source", originalSource, err) + } assigned, err := w.NextAssignment(ctx) if err != nil || assigned.Task.ID != child.Task.ID || assigned.Task.RetryOf == nil || *assigned.Task.RetryOf != unknown.ID { t.Fatal("wire lost retry lineage", assigned, err) diff --git a/sdk/go/schedule_integration_test.go b/sdk/go/schedule_integration_test.go index dbcb9b0..a137b31 100644 --- a/sdk/go/schedule_integration_test.go +++ b/sdk/go/schedule_integration_test.go @@ -90,6 +90,10 @@ func TestIntegrationScheduleLifecycle(t *testing.T) { if err != nil || updated.Schedule.CurrentVersion != 2 || !updated.Schedule.Paused { t.Fatal("publish schedule", err) } + source, err := api.GetTaskSource(ctx, task.ID) + if err != nil || source.Kind != "SCHEDULE" || source.Schedule == nil || source.Schedule.ScheduleID != paused.Schedule.ID || source.Schedule.OccurrenceID != occurrence.ID || source.Schedule.ScheduleVersion != 1 || !source.Schedule.ScheduledAt.Equal(occurrence.ScheduledAt) { + t.Fatal("published version changed task source", source, err) + } old, err := api.GetScheduleVersion(ctx, paused.Schedule.ID, 1) if err != nil || old.Definition.Rule.Kind != "FIXED_RATE" { t.Fatal("old version lost", err) diff --git a/sdk/go/task_source.go b/sdk/go/task_source.go new file mode 100644 index 0000000..c650f11 --- /dev/null +++ b/sdk/go/task_source.go @@ -0,0 +1,68 @@ +package orbit + +import ( + "context" + "errors" + "time" +) + +// TaskSource 只读的直接来源，不遍历整条重试链，也不授予来源资源管理权限。 +type TaskSource struct { + // TaskID 本次已授权查询的任务标识。 + TaskID string `json:"task_id"` + // Kind 当前支持DIRECT、MANUAL_RETRY、SCHEDULE；分别是普通受理、人工派生、计划触发。 + Kind string `json:"kind"` + // RetryOf 人工重试的直接原任务，限同一命名空间及队列；其他种类省略。 + RetryOf *string `json:"retry_of,omitempty"` + // Schedule 计划触发的固定来源，仅SCHEDULE返回，不包含任务模板或计划配置。 + Schedule *TaskScheduleSource `json:"schedule,omitempty"` +} + +// TaskScheduleSource 关联已生成任务的触发记录及当时的不可变计划版本。 +type TaskScheduleSource struct { + // ScheduleID 原计划标识，不因发布新版本而变化。 + ScheduleID string `json:"schedule_id"` + // ScheduleVersion 触发时使用的固定版本，JSON字符串避免精度丢失。 + ScheduleVersion uint64 `json:"schedule_version,string"` + // OccurrenceID 唯一关联该任务的触发记录标识。 + OccurrenceID string `json:"occurrence_id"` + // ScheduledAt 原计划时间，UTC微秒精度，不等于实际受理/执行时间。 + ScheduledAt time.Time `json:"scheduled_at"` + // Decision FIRED为单点触发，COALESCED为合并触发；跳过不会产生任务。 + Decision string `json:"decision"` + // CoveredUntil 合并范围的排除边界，仅COALESCED返回。 + CoveredUntil *time.Time `json:"covered_until,omitempty"` +} + +// GetTaskSource 查询直接来源；不会授予来源计划的读取/管理或任务执行权限。 +func (c *Client) GetTaskSource(ctx context.Context, id string) (TaskSource, error) { + var out TaskSource + if !validID(id) { + return out, errors.New("invalid task id") + } + if err := c.call(ctx, "GET", "/v1/tasks/"+id+"/source", "", nil, &out); err != nil { + return TaskSource{}, err + } + if out.TaskID != id { + return TaskSource{}, errors.New("source does not match requested task") + } + valid := false + switch out.Kind { + case "DIRECT": + valid = out.RetryOf == nil && out.Schedule == nil + case "MANUAL_RETRY": + valid = out.RetryOf != nil && validID(*out.RetryOf) && *out.RetryOf != id && out.Schedule == nil + case "SCHEDULE": + s := out.Schedule + if s != nil && out.RetryOf == nil && validID(s.ScheduleID) && validID(s.OccurrenceID) && s.ScheduleVersion > 0 && validSourceTime(s.ScheduledAt) { + valid = s.Decision == "FIRED" && s.CoveredUntil == nil || s.Decision == "COALESCED" && s.CoveredUntil != nil && validSourceTime(*s.CoveredUntil) && s.CoveredUntil.After(s.ScheduledAt) + } + } + if !valid { + return TaskSource{}, errors.New("invalid or unsupported task source response") + } + return out, nil +} +func validSourceTime(t time.Time) bool { + return t.Year() >= 2000 && t.Year() <= 9998 && t.Nanosecond()%1000 == 0 +} diff --git a/sdk/go/task_source_test.go b/sdk/go/task_source_test.go new file mode 100644 index 0000000..3038009 --- /dev/null +++ b/sdk/go/task_source_test.go @@ -0,0 +1,61 @@ +package orbit + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func TestTaskSourceValidationAndPrecision(t *testing.T) { + const task = "11111111111111111111111111111111" + const parent = "22222222222222222222222222222222" + const occurrence = "33333333333333333333333333333333" + at := time.Date(2026, 9, 17, 0, 0, 0, 123456000, time.UTC) + until := at.Add(time.Minute) + var reply atomic.Value + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.Method != "GET" || r.URL.Path != "/v1/tasks/"+task+"/source" || r.Header.Get("Authorization") != "Bearer example-token" || r.Header.Get("Idempotency-Key") != "" { + t.Error("invalid source request") + } + json.NewEncoder(w).Encode(map[string]any{"request_id": task, "data": reply.Load()}) + })) + defer server.Close() + client, _ := NewClient(server.URL, Options{Token: "example-token"}) + p := parent + valid := []TaskSource{{TaskID: task, Kind: "DIRECT"}, {TaskID: task, Kind: "MANUAL_RETRY", RetryOf: &p}, {TaskID: task, Kind: "SCHEDULE", Schedule: &TaskScheduleSource{ScheduleID: parent, ScheduleVersion: 9007199254740993, OccurrenceID: occurrence, ScheduledAt: at, Decision: "FIRED"}}, {TaskID: task, Kind: "SCHEDULE", Schedule: &TaskScheduleSource{ScheduleID: parent, ScheduleVersion: 1, OccurrenceID: occurrence, ScheduledAt: at, Decision: "COALESCED", CoveredUntil: &until}}} + for _, v := range valid { + reply.Store(v) + out, err := client.GetTaskSource(context.Background(), task) + if err != nil { + t.Fatal(err) + } + if v.Schedule != nil && out.Schedule.ScheduleVersion != v.Schedule.ScheduleVersion { + t.Fatal("version precision lost") + } + } + self := task + invalid := []TaskSource{{TaskID: parent, Kind: "DIRECT"}, {TaskID: task, Kind: "unknown"}, {TaskID: task, Kind: "DIRECT", RetryOf: &p}, {TaskID: task, Kind: "MANUAL_RETRY"}, {TaskID: task, Kind: "MANUAL_RETRY", RetryOf: &self}, {TaskID: task, Kind: "SCHEDULE"}} + for _, change := range []func(*TaskScheduleSource){func(v *TaskScheduleSource) { v.ScheduleVersion = 0 }, func(v *TaskScheduleSource) { v.Decision = "SKIPPED" }, func(v *TaskScheduleSource) { v.Decision = "COALESCED" }, func(v *TaskScheduleSource) { v.CoveredUntil = &until }, func(v *TaskScheduleSource) { v.ScheduledAt = time.Time{} }, func(v *TaskScheduleSource) { v.OccurrenceID = "invalid" }} { + v := valid[2] + s := *v.Schedule + change(&s) + v.Schedule = &s + invalid = append(invalid, v) + } + for _, v := range invalid { + reply.Store(v) + if _, err := client.GetTaskSource(context.Background(), task); err == nil { + t.Fatal("invalid source accepted", v) + } + } + before := calls.Load() + if _, err := client.GetTaskSource(context.Background(), "../tasks"); err == nil || calls.Load() != before { + t.Fatal("unsafe source path sent") + } +}