From a664ec9370ea4924fefbcfc9cfbe82f5f98a1329 Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 23:43:57 +0800 Subject: [PATCH] feat: add durable relative task delays and outage recovery regression --- AGENTS.md | 4 +- README.md | 2 +- docs/01-functional-spec.md | 4 +- docs/05-production-acceptance.md | 2 +- docs/07-data-model.md | 4 + docs/08-development-plan.md | 19 +- docs/README.md | 4 +- docs/api/README.md | 1 + docs/api/retries.md | 4 +- docs/api/schedules.md | 2 +- docs/api/tasks.md | 5 + docs/examples/README.md | 6 + docs/examples/delayed-task.md | 59 ++++++ docs/examples/task-outbox.md | 2 + docs/examples/task-query.md | 6 +- docs/reports/phase-3-delay-recovery.md | 74 +++++++ docs/reports/phase-3.md | 64 ++++++ docs/sdk/README.md | 2 + .../task-outbox/internal/outbox/store_test.go | 3 +- internal/storage/mysql/task_delay_test.go | 34 ++++ .../storage/mysql/task_integration_test.go | 62 ++++++ internal/storage/mysql/task_store.go | 3 +- internal/transport/httpserver/tasks_test.go | 7 + pulse/schedule/service.go | 6 +- pulse/schedule/service_test.go | 25 +++ pulse/service/service.go | 3 + pulse/service/service_test.go | 13 +- pulse/service/types.go | 4 +- scripts/test-mysql.sh | 1 + sdk/go/examples/delayed-task/main.go | 129 ++++++++++++ sdk/go/examples/task-query/main.go | 23 ++- sdk/go/retry_process_integration_test.go | 192 ++++++++++++++++++ sdk/go/task_delay_integration_test.go | 66 ++++++ sdk/go/types.go | 4 +- 34 files changed, 811 insertions(+), 28 deletions(-) create mode 100644 docs/examples/delayed-task.md create mode 100644 docs/reports/phase-3-delay-recovery.md create mode 100644 docs/reports/phase-3.md create mode 100644 internal/storage/mysql/task_delay_test.go create mode 100644 pulse/schedule/service_test.go create mode 100644 sdk/go/examples/delayed-task/main.go create mode 100644 sdk/go/retry_process_integration_test.go create mode 100644 sdk/go/task_delay_integration_test.go diff --git a/AGENTS.md b/AGENTS.md index 2c1825f..d08f5dd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,7 +14,7 @@ - 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 - 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 -- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，任务查询已支持业务键、创建时间及直接来源（普通/人工重试/计划固定版本），业务事务Outbox独立示例已实现并通过全量链路回归；P3仍需原功能范围与验收证据逐项核对，不能因局部功能完成而跳过原范围。 +- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，任务查询已支持业务键、创建时间及直接来源（普通/人工重试/计划固定版本），业务事务Outbox独立示例已实现并通过全量链路回归；P3已完成原功能范围核对与全量阶段回归（docs/reports/phase-3.md），相对延迟及四种重试全停恢复一并验证；继续P4–P7，保留精确A04故障、生产验证和实际文件转换示例等未完成项，不能把P3阶段验收当作全平台交付。 - 批量受理最多32项、512 KiB；每项复用单项身份与事务，未知结果必须原键核对。任务提交按认证→应用额度→提交幂等身份→队列持锁，禁止恢复为先抢提交唯一键后等待额度锁；原回执仍须先于配额上限检查返回。 @@ -26,3 +26,5 @@ - 任务来源通过独立只读路由查询，tasks:read仅授予当前任务的有限来源引用；计划定义仍需schedules:read。显式核对命名空间、队列、计划、触发与固定版本，悬空/错误/未实现所有者不得默认为DIRECT；不依赖外键，也不向派发路径加入来源查询。 - 业务事务Outbox示例使用独立module和业务数据库，核心SDK不得引入SQL驱动。业务记录/待发命令同事务；原请求、操作ID、目标及提交键冻结，HTTP在事务外，确认按最新租约代次与业务关联同事务。24小时/100次预算内原键核对，禁止换键、重置预算或把ACCEPTED解释成业务成功；REVIEW保留待核查数据。 + +- 相对延迟delay_ms只在首次受理事务按数据库时间计算；原时长参与摘要，零值omitempty保持旧摘要。重报先返回原回执，禁止重算执行时间；总时限包含等待。计划模板拒绝非零delay_ms，人工重试不自动继承首次等待。 diff --git a/README.md b/README.md index b9db9cc..547a3da 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场景；[任务来源查询](docs/api/task-source.md) 已提供人工重试父任务、计划触发记录与固定版本；[业务事务Outbox](docs/examples/task-outbox.md) 已提供独立业务库、稳定提交身份、发送恢复与CLI；剩余 Pulse 能力与流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已完成 P0–P3 平台功能阶段验收：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) 已提供人工重试父任务、计划触发记录与固定版本；[业务事务Outbox](docs/examples/task-outbox.md) 已提供独立业务库、稳定提交身份、发送恢复与CLI；[相对延迟执行](docs/examples/delayed-task.md)已提供数据库计时及原请求恢复；完整范围核对见[P3验收](docs/reports/phase-3.md)，流程引擎与生产验证按 [持续开发计划](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 376ddec..630f927 100644 --- a/docs/01-functional-spec.md +++ b/docs/01-functional-spec.md @@ -39,7 +39,7 @@ Weave 有界查询的一轮对应一个 Pulse 任务；查询得到“处理中 ### 3.1 提交方式 - **立即执行**：受理后尽快进入可派发范围，实际开始时间受资源和排队影响。 -- **延迟执行**：在指定等待时长之后才允许开始。 +- **延迟执行**：通过delay_ms在首次受理数据库时间后等待指定时长，原键重报不重新计时；见[调用示例](examples/delayed-task.md)。 - **指定时间执行**：在指定时间之后才允许开始，不承诺硬实时触发。 - **批量提交**：多项普通任务的提交方式；已实现最多32项逐项返回结果，每项独立幂等，失败项独立补交，整批请求不超过512 KiB。 @@ -58,6 +58,8 @@ Weave 有界查询的一轮对应一个 Pulse 任务；查询得到“处理中 | 人工重试 | 校验权限、错误类别及业务确认条件，保留原历史，记录操作人和原因 | | 修改业务参数 | 新建关联任务，原任务参数不变 | +修改参数通过普通提交创建新任务，使用新的提交身份及相同business_key建立业务关联；原任务不自动取消，也不会变更输入。调用方须先决定原任务的执行/取消和业务效果处理。此类任务来源仍为DIRECT，business_key不等同于retry_of或唯一执行约束；[查询示例](examples/task-query.md)展示不同参数的关联新任务和原输入核对。 + **当前契约：人工重试创建带 `retry_of` 的新任务。** 原任务保留终态及结果，避免外部订阅方已经消费终态后任务又悄悄恢复。业务幂等键应按原操作语义保留；平台新任务 ID 不代表允许重复产生业务副作用。每个原任务最多一个直接后继，继续重试沿后继任务推进；需要明确安全声明、版本、证据和审计，详见[人工恢复接口](api/task-recovery.md)。 流程关联任务的人工恢复统一经过 Weave，校验流程版本、步骤归属和当前恢复条件；不能直接从普通任务重试入口生成无人接续的流程任务。 diff --git a/docs/05-production-acceptance.md b/docs/05-production-acceptance.md index 9e33a93..7a0396e 100644 --- a/docs/05-production-acceptance.md +++ b/docs/05-production-acceptance.md @@ -1,6 +1,6 @@ # Orbit 生产验收与发布门槛 -> 版本 0.2 · 2026-09-17。本文是全平台生产发布门槛。P0–P2 已积累阶段证据，见 [开发计划](08-development-plan.md) 与 [P2 验收](reports/phase-2.md)；P3 新增 [计划调度证据](reports/phase-3-schedules.md)。未覆盖的场景继续实施，尚无全平台生产验收或容量结论。 +> 版本 0.2 · 2026-09-17。本文是全平台生产发布门槛。P0–P2 已积累阶段证据，见 [开发计划](08-development-plan.md) 与 [P2 验收](reports/phase-2.md)；P3的逐项功能及故障证据边界见[阶段核对](reports/phase-3.md)，新增[相对延迟与重试全停恢复](reports/phase-3-delay-recovery.md)。未覆盖的场景继续实施，尚无全平台生产验收或容量结论。 ## 1. 验收对象与证据 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 821c6f3..5d826e0 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -182,3 +182,7 @@ HTTP/SDK创建时间范围使用包含下界、不包含上界；v2游标以固 ### 业务事务Outbox示例模型 [独立task-outbox](examples/task-outbox.md)包含sender_outbox和sender_jobs，2个业务实体、22个Go/SQL中文注释字段，无外键；不计入平台35个实体和289字段。Outbox保存固定命名空间/队列、业务操作、原提交键/请求/摘要、状态/次数/代次/到期时间及原受理回执，业务行同事务创建和关联。命名空间与队列范围内按wake_at索引领取，终结状态wake_at为空；发送次数与总发送期限有界。 + +## 相对延迟的持久化边界 + +delay_ms是普通/批量提交命令字段，首次事务以同一个数据库时间写入created_at和available_at；不增加表字段、迁移或外键。原时长参与幂等请求摘要，原回执保存初始时间；重报不能重新计时。后续安全重试可以更新当前available_at，首次受理回执及历史尝试仍保留。Go命令字段提供中文说明，见[任务契约](api/tasks.md)和[回归报告](reports/phase-3-delay-recovery.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index efd89fc..88fcc21 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -27,8 +27,8 @@ | 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与示例已通过全量回归；任务直接来源（普通/人工重试/计划固定版本）及SDK示例已通过全量回归；事务Outbox独立业务示例、CLI及故障恢复已实现并通过全量链路回归，待完成P3整体范围验收 | -| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 已完成平台功能阶段验收：[原范围逐项核对](reports/phase-3.md)、[本轮完整回归](reports/phase-3-delay-recovery.md)。补齐相对延迟、改参数关联示例和四种重试全停恢复场景；20个可运行示例全部通过。全平台生产门槛及SDK首版交付剩余项仍按P6–P7推进 | +| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 下一阶段：先实现定义校验、不可变版本发布及独立SDK，再接实例推进与持久化等待 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | | P7 首版交付核对 | SDK 兼容与发布准备、完整 API 手册、场景示例、全量回归及交付清单 | 新接入者按文档跑通全部示例；所有首版能力可追溯到实现与测试；报告实际生产边界 | 待开发 | @@ -37,21 +37,23 @@ 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验收](reports/phase-3.md)逐项保留实现证据及尚待P4–P7完成的原范围，不按已有模块反推全平台完成。 | 核对项 | 当前证据与后续交付 | | --- | --- | +| 相对延迟与退避全停恢复 | 已实现delay_ms、数据库计时与原回执恢复，[调用示例](examples/delayed-task.md)和[回归记录](reports/phase-3-delay-recovery.md)同步；四种真实进程场景保持原退避、预算、截止与未知效果 | +| 改参数形成关联新任务 | [查询示例](examples/task-query.md)以新提交身份和相同business_key创建不同参数任务，逐项读取确认旧输入保留；不自动取消原任务，也不冒充人工retry_of | | 批量任务受理 | 已实现[批量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)同步记录 | | 任务详情来源追踪 | 已实现[独立来源接口](api/task-source.md)、SDK和普通/计划/人工恢复场景，返回计划及触发固定版本；同命名空间归属、当前权限和来源缺失拒绝已通过[全量回归](reports/phase-3-task-source.md)，流程来源随Weave接入 | | 业务到平台的事务Outbox接入 | 已实现[独立业务模块及完整接入](examples/task-outbox.md)，覆盖业务变更/待发请求同事务、有限发送恢复、CLI及真实HTTP/MySQL/进程故障，见[报告](reports/phase-3-task-outbox.md)；核心SDK保持无ORM/数据库依赖 | -凭据安全重绑、历史清理、全量监控/容量与MySQL故障切换仍按P6执行；Weave的通用执行门控/内部事件接续随P4–P5落实，不将外部结果订阅当作内部流程引擎。 +精确A04派发消息未送达时的进程故障、慢消费者/重连风暴、凭据安全重绑、历史清理、全量监控/容量与MySQL故障切换仍按P6执行；Weave的通用执行门控/内部事件接续随P4–P5落实，不将外部结果订阅当作内部流程引擎。 ### P3 事务Outbox接入的验收范围 -以下内容已由独立业务示例实现并进入回归，实际证据见[Outbox报告](reports/phase-3-task-outbox.md)。P3仍需逐项对照原功能范围完成整体验收，再推进Weave；不能仅依据本表无待开发行即宣布阶段完成： +以下内容已由独立业务示例实现并进入回归，实际证据见[Outbox报告](reports/phase-3-task-outbox.md)。P3整体范围已另行[逐项核对](reports/phase-3.md)，下表保留该子项的验收要求： 1. 独立业务示例module，以业务记录和待发请求同一个MySQL事务提交；核心SDK保持无ORM/数据库依赖，业务表示例也禁止外键，字段提供中文注释。 2. 同时冻结目标命名空间/队列、原幂等键、完整请求及内容摘要。相同业务操作重报不新建待发项，内容不同必须冲突；凭据原文不能进入Outbox或日志。 @@ -59,7 +61,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 4. 平台已受理但响应丢失、业务COMMIT回执未知、发送进程退出，都必须用原身份核对。只确认平台受理，不将受理标记当作业务执行成功。 5. 退避、发送次数及总发送期限均有上限，补发窗口不超过平台幂等保留期；权限、内容冲突、预算耗尽和不能可靠确认的响应需保留可核查记录，不生成新键绕过。 6. 真实数据库回归覆盖业务回滚、并发重复、发送竞争、过期租约、迟到确认和原始数字精度；真实HTTP丢响应、MySQL COMMIT响应丢失和发送进程SIGKILL后恢复一并验收。 -7. 提供完整初始化、提交、运行/恢复、状态核对及失败处理命令；接入全量回归，记录实际测试证据后再完成P3范围核对。 +7. 提供完整初始化、提交、运行/恢复、状态核对及失败处理命令；接入全量回归，记录实际测试证据，并纳入P3范围核对。 ## 4. SDK 与示例交付清单 @@ -67,7 +69,8 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 场景 | 预期调用链 | 所属阶段 | | --- | --- | --- | | 健康检查 | 创建客户端 → 存活/就绪查询 | P0 | -| 文件转换任务 | 提交幂等任务 → Worker 处理 → 查询结果 | P1–P2 | +| 文件转换任务 | 提交幂等任务 → Worker实际转换 → 查询结果 | 当前仅有受理/取消示例；真实转换例子仍须在P7首版交付前补齐 | +| 相对延迟执行 | 受理时按数据库时间等待 → Worker实际执行 → 历史时间核对 → 原键重报不重计时 | P3，已实现完整示例 | | 有限安全重试 | 明确失败 → 持久化退避 → 新尝试 → 查询完整历史 | P3，已实现完整示例 | | 定期数据同步 | 发布计划 → 触发任务 → 暂停/恢复 → 查看触发记录 | P3，已实现通用快照示例 | | 运行中取消 | 保存取消意图 → 停止处理器 → 确认结果 → 查询并排空 | P3，已实现完整示例 | @@ -116,6 +119,8 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P3 业务键与时间查询回归](reports/phase-3-task-queries.md) - [P3 任务直接来源回归](reports/phase-3-task-source.md) - [P3 业务事务Outbox回归](reports/phase-3-task-outbox.md) +- [P3 相对延迟与全停恢复回归](reports/phase-3-delay-recovery.md) +- [P3 原功能范围核对](reports/phase-3.md) - [接口手册](api/README.md) - [SDK 接入](sdk/README.md) - [通用场景示例](examples/README.md) diff --git a/docs/README.md b/docs/README.md index b7e949a..706a0ca 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本：0.2 - 整理日期：2026-09-17 -- 阶段：P0–P2 已通过阶段验收；P3 安全重试、计划调度、运行中取消、队列资源控制和应用共享配额已通过回归，资源键许可及真实进程恢复已通过全量回归，优先级/节点公平派发及真实隔离场景已通过全量回归，批量任务受理、独立任务人工核查/重试及业务键/时间查询已实现，数据库已改为无外键，P3–P7 继续推进，尚未完成全平台生产验收 +- 阶段：P0–P3平台功能已通过阶段回归；[P3逐项验收](reports/phase-3.md)保留原范围与证据边界。[相对延迟及重试全停恢复](reports/phase-3-delay-recovery.md)已通过完整回归。继续P4–P7，尚未完成全平台生产验收，数据库保持无外键。 - 适用范围：Orbit 统一平台，包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 @@ -82,4 +82,4 @@ Orbit 不内置特定行业规则；Pulse 管理执行，Weave 管理步骤依 2026-09-17，文档更新至 0.2：用户确认研发定义发布流程及 Weave 首版功能，明确纳入有界重复查询和暂停/继续等生产管理能力。技术细则与参数继续按建议或待定标注，未将功能确认等同于实现或验收通过。 -持续开发目标已建立。当前实现和验收进度以 [开发计划](08-development-plan.md)、[P0 报告](reports/phase-0.md)、[P1 报告](reports/phase-1.md) 与 [P2 阶段验收](reports/phase-2.md) 、[P3 安全重试报告](reports/phase-3-safe-retry.md) 和 [P3 计划调度报告](reports/phase-3-schedules.md) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 +持续开发目标已建立。当前实现和验收进度以 [开发计划](08-development-plan.md)、[P0 报告](reports/phase-0.md)、[P1 报告](reports/phase-1.md) 与 [P2 阶段验收](reports/phase-2.md) 、[P3 安全重试报告](reports/phase-3-safe-retry.md) 、[P3 计划调度报告](reports/phase-3-schedules.md) 和 [P3 完整范围验收](reports/phase-3.md) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 diff --git a/docs/api/README.md b/docs/api/README.md index d7a7e71..3bc41d3 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -20,6 +20,7 @@ | 任务 `priority` 与节点领取池 | P3 已实现有界优先级加速、应用/队列轮转 | [优先级与公平派发](priority-and-fairness.md) | | 凭据校验与初始化 CLI | P1 已实现 | [鉴权](authentication.md) | | `/v1/worker/connect` | P2 已实现并通过阶段验收 | [Worker 契约](worker.md)、[Schema](worker-schema.md) | +| 任务相对延迟 `delay_ms` | 已实现数据库计时、原键恢复与实际执行例子 | [任务接口](tasks.md)、[延迟示例](../examples/delayed-task.md) | | 任务冻结策略与 Worker `retryable` | P3 安全重试已实现 | [安全重试](retries.md) | | `/v1/tasks/{task_id}:cancel` 与 `task.cancel` 通知 | P3 运行中取消已实现 | [取消及 SDK 场景](cancellation.md) | | `/v1/schedules`、不可变版本、暂停/恢复及触发查询 | P3 已实现 | [计划接口与 SDK 场景](schedules.md) | diff --git a/docs/api/retries.md b/docs/api/retries.md index af7a45e..36d609f 100644 --- a/docs/api/retries.md +++ b/docs/api/retries.md @@ -52,7 +52,9 @@ available_at = 本次结束事务中的数据库时间 + delay_ms - 等待重试期间不占执行许可，任务不带 result/completed_at；上一尝试的错误和结果保留在 [执行历史](execution-history.md)。NEEDS_REVIEW 也没有 completed_at。 - 重启后以数据库 available_at 继续领取，不依赖某一节点内存中的定时器。 -`task.requeued` 等持久化事件记录 decision、delay_ms、next_available_at、attempt_count、max_attempts、effect_uncertain。decision 包括 RETRYABLE_FAILURE、IDEMPOTENT_UNKNOWN、UNSTARTED_LEASE_EXPIRED、ATTEMPT_LIMIT、DEADLINE_EXCEEDED、RECONCILIATION_REQUIRED 等。当前可靠事件订阅尚未开放。 +`task.requeued` 等内部持久化事件记录 decision、delay_ms、next_available_at、attempt_count、max_attempts、effect_uncertain。decision 包括 RETRYABLE_FAILURE、IDEMPOTENT_UNKNOWN、UNSTARTED_LEASE_EXPIRED、ATTEMPT_LIMIT、DEADLINE_EXCEEDED、RECONCILIATION_REQUIRED 等。[可靠结果订阅](subscriptions.md)已开放规定的结果类型；内部重排事件不自动成为外部可订阅类型。 + +提交参数`delay_ms`只控制首次受理的最早执行时间；它与这里按每次失败计算的退避分别定义。真实进程回归已覆盖退避期间唯一节点退出、两个新节点竞争恢复，以及停机期间总时限耗尽；冻结策略、已用次数和未知效果均保留，见[验证记录](../reports/phase-3-delay-recovery.md)。 取消接口支持独立任务和计划任务的 WAITING、LEASED、RUNNING 与 NEEDS_REVIEW。取消意图阻止后续尝试；运行中继续收集结果，普通取消错误或失联保持 UNKNOWN，不因 IDEMPOTENT 声明再次执行。此前未核实的效果不会被本次确认取消清除，详见 [取消状态与结果表](cancellation.md)。 diff --git a/docs/api/schedules.md b/docs/api/schedules.md index cf77fde..5c711b9 100644 --- a/docs/api/schedules.md +++ b/docs/api/schedules.md @@ -54,7 +54,7 @@ Cron 段顺序为分钟、小时、月日、月、星期；支持 `*`、列表 | catchup_window_ms | 必填 1000–2592000000，未处理触发点最多保留的时间窗口 | | grace_ms | 必填 1000–60000，正常派发允许延迟；不能超过 catchup_window_ms | -所有任务策略在 task 内冻结，字段同 SubmitTaskRequest；payload 最大 256 KiB，数字文本保持精确。task 中禁止绝对 available_at / start_deadline_at，每次生成时立即可受理，task_timeout_ms 从本次实际创建时间计算。计划结束不取消已经生成的任务，也不缩短其任务时限。 +所有任务策略在 task 内冻结，字段同 SubmitTaskRequest；payload 最大 256 KiB，数字文本保持精确。task 中禁止绝对 available_at / start_deadline_at 和非零 delay_ms，时间由计划规则定义，不静默忽略这些字段。每次生成时立即可受理，task_timeout_ms 从本次实际创建时间计算。计划结束不取消已经生成的任务，也不缩短其任务时限。 生成任务受 [队列积压上限](queues.md) 约束。满额时保留未受理触发点，本批较早已生成部分仍提交；释放空间后按冻结的漏跑策略接续，超过补偿窗口的点仍按策略跳过。队列 paused 仅停止领取/开始，停止计划生成需使用计划暂停接口。 diff --git a/docs/api/tasks.md b/docs/api/tasks.md index 79f5091..3cd7f8c 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -26,6 +26,7 @@ P2 进展：服务端执行内核、WSS 与独立自动 Worker SDK 已可产生 | business_key | string | 可选 | 最多 191 个 Unicode 字符，无 NUL/换行；用于关联，不提供业务幂等 | | concurrency_key | string | 可选，空为关闭 | 1–191 个 Unicode 字符，无控制字符和首尾空白；显式资源互斥键，同命名空间跨队列共享，见 [资源键契约](keyed-concurrency.md) | | priority | 整数 | 可选，默认 0 | 0–9，冻结优先级，每级30秒排序优势；不提前延时任务，见 [优先级与公平派发](priority-and-fairness.md) | +| delay_ms | uint64 | 可选，默认 0 | 相对首次受理事务数据库时间的等待毫秒数，必须小于 task_timeout_ms；非零不能与 available_at 同用 | | available_at | time / null | 受理时数据库时间 | 最早可执行时间；必须早于总截止时间，仅允许最多 1 秒的过去时间容差 | | start_deadline_at | time / null | 无独立限制 | 必须晚于当前时间，不早于 available_at，不晚于总截止时间 | | task_timeout_ms | uint64 | 0 或省略为 86400000 | 从受理时计算的总时限，最大 30 天（2592000000 毫秒） | @@ -34,6 +35,10 @@ P2 进展：服务端执行内核、WSS 与独立自动 Worker SDK 已可产生 处理器及版本名称以 ASCII 字母或数字开头，后续只能包含字母、数字、`_ . : -`。时间年份为 2000–9998。 +相对延迟通过`delay_ms`提交：首次受理时保存`available_at = created_at + delay_ms`，同键重报返回原受理回执，不能重新计时。`delay_ms`省略或0保留原有立即/绝对时间语义；负数、小数、字符串、null和溢出值均拒绝。总时限包含等待，最晚开始时间仍须满足上表。任务响应不新增delay_ms；当前available_at可能在安全重试后变为下一次退避时间，首次时间以原提交回执为准。见[完整SDK示例](../examples/delayed-task.md)。 + +批量每项支持相同字段，各项使用自己的首次受理数据库时间。相对时长和绝对时刻属于不同请求语义，即使碰巧对应相同时间，也不能在原幂等键下互换。启用delay_ms前先更新服务端；旧服务会拒绝未知字段，省略/零值不改变已有请求摘要。 + | retry 字段 | 默认与限制 | | --- | --- | | max_attempts | 0 或省略为 1；范围 1–100，包含首次尝试 | diff --git a/docs/examples/README.md b/docs/examples/README.md index 1cde70f..2579695 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -10,6 +10,12 @@ - 覆盖：8个并发调用/20项领取、业务回滚、旧代次拒绝、HTTP回执丢失、三处MySQL COMMIT回执丢失及发送进程SIGKILL恢复。 - 边界：业务侧两张表均无外键，22个中文注释字段；固定发送预算，REVIEW保留原记录供业务管理核查，核心SDK不引入数据库依赖。 +## 已可运行：相对延迟执行 + +- [完整命令、输入输出和失败处理](delayed-task.md)，[SDK源码](../../sdk/go/examples/delayed-task/main.go)。 +- delay_ms按数据库首次受理时间计时，实际Worker执行后查询开始时间；重报原请求仍返回原任务及时间。 +- 自动失败重试单独使用retry策略，初始延迟不在每次尝试中重新计时。 + ## 已可运行：健康检查 - 源码：[SDK 健康检查](../../sdk/go/examples/health/main.go)。 diff --git a/docs/examples/delayed-task.md b/docs/examples/delayed-task.md new file mode 100644 index 0000000..fd7f5bb --- /dev/null +++ b/docs/examples/delayed-task.md @@ -0,0 +1,59 @@ +# 相对延迟与原请求恢复 + +源码：[delayed-task/main.go](../../sdk/go/examples/delayed-task/main.go)。示例提交等待1500毫秒的对象任务，启动自动Worker，核对实际执行历史，再在任务完成后重报原请求。处理器原样返回示例JSON，无外部副作用。 + +## 准备与运行 + +先完成[迁移和Worker凭据初始化](../sdk/worker.md#1-可运行的文本规范化示例)，启动当前Orbit。凭据需要任务提交、读取及Worker连接/消费权限，队列须有可用额度。仓库根目录执行： + +```sh +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/worker.credentials.json \ + go run ./examples/delayed-task +``` + +也可用ORBIT_TOKEN和ORBIT_QUEUE_ID覆盖凭据文件中的对应值。示例总上下文20秒、任务总时限15秒、单次执行5秒、最多1次尝试；每次运行使用独立操作及处理器身份，不会消费其他处理器的任务。 + +预期输出： + +```text +task=<任务ID> delay_ms=1500 +state=SUCCEEDED attempts=1 no_early_start=true original_receipt=true worker_drained=true +``` + +检查依据是同一个数据库的CreatedAt、AvailableAt及尝试创建/开始时间，不把客户端墙钟或sleep结束当作执行授权。Worker正常排空后退出；任务、结果和审计保留。 + +## SDK调用 + +以下为函数中的调用片段，完整imports及错误处理见源码： + +```go +request := orbit.SubmitTaskRequest{ + QueueID: queueID, + Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", + Payload: json.RawMessage(`{"object_ref":"objects/input"}`), + DelayMS: 5000, + TaskTimeoutMS: 60000, + Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 10000}, +} +accepted, err := client.SubmitTask(ctx, operationKey, request) +if err != nil { return err } +fmt.Println(accepted.Task.ID, accepted.Task.AvailableAt) +``` + +`delay_ms`相对首次成功受理事务中的数据库时间计算，保存为`available_at`。同一幂等键重报仍提交原等待时长；不按当前时间重算available_at，不生成新键。响应是原受理回执，即使任务后来已经成功，回执中仍可能是WAITING；当前状态通过GetTask查询。 + +指定绝对时刻时使用AvailableAt并令DelayMS为0。非零delay_ms和available_at不能同时提供。等待时长必须小于task_timeout_ms；总时限包含等待时间。start_deadline_at不能早于计算后的available_at。所有时间规则及400/409错误见[任务接口](../api/tasks.md)。 + +服务端停机、队列暂停、没有兼容Worker或额度不足都可能让开始时间晚于available_at；到期后才可领取，不承诺准点执行。自动失败重试使用retry中的退避策略，不再次应用初始delay_ms。 + +## 失败与持久化边界 + +示例网络错误时非零退出，已打印任务ID可用于核对；不在同一次运行中换键重提。真实接入须在第一次请求前保存操作身份和请求，可使用[业务事务Outbox](task-outbox.md)。本例生成的新身份仅用于独立演示，重新运行不是上一次进程的恢复。 + +Outbox若使用delay_ms，等待从Orbit受理开始计算，不从本地业务事务或Outbox创建时间计算。需要固定业务时刻时冻结available_at；若发送时该时刻已超出接口容差，按拒绝结果核查，不静默改成“从现在起再等”。 + +旧服务不识别delay_ms。先更新服务端，再让调用方启用该字段；省略/零值保持旧请求摘要。计划模板拒绝非零delay_ms，时间由计划规则定义。人工重试的新开始时间使用其专门接口参数，不继承原任务的初始延迟。 + +`make integration`执行此例、真实HTTP/Worker延迟用例，以及重试退避期间全节点退出后的恢复测试；结果见[本轮报告](../reports/phase-3-delay-recovery.md)。 diff --git a/docs/examples/task-outbox.md b/docs/examples/task-outbox.md index eb2d26b..ebb94d1 100644 --- a/docs/examples/task-outbox.md +++ b/docs/examples/task-outbox.md @@ -172,3 +172,5 @@ send_count统计本地发送领取，与Pulse的执行尝试次数分别计数 本例的业务效果仅为本地SQL记录及平台受理，不提供外部服务恰好一次执行。业务Worker继续承担业务幂等；执行结果需要[可靠结果订阅与Inbox](result-inbox.md)等衔接。敏感请求和回执留在业务数据库，应纳入业务访问控制、审计、备份及保留策略；日志只输出操作ID和稳定状态，不输出token或请求正文。 REVIEW后的人工核查、证据审批及受控恢复由接入方的业务管理流程承担，示例不提供无证据清空错误/重置预算接口。不要直接修改SQL状态制造成功，也不要在无法确认的情况下换业务操作ID。生产MySQL故障切换、长期归档、跨部署迁移和容量仍按P6验收；本例不启动流程或发送Weave信号，相应Outbox命令类型随流程API演进。 + +相对延迟任务可在SubmitTaskRequest设置DelayMS；Outbox冻结并重发原时长，由Orbit首次受理事务计算执行时间。它不从本地业务创建时间起算；固定业务时刻应使用冻结AvailableAt，详见[延迟接入](delayed-task.md)。 diff --git a/docs/examples/task-query.md b/docs/examples/task-query.md index 8ee8e7d..3737db7 100644 --- a/docs/examples/task-query.md +++ b/docs/examples/task-query.md @@ -17,7 +17,7 @@ ORBIT_URL=http://127.0.0.1:8080 \ ## 实际调用与结果 -1. 提交两个完全相同业务键的任务，以及一个只相差尾部空格的任务。业务键含中文、加号、百分号及重音字符。 +1. 使用不同提交身份创建两个完全相同业务键、format分别为text/json的任务，以及一个业务键只相差尾部空格的任务。业务键含中文、加号、百分号及重音字符。读取前两个任务确认各自输入保留，改参数没有覆盖原任务。 2. 用服务端回执中的CreatedAt设置包含起点、不包含终点的时间范围，避免依赖客户端时钟同步。 3. ListTasks按原键、WAITING状态和创建区间查询，每页1条，沿NextCursor遍历；仅匹配前两个任务。 4. 将created_before设为第二个任务的精确创建时间，确认它被排除，只剩第一个任务。 @@ -26,11 +26,13 @@ ORBIT_URL=http://127.0.0.1:8080 \ 预期输出： ```text -matched=2 pages=2 before_excluded=true cancelled=3 +matched=2 pages=2 before_excluded=true original_payload_preserved=true cancelled=3 ``` 重复运行创建新的独立样本。代码没有外部业务副作用，也不删除数据库历史。失败非零退出，可能留下已受理任务；用回执、业务键和时间范围查询后核对，各任务总时限60秒，当前示例没有Worker会进入未开始超时处理。 +本例通过business_key关联不同参数的任务，不表示覆盖、自动取消或自动代替原工作，来源仍为DIRECT。需要重新执行原冻结参数时使用带证据的[人工重试](../api/task-recovery.md)；不能用新任务身份绕过未知业务效果核查。 + 完整参数、游标兼容和异常说明见[查询契约](../api/task-queries.md)。翻页时保持所有筛选，发生INVALID_ARGUMENT不要丢弃过滤条件“兜底查全部”。状态变化的跨页结果不是快照，不能当作完整审计导出。 `make integration` 会在一次性MySQL中运行此示例及实际HTTP SDK回归，验证原键/尾部空格、时间边界、分页、变更条件拒绝，以及取消后创建时间不改变。 diff --git a/docs/reports/phase-3-delay-recovery.md b/docs/reports/phase-3-delay-recovery.md new file mode 100644 index 0000000..67d0886 --- /dev/null +++ b/docs/reports/phase-3-delay-recovery.md @@ -0,0 +1,74 @@ +# P3 相对延迟与重试全停恢复回归 + +验证日期：2026-09-17。本轮从[原范围核对](phase-3.md)发现并补齐相对等待时长的接口，同时补足“重试退避期间所有服务退出”的真实进程证据。全平台目标仍包含P4–P7，未宣称生产高可用或容量验收。 + +## 1. 实现与兼容 + +- 服务端和独立SDK新增可选delay_ms，默认0；首次受理在同一事务按数据库时间计算available_at，原等待时长参与请求摘要，重报先返回原身份回执，不重新计时。 +- delay_ms小于任务总时限，非零不能与available_at同用；start_deadline_at仍校验计算后的最早执行时间。类型/溢出/非法组合明确拒绝。 +- 零值omitempty维持历史请求摘要；[兼容测试](../../internal/storage/mysql/task_delay_test.go)以新增字段前的规范化JSON形状计算摘要作比较。批量复用同一受理逻辑；Outbox保留时长，不在业务发送端换算。 +- 计划模板明确拒绝非零delay_ms，由已发布规则决定时间；人工重试使用原有新任务时限/绝对时间参数，不重新继承首次延迟。 +- 不新增迁移、外键或持久化字段；仍为35个平台模型/289个直接持久化字段。新增命令字段有中文说明，执行后的退避继续使用既有available_at。 +- [延迟SDK示例](../examples/delayed-task.md)执行实际Worker并查询开始时间；[查询示例](../examples/task-query.md)增加不同参数共享业务关联键及原输入保留核对。接口与持续计划同步。 + +## 2. 本轮测试的实质断言 + +| 测试 | 核对内容 | +| --- | --- | +| 规范化/解码 | 小于总时限边界、最大整数拒绝、相对与绝对互斥、null/负数/小数/字符串/溢出/重复字段拒绝；规范化不能提前将相对时间解析为客户端时刻 | +| HTTP/MySQL受理 | available_at精确等于created_at+5000ms；8个同键并发返回同一ID及原时间；改变时长或改为对应绝对时刻冲突；错误时间关系不受理；显式0兼容旧请求；批量每项使用自己的首次受理时间 | +| 真Worker延迟 | 相对1500ms及绝对时刻两项任务实际完成，尝试创建和开始时间均不早于保存的available_at；等待经过后原键重报仍取原回执，已完成任务不重开 | +| Outbox | 冻结文档保留delay_ms及精确数字，不创建available_at，不改变数据库依赖边界 | +| 改参数的关联任务 | 相同business_key、不同提交身份与不同format，GetTask确认两份输入各自保留；查询后取消三个样本，保留审计 | + +进程测试：[TestIntegrationWorkerProcessRetryBackoffRecovery](../../sdk/go/retry_process_integration_test.go)。四个子场景都先确认第一次结果已入库并进入WAITING，保存完整旧尝试；Process.Kill终止唯一Orbit节点，Wait确认信号退出，然后启动两个不同node_id的新进程。 + +1. 明确可重试失败：8秒上限、半区间抖动，重启后完整任务快照不变；两个节点竞争只取得一次Epoch=2，实际领取时间不早于旧available_at，最终成功。 +2. IDEMPOTENT未知效果：第二次明确可重试失败恰好耗尽2次预算，保留effect_uncertain，进入NEEDS_REVIEW，不能伪装为普通失败或创建第三次尝试。 +3. 明确失败后退避，停机跨过5秒总时限：重启后FAILED，只保留首次尝试，不因恢复开启第二次。 +4. 未知效果后退避，停机跨过总时限：重启后NEEDS_REVIEW且不确定性保留，同样只有首次尝试。 + +全部场景再次读取旧尝试，逐字段确认结果/分类/结束时间未被改写；核对冻结策略、总截止、输入、已用次数及队列未终结数。新节点保持在线并有容量，短窗口再次读取分配确认没有额外尝试。测试不修改数据库的available_at、deadline_at、lease_until；到期自然流逝，时间断言使用数据库记录。它验证进程/数据库逻辑，不验证跨故障域复制或外部系统副作用。 + +## 3. 环境、命令与实际结果 + +Darwin 24.6.0 arm64、12核/32 GiB宿主、Go 1.27.1、Docker内MySQL 8.4.10单实例。此硬件描述不是压测配置或容量结论。所有数据库为脚本创建的一次性容器，随机密码/回环端口，凭据仅存在私有临时文件；强杀仅影响测试拥有的子进程。 + +```sh +make fmt test vet build +make integration +``` + +make integration在停止初始服务后执行进程组，避免残留恢复节点掩盖接管；本轮新增测试已包含在原WorkerProcess匹配规则，新增delayed-task示例也进入正式脚本。 + +定向进程组在隔离库中使用已编译服务和专用Worker凭据执行： + +```sh +cd sdk/go +go test -race -count=3 \ + -run '^TestIntegrationWorkerProcessRetryBackoffRecovery$' -v . +``` + +需要隔离环境提供ORBIT_TEST_MYSQL_DSN、ORBIT_TEST_SERVER_BINARY、ORBIT_TEST_WORKER_URL、ORBIT_TEST_WORKER_CREDENTIALS_FILE；未提供时测试跳过，不能算数据库/进程证据。 + +| 执行 | 实际结果 | +| --- | --- | +| 四种进程场景重复3轮 | PASS，33.42 / 33.05 / 37.18秒，总包104.892秒，命令终态0；随后新增delay_ms再通过全量组验证 | +| 格式/竞态单元/静态检查/四模块构建 | PASS，make终态0；未设集成环境时的Integration跳过不计入真实数据库证据 | +| 全量集成/全部示例/进程组 | PASS，正式脚本终态0，18个核心SDK示例加Inbox/Outbox共20个，10个SDK节点/Worker进程测试（新增测试含4场景），外加Inbox/Outbox各1个进程故障测试 | + +全量组记录：存储105.663秒；Worker91.765秒（新增相对/绝对延迟4.10秒）；计划6.881秒；队列4.506秒；批量1.502秒；查询2.972秒；订阅管理33.485秒；事件消费54.737秒；人工恢复3.358秒；Inbox34.630秒；Outbox33.802秒；应用配额1.322秒；最终进程组95.151秒（新增四场景33.15秒）。所有组均完成，未根据中途PASS推断脚本结束。结束后确认脚本创建的容器已清理。 + +新增例子实际输出state=SUCCEEDED、attempts=1、no_early_start=true、original_receipt=true、worker_drained=true；改参数查询例子输出original_payload_preserved=true，均由程序断言后产生。SDK在GOWORK=off依赖图中无服务端、模型或MySQL依赖。 + +本轮没有发现需修改执行内核的故障；相对延迟是原范围核对发现的遗漏实现。新增回归和示例在同一次最终全量脚本执行前已经定稿，后续只更新文档证据。 + +本次修改前另执行零外键专项回归：MySQL 8.4.10、新建及v7升级、归属关系校验共11.738秒，终态0；本轮全量再次通过对应回归。所有耗时仅为回归耗时，不是性能指标。 + +本地执行日志位于/tmp/orbit-retry-process-focused.log、/tmp/orbit-delay-check-final.log及/tmp/orbit-delay-integration.log；临时日志不是SDK或部署依赖。本报告保留关键结果与可复现入口。 + +## 4. 限制 + +- delay_ms相对Orbit成功受理事务，不相对客户端发起或业务Outbox创建。可执行时间是下界，排队/暂停/故障可能推迟实际开始。 +- 旧服务不支持该输入字段，启用前先升级服务；没有据此宣称混合版本滚动升级通过。 +- 本轮没有MySQL主从切换、备份恢复、持续负载、重连风暴或生产容量数据；完整首版仍按[阶段核对](phase-3.md)与[持续计划](../08-development-plan.md)推进。 diff --git a/docs/reports/phase-3.md b/docs/reports/phase-3.md new file mode 100644 index 0000000..7704539 --- /dev/null +++ b/docs/reports/phase-3.md @@ -0,0 +1,64 @@ +# P3 Pulse 功能范围核对与阶段验收 + +核对日期：2026-09-17。依据[原功能说明](../01-functional-spec.md)、[可靠性设计](../04-reliability-and-data.md)、[开发计划](../08-development-plan.md)及[全平台发布门槛](../05-production-acceptance.md)逐项检查。此报告区分Pulse功能交付和全平台生产验收；后者仍未通过。 + +阶段结论：P3平台功能范围已核对并通过本轮完整回归，可推进P4 Weave。新增相对延迟及四种重试全停场景已包含在最终脚本，20个可运行示例通过，命令终态0。下表明确保留首版尚未完成的生产门槛及示例交付项；不能用阶段验收替代它们。 + +## 1. Pulse功能与证据 + +| 原要求 | 当前行为与主要证据 | +| --- | --- | +| 立即、相对延迟、指定时间提交 | 立即受理、数据库时间计算delay_ms、绝对available_at；[HTTP事务用例](../../internal/storage/mysql/task_integration_test.go)核对同键并发、相对时长、非法组合与批量；[SDK真实执行](../../sdk/go/task_delay_integration_test.go)核对数据库创建/开始时间不早于保存的执行时间 | +| 批量受理、逐项恢复 | 最多32项/512 KiB，单项事务和身份复用，区分明确失败/未知/未尝试；[批量报告](phase-3-batch-tasks.md)记录部分已提交后的真实HTTP回执丢失及原键补交 | +| 幂等及不可变参数/策略 | 任务、身份回执、事件与审计同事务；同语义原回执、变更冲突；[任务用例](../../internal/storage/mysql/task_integration_test.go)核对16并发只创建一次和精确数字；新增可选字段的[旧摘要兼容](../../internal/storage/mysql/task_delay_test.go) | +| 改参数形成关联新任务 | 新提交身份和相同business_key，旧输入保持；[查询示例](../../sdk/go/examples/task-query/main.go)分别读取text/json任务核对输入。该关联不等于业务去重或retry_of，原任务需调用方决定如何处理 | +| 单次/总时限、最晚开始、安全重试 | 冻结策略、有限退避及未知效果；[决策](../../pulse/retry/decision_test.go)、[真实事务](../../internal/storage/mysql/worker_retry_integration_test.go)核对预算/回滚/迟到；[全停恢复](phase-3-delay-recovery.md)保留退避、次数和截止，过期不补发 | +| 等待/运行取消 | 持久化意图、预占阻止开始、运行收集结果、未知效果仍核查；[取消事务](../../internal/storage/mysql/worker_cancel_integration_test.go)验证开始竞争/回滚，[取消报告](phase-3-cancellation.md)包含自动SDK通知、丢通知及不协作处理器 | +| 人工核查与重试 | 专门权限、版本、原因/证据、安全声明；核查不改旧尝试，人工重试创建唯一直接后继；[恢复用例](../../internal/storage/mysql/task_recovery_integration_test.go)验证竞争、审计失败回滚、原输入/结果保留及管理所有权；[接口/故障报告](phase-3-task-recovery.md) | +| Cron/固定间隔/时区/边界 | 五段Cron、fixed-rate、显式时区快照、开始/结束；[规则测试](../../pulse/schedule/rule_test.go)和[调度报告](phase-3-schedules.md)覆盖夏令时缺失/重复时刻；不支持的fixed-delay明确拒绝 | +| 唯一触发、版本与漏跑 | SKIP/COALESCE/CATCH_UP及有限窗口/批次/持久化限速，任务与进度同事务；[调度事务](../../internal/storage/mysql/schedule_integration_test.go)竞争只生成3项、扫描冷却有效、版本和覆盖范围保留 | +| 计划暂停、重叠与恢复 | PARALLEL/SERIAL/SKIP，暂停不取消旧任务，串行遇未知效果继续阻塞；[调度进程恢复](../../sdk/go/worker_process_integration_test.go)先杀唯一节点，再由两个新节点恢复真实积压 | +| WSS、能力、容量、租约、排空 | 独立Worker及自动SDK；[P2验收](phase-2.md)和[执行事务](../../internal/storage/mysql/worker_integration_test.go)核对16个竞争请求只产生2个允许的许可、重连不重置容量、旧会话/旧尝试拒绝 | +| 队列与应用共享额度 | 应用额度→队列统一锁顺序，开始窗口、未终结数、限速及背压；[队列报告](phase-3-queue-resources.md)、[应用报告](phase-3-application-resources.md)含跨命名空间竞争、回滚、原回执及进程恢复 | +| 同资源互斥与旧执行隔离 | 显式concurrency_key、跨队列单许可、递增资源令牌；[资源键用例](../../internal/storage/mysql/keyed_concurrency_integration_test.go)验证竞争、释放一次、序号不回绕；[SDK/资源端示例](phase-3-keyed-concurrency.md)展示拒绝旧令牌 | +| 优先级、项目间派发与缓冲上限 | 0–9冻结优先级、等待补偿、节点内应用/队列轮转、有界池和发送缓冲；[优先级报告](phase-3-priority-fairness.md)包含真实数据库应用行锁阻塞时其他应用仍能领取。多节点不均衡下延迟和容量留P6 | +| 权限、隔离、敏感数据 | 身份/命名空间/动作/队列范围按事务检查；[任务](../../internal/storage/mysql/task_integration_test.go)、[来源](../../internal/storage/mysql/task_source_integration_test.go)及[历史](../../internal/storage/mysql/task_history_integration_test.go)回归验证跨范围拒绝、当前权限和不返回秘密；凭据安全轮换仍待P6 | +| 查询、审计与来源 | ID、状态、队列、精确业务键、创建时间和绑定筛选的分页；[查询](phase-3-task-queries.md)、[来源](phase-3-task-source.md)、[恢复审计](phase-3-task-recovery.md)。流程来源随Weave接入，损坏/未知所有者返回503，不能假装普通任务 | +| 持久化结果、订阅与接收方去重 | 结果流序号按提交顺序分配，有限交付/证明/ACK、暂停与人工补投；[订阅核心](phase-3-result-subscription-core.md)、[通信](phase-3-result-subscription-transport.md)、[消费器](phase-3-event-consumer-sdk.md)、[独立Inbox](phase-3-result-inbox.md)包含真实COMMIT和进程故障 | +| 业务事务到平台的可靠提交 | [独立Outbox](phase-3-task-outbox.md)冻结目标/操作/请求，业务记录同事务，HTTP事务外，原键恢复，24小时/100次预算，三处COMMIT响应丢失和真实发送进程退出 | +| SDK、接口文档与实例 | sdk/go为独立module；每项能力随附Markdown契约及可运行示例。文件转换样例目前只受理/取消，实际处理示例为文本规范化与对象任务；P7仍须交付计划要求的真实文件转换例子，不能把受理样例当作文件生成 | +| 数据模型、无外键 | 35个平台实体/289个直接持久化字段维持中文说明，SQL列同步COMMENT；[真实升级](../../internal/storage/mysql/no_foreign_keys_integration_test.go)核对70→0、原数据与323项未替换索引组件保留。后续禁止外键/级联 | + +## 2. 全平台故障门槛的证据边界 + +“平台逻辑已验证”不表示对应场景在任意生产拓扑达标。此表保留原A编号，未由当前证据覆盖的部分不能打勾为完成。 + +| 门槛 | 当前证据及未覆盖部分 | +| --- | --- | +| A01–A02 | 单项/批量并发幂等已验证；批量、人工恢复、Outbox有真实响应截断后的原身份恢复。首次受理及业务事务COMMIT未知不能更换身份 | +| A03 | 两个逻辑节点的真实MySQL竞争已核对配额/许可；本轮两个真实新进程仅一个拿到重试尝试 | +| A04 | 未开始租约回收及旧证明拒绝有事务回归；尚缺“派发已提交但task.assign尚未到达Worker时强杀节点”的精确进程注入，列入P6，不能用普通节点重启替代 | +| A05 | 真实start/finish确认丢失并先后强杀节点，由其他进程恢复同一尝试/回执，处理器只运行一次 | +| A06 | Worker已写入同步外部标记后强杀，平台保留UNKNOWN/核查且释放许可；幂等声明下有限恢复有事务/SDK证据。真实接入方业务写入的幂等仍需接入验收 | +| A07–A09 | 旧会话/代次、迟到证据、续租/完成/回收竞争、跨节点重连与资源令牌隔离已验证；物理旧进程仍可运行，需要业务资源端防护 | +| A10 | 重复容量、SDK接收缓冲耗尽、有界请求/发送及断线恢复已验证；持续慢消费者和重连风暴负载仍待P6 | +| A11–A12 | 取消竞争和UNKNOWN安全分流已验证；模拟处理器的错误不证明任意外部渠道都满足幂等/核查协议 | +| A13–A14 | 真实MySQL触发竞争、真实服务全停后计划恢复、版本/时区/漏跑已验证；生产大规模集中补偿性能仍待P6 | +| A15 | 结果交付、HTTP/WSS确认丢失、Inbox事务及消费者强杀后去重已验证；保留期清理与真实持久化介质保障仍待P6 | +| A16–A17 | 跨队列资源键和跨命名空间应用额度竞争、当前权限隔离已验证；多节点混合负载下公平性及凭据安全重绑仍待P6 | +| A18 | 受控MySQL COMMIT响应丢失已验证；MySQL停机、复制主节点切换、持久化/复制确认组合尚未验收 | +| A19–A20 | 混合版本滚动升级、清理/归档与重报补投竞争尚未实现完整验收；保留原首版范围，P6必须补齐 | +| W01–W17 | 尚未完成Weave运行引擎。模型和约束测试不能证明流程发布、推进、信号、并行、暂停、恢复或轮询已实现 | + +## 3. 后续仍需完成的原范围 + +1. P4：研发SDK定义与发布、不可变版本、实例、顺序/分支/有限并行汇合、信号与定时等待、同事务创建下一项任务、内部事件持久化接续。 +2. P5：有界轮询、独立轮次/尝试预算、暂停门控、继续、取消、人工恢复、新激活隔离及审计；Weave管理任务不能绕过所属流程。 +3. P6：上表未完成故障门槛、可轮换凭据和安全重绑、历史清理、完整指标/告警/日志策略、负载工具、单点/复制拓扑、备份恢复与发布运行手册。容量和RPO/RTO需真实拓扑证据。 +4. P7：包括实际文件转换在内的完整接入场景、独立SDK远端取包/版本与兼容验证、全部命令/链接/API/示例及首版清单复核。 + +上述内容没有因Pulse阶段验收而移出首版目标。当前仓库的实现及本地回归不能被描述为已上市产品的生产验收。 + +## 4. 复现与最新结果 + +本轮环境、命令、新增测试、完整脚本终态及限制见[延迟与停机恢复报告](phase-3-delay-recovery.md)。全量脚本已确认终态0；历史子阶段报告的“下一步/未实现”描述各自当时状态，当前范围以本表和持续计划为准。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 63f251f..8ac4029 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -4,6 +4,8 @@ module：`git.yuntuo.hk/k2safe/orbit/sdk/go`。SDK 使用独立 go.mod，依赖 Go 标准库与 `github.com/coder/websocket v1.8.15`，不引用服务端模型或数据库实现。 +任务相对延迟使用SubmitTaskRequest.DelayMS，由数据库首次受理时间计算，原键重报不重新计时；[完整延迟示例](../examples/delayed-task.md)包含Worker执行和历史核对。非零DelayMS不能与AvailableAt同用，旧服务须先升级。 + ## 1. 从初始化到运行示例 需要 Go 1.27.1、MySQL 8.4+。在仓库根目录，通过环境设置专用 ORBIT_MYSQL_DSN： diff --git a/examples/task-outbox/internal/outbox/store_test.go b/examples/task-outbox/internal/outbox/store_test.go index 885365b..ed91c91 100644 --- a/examples/task-outbox/internal/outbox/store_test.go +++ b/examples/task-outbox/internal/outbox/store_test.go @@ -26,12 +26,13 @@ func TestFrozenDocumentAndFailureClassification(t *testing.T) { s := &Store{namespace: testID(t), queue: testID(t)} in := testRequest(s.queue) in.TaskTimeoutMS = 0 + in.DelayMS = 1500 raw, err := s.document(in) if err != nil { t.Fatal(err) } var frozen orbit.SubmitTaskRequest - if json.Unmarshal(raw, &frozen) != nil || frozen.TaskTimeoutMS != 86400000 || frozen.Retry.MaxAttempts != 1 || frozen.Retry.AttemptTimeoutMS != 30000 || !bytes.Contains(frozen.Payload, []byte("9007199254740993")) { + if json.Unmarshal(raw, &frozen) != nil || frozen.TaskTimeoutMS != 86400000 || frozen.DelayMS != 1500 || frozen.AvailableAt != nil || frozen.Retry.MaxAttempts != 1 || frozen.Retry.AttemptTimeoutMS != 30000 || !bytes.Contains(frozen.Payload, []byte("9007199254740993")) { t.Fatal("frozen defaults or precision changed") } in.QueueID = testID(t) diff --git a/internal/storage/mysql/task_delay_test.go b/internal/storage/mysql/task_delay_test.go new file mode 100644 index 0000000..87b07e9 --- /dev/null +++ b/internal/storage/mysql/task_delay_test.go @@ -0,0 +1,34 @@ +package mysql + +import ( + "encoding/json" + "testing" + + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func TestRelativeDelayPreservesLegacySubmissionDigest(t *testing.T) { + // 升级前已规范化的实际请求形状；新可选字段为零时不能改变原幂等摘要。 + legacy := json.RawMessage(`{"queue_id":"11111111111111111111111111111111","handler":"object.prepare","handler_version":"v1","payload_version":"v1","payload":{"n":9007199254740993},"task_timeout_ms":86400000,"recovery_safety":"RECONCILE_ONLY","retry":{"max_attempts":1,"initial_delay_ms":0,"max_delay_ms":0,"attempt_timeout_ms":30000}}`) + var in service.SubmitInput + if err := json.Unmarshal(legacy, &in); err != nil { + t.Fatal(err) + } + in, err := service.NormalizeSubmit(in) + if err != nil { + t.Fatal(err) + } + want, err := requestDigest(legacy) + if err != nil { + t.Fatal(err) + } + got, err := requestDigest(in) + if err != nil || got != want { + t.Fatal("zero delay changed a legacy request identity", err) + } + in.DelayMS = 1000 + changed, err := requestDigest(in) + if err != nil || changed == want { + t.Fatal("nonzero delay missing from request identity", err) + } +} diff --git a/internal/storage/mysql/task_integration_test.go b/internal/storage/mysql/task_integration_test.go index b9fc5cb..c8d982b 100644 --- a/internal/storage/mysql/task_integration_test.go +++ b/internal/storage/mysql/task_integration_test.go @@ -393,4 +393,66 @@ func TestIntegrationTaskAPI(t *testing.T) { t.Fatal(err) } }) + t.Run("relative_delay_uses_database_acceptance_and_original_identity", func(t *testing.T) { + delayed := input + delayed.DelayMS = 5000 + key := "delayed-submit" + first := must("POST", "/v1/tasks", a.Token, key, marshal(delayed), 201) + var v service.SubmitResult + if err := json.Unmarshal(first.data, &v); err != nil || !v.Task.AvailableAt.Equal(v.Task.CreatedAt.Add(5*time.Second)) { + t.Fatal("delay not anchored to persisted database acceptance", err) + } + const n = 8 + delayedBody := marshal(delayed) + results := make(chan apiReply, n) + errs := make(chan error, n) + for range n { + go func() { + out, err := call("POST", "/v1/tasks", a.Token, key, delayedBody) + results <- out + errs <- err + }() + } + for range n { + out, err := <-results, <-errs + var duplicate service.SubmitResult + if err != nil || out.status != 200 || json.Unmarshal(out.data, &duplicate) != nil || !duplicate.Deduplicated || duplicate.Task.ID != v.Task.ID || !duplicate.Task.AvailableAt.Equal(v.Task.AvailableAt) || !duplicate.Task.DeadlineAt.Equal(v.Task.DeadlineAt) { + t.Fatal("repeated delay reset execution time or identity", err) + } + } + changed := delayed + changed.DelayMS++ + must("POST", "/v1/tasks", a.Token, key, marshal(changed), 409) + changed = delayed + changed.DelayMS = 0 + changed.AvailableAt = &v.Task.AvailableAt + must("POST", "/v1/tasks", a.Token, key, marshal(changed), 409) + changed.DelayMS = delayed.DelayMS + must("POST", "/v1/tasks", a.Token, "both-timing-modes", marshal(changed), 400) + changed = delayed + tooEarly := v.Task.CreatedAt.Add(time.Second) + changed.StartDeadlineAt = &tooEarly + must("POST", "/v1/tasks", a.Token, "delay-after-start-deadline", marshal(changed), 400) + changed = delayed + changed.DelayMS = 86400000 + must("POST", "/v1/tasks", a.Token, "delay-at-total-deadline", marshal(changed), 400) + if count("pulse_tasks", "namespace_id=? AND id=?", a.NamespaceID, v.Task.ID) != 1 || count("orbit_events", "namespace_id=? AND aggregate_id=?", a.NamespaceID, v.Task.ID) != 1 { + t.Fatal("delayed duplicate emitted another task or event") + } + // 省略与显式零保持原有即刻/绝对时间请求语义，不破坏升级前的摘要。 + zero := bytes.Clone(bytes.TrimSuffix(raw, []byte("}"))) + zero = append(zero, []byte(`,"delay_ms":0}`)...) + must("POST", "/v1/tasks", a.Token, "stable-submit", zero, 200) + batch := service.BatchSubmitInput{Items: []service.BatchSubmitItem{{IdempotencyKey: key, Task: delayed}, {IdempotencyKey: "delay-batch-new", Task: delayed}}} + batchReply := must("POST", "/v1/tasks:batch", a.Token, "", marshal(batch), 200) + var batchOut service.BatchSubmitResult + if json.Unmarshal(batchReply.data, &batchOut) != nil || len(batchOut.Items) != 2 || batchOut.Items[0].Receipt == nil || batchOut.Items[1].Receipt == nil || !batchOut.Items[0].Receipt.Deduplicated || batchOut.Items[0].Receipt.TaskID != v.Task.ID { + t.Fatal("batch did not reuse relative-delay identity") + } + fresh := must("GET", "/v1/tasks/"+string(batchOut.Items[1].Receipt.TaskID), a.Token, "", nil, 200) + var freshTask service.TaskView + if json.Unmarshal(fresh.data, &freshTask) != nil || !freshTask.AvailableAt.Equal(freshTask.CreatedAt.Add(5*time.Second)) { + t.Fatal("batch delay did not use its own acceptance time") + } + }) } diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index 2fb662a..8e9a960 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -221,7 +221,8 @@ func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in if err != nil { return service.TaskView{}, err } - available := now + // 相对时长保留在请求摘要中，只在首次受理时按数据库时钟转换；重报先返回原回执。 + available := now.Add(time.Duration(in.DelayMS) * time.Millisecond) if in.AvailableAt != nil { available = *in.AvailableAt } diff --git a/internal/transport/httpserver/tasks_test.go b/internal/transport/httpserver/tasks_test.go index db87f5b..77abb32 100644 --- a/internal/transport/httpserver/tasks_test.go +++ b/internal/transport/httpserver/tasks_test.go @@ -18,6 +18,13 @@ func TestBodyAndErrorBoundary(t *testing.T) { {`{"retry":{"max_attempts":1},"Retry":{"max_attempts":2}}`, "application/json", "INVALID_ARGUMENT"}, {`{}`, "text/plain", "UNSUPPORTED_MEDIA_TYPE"}, {`{"payload":{"arbitrary_key":9007199254740993}}`, "application/json", ""}, + {`{"delay_ms":1000}`, "application/json", ""}, + {`{"delay_ms":null}`, "application/json", "INVALID_ARGUMENT"}, + {`{"delay_ms":-1}`, "application/json", "INVALID_ARGUMENT"}, + {`{"delay_ms":1.5}`, "application/json", "INVALID_ARGUMENT"}, + {`{"delay_ms":"1000"}`, "application/json", "INVALID_ARGUMENT"}, + {`{"delay_ms":18446744073709551616}`, "application/json", "INVALID_ARGUMENT"}, + {`{"delay_ms":1,"delay_ms":2}`, "application/json", "INVALID_ARGUMENT"}, } { r := httptest.NewRequest(http.MethodPost, "/v1/tasks", strings.NewReader(test.body)) r.Header.Set("Content-Type", test.content) diff --git a/pulse/schedule/service.go b/pulse/schedule/service.go index 44527a7..ccf897e 100644 --- a/pulse/schedule/service.go +++ b/pulse/schedule/service.go @@ -16,7 +16,7 @@ import ( type Definition struct { Rule Rule `json:"rule"` // Rule 时间规则；start_at 是基准，不允许追溯到版本生效前产生新任务。 Policy Policy `json:"policy"` // Policy 明确的漏跑、重叠和补跑边界。 - Task service.SubmitInput `json:"task"` // Task 每次触发的模板，禁止绝对 available_at/start_deadline_at。 + Task service.SubmitInput `json:"task"` // Task 每次触发的模板，禁止 available_at/start_deadline_at 和非零 delay_ms。 } // CreateInput 创建持久化计划，暂停计划也保存完整的触发进度。 @@ -121,8 +121,8 @@ func Normalize(in Definition) (Definition, error) { if err := in.Policy.Validate(); err != nil { return in, fault.Invalid(err.Error()) } - if in.Task.AvailableAt != nil || in.Task.StartDeadlineAt != nil { - return in, fault.Invalid("schedule templates cannot contain absolute task times") + if in.Task.AvailableAt != nil || in.Task.StartDeadlineAt != nil || in.Task.DelayMS != 0 { + return in, fault.Invalid("schedule templates cannot contain task times or a nonzero delay_ms") } in.Task, err = service.NormalizeSubmit(in.Task) return in, err diff --git a/pulse/schedule/service_test.go b/pulse/schedule/service_test.go new file mode 100644 index 0000000..3e85d15 --- /dev/null +++ b/pulse/schedule/service_test.go @@ -0,0 +1,25 @@ +package schedule + +import ( + "encoding/json" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func TestScheduleDoesNotSilentlyIgnoreTaskDelay(t *testing.T) { + in := Definition{ + Rule: Rule{Kind: "FIXED_RATE", Timezone: "UTC", StartAt: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), IntervalMS: 1000}, + Policy: Policy{Misfire: "SKIP", Overlap: "PARALLEL", CatchupLimit: 1, CatchupWindowMS: 30000, GraceMS: 1000}, + Task: service.SubmitInput{QueueID: core.ID("11111111111111111111111111111111"), Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}, + } + if _, err := Normalize(in); err != nil { + t.Fatal(err) + } + in.Task.DelayMS = 1 + if _, err := Normalize(in); err == nil { + t.Fatal("schedule accepted a delay outside its firing rule") + } +} diff --git a/pulse/service/service.go b/pulse/service/service.go index 67d756f..8d96047 100644 --- a/pulse/service/service.go +++ b/pulse/service/service.go @@ -64,6 +64,9 @@ func NormalizeSubmit(in SubmitInput) (SubmitInput, error) { if in.TaskTimeoutMS > 30*24*60*60*1000 { return in, fault.Invalid("task_timeout_ms exceeds 30 days") } + if in.DelayMS >= in.TaskTimeoutMS || (in.DelayMS != 0 && in.AvailableAt != nil) { + return in, fault.Invalid("delay_ms must be less than task_timeout_ms and cannot be combined with available_at") + } if in.RecoverySafety == "" { in.RecoverySafety = pulse.RecoveryReconcileOnly } diff --git a/pulse/service/service_test.go b/pulse/service/service_test.go index 8e8c696..de191ba 100644 --- a/pulse/service/service_test.go +++ b/pulse/service/service_test.go @@ -24,7 +24,14 @@ func TestSubmitNormalizationAndBounds(t *testing.T) { "unbounded attempts": func(v *SubmitInput) { v.Retry.MaxAttempts = 101 }, "unknown recovery policy": func(v *SubmitInput) { v.RecoverySafety = "UNKNOWN" }, "excess timeout": func(v *SubmitInput) { v.TaskTimeoutMS = 30*86400000 + 1 }, - "sub-microsecond time": func(v *SubmitInput) { now := time.Date(2026, 1, 1, 0, 0, 0, 1, time.UTC); v.AvailableAt = &now }, + "delay equals deadline": func(v *SubmitInput) { v.DelayMS = 86400000 }, + "delay overflow": func(v *SubmitInput) { v.DelayMS = ^uint64(0) }, + "both timing modes": func(v *SubmitInput) { + now := time.Now().UTC().Truncate(time.Microsecond) + v.AvailableAt = &now + v.DelayMS = 1 + }, + "sub-microsecond time": func(v *SubmitInput) { now := time.Date(2026, 1, 1, 0, 0, 0, 1, time.UTC); v.AvailableAt = &now }, } { t.Run(name, func(t *testing.T) { in := base @@ -34,6 +41,10 @@ func TestSubmitNormalizationAndBounds(t *testing.T) { } }) } + base.DelayMS = 86400000 - 1 + if out, err := NormalizeSubmit(base); err != nil || out.DelayMS != base.DelayMS || out.AvailableAt != nil { + t.Fatal("relative delay was resolved before database acceptance", err) + } for _, key := range []string{"", strings.Repeat("a", 129), "with space", "带中文", "line\n"} { if ValidateKey(key) == nil { t.Fatal("accepted invalid key") diff --git a/pulse/service/types.go b/pulse/service/types.go index 380ae23..e2aa6ab 100644 --- a/pulse/service/types.go +++ b/pulse/service/types.go @@ -29,7 +29,9 @@ type SubmitInput struct { ConcurrencyKey string `json:"concurrency_key,omitempty"` // Priority 冻结优先级，0至9，每级30秒排序优势，不改变最早执行时间。 Priority int16 `json:"priority,omitempty"` - // AvailableAt 最早执行时间；为空表示受理后即可进入派发范围。 + // DelayMS 相对首次受理数据库时间的等待毫秒数；零为不延迟，非零不能与 AvailableAt 同用。 + DelayMS uint64 `json:"delay_ms,omitempty"` + // AvailableAt 最早执行时间；为空且 DelayMS 为零表示受理后即可进入派发范围。 AvailableAt *time.Time `json:"available_at,omitempty"` // StartDeadlineAt 最晚开始时间；为空表示使用任务总截止时间。 StartDeadlineAt *time.Time `json:"start_deadline_at,omitempty"` diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index fcb57d3..f9b5639 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -61,6 +61,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/worker) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/history) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/retry) +(cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/delayed-task) (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) diff --git a/sdk/go/examples/delayed-task/main.go b/sdk/go/examples/delayed-task/main.go new file mode 100644 index 0000000..5736193 --- /dev/null +++ b/sdk/go/examples/delayed-task/main.go @@ -0,0 +1,129 @@ +// delayed-task 演示数据库计时的相对延迟、原键核对和实际 Worker 执行。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run() error { + token, queue := os.Getenv("ORBIT_TOKEN"), os.Getenv("ORBIT_QUEUE_ID") + if path := os.Getenv("ORBIT_CREDENTIALS_FILE"); path != "" { + raw, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read private credentials: %w", err) + } + var credentials struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if json.Unmarshal(raw, &credentials) != nil { + return errors.New("invalid credentials file") + } + if token == "" { + token = credentials.Token + } + if queue == "" { + queue = credentials.QueueID + } + } + client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + if err != nil { + return err + } + op, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + in := orbit.SubmitTaskRequest{QueueID: queue, Handler: "object.delayed." + op, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"object_ref":"examples/input","sequence":9007199254740993}`), DelayMS: 1500, TaskTimeoutMS: 15000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 5000}} + accepted, err := client.SubmitTask(ctx, op, in) + if err != nil { + return err + } + // 先输出任务身份，后续失败可据此核对；不输出凭据或任意业务正文。 + fmt.Printf("task=%s delay_ms=%d\n", accepted.Task.ID, in.DelayMS) + if !accepted.Task.AvailableAt.Equal(accepted.Task.CreatedAt.Add(1500 * time.Millisecond)) { + return errors.New("delay was not based on database acceptance") + } + w, err := orbit.NewWorker(client, orbit.WorkerOptions{QueueID: queue, Capacity: 1, ShutdownTimeout: 5 * time.Second, Handlers: []orbit.WorkerHandler{{Capability: orbit.WorkerCapability{Handler: in.Handler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + // 无外部副作用的示例处理器，保持精确数字文本。 + return e.Task.Payload, nil + }}}}) + if err != nil { + return err + } + work, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- w.Run(work) }() + waited := false + defer func() { + stop() + if !waited { + <-done + } + }() + for { + current, err := client.GetTask(ctx, accepted.Task.ID) + if err != nil { + return err + } + if current.State == orbit.TaskSucceeded { + history, err := client.ListTaskAttempts(ctx, current.ID, orbit.HistoryPageRequest{}) + if err != nil { + return err + } + if len(history.Items) != 1 || history.Items[0].AuthorizedAt == nil || history.Items[0].CreatedAt.Before(accepted.Task.AvailableAt) || history.Items[0].AuthorizedAt.Before(accepted.Task.AvailableAt) { + return errors.New("execution preceded the saved release time") + } + break + } + if current.State == orbit.TaskFailed || current.State == orbit.TaskNeedsReview || current.State == orbit.TaskCancelled { + return fmt.Errorf("task requires attention: %s", current.State) + } + select { + case err := <-done: + waited = true + return fmt.Errorf("Worker stopped before completion: %v", err) + case <-ctx.Done(): + return ctx.Err() + case <-time.After(100 * time.Millisecond): + } + } + duplicate, err := client.SubmitTask(ctx, op, in) + if err != nil { + return err + } + if !duplicate.Deduplicated || duplicate.Task.ID != accepted.Task.ID || !duplicate.Task.AvailableAt.Equal(accepted.Task.AvailableAt) { + return errors.New("replay changed the original delayed task") + } + stop() + err = <-done + waited = true + if err != nil { + return err + } + stats := w.Stats() + if stats.Started != 1 || stats.ReportsConfirmed != 1 || stats.Running != 0 { + return errors.New("Worker did not confirm and drain") + } + fmt.Println("state=SUCCEEDED attempts=1 no_early_start=true original_receipt=true worker_drained=true") + return nil +} diff --git a/sdk/go/examples/task-query/main.go b/sdk/go/examples/task-query/main.go index 5b7150d..0c6adfd 100644 --- a/sdk/go/examples/task-query/main.go +++ b/sdk/go/examples/task-query/main.go @@ -52,15 +52,32 @@ func run() error { if err != nil { return err } - businessKey := "object /+% é " + op + " " + businessKey := "对象 /+% é " + op + " " var tasks []orbit.Task for i, key := range []string{businessKey, businessKey, businessKey[:len(businessKey)-1]} { - out, err := client.SubmitTask(ctx, fmt.Sprintf("query-%s-%d", op, i), orbit.SubmitTaskRequest{QueueID: queue, Handler: "query." + op, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"object_ref":"examples/input"}`), BusinessKey: key, TaskTimeoutMS: 60000}) + payload := json.RawMessage(`{"object_ref":"examples/input","format":"text"}`) + if i == 1 { + payload = json.RawMessage(`{"object_ref":"examples/input","format":"json"}`) + } + out, err := client.SubmitTask(ctx, fmt.Sprintf("query-%s-%d", op, i), orbit.SubmitTaskRequest{QueueID: queue, Handler: "query." + op, HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, BusinessKey: key, TaskTimeoutMS: 60000}) if err != nil { return err } tasks = append(tasks, out.Task) } + // 改参数使用新身份与相同关联键，读取旧任务确认原输入仍保留。 + for _, task := range tasks[:2] { + current, err := client.GetTask(ctx, task.ID) + if err != nil { + return err + } + if string(current.Payload) != string(task.Payload) { + return errors.New("new linked submission changed existing input") + } + } + if tasks[0].ID == tasks[1].ID || string(tasks[0].Payload) == string(tasks[1].Payload) { + return errors.New("parameter change did not create an independent task") + } from, before := tasks[0].CreatedAt, tasks[2].CreatedAt.Add(time.Microsecond) query := orbit.ListTasksRequest{QueueID: queue, State: orbit.TaskWaiting, BusinessKey: businessKey, CreatedFrom: &from, CreatedBefore: &before, Limit: 1} matched, pages := 0, 0 @@ -99,6 +116,6 @@ func run() error { return err } } - fmt.Printf("matched=%d pages=%d before_excluded=true cancelled=%d\n", matched, pages, len(tasks)) + fmt.Printf("matched=%d pages=%d before_excluded=true original_payload_preserved=true cancelled=%d\n", matched, pages, len(tasks)) return nil } diff --git a/sdk/go/retry_process_integration_test.go b/sdk/go/retry_process_integration_test.go new file mode 100644 index 0000000..2e838fb --- /dev/null +++ b/sdk/go/retry_process_integration_test.go @@ -0,0 +1,192 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "reflect" + "testing" + "time" +) + +// 首次失败已提交并进入退避后，强杀唯一服务进程；两个新进程仅靠持久化状态接管。 +// 所有时间自然流逝，不修改数据库的退避、截止或租约字段。 +func TestIntegrationWorkerProcessRetryBackoffRecovery(t *testing.T) { + for _, tc := range []struct { + name string + unknown bool + expire bool + final TaskState + }{ + {"known_then_success", false, false, TaskSucceeded}, + {"unknown_then_exhausted", true, false, TaskNeedsReview}, + {"known_deadline_during_outage", false, true, TaskFailed}, + {"unknown_deadline_during_outage", true, true, TaskNeedsReview}, + } { + t.Run(tc.name, func(t *testing.T) { + a := startOrbitProcess(t) + _, _, token := integrationWorkerClient(t) + api := processClient(t, a, token) + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + name := "retry-crash." + workerTestID(t) + q, err := api.CreateQueue(ctx, name, CreateQueueRequest{Name: name, MaxConcurrency: 2, RatePerSecond: 100}) + if err != nil { + t.Fatal(err) + } + connect := func(client *Client) *WorkerConnection { + t.Helper() + r, err := NewWorkerRegistration(q.Queue.ID, 1, []WorkerCapability{{Handler: name, HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + t.Fatal(err) + } + w, err := client.DialWorker(ctx, r) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = w.Close() }) + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + t.Fatal(err) + } + return w + } + in := SubmitTaskRequest{QueueID: q.Queue.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`), RecoverySafety: RecoveryReconcileOnly, TaskTimeoutMS: 40000, Retry: RetryPolicy{MaxAttempts: 2, AttemptTimeoutMS: 3000, InitialDelayMS: 8000, MaxDelayMS: 8000}} + if tc.unknown { + in.RecoverySafety = RecoveryIdempotent + } + if tc.expire { + in.TaskTimeoutMS = 5000 + in.Retry.InitialDelayMS, in.Retry.MaxDelayMS = 1000, 1000 + } + accepted, err := api.SubmitTask(ctx, workerTestID(t), in) + if err != nil { + t.Fatal(err) + } + w := connect(api) + first, err := w.NextAssignment(ctx) + if err != nil || first.Task.ID != accepted.Task.ID || first.Epoch != 1 { + t.Fatal("first attempt identity", err) + } + if _, err := w.Start(ctx, workerTestID(t), first.Attempt()); err != nil { + t.Fatal(err) + } + finish := WorkerFinish{Ref: first.Attempt(), Outcome: WorkerFailed, Retryable: true, ErrorCode: "TEMPORARY_REJECTION", Result: json.RawMessage(`{"evidence_ref":"diagnostics/first"}`)} + if tc.unknown { + finish.Outcome, finish.Retryable, finish.ErrorCode = WorkerUnknown, false, "RESULT_UNKNOWN" + } + receipt, err := w.Finish(ctx, workerTestID(t), finish) + if err != nil || receipt.Receipt.TaskState != TaskWaiting || receipt.Lease.Active { + t.Fatal("first result did not durably enter backoff", err) + } + pending, err := api.GetTask(ctx, accepted.Task.ID) + if err != nil || pending.State != TaskWaiting || pending.AttemptCount != 1 || pending.EffectUncertain != tc.unknown || !pending.AvailableAt.After(receipt.Lease.ServerTime) { + t.Fatal("retry facts were not persisted", err) + } + old, err := api.GetTaskAttempt(ctx, pending.ID, first.AttemptID) + if err != nil || old.EndedAt == nil || old.Retryable == tc.unknown || string(old.Result) != string(finish.Result) { + t.Fatal("first attempt evidence missing", err) + } + if delay := pending.AvailableAt.Sub(*old.EndedAt); delay < time.Duration(in.Retry.InitialDelayMS/2)*time.Millisecond || delay > time.Duration(in.Retry.MaxDelayMS)*time.Millisecond { + t.Fatal("persisted backoff outside policy") + } + a.kill(t) + _ = w.Close() + if tc.expire { + // 用已确认的数据库时间差等待，恢复后的状态/历史仍以数据库事实作断言。 + timer := time.NewTimer(pending.DeadlineAt.Sub(receipt.Lease.ServerTime) + 100*time.Millisecond) + defer timer.Stop() + select { + case <-timer.C: + case <-ctx.Done(): + t.Fatal("outage exceeded test budget") + } + } + b, c := startOrbitProcess(t), startOrbitProcess(t) + if a.nodeID == b.nodeID || a.nodeID == c.nodeID || b.nodeID == c.nodeID { + t.Fatal("replacement reused node identity") + } + apiB, apiC := processClient(t, b, token), processClient(t, c, token) + if !tc.expire { + recovered, err := apiB.GetTask(ctx, pending.ID) + if err != nil || !reflect.DeepEqual(recovered, pending) { + t.Fatal("restart changed frozen backoff, budget, deadline or uncertainty", err) + } + } + workers := []*WorkerConnection{connect(apiB), connect(apiC)} + var current Task + wantAttempts := uint32(1) + if !tc.expire { + type assignmentResult struct { + index int + assignment WorkerAssignment + err error + } + wait, stop := context.WithCancel(ctx) + defer stop() + results := make(chan assignmentResult, 2) + for i, worker := range workers { + go func() { + assignment, err := worker.NextAssignment(wait) + results <- assignmentResult{i, assignment, err} + }() + } + winner := <-results + second := winner.assignment + if winner.err != nil || second.Task.ID != pending.ID || second.AttemptID == first.AttemptID || second.Epoch != 2 { + t.Fatal("retry did not preserve task identity and consumed budget", winner.err) + } + history, err := apiB.GetTaskAttempt(ctx, pending.ID, second.AttemptID) + if err != nil || history.CreatedAt.Before(pending.AvailableAt) || !history.CreatedAt.Before(pending.DeadlineAt) { + t.Fatal("retry ignored persisted timing", err) + } + owner := workers[winner.index] + if _, err := owner.Start(ctx, workerTestID(t), second.Attempt()); err != nil { + t.Fatal(err) + } + last := WorkerFinish{Ref: second.Attempt(), Outcome: WorkerSucceeded, Result: json.RawMessage(`{"n":9007199254740993}`)} + if tc.unknown { + last.Outcome, last.Retryable, last.ErrorCode = WorkerFailed, true, "TEMPORARY_REJECTION" + } + if _, err := owner.Finish(ctx, workerTestID(t), last); err != nil { + t.Fatal(err) + } + stop() + loser := <-results + if !errors.Is(loser.err, context.Canceled) { + t.Fatal("two replacement nodes received a retry assignment", loser.err) + } + wantAttempts = 2 + } + current = waitRuntimeState(t, ctx, apiB, pending.ID, tc.final) + if current.AttemptCount != wantAttempts || current.EffectUncertain != tc.unknown || current.Retry != pending.Retry || !current.DeadlineAt.Equal(pending.DeadlineAt) || current.RecoverySafety != pending.RecoverySafety || string(current.Payload) != string(pending.Payload) { + t.Fatal("recovery reset policy, input, deadline, attempts or uncertainty") + } + // 两个节点仍在线且都有可用容量，预算/截止应阻止任何额外领取。 + for _, worker := range workers { + probe, stop := context.WithTimeout(ctx, 250*time.Millisecond) + _, err := worker.NextAssignment(probe) + stop() + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatal("unexpected assignment after final decision", err) + } + } + attempts, err := apiC.ListTaskAttempts(ctx, pending.ID, HistoryPageRequest{}) + if err != nil || len(attempts.Items) != int(wantAttempts) || attempts.NextCursor != "" { + t.Fatal("unexpected execution history after failover", err) + } + preserved, err := apiC.GetTaskAttempt(ctx, pending.ID, first.AttemptID) + if err != nil || !reflect.DeepEqual(preserved, old) { + t.Fatal("restart or subsequent execution rewrote first result", err) + } + queue, err := apiC.GetQueue(ctx, q.Queue.ID) + wantUnfinished := uint64(0) + if tc.final == TaskNeedsReview { + wantUnfinished = 1 + } + if err != nil || queue.UnfinishedTasks != wantUnfinished { + t.Fatal("recovery lost unfinished-task accounting", err) + } + t.Logf("sole node killed; two fresh nodes recovered state=%s attempts=%d uncertainty=%t; original evidence and frozen policy preserved", current.State, wantAttempts, current.EffectUncertain) + }) + } +} diff --git a/sdk/go/task_delay_integration_test.go b/sdk/go/task_delay_integration_test.go new file mode 100644 index 0000000..5bd8d38 --- /dev/null +++ b/sdk/go/task_delay_integration_test.go @@ -0,0 +1,66 @@ +package orbit + +import ( + "context" + "encoding/json" + "testing" + "time" +) + +func TestIntegrationWorkerRuntimeDelayedSubmission(t *testing.T) { + api, queue, _ := integrationWorkerClient(t) + ctx, stop := context.WithTimeout(context.Background(), 20*time.Second) + defer stop() + name := "delay." + workerTestID(t) + key := workerTestID(t) + in := SubmitTaskRequest{QueueID: queue, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`), DelayMS: 1500, TaskTimeoutMS: 15000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 5000}} + delayed, err := api.SubmitTask(ctx, key, in) + if err != nil || !delayed.Task.AvailableAt.Equal(delayed.Task.CreatedAt.Add(1500*time.Millisecond)) { + t.Fatal("relative delay not persisted", err) + } + at := delayed.Task.CreatedAt.Add(3 * time.Second) + absoluteInput := in + absoluteInput.DelayMS, absoluteInput.AvailableAt = 0, &at + absolute, err := api.SubmitTask(ctx, workerTestID(t), absoluteInput) + if err != nil || !absolute.Task.AvailableAt.Equal(at) { + t.Fatal("absolute release time changed", err) + } + r, err := NewWorkerRegistration(queue, 1, []WorkerCapability{{Handler: name, HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + t.Fatal(err) + } + w, err := api.DialWorker(ctx, r) + if err != nil { + t.Fatal(err) + } + defer w.Close() + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + t.Fatal(err) + } + for _, want := range []Task{delayed.Task, absolute.Task} { + a, err := w.NextAssignment(ctx) + if err != nil || a.Task.ID != want.ID || a.ServerTime.Before(want.AvailableAt) || a.Epoch != 1 { + t.Fatal("task dispatched before its persisted release time", err) + } + if _, err := w.Start(ctx, workerTestID(t), a.Attempt()); err != nil { + t.Fatal(err) + } + if _, err := w.Finish(ctx, workerTestID(t), WorkerFinish{Ref: a.Attempt(), Outcome: WorkerSucceeded, Result: a.Task.Payload}); err != nil { + t.Fatal(err) + } + history, err := api.GetTaskAttempt(ctx, want.ID, a.AttemptID) + if err != nil || history.CreatedAt.Before(want.AvailableAt) || history.AuthorizedAt == nil || history.AuthorizedAt.Before(want.AvailableAt) { + t.Fatal("persisted history allowed an early start", err) + } + } + // 原等待时长已经经过，原键重报仍返回首次回执，不能建立新的计时或任务。 + dupe, err := api.SubmitTask(ctx, key, in) + if err != nil || !dupe.Deduplicated || dupe.Task.ID != delayed.Task.ID || !dupe.Task.AvailableAt.Equal(delayed.Task.AvailableAt) || !dupe.Task.DeadlineAt.Equal(delayed.Task.DeadlineAt) { + t.Fatal("late replay reset original delay", err) + } + current, err := api.GetTask(ctx, delayed.Task.ID) + if err != nil || current.State != TaskSucceeded || current.AttemptCount != 1 { + t.Fatal("late replay reopened a completed task", err) + } + t.Log("relative and absolute release times obey database history; replay after elapsed delay returns original receipt without reopening") +} diff --git a/sdk/go/types.go b/sdk/go/types.go index 08458c1..36d0e17 100644 --- a/sdk/go/types.go +++ b/sdk/go/types.go @@ -24,7 +24,9 @@ type SubmitTaskRequest struct { ConcurrencyKey string `json:"concurrency_key,omitempty"` // Priority 冻结优先级，0至9，每级30秒排序优势，不改变最早执行时间。 Priority int16 `json:"priority,omitempty"` - // AvailableAt 最早执行时间；为空表示受理后即可进入派发范围。 + // DelayMS 相对首次受理数据库时间的等待毫秒数；原键重报不重新计时，非零不能与 AvailableAt 同用。 + DelayMS uint64 `json:"delay_ms,omitempty"` + // AvailableAt 最早执行时间；为空且 DelayMS 为零表示受理后即可进入派发范围。 AvailableAt *time.Time `json:"available_at,omitempty"` // StartDeadlineAt 最晚开始时间；为空表示使用任务总截止时间。 StartDeadlineAt *time.Time `json:"start_deadline_at,omitempty"`