diff --git a/AGENTS.md b/AGENTS.md index a076191..291bd45 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,14 +31,16 @@ - P4定义发布使用schema_version=1有限结构,词法作用域禁止跨分支/前向读取,任务策略显式冻结;SDK类型独立。发布按认证→应用额度→操作身份→定义→排序队列持锁;原键与同版本去重仍校验当前权限。定义/版本/revision/审计/回执同事务;同版本不同内容不得覆盖。保留精确数字,校验存储原文与结构摘要;不得把发布通过或DSL支持误称为实例执行已实现。 -- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归(docs/reports/phase-4-definitions.md)。实例启动/固定版本/首批任务同事务的内部存储已补齐;内部事件消费、顺序与组合推进及根接续已接入;继续持久化等待及实例公共SDK;P4整体和P5–P7仍未完成。 +- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归(docs/reports/phase-4-definitions.md)。实例启动/固定版本/首批任务同事务的内部存储已补齐;内部事件消费、顺序与组合推进及根接续已接入;TIMER与后续POLL时钟已接入,继续SIGNAL及实例公共SDK;P4整体和P5–P7仍未完成。 - 内部结果交付使用workflow_activation所有者并固定激活ID,TASK经activation.task_id、POLL经轮次校验归属;weave_inbox来源及独立快照与Pulse结果同事务,先于外部结果流行锁,不在结果路径反锁实例/步骤/订阅。迁移000017隔离缺快照历史,禁止用当前任务重建旧事件。 - weave/engine仅计算有限决策;实例受理、Inbox消费与根接续已接入,完整等待/控制运行器继续开发。Plan最多32个变更,More必须持久化待续;耗尽变更预算仍须遍历其他已活动分支,不能误判在途工作不可达。冻结分支选择,精确数值,不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务,不能把纯快照往返测试当作真实进程恢复。 -- 实例内部受理已实现,公共实例HTTP/SDK及等待运行器尚未开放;结果消费和根待续扫描已接入后台。Start按认证→应用额度→原身份→排序队列建立门控/固定版本/全部步骤/首批工作,所有失败同事务回滚。首批推进递增revision至2,原键返回冻结回执且仍核验当前权限,不重算截止。 +- 实例内部受理已实现,公共实例HTTP/SDK及信号运行器尚未开放;结果消费和根待续扫描已接入后台。Start按认证→应用额度→原身份→排序队列建立门控/固定版本/全部步骤/首批工作,所有失败同事务回滚。首批推进递增revision至2,原键返回冻结回执且仍核验当前权限,不重算截止。 - 实例根步骤`$root`/JOIN不属于用户定义,Steps查询排除它;More对应根步骤版本化持久检查点。TASK直接关联激活task_id,POLL通过轮次关联,实际任务截止必须受实例/查询绝对截止限制。Wait名称及契约版本冻结,旧NULL等待须核实,不能盲目解释;新关联512字节、存储兼容历史764字节。 - 结果消费按身份→应用额度→排序队列→实例→门控→步骤/Inbox持锁,禁止先锁Inbox再锁父资源。来源使用独立冻结快照,旧激活/低版本留痕;数据库故障不归类为业务失败。消费确认、步骤、后继或根待续、事件及审计同事务。 - 后继满额用SAVEPOINT回滚整轮新工作,保留已接受结果和待续;恢复必须重读持久图。多次根推进使用独立系统审计操作身份;迟到证据也递增实例历史版本,禁止复用已有workflow.advanced唯一身份。 -- 本轮消费运行器和真实Orbit强杀/双节点接管已验证,见docs/reports/phase-4-runtime.md。POLL仅当前轮次分类与next_poll_at保存;下一轮派发、等待到期/信号、FAIL_AFTER_SETTLED分支结清、控制API和实例SDK仍须完成,不得按部分自动运行宣称P4/P5完成。 +- 本轮消费运行器和真实Orbit强杀/双节点接管已验证,见docs/reports/phase-4-runtime.md。TIMER到期与POLL后续轮次已接入,见docs/13-workflow-clock.md;SIGNAL及最终信号竞争、FAIL_AFTER_SETTLED分支结清、控制API和实例SDK仍须完成,不得按部分自动运行宣称P4/P5完成。 + +- TIMER/POLL复用weave_pending_advances及到期索引,时钟固定激活ID;迁移000019只回填原等待/查询时钟,不改截止和预算。后续POLL只由已确认POLL_PENDING开启,首轮精确输入/资源键冻结,每轮Pulse尝试独立计数;满额/暂停不消耗轮数。截止遇到已落库结果先等待Inbox,仍在途则关闭门控但保留ACTIVE和当前任务,许可由Pulse结清;不得由时钟假定外部请求已结束。 diff --git a/README.md b/README.md index f6ae344..cc1faab 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台, 已完成 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) 继续实现。 -P4已实现[流程定义与不可变版本发布](docs/api/workflow-definitions.md)、[结构版本1](docs/api/workflow-definition-format.md)及[SDK发布例子](docs/examples/workflow-definition.md)。[有限步骤决策器与内部结果交付](docs/10-workflow-engine-core.md)已实现;已补[实例受理及首批工作存储](docs/11-workflow-admission.md)。[结果消费、顺序/分支/并行推进及根检查点接续](docs/12-workflow-runtime.md)已接入后台运行器。持久化等待、后续查询轮次、失败结清及实例公共API/SDK继续开发,发布定义不会启动任务。 +P4已实现[流程定义与不可变版本发布](docs/api/workflow-definitions.md)、[结构版本1](docs/api/workflow-definition-format.md)及[SDK发布例子](docs/examples/workflow-definition.md)。[有限步骤决策器与内部结果交付](docs/10-workflow-engine-core.md)已实现;已补[实例受理及首批工作存储](docs/11-workflow-admission.md)。[结果消费、顺序/分支/并行推进及根检查点接续](docs/12-workflow-runtime.md)已接入后台运行器。[定时等待、后续查询轮次及原预算恢复](docs/13-workflow-clock.md)已接入;信号、失败结清及实例公共API/SDK继续开发,发布定义不会启动任务。 数据库禁止外键,关联一致性通过服务端事务校验,索引、唯一约束和 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/06-workflow-spec.md b/docs/06-workflow-spec.md index da16772..2e71198 100644 --- a/docs/06-workflow-spec.md +++ b/docs/06-workflow-spec.md @@ -1,6 +1,6 @@ # Weave 流程编排功能与设计 -> 版本 0.3 · 2026-09-18。首版功能范围已确认。定义结构、发布校验与不可变版本已实现,见[定义格式](api/workflow-definition-format.md)和[发布API](api/workflow-definitions.md);已补[步骤决策与内部结果交付](10-workflow-engine-core.md),[实例受理与元数据查询存储](11-workflow-admission.md)已实现;事件消费、等待运行器、公共API/SDK与管理仍待P4–P5实现验收。 +> 版本 0.4 · 2026-09-18。首版功能范围已确认。定义结构、发布校验与不可变版本已实现,见[定义格式](api/workflow-definition-format.md)和[发布API](api/workflow-definitions.md);[实例受理](11-workflow-admission.md)、[结果消费与根推进](12-workflow-runtime.md)、[TIMER与POLL时钟](13-workflow-clock.md)已接入。信号、完整管理及实例公共API/SDK仍待P4–P5实现验收。 ## 1. 职责与边界 @@ -18,7 +18,7 @@ Weave 不执行特定业务逻辑,也不把整段业务代码放入平台进 | 条件分支 | 基于已保存、经过类型校验的结果或信号选择路径;不执行任意脚本 | | 有限并行与汇合 | 建议先支持结构化并行分支、全部成功后汇合;限制分支数,不默认支持任意交叉汇合 | | 等待业务信号 | 持久化等待回执、审核结果等外部输入,不长期占用 Worker | -| 定时等待 | 持久化截止时间,通过 Pulse 的内部定时任务唤醒 | +| 定时等待 | 持久化截止时间,由平台到期索引扫描唤醒,不占业务Worker | | 有界重复查询 | 外部业务尚在处理中时按间隔查询,达到次数或总时限后转人工核查;不与 Pulse 执行重试混用 | | 暂停与继续 | 暂停后停止推进新步骤,继续接收在途结果;恢复后从已保存进度继续 | | 异常与人工恢复 | 保存失败或未知结果,按定义暂停相关推进并核查;干预必须有权限、原因与证据 | @@ -136,7 +136,9 @@ Weave 不执行特定业务逻辑,也不把整段业务代码放入平台进 查询任务自身成功表示取得了一份有效结果。其中“处理中”使 Weave 保存等待与下一次唤醒;最终业务结果则结束查询节点,按定义推进或进入失败处理。单轮请求重试耗尽应转核查,不能通过新建一轮绕过已耗尽的重试预算。 -每轮只创建一个查询任务,以命名空间、实例、步骤激活和轮次形成稳定去重身份。轮次之间释放业务 Worker 槽位,等待依赖数据库记录与 Pulse 内部定时唤醒;SDK 不在处理器中长期 sleep 或维护内存轮询循环。 +每轮只创建一个查询任务,以命名空间、实例、步骤激活和轮次形成稳定去重身份。轮次之间释放业务 Worker 槽位,等待依赖数据库记录与平台到期索引扫描;SDK 不在处理器中长期 sleep 或维护内存轮询循环。 + +2026-09-18实现决策:TIMER/POLL直接复用`weave_pending_advances`检查点和后台扫描,替代早期草案的Pulse内部唤醒任务。唤醒只改变流程持久状态,不调用业务处理器,因此无需额外占用任务/尝试/许可记录;业务查询仍由Pulse执行。当前schema使用固定间隔,不提供业务轮次抖动或变长退避,具体冻结和升级行为见[运行约定](13-workflow-clock.md)。 发布和受理时校验轮次、单轮尝试、超时和总时限的组合上限,避免两层次数相乘后超过外部渠道或平台预算。未完成上一轮前不得并发启动下一轮。最后一轮仍为处理中、等待到期或当前轮次重试耗尽时,保存核查原因及已知结果。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index fb8ba94..9e7d226 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -75,7 +75,7 @@ | weave | `InboxEvent` | `weave_inbox` | | weave | `PendingAdvance` | `weave_pending_advances` | -`orbit_events` 保存原始状态事件,`orbit_result_events` 使用相同事件 ID 保存队列、提交序号和冻结快照,`orbit_event_streams` 串行分配命名空间内的结果序号;对外投递使用 `orbit_deliveries`,内部流程结果已原子写入 `weave_inbox`及其独立冻结快照(实例消费推进尚未实现)。`weave_pending_advances` 保存暂停或分批推进后的续接检查点。 +`orbit_events` 保存原始状态事件,`orbit_result_events` 使用相同事件 ID 保存队列、提交序号和冻结快照,`orbit_event_streams` 串行分配命名空间内的结果序号;对外投递使用 `orbit_deliveries`,内部流程结果原子写入 `weave_inbox`及其独立冻结快照,已接入[实例消费推进](12-workflow-runtime.md)。`weave_pending_advances` 保存根步骤续接和TIMER/POLL时钟检查点。 ## 4. 已落实的数据库约束 @@ -196,3 +196,7 @@ delay_ms是普通/批量提交命令字段,首次事务以同一个数据库 ## 实例受理与等待身份(000018) Wait新增SignalName和PayloadVersion两个中文注释字段;correlation_key转为VARBINARY(764),新受理仍限制512字节,兼容历史191个四字节字符而不截断。实例business_key使用utf8mb4_0900_bin保留精确尾部空格;仍无外键和级联。35个实体,含公共Record的295个直接持久化字段声明均有中文注释。内部`$root`步骤作为实例级待续检查点,不属于定义静态节点或公共步骤列表。具体初始事务、预算和未接通的运行边界见[实例受理](11-workflow-admission.md)。 + +## 等待时钟回填(000019) + +复用PendingAdvance保存ROOT、TIMER、POLL检查点;新增内部JSON结构字段均有中文注释,不新增持久化列或实体,仍为35个实体/295个直接持久化字段。TIMER/POLL检查点固定激活ID,实际推进核验同命名空间归属、原截止及剩余预算。迁移只回填历史有效等待的唤醒身份和时间,保留已有检查点;无外键、无级联,见[时钟运行约定](13-workflow-clock.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index cfe2ad7..98d1b77 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -28,13 +28,15 @@ | P1 任务受理闭环 | 应用凭据与作用域权限、队列、幂等提交、查询、待执行取消、受理事件;SDK 客户端 | API/SDK 创建查询闭环;重复键与内容冲突、响应未确认重报、跨命名空间拒绝、参数边界 | 已完成并通过阶段回归(见 P1 报告;真实断路/提交 ACK 丢失注入留后续故障阶段) | | P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复;Worker SDK | 真 Worker 执行;失联/重连/重复确认/旧尝试拒绝;业务效果未知转核查;节点故障接管 | 已完成并通过阶段回归:服务端、自动 SDK、历史查询、完整 Schema 与真实进程强杀恢复;测试拓扑和剩余生产故障矩阵见 P2 验收报告 | | P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 已完成平台功能阶段验收:[原范围逐项核对](reports/phase-3.md)、[本轮完整回归](reports/phase-3-delay-recovery.md)。补齐相对延迟、改参数关联示例和四种重试全停恢复场景;20个可运行示例全部通过。全平台生产门槛及SDK首版交付剩余项仍按P6–P7推进 | -| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容;端到端通用流程 | 定义格式/词法类型校验、不可变发布、独立SDK及六类节点定义例子已实现,已通过[子项验收](reports/phase-4-definitions.md);已补[有限决策器及内部结果交付](reports/phase-4-engine-core.md);[实例受理及首批事务](reports/phase-4-admission.md)已实现;[结果消费和根接续](reports/phase-4-runtime.md)已接入后台,顺序/分支/并行及真实进程恢复已验证;持久化等待、完整控制和实例公共API/SDK继续开发 | -| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算;暂停/开始/回执/超时竞争;恢复保留预算;未知结果不盲目重复 | 待开发 | +| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容;端到端通用流程 | 定义格式/词法类型校验、不可变发布、独立SDK及六类节点定义例子已实现,已通过[子项验收](reports/phase-4-definitions.md);已补[有限决策器及内部结果交付](reports/phase-4-engine-core.md);[实例受理及首批事务](reports/phase-4-admission.md)已实现;[结果消费和根接续](reports/phase-4-runtime.md)已接入后台,顺序/分支/并行及真实进程恢复已验证;[TIMER与后续POLL轮次](13-workflow-clock.md)已接入;信号、完整控制和实例公共API/SDK继续开发 | +| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算;暂停/开始/回执/超时竞争;恢复保留预算;未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器;信号竞争、完整控制/人工恢复及公共API/SDK继续开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行,报告环境/指标/遗留项;容量、RPO/RTO 按真实拓扑验证 | 待开发 | | P7 首版交付核对 | SDK 兼容与发布准备、完整 API 手册、场景示例、全量回归及交付清单 | 新接入者按文档跑通全部示例;所有首版能力可追溯到实现与测试;报告实际生产边界 | 待开发 | SDK、文档与示例随 P1–P5 同步开发,P7 是全量核对,不是最后才开始补文档。 +P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告](reports/phase-4-clock.md)。实例公共入口继续与信号/控制能力同步交付,不能将内部存储链路作为完整SDK验收。 + ### P3 功能范围核对 可靠订阅链路完成后,重新对照 [原功能范围](01-functional-spec.md) 和 [事务衔接设计](04-reliability-and-data.md),以下原定首版能力已按核对结果补齐;进一步发现的相对延迟与改参数关联示例也已交付。[P3验收](reports/phase-3.md)逐项保留实现证据及尚待P4–P7完成的原范围,不按已有模块反推全平台完成。 diff --git a/docs/10-workflow-engine-core.md b/docs/10-workflow-engine-core.md index e017786..f49bacc 100644 --- a/docs/10-workflow-engine-core.md +++ b/docs/10-workflow-engine-core.md @@ -1,6 +1,6 @@ # Weave 执行决策与内部结果交付 -状态:P4 内部基础已实现。已接通Pulse结果到Weave持久化Inbox,并提供有限步骤决策器;[实例受理和首批存储适配](11-workflow-admission.md)已补齐。[Inbox消费推进及根检查点运行器](12-workflow-runtime.md)已接入;定时/信号运行器及实例公共API/SDK仍待实现。不能把本页能力解释成完整公开端到端流程已交付。 +状态:P4 内部基础已实现。已接通Pulse结果到Weave持久化Inbox,并提供有限步骤决策器;[实例受理和首批存储适配](11-workflow-admission.md)已补齐。[Inbox消费推进及根检查点运行器](12-workflow-runtime.md)已接入;[定时和查询时钟](13-workflow-clock.md)已接入,信号运行器及实例公共API/SDK仍待实现。不能把本页能力解释成完整公开端到端流程已交付。 ## 1. 为什么结果要独立持久化 @@ -74,7 +74,7 @@ v16及以前的未处理记录缺少冻结来源,升级后统一标为QUARANTI ## 4. 后续存储适配必须满足的条件 -以下保留存储运行器的完整要求。第1项已按[受理报告](reports/phase-4-admission.md)验证;第2–4项结果消费和根接续已接入[实际运行器](12-workflow-runtime.md),第5–7项的完整等待、控制、接口和SDK仍继续实现: +以下保留存储运行器的完整要求。第1项已按[受理报告](reports/phase-4-admission.md)验证;第2–4项结果消费和根接续已接入[实际运行器](12-workflow-runtime.md),第5项的TIMER/POLL已接入[时钟运行器](13-workflow-clock.md),其余信号、完整控制、接口和SDK仍继续实现: 1. 启动时认证、固定不可变版本/输入和策略、校验全部队列、计算实例截止,并在同事务创建实例、门控、首批激活/任务及审计回执。相同身份返回原实例。 2. 消费者领取内部事件后核对命名空间、来源任务版本、激活/轮次及当前步骤版本;消费确认、步骤变化、后续任务或持久化待续记录同事务提交。 diff --git a/docs/11-workflow-admission.md b/docs/11-workflow-admission.md index dc88d3a..4382dab 100644 --- a/docs/11-workflow-admission.md +++ b/docs/11-workflow-admission.md @@ -1,6 +1,6 @@ # Weave 实例受理与首批工作 -状态:内部Go服务与MySQL存储适配已实现实例Start/Get/Steps、固定版本、首批决策落库和可靠结果交付。[后台Inbox消费和根待续扫描](12-workflow-runtime.md)已接通。**尚未开放实例HTTP路由、启动SDK或命令行入口;等待运行器、后续轮询和完整控制仍继续开发。** Weave相关公开SDK目前仅支持定义发布。TASK及组合节点已可自动推进,不能将此解释成全部等待/控制能力已实现。 +状态:内部Go服务与MySQL存储适配已实现实例Start/Get/Steps、固定版本、首批决策落库和可靠结果交付。[后台Inbox消费和根待续扫描](12-workflow-runtime.md)已接通。**尚未开放实例HTTP路由、启动SDK或命令行入口;信号运行器和完整控制仍继续开发。** Weave相关公开SDK目前仅支持定义发布。[TIMER和后续POLL轮次](13-workflow-clock.md)已接入,TASK及组合节点已可自动推进,不能将此解释成全部等待/控制能力已实现。 ## 1. 受理事务 @@ -49,14 +49,14 @@ Get返回当前实例视图;Start重报返回受理时冻结的原始视图, | 活跃实例 | 每个namespace最多10000个RUNNING/WAITING/CANCELLING/NEEDS_REVIEW实例;终态不计入;满额仍可恢复原受理回执 | | 信号关联 | 新受理最多512字节;数据库兼容旧191字符最长764字节,升级不截断历史 | | 截止 | 数据库时间生成;任务、查询与等待均不超过实例总截止 | -| 自动运行 | Inbox消费和根待续扫描已接入;信号消费、定时到期处理及后续轮询仍待实现 | +| 自动运行 | Inbox消费、根待续、TIMER到期及后续POLL轮次已接入;信号消费及完整控制仍待实现 | 迁移000018为Wait增加signal_name和payload_version两个带中文注释的可空字段;NULL仅表示TIMER或尚未核实的历史元数据,未来消费者不得把历史NULL自动解释为有效的新信号等待。correlation_key转为精确字节存储;实例business_key改为utf8mb4_0900_bin。35个实体、含公共Record的295个直接持久化字段声明,全部有中文注释,继续保持零外键。 ## 5. 下一步接入 -1. 结果消费与根待续运行器已按[运行器报告](reports/phase-4-runtime.md)验证;继续补齐数据库时间驱动的信号/定时等待。 -2. 已有扫描核验暂停、截止与所属应用停用;继续后续轮次派发、回执竞争及失败分支结清,恢复保留原截止和次数。 +1. 结果消费与根待续运行器已按[运行器报告](reports/phase-4-runtime.md)验证;[定时和查询时钟](13-workflow-clock.md)已接入,继续补齐信号等待。 +2. 已有扫描核验暂停、截止与所属应用停用;继续信号/查询回执竞争及失败分支结清,恢复保留原截止和次数。 3. 接通实例公共HTTP、独立SDK、来源查询和可运行端到端场景,同时实施真实节点退出与接管回归。 本阶段验证见[实例受理回归记录](reports/phase-4-admission.md)。纯受理测试不是流程运行验收,P4–P7仍按[完整计划](08-development-plan.md)推进。 diff --git a/docs/12-workflow-runtime.md b/docs/12-workflow-runtime.md index 205293b..736db3b 100644 --- a/docs/12-workflow-runtime.md +++ b/docs/12-workflow-runtime.md @@ -1,6 +1,6 @@ # Weave 结果消费与自动推进 -状态:内部实例的TASK结果消费、顺序、分支、有限并行汇合和根检查点接续已接入Orbit后台运行器。实例公共HTTP/SDK仍未开放。信号/定时等待、下一轮查询、失败后的分支取消结清及人工恢复仍继续开发,不能将本阶段等同于完整Weave交付。 +状态:内部实例的TASK结果消费、顺序、分支、有限并行汇合和根检查点接续已接入Orbit后台运行器。[定时等待和后续查询轮次](13-workflow-clock.md)也已接入。实例公共HTTP/SDK仍未开放,信号、失败后的分支取消结清及人工恢复继续开发,不能将本阶段等同于完整Weave交付。 ## 1. 后台运行和事务边界 @@ -45,7 +45,7 @@ flowchart LR 暂停、关闭门控或停用应用/命名空间时,继续收集已经落库的结果,但不激活新步骤;若还有后继,将其保存为待续。重新允许推进后,从持久记录继续。测试通过直接设置控制状态验证该行为;暂停/继续公共接口尚未交付。 -已过实例总截止时不创建后继,未能全部完成则进入核查。最后的有效任务结果在截止前已由Pulse接受,而消费发生较晚时,可以整理完整结果并完成实例。暂停或重启不延长截止。纯等待实例到期及信号裁决仍需后续等待扫描器。 +已过实例总截止时不创建后继,未能全部完成则进入核查。最后的有效任务结果在截止前已由Pulse接受,而消费发生较晚时,可以整理完整结果并完成实例。暂停或重启不延长截止。TIMER与POLL的时钟已接入;SIGNAL及其超期裁决仍需后续实现。 ## 4. POLL与失败策略的当前边界 @@ -56,7 +56,7 @@ flowchart LR - 轮次耗尽、查询截止、未知/非法值或单轮请求失败:进入核查,不新建轮次绕过Pulse重试预算。 - 业务明确失败:按固定实例失败策略处理;REVIEW进入核查,FAIL_AFTER_SETTLED进入CANCELLING并关闭门控。 -**尚未实现下一轮查询派发及其到期索引扫描,也未实现FAIL_AFTER_SETTLED的剩余分支取消/收集/失败终态判定。** CANCELLING保留未结清事实,不提前宣称FAILED。信号抢先完成时与当前在途查询的结清竞争,也属于后续必须实现的功能。 +下一轮查询派发及到期扫描已实现,详见[时钟、预算与恢复](13-workflow-clock.md)。**尚未实现FAIL_AFTER_SETTLED的剩余分支取消/收集/失败终态判定。** CANCELLING保留未结清事实,不提前宣称FAILED。信号抢先完成时与当前在途查询的结清竞争,也属于后续必须实现的功能。 ## 5. 回归与后续工作 @@ -64,4 +64,4 @@ flowchart LR 进程回归由`make integration`构建二进制后显式执行。缺少一次性数据库或`ORBIT_TEST_SERVER_BINARY`时测试会跳过;跳过不构成进程恢复证据。验证记录见[本阶段报告](reports/phase-4-runtime.md)。 -下一步仍包括持久化定时/信号等待、轮询后续轮次、完整控制与人工恢复、实例/来源公共API、独立SDK和通用端到端场景。P4整体及P5–P7继续按[完整计划](08-development-plan.md)执行。 +下一步仍包括持久化信号等待及查询最终信号竞争、完整控制与人工恢复、实例/来源公共API、独立SDK和通用端到端场景。P4整体及P5–P7继续按[完整计划](08-development-plan.md)执行。 diff --git a/docs/13-workflow-clock.md b/docs/13-workflow-clock.md new file mode 100644 index 0000000..5ba0434 --- /dev/null +++ b/docs/13-workflow-clock.md @@ -0,0 +1,63 @@ +# Weave 持久化定时等待与查询轮次 + +状态:TIMER到期推进和POLL后续业务轮次已接入Orbit后台运行器。复用MySQL和既有待续索引,不增加外部调度服务或数据库外键。实例公共HTTP/SDK、SIGNAL消费、查询最终信号竞争以及完整控制/人工恢复仍未交付。 + +## 1. 时钟和恢复身份 + +`weave_pending_advances`同时保存ROOT、TIMER、POLL三类检查点。唯一范围为命名空间、实例和步骤;时钟检查点额外固定当前激活ID,避免旧激活唤醒新工作。扫描只提供候选,实际推进仍须持锁核验固定版本、父记录、步骤/激活状态、任务所有者、门控和原始预算。 + +| 检查点 | 到期时间来源 | 推进结果 | +| --- | --- | --- | +| ROOT | 分批推进或暂停/满额后的下一检查时间 | 重新计算最多32项步骤变更 | +| TIMER | 激活时保存的等待截止 | 完成原等待或记录超期核查,再有限推进 | +| POLL当前轮次仍在途 | 查询节点的原绝对截止 | 保留结果收集或关闭新执行门控,不启动重叠轮次 | +| POLL上一轮业务PENDING | 上一轮权威结果发生时间加固定间隔,受查询截止限制 | 若允许且有额度,同事务创建下一轮任务 | + +每个候选遵守[运行器事务顺序](12-workflow-runtime.md):身份、应用额度、排序队列、实例、门控和子记录。多个Orbit节点竞争同一数据库记录,不能凭本地时间或内存缓存取得推进权。扫描失败回滚并按候选版本延后重查,既有结果和预算仍保留。 + +TIMER由待续表直接唤醒,不创建Pulse业务任务,也不占Worker执行许可。旧模型的`wake_task_id`当前保持空;后续若改变这一约定,需要独立迁移与兼容验证。 + +## 2. 定时等待 + +激活时保存开始时间及`min(开始时间 + delay_ms, 实例总截止)`。恢复时重新验证这条等式,不能从重启或继续时间重新计时。 + +- 原计划等待时长能在实例总截止内结束:到期保存SATISFIED及步骤SUCCEEDED,输出为JSON `null`。 +- 后续步骤才激活TIMER,导致剩余实例时长不足:到总截止保存EXPIRED、步骤及实例NEEDS_REVIEW,不能把被截短的等待当作成功。 +- 暂停期间已到期的等待仍可收集完成;后继保留ROOT待续,直到控制状态允许。暂停不延长截止。 +- 全部进程停止后才被扫描:使用原到期时间。若整个流程已能由有效的到期事实完成,可以整理完成;若还需启动新步骤且实例总截止已过,则进入核查。 + +等待结清、步骤状态、后继任务或ROOT检查点、事件和审计同事务。审计失败不会留下已完成等待或部分后继。 + +## 3. 有界查询与请求重试 + +一次业务查询轮次对应一个Pulse任务。当前任务成功返回PENDING后,内部结果事务保存WAITING、清除`current_task_id`、记录`next_poll_at`并更新POLL检查点。成功查询仍在处理和请求执行失败使用不同预算: + +| 场景 | 业务轮次 | Pulse尝试 | +| --- | --- | --- | +| 已知可重试请求失败 | 不增加,仍为原任务 | 按该任务冻结策略增加 | +| 查询成功返回PENDING且未耗尽预算 | 到固定间隔后创建下一任务,增加1 | 新一轮从自身首次尝试开始 | +| DONE等声明成功结果 | 结清当前查询节点 | 不再启动下一轮 | +| 单轮失败/效果未知/非法结果/业务核查值 | 停止正常推进并保存核查事实 | 不通过创建新轮次绕过原尝试上限 | +| 轮次耗尽或总截止到达 | 不增加,进入核查 | 原在途任务仍需收集实际结果 | + +创建下一轮前校验轮次连续、当前无在途任务、上一轮已经确认为POLL_PENDING。复用首轮已解析的精确输入、业务键、资源互斥键以及固定版本的处理器、契约、优先级和重试策略;不重新解释已完成步骤的表达式。任务绝对截止仍取自身时限、查询总截止和实例总截止的较早者。 + +新任务、所有者和门控、轮次记录、已用轮数、当前任务、步骤状态、下一检查点、事件及审计同事务。插入轮次或审计失败会连同任务及额度一起回滚。应用/队列满额、暂停或身份停用仅延后检查;不消耗新轮次,不改变原截止。 + +## 4. 截止与已落库结果的竞争 + +截止扫描发现当前任务已经有终态或核查结果,但Inbox还未消费时,保留时钟并优先等Inbox按原事件时间裁决。不能用扫描发生时间否定已按时可靠受理的成功结果。 + +当前任务仍在途且查询总截止已到时,实例转NEEDS_REVIEW并关闭新执行门控;步骤保持ACTIVE、轮次保持当前任务引用,继续收集Pulse结果。时钟不释放执行许可,也不假定外部请求已经停止。Pulse恢复将失联执行记录为UNKNOWN后,内部消费再结清当前查询记录,保留核查状态。 + +最终信号与在途查询的竞争尚未实现。定义可以描述`final_signal`并建立等待,但当前只能通过查询结果结清;不能据此承诺外部信号已经可以消费。 + +## 5. 升级和验证 + +[迁移000019](../internal/storage/mysql/migrations/000019_workflow_clock_checkpoints.up.sql)仅回填旧实例当前激活的TIMER/POLL检查点:定时等待使用原deadline,等待查询使用原next_poll_at,在途查询使用原查询deadline。已有检查点不覆盖,终态实例及旧激活不回填;不新增表、字段、索引、外键或级联。 + +升级顺序为停止旧写入节点 → 应用全部迁移 → 启动新节点。旧节点可能消费PENDING结果却不会创建新的时钟检查点,因此不能与新版混跑。历史结构损坏或缺少完整定义的记录即使形成候选,也必须经运行器校验才能推进。 + +[存储回归](../internal/storage/mysql/workflow_clock_integration_test.go)覆盖到期、竞争、预算、暂停/满额、回滚、归属损坏与未知结果;[v18升级回归](../internal/storage/mysql/workflow_clock_upgrade_integration_test.go)验证回填及原预算保留;[真实进程回归](../internal/storage/mysql/workflow_runtime_process_integration_test.go)覆盖全部控制进程退出后的定时和查询接续。执行记录见[阶段报告](reports/phase-4-clock.md)。 + +下一阶段继续SIGNAL早到/重复/超期处理、信号与查询竞争、失败分支结清、公共控制及实例SDK,完整范围见[持续计划](08-development-plan.md)。 diff --git a/docs/README.md b/docs/README.md index 63182a2..51020c7 100644 --- a/docs/README.md +++ b/docs/README.md @@ -21,6 +21,7 @@ | [10 Weave决策与交付](10-workflow-engine-core.md) | 有限步骤决策和内部结果Inbox | | [11 Weave实例受理](11-workflow-admission.md) | 固定版本、初始图和首批任务事务 | | [12 Weave结果消费与推进](12-workflow-runtime.md) | 后台消费、根接续、事务与进程恢复,及尚未完成的等待/控制边界 | +| [13 Weave定时等待与查询轮次](13-workflow-clock.md) | 原截止、轮次/尝试预算、时钟检查点、升级与全停恢复 | | [接口手册](api/README.md) | 实际接口状态、调用说明与后续接口规划 | | [SDK 与场景](sdk/README.md) | 可运行示例、接入方法及分阶段 SDK 交付 | diff --git a/docs/reports/phase-4-clock.md b/docs/reports/phase-4-clock.md new file mode 100644 index 0000000..d288615 --- /dev/null +++ b/docs/reports/phase-4-clock.md @@ -0,0 +1,53 @@ +# P4/P5 定时等待与查询轮次回归 + +日期:2026-09-18。范围为内部持久化等待运行器;SIGNAL、完整管理控制、实例公共API/SDK和生产验证继续按原计划执行。 + +## 实现范围 + +- TIMER与POLL复用待续表到期索引和当前激活身份;首批创建、结果消费、时钟推进及审计同事务。 +- 原等待时间、总截止、已用轮次和单轮尝试预算不因暂停、满额或重启重置。 +- 后续查询复用首轮精确输入、业务键、互斥键、处理器与冻结策略;上一轮必须已经成功查询并分类为PENDING。 +- 截止扫描让已落库Inbox结果先完成裁决;仍在途时保留步骤ACTIVE与当前任务,由Pulse收集实际结果和释放许可。 +- 迁移000019从真实v18结构回填等待时钟,不覆盖原检查点、不改变预算、不回填终态实例或旧激活;无新增实体、列、索引或外键。 + +具体语义见[定时与查询文档](../13-workflow-clock.md)。模型维持35个实体、295个直接持久化字段声明,中文注释保留;内部检查点JSON字段同步提供中文说明。 + +## 环境与证据 + +本机darwin/arm64、Go 1.27.1、Docker MySQL 8.4.10。全部数据库为临时隔离测试库,未连接业务库或远端生产库。 + +| 检查 | 结果 | +| --- | --- | +| `make fmt test vet build` | 已通过,退出0,覆盖服务端、SDK与独立示例module | +| 时钟及相关真实MySQL定向回归 | 14组时钟场景、v18升级、原运行器/实例受理、新建结构及v7无外键升级通过,`-race`,44.745秒 | +| 真实Orbit进程恢复 | 已通过,15.093秒;强杀/全停后由两个新进程接续原结果、到期TIMER及下一轮POLL | +| 全量隔离集成回归 | 已通过,退出0;全存储160.211秒、流程进程14.928秒、Worker SDK 92.976秒、SDK进程故障94.972秒;19个SDK示例和2个独立业务示例全部通过 | + +定向入口: + +```sh +go test -race -count=1 -run '^TestIntegration(WorkflowClock|WorkflowClockUpgrade|WorkflowRuntime|WorkflowInstanceAdmission|SchemaAndConstraints|NoForeignKeysUpgrade)$' -v ./internal/storage/mysql +go build -o /tmp/orbit-clock-test-server ./cmd/orbit +ORBIT_TEST_SERVER_BINARY=/tmp/orbit-clock-test-server go test -race -count=1 -run '^TestIntegrationWorkflowRuntimeProcess$' -v ./internal/storage/mysql +``` + +以上命令须设置`ORBIT_TEST_MYSQL_DSN`指向一次性数据库;未配置时用例跳过,不能据此声称通过。[全量脚本](../../scripts/test-mysql.sh)负责隔离MySQL、构建二进制并显式运行进程测试。 + +数据库直接复核外键0、缺少COMMENT的业务字段0;代码复核295个直接持久化字段声明均含中文注释。文档复核847个本地链接和53个JSON代码块,错误0;`git diff --check`通过。 + +存储测试覆盖12个并发定时消费者只创建一个后继、8个并发查询消费者只增加一轮、暂停/满额保留预算、等待被总截止截短、延迟消费已到期事实、审计及轮次写入故障回滚、单轮重试与业务轮次区分、超时收集已落库结果、在途UNKNOWN结清、首轮输入及所有者损坏拒绝。v7升级验证70个历史外键全部移除,323项原索引组件和任务数据保留,额外添加外键时启动检查拒绝。 + +进程测试先确认所有控制进程已经退出,再让TIMER自然到期或提交查询PENDING结果;启动两个全新进程后核对原唤醒时间/绝对截止未改变、无重复任务,查询只产生两轮后由DONE完成。业务任务由真实Pulse执行存储授权与结束,实例使用内部Start;这不替代尚未开放的实例HTTP/SDK端到端回归。 + +开发过程中修正了两个测试夹具问题:业务键保留尾部空格而互斥键禁止首尾空格,因此互斥键改用实例ID;带互斥键的任务同时占用多类许可,因此核验原许可数量未变并在Pulse结清后归零。生产输入校验与许可规则未因此放宽。 + +首次全量存储回归因新故障注入用例未恢复原任务所有者而失败:后续全局租约恢复扫描正确拒绝了这条损坏关联。已为输入/所有者故障注入补齐退出恢复,最终完整重跑通过;不放宽生产关联校验,也不将首次失败计为通过。最后基础检查也已重跑通过,上表全量记录包含该修订。 + +## 剩余原定范围 + +1. SIGNAL早到、重复、超时、契约与关联校验;最终信号和在途查询竞争。 +2. FAIL_AFTER_SETTLED分支取消/收集/终态判断,以及暂停、继续、取消和人工恢复公共操作。 +3. 实例、步骤、来源、信号和审计HTTP/独立SDK、通用场景示例。 +4. P6–P7监控、完整故障矩阵、性能容量、MySQL高可用与备份恢复、SDK发布核对。 + +这些回归验证单个MySQL下的状态与进程故障恢复,不给出生产容量、RPO/RTO或数据库高可用承诺。 diff --git a/internal/storage/mysql/migrations/000019_workflow_clock_checkpoints.up.sql b/internal/storage/mysql/migrations/000019_workflow_clock_checkpoints.up.sql new file mode 100644 index 0000000..6755bc2 --- /dev/null +++ b/internal/storage/mysql/migrations/000019_workflow_clock_checkpoints.up.sql @@ -0,0 +1,20 @@ +-- 为已有定时等待和查询补齐持久化唤醒;检查点只产生候选,运行器仍核验固定版本和全部归属。 +-- 复用既有到期索引及步骤唯一键,不引入外键,也不重算原截止或重置轮次。 +INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at) + SELECT s.namespace_id,RANDOM_BYTES(16),s.workflow_id,s.id,s.revision, + JSON_OBJECT('schema_version',1,'kind','TIMER','activation_id',LOWER(HEX(a.id))),w.deadline_at + FROM weave_steps s JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id + JOIN weave_instances i ON i.namespace_id=s.namespace_id AND i.id=s.workflow_id + JOIN weave_waits w ON w.namespace_id=s.namespace_id AND w.workflow_id=s.workflow_id AND w.activation_id=a.id + WHERE s.kind='TIMER' AND s.state='WAITING' AND a.state='WAITING' AND w.kind='TIMER' AND w.state='WAITING' AND i.state IN ('RUNNING','WAITING','CANCELLING','NEEDS_REVIEW') + ON DUPLICATE KEY UPDATE id=weave_pending_advances.id; + +INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at) + SELECT s.namespace_id,RANDOM_BYTES(16),s.workflow_id,s.id,s.revision, + JSON_OBJECT('schema_version',1,'kind','POLL','activation_id',LOWER(HEX(a.id))), + CASE WHEN p.current_task_id IS NOT NULL THEN p.deadline_at ELSE COALESCE(p.next_poll_at,p.deadline_at) END + FROM weave_steps s JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id + JOIN weave_instances i ON i.namespace_id=s.namespace_id AND i.id=s.workflow_id + JOIN weave_poll_states p ON p.namespace_id=s.namespace_id AND p.workflow_id=s.workflow_id AND p.activation_id=a.id + WHERE s.kind='POLL' AND s.state IN ('ACTIVE','WAITING') AND a.state=s.state AND p.resolved_at IS NULL AND i.state IN ('RUNNING','WAITING','CANCELLING','NEEDS_REVIEW') + ON DUPLICATE KEY UPDATE id=weave_pending_advances.id; diff --git a/internal/storage/mysql/workflow_apply.go b/internal/storage/mysql/workflow_apply.go index 77e9ecf..d7d6314 100644 --- a/internal/storage/mysql/workflow_apply.go +++ b/internal/storage/mysql/workflow_apply.go @@ -251,6 +251,11 @@ func activateWorkflowStep(ctx context.Context, tx *sql.Tx, g *workflowGraph, row if _, err := tx.ExecContext(ctx, `INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at,signal_name,payload_version,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, g.view.NamespaceID, waitID, g.view.ID, id, c.Wait.Kind, []byte(c.Wait.Correlation), until, name, version, now, now); err != nil { return err } + if c.Wait.Kind == "TIMER" { + if err := saveWorkflowClock(ctx, tx, g, row, "TIMER", row.view.Revision+1, until); err != nil { + return err + } + } } if c.Task != nil { if taskID == nil { @@ -268,5 +273,8 @@ func activateWorkflowStep(ctx context.Context, tx *sql.Tx, g *workflowGraph, row return err } } + if c.Kind == "POLL" { + return saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision+1, deadline) + } return nil } diff --git a/internal/storage/mysql/workflow_clock.go b/internal/storage/mysql/workflow_clock.go new file mode 100644 index 0000000..455cc85 --- /dev/null +++ b/internal/storage/mysql/workflow_clock.go @@ -0,0 +1,114 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +// workflowCheckpoint 是待续表中版本化的内部唤醒身份;不保存内存执行栈或新的业务预算。 +type workflowCheckpoint struct { + SchemaVersion uint32 `json:"schema_version"` // SchemaVersion 当前结构版本为1,未知版本禁止推进。 + Kind string `json:"kind"` // Kind 为ROOT、TIMER或POLL,决定实际权威记录的读取方式。 + ActivationID *core.ID `json:"activation_id,omitempty"` // ActivationID 固定等待激活;ROOT不使用该字段。 +} + +func saveWorkflowClock(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, kind string, expected uint64, due time.Time) error { + id, err := core.NewID() + if err != nil { + return err + } + if row.view.ActivationID == nil || kind != "TIMER" && kind != "POLL" { + return errors.New("workflow clock identity invalid") + } + raw, err := json.Marshal(workflowCheckpoint{1, kind, row.view.ActivationID}) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at) VALUES (?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE expected_step_revision=VALUES(expected_step_revision),checkpoint=VALUES(checkpoint),available_at=VALUES(available_at),revision=revision+1,updated_at=UTC_TIMESTAMP(6)`, g.view.NamespaceID, id, g.view.ID, row.view.ID, expected, string(raw), due) + return err +} +func deleteWorkflowClock(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep) error { + _, err := tx.ExecContext(ctx, `DELETE FROM weave_pending_advances WHERE namespace_id=? AND workflow_id=? AND step_id=?`, g.view.NamespaceID, g.view.ID, row.view.ID) + return err +} + +func (r *WorkflowRuntime) advanceClock(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, step core.ID, expected uint64, checkpoint workflowCheckpoint, allow bool, now time.Time) (bool, error) { + var row *workflowStep + for _, candidate := range g.steps { + if candidate.view.ID == step { + row = candidate + break + } + } + if row == nil || checkpoint.ActivationID == nil || string(row.view.Kind) != checkpoint.Kind || expected > row.view.Revision { + return false, errors.New("workflow clock checkpoint mismatch") + } + if row.view.ActivationID == nil || *row.view.ActivationID != *checkpoint.ActivationID || g.view.State.Terminal() { + return false, deleteWorkflowClock(ctx, tx, g, row) + } + if row.view.State != weave.StepActive && row.view.State != weave.StepWaiting { + return false, deleteWorkflowClock(ctx, tx, g, row) + } + var activated time.Time + if err := tx.QueryRowContext(ctx, `SELECT created_at FROM weave_activations WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=?`, c.ns, checkpoint.ActivationID, c.workflow, step).Scan(&activated); err != nil { + return false, err + } + if checkpoint.Kind == "POLL" { + return r.advancePoll(ctx, tx, g, row, c, activated, allow, now) + } + if checkpoint.Kind != "TIMER" || row.view.State != weave.StepWaiting { + return false, errors.New("workflow timer state invalid") + } + var wait core.ID + var deadline time.Time + var kind, state string + var name, version sql.NullString + var correlation []byte + var count int + if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, c.ns, c.workflow, checkpoint.ActivationID).Scan(&count); err != nil { + return false, err + } + if count != 1 { + return false, errors.New("workflow timer wait set invalid") + } + if err := tx.QueryRowContext(ctx, `SELECT id,kind,state,deadline_at,signal_name,payload_version,correlation_key FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, c.ns, c.workflow, checkpoint.ActivationID).Scan(&wait, &kind, &state, &deadline, &name, &version, &correlation); err != nil { + return false, err + } + node, _ := g.program.Step(row.view.Key) + intended := activated.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond) + if kind != "TIMER" || state != "WAITING" || name.Valid || version.Valid || len(correlation) != 0 || !deadline.Equal(minimumTime(intended, g.view.DeadlineAt)) { + return false, errors.New("workflow timer wait invalid") + } + if now.Before(deadline) { + return false, saveWorkflowClock(ctx, tx, g, row, "TIMER", row.view.Revision, deadline) + } + state, reason, stepState := "SATISFIED", "TIMER_ELAPSED", weave.StepSucceeded + output := json.RawMessage(`null`) + if intended.After(g.view.DeadlineAt) { + state, reason, stepState, output = "EXPIRED", "TIMER_WORKFLOW_DEADLINE", weave.StepNeedsReview, nil + } + res, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state=?,resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state='WAITING'`, state, now, now, c.ns, wait) + if err := exactlyOne(res, err); err != nil { + return false, err + } + if err := saveWorkflowResultStep(ctx, tx, g, row, stepState, output, now); err != nil { + return false, err + } + if err := deleteWorkflowClock(ctx, tx, g, row); err != nil { + return false, err + } + if stepState == weave.StepNeedsReview { + if err := holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, reason, now); err != nil { + return false, err + } + } else if err := r.progress(ctx, tx, g, allow, now); err != nil { + return false, err + } + return true, r.audit(ctx, tx, g, c.id, reason, now) +} diff --git a/internal/storage/mysql/workflow_clock_integration_test.go b/internal/storage/mysql/workflow_clock_integration_test.go new file mode 100644 index 0000000..1b6953a --- /dev/null +++ b/internal/storage/mysql/workflow_clock_integration_test.go @@ -0,0 +1,454 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "os" + "strings" + "sync" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" + "git.yuntuo.hk/k2safe/orbit/weave/definition" +) + +func newClockFixture(t *testing.T, ctx context.Context, db *sql.DB) *executionFixture { + t.Helper() + f := newExecutionFixture(t, ctx, db) + return keyedWorker(t, f, f.resources.QueueID, f.a) +} +func clockCandidate(t *testing.T, f *executionFixture, workflow core.ID, key string) workflowCandidate { + t.Helper() + c := workflowCandidate{kind: "ROOT", ns: f.resources.NamespaceID, workflow: workflow} + if err := f.db.QueryRowContext(f.ctx, "SELECT p.id,p.revision FROM weave_pending_advances p JOIN weave_steps s ON s.namespace_id=p.namespace_id AND s.id=p.step_id AND s.workflow_id=p.workflow_id WHERE p.namespace_id=? AND p.workflow_id=? AND s.`key`=?", c.ns, workflow, key).Scan(&c.id, &c.revision); err != nil { + t.Fatal(err) + } + return c +} +func waitClockDue(t *testing.T, f *executionFixture, c workflowCandidate) { + t.Helper() + deadline := time.Now().Add(8 * time.Second) + for time.Now().Before(deadline) { + var due bool + if err := f.db.QueryRowContext(f.ctx, `SELECT available_at<=UTC_TIMESTAMP(6) FROM weave_pending_advances WHERE namespace_id=? AND id=?`, c.ns, c.id).Scan(&due); err != nil { + t.Fatal(err) + } + if due { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("clock did not become due") +} +func runClock(t *testing.T, f *executionFixture, c workflowCandidate) bool { + t.Helper() + done, err := newRuntime(t, f.db).process(f.ctx, c, false) + if err != nil { + t.Fatal(err) + } + return done +} +func timerSpec(queue core.ID, delay uint64, tail bool) definition.Specification { + s := runtimeSpec(queue, 1) + step := s.Body.Steps[0] + s.Body.Steps = []definition.Step{{Key: "timer", Kind: "TIMER", Timer: &definition.Timer{DelayMS: delay}}} + s.Body.Output = definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "timer"}} + s.Body.OutputType = definition.Type{Kind: "NULL"} + if tail { + s.Body.Steps = append(s.Body.Steps, step) + s.Body.Output = definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: step.Key}} + s.Body.OutputType = s.InputType + } + return s +} +func pollingSpec(queue core.ID) definition.Specification { + s := runtimeSpec(queue, 1) + task := *s.Body.Steps[0].Task + task.TimeoutMS = 6000 + task.Retry.AttemptTimeoutMS = 3000 + task.ResultType = definition.Type{Kind: "STRING", Enum: []string{"PENDING", "DONE"}} + key := definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "workflow", Path: []string{"business_key"}}} + concurrencyKey := definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "workflow", Path: []string{"id"}}} + task.BusinessKey = &key + task.ConcurrencyKey = &concurrencyKey + p := definition.Poll{Task: task, MaxRounds: 3, TimeoutMS: 12000, IntervalMS: 1000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}} + s.Body = definition.Block{Steps: []definition.Step{{Key: "query", Kind: "POLL", Poll: &p}}, OutputType: task.ResultType, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "query"}}} + return s +} +func finishPoll(t *testing.T, f *executionFixture, ref worker.AttemptRef, result string) { + t.Helper() + raw, _ := json.Marshal(result) + if _, err := f.a.Finish(f.ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: raw}); err != nil { + t.Fatal(err) + } + runtimeConsume(t, f, newRuntime(t, f.db), ref.TaskID) +} + +func TestIntegrationWorkflowClock(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, stop := context.WithTimeout(context.Background(), 180*time.Second) + defer stop() + db, err := Open(ctx, dsn, 32, 8) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + t.Run("timer_due_duplicate_consumers_and_task_successor", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, timerSpec(f.resources.QueueID, 1000, true)) + c := clockCandidate(t, f, v.ID, "timer") + if runClock(t, f, c) || f.count("pulse_tasks", "1=1") != 0 { + t.Fatal("timer executed before persisted due time") + } + waitClockDue(t, f, c) + var wg sync.WaitGroup + outcomes := make(chan bool, 12) + errs := make(chan error, 12) + for i := 0; i < 12; i++ { + r := newRuntime(t, db) + wg.Add(1) + go func() { defer wg.Done(); done, err := r.process(ctx, c, false); outcomes <- done; errs <- err }() + } + wg.Wait() + close(outcomes) + close(errs) + done := 0 + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + for ok := range outcomes { + if ok { + done++ + } + } + if done != 1 || f.count("pulse_tasks", "1=1") != 1 || f.count("weave_waits", "state='SATISFIED'") != 1 { + t.Fatal("timer duplicated successor", done) + } + ref := retryStart(t, f) + retryFinish(t, f, ref, "SUCCEEDED", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + runtimeView(t, f, v.ID, "SUCCEEDED") + }) + t.Run("timer_pause_collects_and_keeps_root_continuation", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, timerSpec(f.resources.QueueID, 30, true)) + f.exec(`UPDATE weave_instances SET paused=TRUE WHERE namespace_id=?`, f.resources.NamespaceID) + f.exec(`UPDATE pulse_execution_gates SET paused=TRUE WHERE namespace_id=?`, f.resources.NamespaceID) + c := clockCandidate(t, f, v.ID, "timer") + waitClockDue(t, f, c) + if !runClock(t, f, c) || f.count("pulse_tasks", "1=1") != 0 || f.count("weave_waits", "state='SATISFIED'") != 1 { + t.Fatal("paused timer not collected") + } + f.exec(`UPDATE weave_instances SET paused=FALSE WHERE namespace_id=?`, f.resources.NamespaceID) + f.exec(`UPDATE pulse_execution_gates SET paused=FALSE WHERE namespace_id=?`, f.resources.NamespaceID) + runtimeAdvance(t, f, newRuntime(t, db), v.ID) + if f.count("pulse_tasks", "1=1") != 1 { + t.Fatal("timer successor not resumed") + } + }) + t.Run("timer_audit_failure_rolls_back", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, timerSpec(f.resources.QueueID, 20, true)) + c := clockCandidate(t, f, v.ID, "timer") + waitClockDue(t, f, c) + f.exec(`CREATE TRIGGER test_clock_audit_failure BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected clock audit failure'`) + _, err := newRuntime(t, db).process(ctx, c, false) + f.exec(`DROP TRIGGER test_clock_audit_failure`) + if err == nil || f.count("pulse_tasks", "1=1") != 0 || f.count("weave_waits", "state='WAITING'") != 1 || f.count("weave_pending_advances", "1=1") != 1 { + t.Fatal("clock failure partially committed") + } + if !runClock(t, f, c) { + t.Fatal("clock did not recover") + } + }) + t.Run("timer_late_processing_preserves_original_elapsed_time", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + spec := timerSpec(f.resources.QueueID, 20, false) + spec.TimeoutMS = 250 + v := runtimeStart(t, f, spec) + c := clockCandidate(t, f, v.ID, "timer") + time.Sleep(300 * time.Millisecond) + if !runClock(t, f, c) { + t.Fatal("elapsed timer did not complete") + } + runtimeView(t, f, v.ID, "SUCCEEDED") + }) + t.Run("timer_clamped_by_instance_deadline_is_not_early_success", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + spec := runtimeSpec(f.resources.QueueID, 1) + spec.TimeoutMS = 2000 + spec.Body.Steps[0].Task.TimeoutMS = 1000 + spec.Body.Steps[0].Task.Retry.AttemptTimeoutMS = 1000 + spec.Body.Steps = append(spec.Body.Steps, definition.Step{Key: "timer", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 1800}}) + spec.Body.Output = definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "timer"}} + spec.Body.OutputType = definition.Type{Kind: "NULL"} + v := runtimeStart(t, f, spec) + ref := retryStart(t, f) + time.Sleep(250 * time.Millisecond) + retryFinish(t, f, ref, "SUCCEEDED", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + c := clockCandidate(t, f, v.ID, "timer") + waitClockDue(t, f, c) + if !runClock(t, f, c) { + t.Fatal("clamped timer not handled") + } + runtimeView(t, f, v.ID, "NEEDS_REVIEW") + if f.count("weave_waits", "state='EXPIRED'") != 1 { + t.Fatal("clamped duration treated as complete") + } + }) + t.Run("poll_rounds_fixed_payload_concurrency_and_budget", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, pollingSpec(f.resources.QueueID)) + first := retryStart(t, f) + finishPoll(t, f, first, "PENDING") + c := clockCandidate(t, f, v.ID, "query") + if runClock(t, f, c) { + t.Fatal("new round before interval") + } + waitClockDue(t, f, c) + var wg sync.WaitGroup + errs := make(chan error, 8) + for i := 0; i < 8; i++ { + r := newRuntime(t, db) + wg.Add(1) + go func() { defer wg.Done(); _, err := r.process(ctx, c, false); errs <- err }() + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + if f.count("pulse_tasks", "1=1") != 2 || f.count("weave_poll_states", "used_rounds=2 AND current_task_id IS NOT NULL") != 1 || f.count("weave_poll_rounds", "1=1") != 2 { + t.Fatal("poll round duplicated or budget reset") + } + second := retryStart(t, f) + task := f.get(second.TaskID) + if task.BusinessKey != v.BusinessKey || task.ConcurrencyKey != string(v.ID) || !strings.Contains(string(task.Payload), "9007199254740993") || task.Retry.MaxAttempts != 1 { + t.Fatal("poll parameters changed") + } + finishPoll(t, f, second, "PENDING") + c = clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + if !runClock(t, f, c) { + t.Fatal("third round absent") + } + third := retryStart(t, f) + finishPoll(t, f, third, "PENDING") + runtimeView(t, f, v.ID, "NEEDS_REVIEW") + if f.count("pulse_tasks", "1=1") != 3 || f.count("weave_pending_advances", "1=1") != 0 || f.count("weave_poll_states", "used_rounds=3 AND resolved_at IS NOT NULL") != 1 { + t.Fatal("exhaustion reset loop") + } + }) + t.Run("poll_success_and_failure_rollback_of_round_creation", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, pollingSpec(f.resources.QueueID)) + first := retryStart(t, f) + finishPoll(t, f, first, "PENDING") + c := clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + f.exec(`CREATE TRIGGER test_poll_round_failure BEFORE INSERT ON weave_poll_rounds FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected round failure'`) + _, err := newRuntime(t, db).process(ctx, c, false) + f.exec(`DROP TRIGGER test_poll_round_failure`) + if err == nil || f.count("pulse_tasks", "1=1") != 1 || f.count("weave_poll_states", "used_rounds=1 AND current_task_id IS NULL") != 1 || instanceCount(f, `SELECT unfinished_tasks FROM pulse_queues WHERE namespace_id=?`, f.resources.NamespaceID) != 0 { + t.Fatal("partial poll task or budget committed") + } + if !runClock(t, f, c) { + t.Fatal("poll failed to resume after rollback") + } + second := retryStart(t, f) + finishPoll(t, f, second, "DONE") + runtimeView(t, f, v.ID, "SUCCEEDED") + if f.count("weave_pending_advances", "1=1") != 0 { + t.Fatal("resolved poll retained clock") + } + }) + t.Run("poll_pause_and_backpressure_keep_round_and_deadline", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, pollingSpec(f.resources.QueueID)) + ref := retryStart(t, f) + finishPoll(t, f, ref, "PENDING") + var original time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_poll_states WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&original); err != nil { + t.Fatal(err) + } + c := clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + f.exec(`UPDATE weave_instances SET paused=TRUE WHERE namespace_id=?`, f.resources.NamespaceID) + f.exec(`UPDATE pulse_execution_gates SET paused=TRUE WHERE namespace_id=?`, f.resources.NamespaceID) + if runClock(t, f, c) || f.count("pulse_tasks", "1=1") != 1 { + t.Fatal("paused poll started round") + } + f.exec(`UPDATE weave_instances SET paused=FALSE WHERE namespace_id=?`, f.resources.NamespaceID) + f.exec(`UPDATE pulse_execution_gates SET paused=FALSE WHERE namespace_id=?`, f.resources.NamespaceID) + f.exec(`UPDATE pulse_queues SET max_unfinished_tasks=1 WHERE namespace_id=?`, f.resources.NamespaceID) + blocker := f.submit(1) + waitClockDue(t, f, c) + if runClock(t, f, c) || f.count("weave_poll_states", "used_rounds=1 AND current_task_id IS NULL") != 1 { + t.Fatal("quota consumed poll budget") + } + f.exec(`UPDATE pulse_queues SET max_unfinished_tasks=2 WHERE namespace_id=?`, f.resources.NamespaceID) + waitClockDue(t, f, c) + if !runClock(t, f, c) { + t.Fatal("backpressure did not resume") + } + var after time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_poll_states WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&after); err != nil { + t.Fatal(err) + } + if !after.Equal(original) || f.count("pulse_tasks", "id<>?", blocker.ID) != 2 { + t.Fatal("recovery extended deadline or duplicated round") + } + }) + t.Run("expired_inflight_poll_collects_existing_result_first", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + spec := pollingSpec(f.resources.QueueID) + spec.Body.Steps[0].Poll.TimeoutMS = 2000 + spec.Body.Steps[0].Poll.Task.TimeoutMS = 1500 + spec.Body.Steps[0].Poll.Task.Retry.AttemptTimeoutMS = 1000 + v := runtimeStart(t, f, spec) + ref := retryStart(t, f) + if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`"DONE"`)}); err != nil { + t.Fatal(err) + } + c := clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + if runClock(t, f, c) { + t.Fatal("deadline scanner preempted committed result") + } + runtimeView(t, f, v.ID, "RUNNING") + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + runtimeView(t, f, v.ID, "SUCCEEDED") + }) + t.Run("poll_deadline_does_not_grant_extra_round", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + spec := pollingSpec(f.resources.QueueID) + spec.Body.Steps[0].Poll.TimeoutMS = 2000 + spec.Body.Steps[0].Poll.Task.TimeoutMS = 1500 + spec.Body.Steps[0].Poll.Task.Retry.AttemptTimeoutMS = 1000 + v := runtimeStart(t, f, spec) + ref := retryStart(t, f) + finishPoll(t, f, ref, "PENDING") + c := clockCandidate(t, f, v.ID, "query") + time.Sleep(2100 * time.Millisecond) + if !runClock(t, f, c) { + t.Fatal("poll timeout absent") + } + runtimeView(t, f, v.ID, "NEEDS_REVIEW") + if f.count("pulse_tasks", "1=1") != 1 || f.count("weave_poll_states", "used_rounds=1 AND resolved_at IS NOT NULL") != 1 { + t.Fatal("late poll started extra work") + } + }) + t.Run("request_retry_stays_in_same_business_round", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + spec := pollingSpec(f.resources.QueueID) + p := spec.Body.Steps[0].Poll + p.Task.Retry.MaxAttempts = 2 + p.Task.Retry.InitialDelayMS = 1000 + p.Task.Retry.MaxDelayMS = 1000 + v := runtimeStart(t, f, spec) + first := retryStart(t, f) + retryFinish(t, f, first, "FAILED", true) + if f.count("weave_inbox", "1=1") != 0 || f.count("weave_poll_states", "used_rounds=1") != 1 { + t.Fatal("network retry advanced business round") + } + c := clockCandidate(t, f, v.ID, "query") + if runClock(t, f, c) { + t.Fatal("in-flight round overlapped") + } + time.Sleep(1050 * time.Millisecond) + retry := retryStart(t, f) + if retry.TaskID != first.TaskID { + t.Fatal("request retry created a new task") + } + finishPoll(t, f, retry, "PENDING") + c = clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + if !runClock(t, f, c) { + t.Fatal("business round did not advance after successful pending result") + } + if f.count("weave_poll_states", "used_rounds=2 AND max_attempts_per_round=2") != 1 || f.count("pulse_tasks", "attempt_count=2") != 1 || f.count("pulse_tasks", "attempt_count=0 AND max_attempts=2") != 1 { + t.Fatal("round/attempt budgets were mixed") + } + }) + t.Run("poll_deadline_keeps_inflight_evidence_until_pulse_recovery", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + spec := pollingSpec(f.resources.QueueID) + spec.Body.Steps[0].Poll.TimeoutMS = 2000 + spec.Body.Steps[0].Poll.Task.TimeoutMS = 1500 + spec.Body.Steps[0].Poll.Task.Retry.AttemptTimeoutMS = 1000 + v := runtimeStart(t, f, spec) + ref := retryStart(t, f) + permits := f.used() + if permits == 0 { + t.Fatal("started query did not reserve permits") + } + c := clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + if !runClock(t, f, c) { + t.Fatal("in-flight deadline did not close workflow gate") + } + runtimeView(t, f, v.ID, "NEEDS_REVIEW") + if f.count("weave_steps", "kind='POLL' AND state='ACTIVE'") != 1 || f.count("weave_poll_states", "current_task_id=? AND resolved_at IS NULL", ref.TaskID) != 1 || f.count("pulse_tasks", "state='RUNNING'") != 1 || f.used() != permits { + t.Fatal("clock falsely settled in-flight work or released its permit") + } + if _, err := f.storeB.Recover(ctx, 100); err != nil { + t.Fatal(err) + } + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + runtimeView(t, f, v.ID, "NEEDS_REVIEW") + if f.count("weave_steps", "kind='POLL' AND state='NEEDS_REVIEW'") != 1 || f.count("weave_poll_states", "current_task_id IS NULL AND used_rounds=1 AND resolved_at IS NOT NULL") != 1 || f.count("weave_pending_advances", "1=1") != 0 || f.used() != 0 { + t.Fatal("Pulse unknown result was lost or created another round") + } + }) + t.Run("poll_continuation_rejects_corrupt_frozen_task", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, pollingSpec(f.resources.QueueID)) + ref := retryStart(t, f) + finishPoll(t, f, ref, "PENDING") + c := clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + original := f.get(ref.TaskID).Payload + defer f.exec(`UPDATE pulse_tasks SET payload=? WHERE namespace_id=? AND id=?`, string(original), f.resources.NamespaceID, ref.TaskID) + f.exec(`UPDATE pulse_tasks SET payload='{"n":1}' WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, ref.TaskID) + if _, err := newRuntime(t, db).process(ctx, c, false); err == nil { + t.Fatal("corrupt first-round parameters granted new work") + } + if f.count("pulse_tasks", "1=1") != 1 || f.count("weave_poll_states", "used_rounds=1 AND current_task_id IS NULL") != 1 || f.count("weave_pending_advances", "1=1") != 1 { + t.Fatal("rejected frozen parameters changed budget") + } + }) + t.Run("poll_clock_validates_owner_without_foreign_keys", func(t *testing.T) { + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, pollingSpec(f.resources.QueueID)) + ref := retryStart(t, f) + c := clockCandidate(t, f, v.ID, "query") + var original core.ID + if err := db.QueryRowContext(ctx, `SELECT owner_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, ref.TaskID).Scan(&original); err != nil { + t.Fatal(err) + } + defer f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, original, f.resources.NamespaceID, ref.TaskID) + f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, ref.TaskID) + // 提前形成到期候选不能绕过实际父记录与所有者校验。 + f.exec(`UPDATE weave_pending_advances SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, c.ns, c.id) + if _, err := newRuntime(t, db).process(ctx, c, false); err == nil { + t.Fatal("wrong activation owner accepted without a foreign key") + } + if f.count("weave_poll_states", "used_rounds=1 AND current_task_id=?", ref.TaskID) != 1 || f.count("pulse_tasks", "1=1") != 1 { + t.Fatal("corrupt owner changed round state") + } + }) +} diff --git a/internal/storage/mysql/workflow_clock_upgrade_integration_test.go b/internal/storage/mysql/workflow_clock_upgrade_integration_test.go new file mode 100644 index 0000000..a9eaa28 --- /dev/null +++ b/internal/storage/mysql/workflow_clock_upgrade_integration_test.go @@ -0,0 +1,162 @@ +package mysql + +import ( + "context" + "encoding/json" + "os" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + driver "github.com/go-sql-driver/mysql" +) + +// 用真实v18结构和直接写入的历史等待记录验证数据回填,不冒充完整实例的执行回归。 +func TestIntegrationWorkflowClockUpgrade(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + admin, err := Open(ctx, dsn, 2, 1) + if err != nil { + t.Fatal(err) + } + defer admin.Close() + name := "orbit_clock_upgrade_" + string(executionID(t)) + if _, err := admin.ExecContext(ctx, "CREATE DATABASE `"+name+"`"); err != nil { + t.Fatal(err) + } + defer func() { + cleanup, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + if _, err := admin.ExecContext(cleanup, "DROP DATABASE `"+name+"`"); err != nil { + t.Error(err) + } + }() + cfg, err := driver.ParseDSN(dsn) + if err != nil { + t.Fatal("invalid test DSN") + } + cfg.DBName = name + db, err := Open(ctx, cfg.FormatDSN(), 4, 2) + if err != nil { + t.Fatal(err) + } + defer db.Close() + exec := func(q string, args ...any) { + t.Helper() + if _, err := db.ExecContext(ctx, q, args...); err != nil { + t.Fatal(err) + } + } + exec(`CREATE TABLE orbit_schema_migrations (version BIGINT UNSIGNED PRIMARY KEY,name VARCHAR(191) NOT NULL,checksum CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,dirty BOOLEAN NOT NULL,applied_at DATETIME(6) NULL) ENGINE=InnoDB`) + migrations, err := Migrations() + if err != nil { + t.Fatal(err) + } + for _, m := range migrations { + if m.Version > 18 { + break + } + statements, err := splitSQL(m.SQL) + if err != nil { + t.Fatal(err) + } + for _, q := range statements { + exec(q) + } + exec(`INSERT INTO orbit_schema_migrations(version,name,checksum,dirty,applied_at) VALUES (?,?,?,FALSE,UTC_TIMESTAMP(6))`, m.Version, m.Name, m.Checksum) + } + var now time.Time + if err := db.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil { + t.Fatal(err) + } + ns := executionID(t) + deadline, next := now.Add(time.Hour), now.Add(time.Minute) + type fixture struct { + workflow, step, activation core.ID + kind string + due time.Time + } + seed := func(key, kind, state string, terminal bool) fixture { + t.Helper() + f := fixture{executionID(t), executionID(t), executionID(t), kind, deadline} + instanceState := "WAITING" + var completed *time.Time + if terminal { + instanceState, completed = "CANCELLED", &now + } + exec(`INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,state,input,policy_snapshot,deadline_at,completed_at) VALUES (?,?,?,?,?,?,'{}','{}',?,?)`, ns, f.workflow, executionID(t), executionID(t), executionID(t), instanceState, deadline, completed) + exec("INSERT INTO weave_steps(namespace_id,id,workflow_id,`key`,kind,state,current_activation_id,revision) VALUES (?,?,?,?,?,?,?,7)", ns, f.step, f.workflow, key, kind, state, f.activation) + exec(`INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,input) VALUES (?,?,?,?,1,?,'null')`, ns, f.activation, f.workflow, f.step, state) + if kind == "TIMER" { + exec(`INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at) VALUES (?,?,?,?,'TIMER','',?)`, ns, executionID(t), f.workflow, f.activation, deadline) + } else { + var current *core.ID + var nextPoll *time.Time + if state == "ACTIVE" { + id := executionID(t) + current = &id + } else { + nextPoll, f.due = &next, next + } + exec(`INSERT INTO weave_poll_states(namespace_id,id,workflow_id,activation_id,max_rounds,used_rounds,max_attempts_per_round,interval_ms,deadline_at,next_poll_at,current_task_id) VALUES (?,?,?,?,10,3,2,1000,?,?,?)`, ns, executionID(t), f.workflow, f.activation, deadline, nextPoll, current) + } + return f + } + timer := seed("timer", "TIMER", "WAITING", false) + poll := seed("query", "POLL", "WAITING", false) + inflight := seed("query", "POLL", "ACTIVE", false) + seed("terminal", "TIMER", "WAITING", true) + stale := seed("stale", "TIMER", "WAITING", false) + exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, executionID(t), ns, stale.step) + // 已有时钟与根待续均不能被数据回填重写;模拟提前部署已产生的检查点。 + existing := seed("existing", "TIMER", "WAITING", false) + existingID, rootID, rootStep := executionID(t), executionID(t), executionID(t) + original := `{"schema_version":1,"kind":"TIMER","activation_id":"` + string(existing.activation) + `"}` + exec(`INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at,revision) VALUES (?,?,?,?,7,?,?,11)`, ns, existingID, existing.workflow, existing.step, original, next) + exec(`INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at) VALUES (?,?,?,?,4,'{"schema_version":1,"kind":"ROOT"}',?)`, ns, rootID, timer.workflow, rootStep, now) + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if err := CheckSchema(ctx, db); err != nil { + t.Fatal(err) + } + for _, f := range []fixture{timer, poll, inflight} { + var raw []byte + var due time.Time + var expected uint64 + if err := db.QueryRowContext(ctx, `SELECT checkpoint,available_at,expected_step_revision FROM weave_pending_advances WHERE namespace_id=? AND workflow_id=? AND step_id=?`, ns, f.workflow, f.step).Scan(&raw, &due, &expected); err != nil { + t.Fatal(err) + } + var checkpoint workflowCheckpoint + if err := json.Unmarshal(raw, &checkpoint); err != nil || checkpoint.SchemaVersion != 1 || checkpoint.Kind != f.kind || checkpoint.ActivationID == nil || *checkpoint.ActivationID != f.activation || expected != 7 || !due.Equal(f.due) { + t.Fatalf("incorrect v18 clock backfill: %s %v", raw, err) + } + } + var count int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_pending_advances`).Scan(&count); err != nil || count != 5 { + t.Fatal("missing/duplicate/terminal/stale checkpoint", count, err) + } + var raw string + var revision uint64 + var due time.Time + if err := db.QueryRowContext(ctx, `SELECT checkpoint,revision,available_at FROM weave_pending_advances WHERE namespace_id=? AND id=?`, ns, existingID).Scan(&raw, &revision, &due); err != nil || raw != original || revision != 11 || !due.Equal(next) { + t.Fatal("existing clock overwritten", err) + } + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_pending_advances WHERE namespace_id=? AND id=? AND checkpoint='{"schema_version":1,"kind":"ROOT"}' AND expected_step_revision=4 AND available_at=?`, ns, rootID, now).Scan(&count); err != nil || count != 1 { + t.Fatal("root checkpoint overwritten", err) + } + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_poll_states WHERE max_rounds=10 AND used_rounds=3 AND max_attempts_per_round=2 AND interval_ms=1000 AND deadline_at=? AND resolved_at IS NULL`, deadline).Scan(&count); err != nil || count != 2 { + t.Fatal("poll budget or deadline changed", err) + } + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_pending_advances`).Scan(&count); err != nil || count != 5 { + t.Fatal("repeated migration duplicated checkpoints", err) + } + t.Log("v18 timer/pending-query/in-flight-query checkpoints backfilled at original time; existing ROOT/clock preserved; terminal and stale activations excluded; budgets unchanged; zero foreign keys") +} diff --git a/internal/storage/mysql/workflow_consume.go b/internal/storage/mysql/workflow_consume.go index a08b528..5f05ff5 100644 --- a/internal/storage/mysql/workflow_consume.go +++ b/internal/storage/mysql/workflow_consume.go @@ -208,6 +208,13 @@ func (r *WorkflowRuntime) consume(ctx context.Context, tx *sql.Tx, g *workflowGr if _, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET current_task_id=NULL,next_poll_at=?,resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, next, settled, now, c.ns, c.workflow, source.activation); err != nil { return false, err } + if next != nil { + if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision+1, *next); err != nil { + return false, err + } + } else if err := deleteWorkflowClock(ctx, tx, g, row); err != nil { + return false, err + } if settled != nil { waitState := "CANCELLED" if state == weave.StepSucceeded { diff --git a/internal/storage/mysql/workflow_poll_clock.go b/internal/storage/mysql/workflow_poll_clock.go new file mode 100644 index 0000000..d3ef1c5 --- /dev/null +++ b/internal/storage/mysql/workflow_poll_clock.go @@ -0,0 +1,211 @@ +package mysql + +import ( + "bytes" + "context" + "database/sql" + "errors" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/engine" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +func (r *WorkflowRuntime) advancePoll(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, c workflowCandidate, activated time.Time, allow bool, now time.Time) (bool, error) { + var p weave.PollState + err := tx.QueryRowContext(ctx, `SELECT id,revision,max_rounds,used_rounds,max_attempts_per_round,interval_ms,deadline_at,next_poll_at,current_task_id,resolved_at FROM weave_poll_states WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, c.ns, c.workflow, row.view.ActivationID).Scan(&p.ID, &p.Revision, &p.MaxRounds, &p.UsedRounds, &p.MaxAttemptsPerRound, &p.IntervalMS, &p.DeadlineAt, &p.NextPollAt, &p.CurrentTaskID, &p.ResolvedAt) + if err != nil { + return false, err + } + node, _ := g.program.Step(row.view.Key) + policy := node.Poll + if policy == nil || p.MaxRounds != policy.MaxRounds || p.MaxAttemptsPerRound != policy.Task.Retry.MaxAttempts || p.IntervalMS != policy.IntervalMS || p.UsedRounds < 1 || !p.DeadlineAt.Equal(minimumTime(g.view.DeadlineAt, activated.Add(time.Duration(policy.TimeoutMS)*time.Millisecond))) { + return false, errors.New("workflow poll policy mismatch") + } + if p.ResolvedAt != nil { + return false, errors.New("unsettled poll step has resolved budget") + } + var count, minRound, maxRound uint32 + if err := tx.QueryRowContext(ctx, "SELECT COUNT(*),COALESCE(MIN(`round`),0),COALESCE(MAX(`round`),0) FROM weave_poll_rounds WHERE namespace_id=? AND workflow_id=? AND activation_id=?", c.ns, c.workflow, row.view.ActivationID).Scan(&count, &minRound, &maxRound); err != nil { + return false, err + } + if count != p.UsedRounds || minRound != 1 || maxRound != p.UsedRounds { + return false, errors.New("workflow poll rounds incomplete") + } + var lastTask core.ID + if err := tx.QueryRowContext(ctx, "SELECT task_id FROM weave_poll_rounds WHERE namespace_id=? AND workflow_id=? AND activation_id=? AND `round`=?", c.ns, c.workflow, row.view.ActivationID, p.UsedRounds).Scan(&lastTask); err != nil { + return false, err + } + if p.CurrentTaskID != nil { + if *p.CurrentTaskID != lastTask || p.NextPollAt != nil || row.view.State != weave.StepActive { + return false, errors.New("workflow in-flight poll mismatch") + } + if err := validatePollTaskOwner(ctx, tx, g, row, lastTask); err != nil { + return false, err + } + if now.Before(p.DeadlineAt) { + return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, p.DeadlineAt) + } + task, err := readTask(ctx, tx, c.ns, lastTask, false) + if err != nil { + return false, err + } + if task.QueueID != policy.Task.QueueID { + return false, errors.New("workflow in-flight poll queue mismatch") + } + if task.State.Terminal() || task.State == "NEEDS_REVIEW" { + // 原结果已可靠落库时,先让Inbox按事件时间裁决,不能用扫描器的当前时间覆盖它。 + var available time.Time + err := tx.QueryRowContext(ctx, `SELECT available_at FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND state='PENDING' ORDER BY source_revision DESC LIMIT 1`, c.ns, c.workflow, lastTask, row.view.ActivationID).Scan(&available) + if err != nil { + return false, err + } + if available.Before(now.Add(time.Second)) { + available = now.Add(time.Second) + } + return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, available) + } + // 到期但仍在途时先关闭门控,保留当前轮次和ACTIVE步骤继续接收Pulse结清结果。 + changed := g.view.State != weave.InstanceNeedsReview + if changed { + if err := holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, "POLL_INFLIGHT_DEADLINE", now); err != nil { + return false, err + } + if err := r.audit(ctx, tx, g, c.id, "POLL_INFLIGHT_DEADLINE", now); err != nil { + return false, err + } + } + return changed, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, now.Add(time.Second)) + } + if row.view.State != weave.StepWaiting || p.NextPollAt == nil { + return false, errors.New("workflow poll continuation missing") + } + if !now.Before(p.DeadlineAt) || p.UsedRounds >= p.MaxRounds { + reason := "POLL_DEADLINE" + if p.UsedRounds >= p.MaxRounds { + reason = "POLL_ROUNDS_EXHAUSTED" + } + if _, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET resolved_at=?,next_poll_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, now, c.ns, p.ID); err != nil { + return false, err + } + if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='EXPIRED',resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND activation_id=? AND state='WAITING'`, now, now, c.ns, c.workflow, row.view.ActivationID); err != nil { + return false, err + } + if err := saveWorkflowResultStep(ctx, tx, g, row, weave.StepNeedsReview, nil, now); err != nil { + return false, err + } + if err := deleteWorkflowClock(ctx, tx, g, row); err != nil { + return false, err + } + if err := holdWorkflow(ctx, tx, g, weave.InstanceNeedsReview, reason, now); err != nil { + return false, err + } + return true, r.audit(ctx, tx, g, c.id, reason, now) + } + if now.Before(*p.NextPollAt) { + return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, *p.NextPollAt) + } + if !allow { + return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, minimumTime(p.DeadlineAt, now.Add(time.Second))) + } + round, err := p.NextRound(now) + if err != nil { + return false, err + } + // 只有上一轮成功查询并已分类为PENDING才可以消耗新轮次,不能用时间推进绕过失败预算。 + var disposition string + if err := tx.QueryRowContext(ctx, `SELECT error_code FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND state='PROCESSED' ORDER BY source_revision DESC LIMIT 1`, c.ns, c.workflow, lastTask, row.view.ActivationID).Scan(&disposition); err != nil { + return false, err + } + if disposition != "POLL_PENDING" { + return false, errors.New("poll previous round was not pending") + } + in, err := pollTaskInput(ctx, tx, g, row, policy.Task) + if err != nil { + return false, err + } + taskID, err := core.NewID() + if err != nil { + return false, err + } + task, err := insertAcceptedTaskUntil(ctx, tx, c.ns, taskID, in, nil, &p.DeadlineAt) + if errors.Is(err, fault.QueueFull) || errors.Is(err, fault.ApplicationFull) { + return false, saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, minimumTime(p.DeadlineAt, now.Add(time.Second))) + } + if err != nil { + return false, err + } + res, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET owner_kind=?,owner_id=?,gate_id=? WHERE namespace_id=? AND id=? AND owner_kind IS NULL`, engine.TaskOwnerKind, row.view.ActivationID, g.gate, c.ns, taskID) + if err := exactlyOne(res, err); err != nil { + return false, err + } + roundID, err := core.NewID() + if err != nil { + return false, err + } + if _, err := tx.ExecContext(ctx, "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id) VALUES (?,?,?,?,?,?)", c.ns, roundID, c.workflow, row.view.ActivationID, round, taskID); err != nil { + return false, err + } + res, err = tx.ExecContext(ctx, `UPDATE weave_poll_states SET used_rounds=?,current_task_id=?,next_poll_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_task_id IS NULL`, round, taskID, now, c.ns, p.ID, p.Revision) + if err := exactlyOne(res, err); err != nil { + return false, err + } + if err := saveWorkflowResultStep(ctx, tx, g, row, weave.StepActive, nil, now); err != nil { + return false, err + } + if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, p.DeadlineAt); err != nil { + return false, err + } + if err := emitEvent(ctx, tx, c.ns, taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": taskID, "state": task.State, "workflow_id": c.workflow, "activation_id": row.view.ActivationID, "round": round}); err != nil { + return false, err + } + // 新查询已持久化,实例从纯等待恢复RUNNING;组合推进仍受同一个32步预算。 + if err := r.progress(ctx, tx, g, allow, now); err != nil { + return false, err + } + return true, r.audit(ctx, tx, g, c.id, "POLL_ROUND_ACCEPTED", now) +} + +func validatePollTaskOwner(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, id core.ID) error { + var owner, gate *core.ID + var kind sql.NullString + if err := tx.QueryRowContext(ctx, `SELECT owner_kind,owner_id,gate_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, g.view.NamespaceID, id).Scan(&kind, &owner, &gate); err != nil { + return err + } + if kind.String != engine.TaskOwnerKind || owner == nil || row.view.ActivationID == nil || *owner != *row.view.ActivationID || gate == nil || *gate != g.gate { + return errors.New("poll task ownership invalid") + } + return nil +} + +// pollTaskInput 复用首轮已经解析并冻结的输入、业务键及互斥键;不重算依赖可能已改变的表达式。 +func pollTaskInput(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, spec definition.Task) (service.SubmitInput, error) { + var first core.ID + if err := tx.QueryRowContext(ctx, "SELECT task_id FROM weave_poll_rounds WHERE namespace_id=? AND workflow_id=? AND activation_id=? AND `round`=1", g.view.NamespaceID, g.view.ID, row.view.ActivationID).Scan(&first); err != nil { + return service.SubmitInput{}, err + } + if err := validatePollTaskOwner(ctx, tx, g, row, first); err != nil { + return service.SubmitInput{}, err + } + base, err := readTask(ctx, tx, g.view.NamespaceID, first, false) + if err != nil { + return service.SubmitInput{}, err + } + input, err := jsonvalue.Canonical(base.Payload) + if err != nil { + return service.SubmitInput{}, err + } + activationInput, err := jsonvalue.Canonical(row.input) + if err != nil { + return service.SubmitInput{}, err + } + if !bytes.Equal(input, activationInput) || definition.ValidateData(spec.InputType, base.Payload) != nil || base.QueueID != spec.QueueID || base.Handler != spec.Handler || base.HandlerVersion != spec.HandlerVersion || base.PayloadVersion != spec.PayloadVersion || base.Retry != spec.Retry || base.Priority != spec.Priority || base.RecoverySafety != spec.RecoverySafety { + return service.SubmitInput{}, errors.New("poll frozen task parameters mismatch") + } + return service.NormalizeSubmit(service.SubmitInput{QueueID: base.QueueID, Handler: base.Handler, HandlerVersion: base.HandlerVersion, PayloadVersion: base.PayloadVersion, Payload: base.Payload, BusinessKey: base.BusinessKey, ConcurrencyKey: base.ConcurrencyKey, Priority: base.Priority, TaskTimeoutMS: spec.TimeoutMS, Retry: base.Retry, RecoverySafety: base.RecoverySafety}) +} diff --git a/internal/storage/mysql/workflow_runtime.go b/internal/storage/mysql/workflow_runtime.go index f333b48..f005ec2 100644 --- a/internal/storage/mysql/workflow_runtime.go +++ b/internal/storage/mysql/workflow_runtime.go @@ -13,7 +13,7 @@ import ( weave "git.yuntuo.hk/k2safe/orbit/weave/model" ) -// WorkflowRuntime 消费持久化结果与根推进检查点;不持有业务执行租约或调用业务系统。 +// WorkflowRuntime 消费持久化结果、根推进与等待时钟检查点;不持有业务执行租约或调用业务系统。 // 每个候选独立短事务,数据库行锁决定唯一推进权,节点内存不保存恢复游标。 type WorkflowRuntime struct { db *sql.DB @@ -30,7 +30,7 @@ func NewWorkflowRuntime(db *sql.DB, node core.ID) (*WorkflowRuntime, error) { type WorkflowRuntimeStats struct { Examined int // Examined 本批取出的持久化候选数,包含被其他节点先处理或跳过的记录。 Consumed int // Consumed 已提交确认或隔离的内部结果数,事务回滚不计入。 - Advanced int // Advanced 已提交推进的根检查点数,不等于业务任务执行次数。 + Advanced int // Advanced 已提交推进的根或等待检查点数,不等于业务任务执行次数。 } type workflowCandidate struct { @@ -129,12 +129,16 @@ func (r *WorkflowRuntime) process(ctx context.Context, c workflowCandidate, skip if due.After(now) { return nil } - var checkpoint struct { - SchemaVersion uint32 `json:"schema_version"` - Kind string `json:"kind"` + var checkpoint workflowCheckpoint + if json.Unmarshal(raw, &checkpoint) != nil || checkpoint.SchemaVersion != 1 { + return errors.New("workflow checkpoint invalid") + } + if checkpoint.Kind == "TIMER" || checkpoint.Kind == "POLL" { + done, err = r.advanceClock(ctx, tx, g, c, step, expected, checkpoint, allow, now) + return err } root := g.steps[workflowRootKey] - if json.Unmarshal(raw, &checkpoint) != nil || checkpoint.SchemaVersion != 1 || checkpoint.Kind != "ROOT" || step != root.view.ID || expected > root.view.Revision { + if checkpoint.Kind != "ROOT" || checkpoint.ActivationID != nil || step != root.view.ID || expected > root.view.Revision { return errors.New("workflow checkpoint invalid") } if g.view.State.Terminal() { diff --git a/internal/storage/mysql/workflow_runtime_process_integration_test.go b/internal/storage/mysql/workflow_runtime_process_integration_test.go index 453d814..951e2ab 100644 --- a/internal/storage/mysql/workflow_runtime_process_integration_test.go +++ b/internal/storage/mysql/workflow_runtime_process_integration_test.go @@ -2,12 +2,14 @@ package mysql import ( "context" + "encoding/json" "os" "os/exec" "path/filepath" "testing" "time" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" "git.yuntuo.hk/k2safe/orbit/weave/instance" mysqldriver "github.com/go-sql-driver/mysql" ) @@ -145,4 +147,68 @@ func TestIntegrationWorkflowRuntimeProcess(t *testing.T) { t.Fatal("process restart duplicated work or lost confirmation") } t.Log("sole Orbit process SIGKILL confirmed; result committed with all controllers stopped; two new Orbit processes completed exactly three tasks and three result confirmations") + for _, p := range processes { + stop(p) + } + timer := runtimeStart(t, f, timerSpec(f.resources.QueueID, 100, true)) + var originalWake time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, timer.ID).Scan(&originalWake); err != nil { + t.Fatal(err) + } + time.Sleep(150 * time.Millisecond) // 全部控制进程保持退出,定时等待自然到期。 + if f.count("pulse_tasks", "1=1") != 3 { + t.Fatal("timer advanced with every controller stopped") + } + start("timer-replacement-a") + start("timer-replacement-b") + wait(func() bool { return f.count("pulse_tasks", "1=1") == 4 }, "expired timer takeover") + ref = retryStart(t, f) + if f.get(ref.TaskID).CreatedAt.Before(originalWake) { + t.Fatal("timer successor accepted before original wake") + } + retryFinish(t, f, ref, "SUCCEEDED", false) + wait(func() bool { + current, err := api.Get(ctx, f.token, timer.ID) + return err == nil && current.State == "SUCCEEDED" + }, "timer workflow completion") + var afterWake time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, timer.ID).Scan(&afterWake); err != nil || !afterWake.Equal(originalWake) { + t.Fatal("timer wake was reset on restart", err) + } + for _, p := range processes { + stop(p) + } + f = keyedWorker(t, f, f.resources.QueueID, f.a) + poll := runtimeStart(t, f, pollingSpec(f.resources.QueueID)) + var originalDeadline time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_poll_states WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, poll.ID).Scan(&originalDeadline); err != nil { + t.Fatal(err) + } + ref = retryStart(t, f) + if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`"PENDING"`)}); err != nil { + t.Fatal(err) + } + time.Sleep(1100 * time.Millisecond) + if f.count("pulse_tasks", "1=1") != 5 || f.count("weave_poll_states", "used_rounds=1") != 1 { + t.Fatal("poll continued while all controllers stopped") + } + start("poll-replacement-a") + start("poll-replacement-b") + wait(func() bool { return f.count("pulse_tasks", "1=1") == 6 }, "next poll round after all-stop") + ref = retryStart(t, f) + if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`"DONE"`)}); err != nil { + t.Fatal(err) + } + wait(func() bool { + current, err := api.Get(ctx, f.token, poll.ID) + return err == nil && current.State == "SUCCEEDED" + }, "poll workflow completion") + var afterDeadline time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_poll_states WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, poll.ID).Scan(&afterDeadline); err != nil || !afterDeadline.Equal(originalDeadline) { + t.Fatal("poll deadline reset", err) + } + if f.count("weave_poll_states", "used_rounds=2 AND resolved_at IS NOT NULL") != 1 || f.count("weave_poll_rounds", "1=1") != 2 || f.count("weave_pending_advances", "1=1") != 0 { + t.Fatal("poll takeover lost budget or duplicated round") + } + t.Log("all controllers exited before timer expiry and poll PENDING delivery; two fresh processes preserved original wake/deadline and completed timer plus exactly two query rounds") } diff --git a/weave/model/pending_advance.go b/weave/model/pending_advance.go index 0917ca1..122d87b 100644 --- a/weave/model/pending_advance.go +++ b/weave/model/pending_advance.go @@ -6,7 +6,7 @@ import ( core "git.yuntuo.hk/k2safe/orbit/model" ) -// PendingAdvance 表示暂停或分批推进后的续接检查点,避免仅依赖内存唤醒。 +// PendingAdvance 保存根步骤续接或等待时钟检查点,避免仅依赖内存唤醒。 // 对应表 weave_pending_advances;这是服务端持久化模型,不直接作为 API 响应。 type PendingAdvance struct { // Record 包含命名空间、记录标识、并发版本和 UTC 时间。 @@ -17,7 +17,7 @@ type PendingAdvance struct { StepID core.ID `db:"step_id"` // ExpectedStepRevision 恢复推进时要求的步骤版本,避免使用过期检查点。 ExpectedStepRevision uint64 `db:"expected_step_revision"` - // Checkpoint 分批或暂停后继续推进所需的持久化 JSON 检查点。 + // Checkpoint 版本化 JSON;ROOT用于分批/暂停续接,TIMER/POLL另固定激活ID以隔离旧唤醒。 Checkpoint core.JSON `db:"checkpoint"` // AvailableAt 最早可被领取或处理的时间,统一 UTC。 AvailableAt time.Time `db:"available_at"`