From af1ae4cb21438f3b93082b76282c04da7879427a Mon Sep 17 00:00:00 2001 From: k2safe Date: Fri, 18 Sep 2026 06:10:04 +0800 Subject: [PATCH] feat: recover workflow waits within original deadlines --- AGENTS.md | 14 +- README.md | 2 +- docs/06-workflow-spec.md | 4 +- docs/07-data-model.md | 7 +- docs/08-development-plan.md | 19 +- docs/10-workflow-engine-core.md | 2 +- docs/11-workflow-admission.md | 4 +- docs/12-workflow-runtime.md | 4 +- docs/14-workflow-signals.md | 2 +- docs/15-workflow-controls.md | 4 +- docs/16-workflow-recovery.md | 12 +- docs/17-workflow-poll-recovery.md | 6 +- docs/18-workflow-wait-recovery.md | 55 ++++ docs/reports/phase-5-wait-recovery.md | 39 +++ internal/storage/mysql/workflow_clock.go | 10 +- internal/storage/mysql/workflow_consume.go | 7 +- internal/storage/mysql/workflow_poll_clock.go | 6 +- .../storage/mysql/workflow_poll_continue.go | 159 +++++++++ internal/storage/mysql/workflow_recovery.go | 20 +- ...rkflow_runtime_process_integration_test.go | 85 +++++ .../storage/mysql/workflow_signal_runtime.go | 6 +- internal/storage/mysql/workflow_wait_rearm.go | 159 +++++++++ ...workflow_wait_recovery_integration_test.go | 305 ++++++++++++++++++ weave/recovery/service.go | 40 +++ weave/recovery/service_test.go | 22 +- 25 files changed, 941 insertions(+), 52 deletions(-) create mode 100644 docs/18-workflow-wait-recovery.md create mode 100644 docs/reports/phase-5-wait-recovery.md create mode 100644 internal/storage/mysql/workflow_poll_continue.go create mode 100644 internal/storage/mysql/workflow_wait_rearm.go create mode 100644 internal/storage/mysql/workflow_wait_recovery_integration_test.go diff --git a/AGENTS.md b/AGENTS.md index 7882122..f015622 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,17 +31,17 @@ - P4定义发布使用schema_version=1有限结构，词法作用域禁止跨分支/前向读取，任务策略显式冻结；SDK类型独立。发布按认证→应用额度→操作身份→定义→排序队列持锁；原键与同版本去重仍校验当前权限。定义/版本/revision/审计/回执同事务；同版本不同内容不得覆盖。保留精确数字，校验存储原文与结构摘要；不得把发布通过或DSL支持误称为实例执行已实现。 -- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归（docs/reports/phase-4-definitions.md）。实例启动/固定版本/首批任务同事务的内部存储已补齐；内部事件消费、顺序与组合推进及根接续已接入；TIMER与后续POLL时钟已接入，SIGNAL及最终查询竞争已接入，内部控制与停止结清已接入，继续独立SIGNAL/TIMER等待重建及实例公共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消费与根接续已接入，等待和停止结清运行器已接入，内部核查/TASK恢复已接入，独立SIGNAL/TIMER等待重建继续开发。Plan最多32个变更，More必须持久化待续；耗尽变更预算仍须遍历其他已活动分支，不能误判在途工作不可达。冻结分支选择，精确数值，不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务，不能把纯快照往返测试当作真实进程恢复。 +- weave/engine仅计算有限决策；实例受理、Inbox消费与根接续已接入，等待和停止结清运行器已接入，内部核查/TASK恢复已接入，等待恢复已接入内部服务，公共API/SDK继续开发。Plan最多32个变更，More必须持久化待续；耗尽变更预算仍须遍历其他已活动分支，不能误判在途工作不可达。冻结分支选择，精确数值，不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务，不能把纯快照往返测试当作真实进程恢复。 - 实例内部受理已实现，信号内部运行器已接入，公共实例/信号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。TIMER到期与POLL后续轮次已接入，见docs/13-workflow-clock.md；SIGNAL及最终信号竞争已接入（docs/14-workflow-signals.md）；内部FAIL_AFTER_SETTLED分支结清与控制已接入（docs/15-workflow-controls.md）；独立SIGNAL/TIMER等待重建、公共控制API和实例SDK仍须完成，不得按部分自动运行宣称P4/P5完成。 +- 本轮消费运行器和真实Orbit强杀/双节点接管已验证，见docs/reports/phase-4-runtime.md。TIMER到期与POLL后续轮次已接入，见docs/13-workflow-clock.md；SIGNAL及最终信号竞争已接入（docs/14-workflow-signals.md）；内部FAIL_AFTER_SETTLED分支结清与控制已接入（docs/15-workflow-controls.md）；公共控制API和实例SDK仍须完成，不得按部分自动运行宣称P4/P5完成。 - TIMER/POLL复用weave_pending_advances及到期索引，时钟固定激活ID；迁移000019只回填原等待/查询时钟，不改截止和预算。后续POLL只由已确认POLL_PENDING开启，首轮精确输入/资源键冻结，每轮Pulse尝试独立计数；满额/暂停不消耗轮数。截止遇到已落库结果先等待Inbox，仍在途则关闭门控但保留ACTIVE和当前任务，许可由Pulse结清；不得由时钟假定外部请求已结束。 @@ -52,9 +52,11 @@ - StopKind/StopRequestedAt冻结首次CANCEL或FAILURE目标，后来的用户取消不覆盖原失败目标。结清复用ROOT，每事务最多一个Pulse取消、32项步骤收尾；结果流锁取得后不能再锁另一任务。终态前校验全部历史任务的双向归属、未结束尝试和未释放许可，UNKNOWN不得终结。迁移000021恢复旧停止意图，不添加外键。 - 人工核查使用显式workflows:recover、当前实例revision、步骤/任务版本及激活身份，必须有原因和证据；执行结论、步骤结论、解除核查分别确认。新决定拒绝PENDING/QUARANTINED Inbox，原键回执仍核验权限。核查回执/额度/双审计/结果流同事务，结果流锁最后取得。 -- TASK人工重试保留首次绝对截止和精确输入，以新激活/关联任务继续消耗原max_attempts；激活数也受该上限限制，不能无限制造零尝试任务。成功步骤、未知执行、原停止意图或预算耗尽均不得重试。Continue不解除paused、重置预算或撤销停止。POLL跨激活恢复已实现（docs/17-workflow-poll-recovery.md），独立SIGNAL/TIMER等待重建尚未实现，公共HTTP/SDK也未开放，不能据此宣称P5完成。 +- TASK人工重试保留首次绝对截止和精确输入，以新激活/关联任务继续消耗原max_attempts；激活数也受该上限限制，不能无限制造零尝试任务。成功步骤、未知执行、原停止意图或预算耗尽均不得重试。Continue不解除paused、重置预算或撤销停止。POLL跨激活恢复已实现（docs/17-workflow-poll-recovery.md），等待恢复已接入内部服务，公共HTTP/SDK也未开放，不能据此宣称P5完成。 - POLL人工恢复以新激活重查同一业务轮，FirstRound/FirstRoundAttemptsUsed保留原轮次和已用尝试；实际尝试行、retry_of链和冻结参数必须相符。同轮任务数也受原max_attempts限制，PENDING不得人工重查绕过轮次上限。查询截止按首次激活、请求截止按该轮首任务，不允许重计时；下一业务轮才获得完整单轮策略。 -- POLL恢复的最终信号等待保留原契约/关联/截止并更换代次，旧winner和已消费信号只留历史；存在当前代次PENDING信号时拒绝重建。迁移000022无外键，新增2个中文注释字段。独立SIGNAL/TIMER等待重建、公共HTTP/SDK与生产验收继续推进。 +- POLL恢复的最终信号等待保留原契约/关联/截止并更换代次，旧winner和已消费信号只留历史；存在当前代次PENDING信号时拒绝重建。迁移000022无外键，新增2个中文注释字段。公共HTTP/SDK与生产验收继续推进。 -- 核查将UNKNOWN请求确认成成功PENDING时，当前RetryStep仍拒绝重查同轮；显式恢复等待/后续业务轮的管理操作尚待实现，必须保留原间隔、轮次与截止并使用新的审计决定，不能由ReviewTask自动解除核查。 +- 核查将UNKNOWN请求确认成成功PENDING时，当前RetryStep仍拒绝重查同轮；ContinuePoll显式恢复原等待，必须核验当前SUCCEEDED/PENDING任务、已处理Inbox及不可变来源。新审计固定来源事件/版本、原间隔与截止，时钟/最终信号核验证明后才推进；不改写旧Inbox处置，ReviewTask不自动解除核查。 + +- RearmWait仅重建独立SIGNAL/TIMER，显式wait_verified及最新步骤/激活/实例版本，连续激活上限32。旧WAITING激活/等待关闭，新代次保留首次输入、契约、关联及截止；PENDING信号、已有赢家、停止意图及过期均拒绝。TIMER运行和成功核查按首次激活计时，恢复后仍须Continue；无新持久化字段或外键。 diff --git a/README.md b/README.md index 4d18483..1d72f26 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)已接入后台运行器。[定时等待、后续查询轮次及原预算恢复](docs/13-workflow-clock.md)已接入；[外部信号和查询最终信号竞争](docs/14-workflow-signals.md)已接入内部运行器；[暂停/继续/取消与失败结清](docs/15-workflow-controls.md)已接入内部运行器；[人工核查、解除核查与TASK新激活](docs/16-workflow-recovery.md)已接入内部服务；[POLL跨激活恢复](docs/17-workflow-poll-recovery.md)已接入；独立SIGNAL/TIMER等待重建及实例公共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)已接入；[外部信号和查询最终信号竞争](docs/14-workflow-signals.md)已接入内部运行器；[暂停/继续/取消与失败结清](docs/15-workflow-controls.md)已接入内部运行器；[人工核查、解除核查与TASK新激活](docs/16-workflow-recovery.md)已接入内部服务；[POLL跨激活恢复](docs/17-workflow-poll-recovery.md)已接入；[PENDING接续和独立等待恢复](docs/18-workflow-wait-recovery.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 d4e302a..f7dad3f 100644 --- a/docs/06-workflow-spec.md +++ b/docs/06-workflow-spec.md @@ -1,6 +1,6 @@ # Weave 流程编排功能与设计 -> 版本 0.8 · 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)已接入。[信号等待及最终查询竞争](14-workflow-signals.md)已接入内部运行器；[暂停/继续/取消与停止结清](15-workflow-controls.md)已实现内部链路；[人工核查与TASK恢复](16-workflow-recovery.md)已接入内部服务；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；独立SIGNAL/TIMER等待重建及实例公共API/SDK仍待P4–P5实现验收。 +> 版本 0.9 · 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)已接入。[信号等待及最终查询竞争](14-workflow-signals.md)已接入内部运行器；[暂停/继续/取消与停止结清](15-workflow-controls.md)已实现内部链路；[人工核查与TASK恢复](16-workflow-recovery.md)已接入内部服务；[POLL跨激活恢复](17-workflow-poll-recovery.md)和[等待恢复](18-workflow-wait-recovery.md)已接入；实例公共API/SDK仍待P4–P5实现验收。 ## 1. 职责与边界 @@ -174,7 +174,7 @@ Weave 不执行特定业务逻辑，也不把整段业务代码放入平台进 3. 在一个事务内记录管理操作，创建新的步骤激活及关联 Pulse 任务，并固定关联关系。 4. 保留旧激活与旧任务结果，沿用恰当的业务幂等身份。 -执行核查、步骤结论、解除核查及TASK新激活已按[人工恢复约定](16-workflow-recovery.md)实现内部服务，原预算和绝对截止不重置；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入，独立SIGNAL/TIMER等待重建仍需补齐，公共API另行交付。普通任务人工重试入口应拒绝绕过流程管理的关联任务请求。已经确认成功的步骤不能通过“重试流程”随意再执行一次。 +执行核查、步骤结论、解除核查及TASK新激活已按[人工恢复约定](16-workflow-recovery.md)实现内部服务，原预算和绝对截止不重置；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入，等待恢复已接入内部服务，公共API另行交付。普通任务人工重试入口应拒绝绕过流程管理的关联任务请求。已经确认成功的步骤不能通过“重试流程”随意再执行一次。 取消和补偿是不同业务操作。若未来增加补偿，补偿本身也必须是有权限、可审计、可重试或可核查的业务步骤；不能宣称撤销流程就能撤回已经生效的外部业务操作。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index a75d290..743b7a8 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -211,9 +211,14 @@ Instance新增StopKind和StopRequestedAt，冻结首次用户取消或失败结 ## 流程人工核查与TASK新激活 -复用现有激活sequence、任务retry_of、审计和幂等回执，不增加表、字段或外键。TASK人工重试以新激活保留旧执行，原尝试预算跨激活扣除，原绝对截止不改变；执行核查和步骤逻辑结论分别保存。35个实体、303个中文注释字段。当前内部支持范围与独立SIGNAL/TIMER等待重建剩余项见[核查约定](16-workflow-recovery.md)。 +复用现有激活sequence、任务retry_of、审计和幂等回执，不增加表、字段或外键。TASK人工重试以新激活保留旧执行，原尝试预算跨激活扣除，原绝对截止不改变；执行核查和步骤逻辑结论分别保存。此阶段为35个实体、303个中文注释字段；后续POLL恢复增加2个预算字段，见下节。核查操作见[核查约定](16-workflow-recovery.md)，独立等待恢复见[等待恢复](18-workflow-wait-recovery.md)。 ### 000022：跨激活保留查询轮次与尝试 PollState新增FirstRound和FirstRoundAttemptsUsed，记录当前激活的首个业务轮及该轮在旧激活已消耗的尝试。UsedRounds继续使用首次激活起的业务轮编号；原单轮尝试和查询/请求绝对截止均不重置。新业务轮恢复原单轮策略，人工恢复同轮只分配剩余尝试。当前35个实体、305个直接持久化字段声明，Go/SQL均有中文注释，无外键。v21升级保留旧数据、时间和预算，详情见[查询恢复](17-workflow-poll-recovery.md)。 + + +### 等待恢复与PENDING接续 + +复用现有激活、等待、查询状态和不可变审计，不新增迁移或字段。RearmWait以新代次保留首次等待截止；ContinuePoll保留当前代次和原轮次，以权威已处理结果及管理审计恢复间隔。仍为35个实体/305个中文注释字段，无外键，见[等待恢复](18-workflow-wait-recovery.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 546fc5b..23c6ab3 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -28,8 +28,8 @@ | 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)已接入后台，顺序/分支/并行及真实进程恢复已验证；[TIMER与后续POLL轮次](13-workflow-clock.md)已接入；[信号消费和最终查询竞争](14-workflow-signals.md)已接入内部运行器；[内部控制与停止结清](15-workflow-controls.md)已接入；[内部核查与TASK新激活](16-workflow-recovery.md)已接入；独立SIGNAL/TIMER等待重建和实例公共API/SDK继续开发 | -| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器；最终信号竞争及在途查询结清已接入；内部暂停/继续/取消与失败结清已接入；执行核查、步骤结论、解除核查及TASK新激活已接入；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；独立SIGNAL/TIMER等待重建及公共API/SDK继续开发 | +| 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)已接入；[信号消费和最终查询竞争](14-workflow-signals.md)已接入内部运行器；[内部控制与停止结清](15-workflow-controls.md)已接入；[内部核查与TASK新激活](16-workflow-recovery.md)已接入；等待恢复已接入内部服务；实例公共API/SDK继续开发 | +| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器；最终信号竞争及在途查询结清已接入；内部暂停/继续/取消与失败结清已接入；执行核查、步骤结论、解除核查及TASK新激活已接入；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；等待恢复已接入内部服务；公共API/SDK继续开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | | P7 首版交付核对 | SDK 兼容与发布准备、完整 API 手册、场景示例、全量回归及交付清单 | 新接入者按文档跑通全部示例；所有首版能力可追溯到实现与测试；报告实际生产边界 | 待开发 | @@ -41,10 +41,21 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( 1. 已实现FAIL_AFTER_SETTLED与显式取消的内部分支结清，见[控制报告](reports/phase-5-controls.md)：持久取消意图、复用Pulse取消、收集已发出查询/任务结果、等待全部已激活工作结清后才进入失败或取消终态。效果未知仍保留核查。 2. 已实现实例暂停/继续/取消内部操作，以control_revision、操作幂等身份、明确原因及审计保护控制决定；暂停不延长原截止，继续不覆盖核查或取消状态。 -3. 已实现内部执行核查、步骤结论、解除核查及TASK新激活和旧激活隔离，见[核查报告](reports/phase-5-recovery.md)。POLL跨激活恢复已实现（见[查询恢复](17-workflow-poll-recovery.md)）；继续核查后已确认PENDING的有界等待接续及独立SIGNAL/TIMER等待重建，保留原预算、轮次、结果和证据，不能通过换激活/新键绕过未知效果或执行上限。 +3. 已实现内部执行核查、步骤结论、解除核查及TASK新激活和旧激活隔离，见[核查报告](reports/phase-5-recovery.md)。POLL跨激活恢复已实现（见[查询恢复](17-workflow-poll-recovery.md)）；[核查后PENDING接续和独立等待重建](18-workflow-wait-recovery.md)也已接入内部服务，保留原预算、轮次、结果和证据。 4. 同步开放实例/步骤/信号/来源/审计HTTP、独立SDK和端到端场景，覆盖原键恢复、权限、版本冲突、竞争与进程全停；公共步骤/激活查询须返回目标激活代次，调用方据此发送恢复后的信号，不能要求调用方自行猜测sequence。公共能力逐项记录实际验收结果。 5. 转入P6/P7的生产故障、容量、运维、发布与全部文档核对。内部存储阶段通过不会提前勾选公共接口或生产验收。 +公共接入下一批按以下调用链交付，逐项补齐服务端、独立SDK、Markdown手册和真实HTTP回归： + +| 调用链 | 需要补齐的内容 | 验收重点 | +| --- | --- | --- | +| 启动与读取 | 显式固定版本启动、当前实例、步骤及当前激活代次；后续补筛选分页和单次有界历史读取 | 原键丢回执恢复、当前权限、读取快照一致性、未激活步骤语义 | +| 控制与信号 | 暂停/继续/取消、固定步骤和代次的信号提交及原受理回执 | control_revision与通用revision区分、迟到/旧代次、暂停期间结果收集 | +| 核查与追踪 | 六类人工恢复操作、激活/轮次/等待/信号及管理审计读取、Pulse任务流程来源 | 当前实例/步骤/任务版本、证据和显式声明、旧历史保留、有界响应和权限隔离 | +| 通用接入场景 | 并行汇合、外部回执、有限轮询、暂停核查恢复的可运行SDK示例 | 真实Worker执行、完整进程退出后接续、未知效果保留、全链路回归 | + +接口只使用已经落地的内部事务规则；SDK保持独立module，不能引用服务端模型或数据库实现。生产故障及容量结论仍由P6专门验收。 + ### P3 功能范围核对 可靠订阅链路完成后，重新对照 [原功能范围](01-functional-spec.md) 和 [事务衔接设计](04-reliability-and-data.md)，以下原定首版能力已按核对结果补齐；进一步发现的相对延迟与改参数关联示例也已交付。[P3验收](reports/phase-3.md)逐项保留实现证据及尚待P4–P7完成的原范围，不按已有模块反推全平台完成。 @@ -119,6 +130,8 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( ## 7. 文档入口 +- [等待恢复](18-workflow-wait-recovery.md) +- [等待恢复验证记录](reports/phase-5-wait-recovery.md) - [有界查询恢复](17-workflow-poll-recovery.md) - [POLL恢复验证记录](reports/phase-5-poll-recovery.md) - [数据模型](07-data-model.md) diff --git a/docs/10-workflow-engine-core.md b/docs/10-workflow-engine-core.md index 534afc7..75dc2c5 100644 --- a/docs/10-workflow-engine-core.md +++ b/docs/10-workflow-engine-core.md @@ -74,7 +74,7 @@ v16及以前的未处理记录缺少冻结来源，升级后统一标为QUARANTI ## 4. 后续存储适配必须满足的条件 -以下保留存储运行器的完整要求。第1项已按[受理报告](reports/phase-4-admission.md)验证；第2–4项结果消费和根接续已接入[实际运行器](12-workflow-runtime.md)，第5项的TIMER/POLL已接入[时钟运行器](13-workflow-clock.md)，第5项的信号等待已接入[信号运行器](14-workflow-signals.md)，内部控制与结清已接入[控制运行器](15-workflow-controls.md)，[核查与TASK新激活](16-workflow-recovery.md)已接入；独立SIGNAL/TIMER等待重建、接口和SDK仍继续实现： +以下保留存储运行器的完整要求。第1项已按[受理报告](reports/phase-4-admission.md)验证；第2–4项结果消费和根接续已接入[实际运行器](12-workflow-runtime.md)，第5项的TIMER/POLL已接入[时钟运行器](13-workflow-clock.md)，第5项的信号等待已接入[信号运行器](14-workflow-signals.md)，内部控制与结清已接入[控制运行器](15-workflow-controls.md)，[核查与TASK新激活](16-workflow-recovery.md)已接入；接口和SDK仍继续实现： 1. 启动时认证、固定不可变版本/输入和策略、校验全部队列、计算实例截止，并在同事务创建实例、门控、首批激活/任务及审计回执。相同身份返回原实例。 2. 消费者领取内部事件后核对命名空间、来源任务版本、激活/轮次及当前步骤版本；消费确认、步骤变化、后续任务或持久化待续记录同事务提交。 diff --git a/docs/11-workflow-admission.md b/docs/11-workflow-admission.md index d6c610b..9072489 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或命令行入口；[内部控制和结清](15-workflow-controls.md)已接入，[内部核查及TASK恢复](16-workflow-recovery.md)已接入，独立SIGNAL/TIMER等待重建继续开发；[信号运行器](14-workflow-signals.md)已接入。** Weave相关公开SDK目前仅支持定义发布。[TIMER和后续POLL轮次](13-workflow-clock.md)已接入，TASK及组合节点已可自动推进，不能将此解释成全部等待/控制能力已实现。 +状态：内部Go服务与MySQL存储适配已实现实例Start/Get/Steps、固定版本、首批决策落库和可靠结果交付。[后台Inbox消费和根待续扫描](12-workflow-runtime.md)已接通。**尚未开放实例HTTP路由、启动SDK或命令行入口；[内部控制和结清](15-workflow-controls.md)已接入，[内部核查及TASK恢复](16-workflow-recovery.md)已接入，等待恢复已接入内部服务，公共API/SDK继续开发；[信号运行器](14-workflow-signals.md)已接入。** Weave相关公开SDK目前仅支持定义发布。[TIMER和后续POLL轮次](13-workflow-clock.md)已接入，TASK及组合节点已可自动推进，不能将此解释成全部等待/控制能力已实现。 ## 1. 受理事务 @@ -49,7 +49,7 @@ Get返回当前实例视图；Start重报返回受理时冻结的原始视图， | 活跃实例 | 每个namespace最多10000个RUNNING/WAITING/CANCELLING/NEEDS_REVIEW实例；终态不计入；满额仍可恢复原受理回执 | | 信号关联 | 新受理最多512字节；数据库兼容旧191字符最长764字节，升级不截断历史 | | 截止 | 数据库时间生成；任务、查询与等待均不超过实例总截止 | -| 自动运行 | Inbox消费、根待续、TIMER到期及后续POLL轮次已接入；SIGNAL与最终查询竞争已接入，内部控制与结清已接入，内部核查及TASK恢复已接入，独立SIGNAL/TIMER等待重建仍待实现 | +| 自动运行 | Inbox消费、根待续、TIMER到期及后续POLL轮次已接入；SIGNAL与最终查询竞争已接入，内部控制与结清已接入，内部核查及TASK恢复已接入，等待恢复已接入内部服务，公共API/SDK仍待实现 | 迁移000018为Wait增加signal_name和payload_version两个带中文注释的可空字段；NULL仅表示TIMER或尚未核实的历史元数据，未来消费者不得把历史NULL自动解释为有效的新信号等待。correlation_key转为精确字节存储；实例business_key改为utf8mb4_0900_bin。该迁移时为35个实体、含公共Record的295个直接持久化字段声明，全部有中文注释，继续保持零外键。 diff --git a/docs/12-workflow-runtime.md b/docs/12-workflow-runtime.md index 0736930..171efe7 100644 --- a/docs/12-workflow-runtime.md +++ b/docs/12-workflow-runtime.md @@ -1,6 +1,6 @@ # Weave 结果消费与自动推进 -状态：内部实例的TASK结果消费、顺序、分支、有限并行汇合和根检查点接续已接入Orbit后台运行器。[定时等待和后续查询轮次](13-workflow-clock.md)也已接入。[信号等待与最终查询竞争](14-workflow-signals.md)已接入。实例公共HTTP/SDK仍未开放，[失败后的分支取消结清与内部控制](15-workflow-controls.md)已接入，[内部核查及TASK恢复](16-workflow-recovery.md)已接入，独立SIGNAL/TIMER等待重建继续开发，不能将本阶段等同于完整Weave交付。 +状态：内部实例的TASK结果消费、顺序、分支、有限并行汇合和根检查点接续已接入Orbit后台运行器。[定时等待和后续查询轮次](13-workflow-clock.md)也已接入。[信号等待与最终查询竞争](14-workflow-signals.md)已接入。实例公共HTTP/SDK仍未开放，[失败后的分支取消结清与内部控制](15-workflow-controls.md)已接入，[内部核查及TASK恢复](16-workflow-recovery.md)已接入，等待恢复已接入内部服务，公共API/SDK继续开发，不能将本阶段等同于完整Weave交付。 ## 1. 后台运行和事务边界 @@ -64,4 +64,4 @@ flowchart LR 进程回归由`make integration`构建二进制后显式执行。缺少一次性数据库或`ORBIT_TEST_SERVER_BINARY`时测试会跳过；跳过不构成进程恢复证据。验证记录见[本阶段报告](reports/phase-4-runtime.md)。 -持久化信号等待及查询最终信号竞争已按[信号运行约定](14-workflow-signals.md)实现。内部控制及停止结清也已接入。下一步仍包括独立SIGNAL/TIMER等待重建、实例/来源公共API、独立SDK和通用端到端场景。P4整体及P5–P7继续按[完整计划](08-development-plan.md)执行。 +持久化信号等待及查询最终信号竞争已按[信号运行约定](14-workflow-signals.md)实现。内部控制及停止结清也已接入。下一步仍包括实例/来源公共API、独立SDK和通用端到端场景。P4整体及P5–P7继续按[完整计划](08-development-plan.md)执行。 diff --git a/docs/14-workflow-signals.md b/docs/14-workflow-signals.md index 9ee0112..fcf92ee 100644 --- a/docs/14-workflow-signals.md +++ b/docs/14-workflow-signals.md @@ -65,4 +65,4 @@ [受理回归](../internal/storage/mysql/workflow_signals_integration_test.go)、[消费回归](../internal/storage/mysql/workflow_signal_runtime_integration_test.go)、[真实v19升级](../internal/storage/mysql/workflow_signal_upgrade_integration_test.go)和[实际进程恢复](../internal/storage/mysql/workflow_runtime_process_integration_test.go)对应受理/并发/故障/升级边界，实际运行结果记录在[阶段报告](reports/phase-4-signals.md)。 -内部失败分支取消结清、实例暂停/继续/取消已接入。人工核查及[POLL新激活的最终信号换代](17-workflow-poll-recovery.md)已接入；下一步完成独立等待重建，再交付实例与信号HTTP、独立SDK及完整场景。当前内部存储测试不替代尚未开放的公共接口验收，也不替代生产MySQL切换与容量测试。 +内部失败分支取消结清、实例暂停/继续/取消已接入。人工核查及[POLL新激活的最终信号换代](17-workflow-poll-recovery.md)已接入；[PENDING接续与独立等待重建](18-workflow-wait-recovery.md)已接入内部服务，下一步交付实例与信号HTTP、独立SDK及完整场景。当前内部存储测试不替代尚未开放的公共接口验收，也不替代生产MySQL切换与容量测试。 diff --git a/docs/15-workflow-controls.md b/docs/15-workflow-controls.md index 70927e4..d100ee7 100644 --- a/docs/15-workflow-controls.md +++ b/docs/15-workflow-controls.md @@ -1,6 +1,6 @@ # Weave 控制与停止结清 -状态：内部Go暂停、继续、取消服务及后台失败/取消结清运行器已接入。**公共实例/控制HTTP与独立SDK尚未开放，[内部核查及TASK新激活](16-workflow-recovery.md)已接入，独立SIGNAL/TIMER等待重建仍在后续交付范围**。本文说明已实现的持久化规则，不把内部存储回归当作全部Weave或生产验收。 +状态：内部Go暂停、继续、取消服务及后台失败/取消结清运行器已接入。**公共实例/控制HTTP与独立SDK尚未开放，[内部核查及TASK新激活](16-workflow-recovery.md)已接入，等待恢复已接入内部服务，公共入口仍在后续交付范围**。本文说明已实现的持久化规则，不把内部存储回归当作全部Weave或生产验收。 ## 1. 控制操作 @@ -60,4 +60,4 @@ CANCELLED仅表示本次流程停止已结清，不撤销先前成功的业务 [控制及结清回归](../internal/storage/mysql/workflow_controls_integration_test.go)覆盖控制版本、权限重报、开始竞争、失败/未知、审计回滚、信号与查询及超过32步骤接续；[真实v20升级](../internal/storage/mysql/workflow_stop_upgrade_integration_test.go)验证兼容回填；[实际Orbit进程回归](../internal/storage/mysql/workflow_runtime_process_integration_test.go)覆盖停止意图和结果在全停期间保存后由两个新进程接续。实际结果见[阶段报告](reports/phase-5-controls.md)。 -下一步仍需完成独立SIGNAL/TIMER等待重建、公共实例/控制/信号/审计接口、独立SDK及可运行场景，再完成P6–P7生产故障、性能、运维和发布验收。当前测试不替代生产MySQL复制、故障切换、备份恢复或RPO/RTO验证。 +下一步仍需完成公共实例/控制/信号/审计接口、独立SDK及可运行场景，再完成P6–P7生产故障、性能、运维和发布验收。当前测试不替代生产MySQL复制、故障切换、备份恢复或RPO/RTO验证。 diff --git a/docs/16-workflow-recovery.md b/docs/16-workflow-recovery.md index c9dede2..f639582 100644 --- a/docs/16-workflow-recovery.md +++ b/docs/16-workflow-recovery.md @@ -1,15 +1,17 @@ # Weave 人工核查与接续 -状态：内部Go服务已提供执行核查、步骤结论确认、解除核查，以及TASK/POLL新激活重试。POLL跨激活预算与附属信号换代见[查询恢复](17-workflow-poll-recovery.md)。公共HTTP、独立SDK和通用接入示例尚未交付；独立SIGNAL/TIMER等待重建继续实现。完整交付继续按[开发计划](08-development-plan.md)推进。 +状态：内部Go服务已提供执行核查、步骤结论确认、解除核查，以及TASK/POLL新激活重试。POLL跨激活预算与附属信号换代见[查询恢复](17-workflow-poll-recovery.md)。公共HTTP、独立SDK和通用接入示例尚未交付；[PENDING接续及独立等待重建](18-workflow-wait-recovery.md)已接入内部服务。完整交付继续按[开发计划](08-development-plan.md)推进。 -## 1. 四类操作 +## 1. 六类操作 -[内部服务](../weave/recovery/service.go)要求显式`workflows:recover`权限，普通Worker、任务核查和流程暂停权限均不自动包含它。四个方法都接收认证身份、实例ID、稳定幂等键和命令，命名空间与操作者从认证取得。 +[内部服务](../weave/recovery/service.go)要求显式`workflows:recover`权限，普通Worker、任务核查和流程暂停权限均不自动包含它。六个方法都接收认证身份、实例ID、稳定幂等键和命令，命名空间与操作者从认证取得。 | 方法 | 作用 | 不会自动发生的事情 | | --- | --- | --- | | ReviewTask | 核实本实例历史激活拥有的NEEDS_REVIEW任务，写入SUCCEEDED/FAILED、明确执行已结束和证据 | 不把查询请求成功解释成业务成功，不解除步骤或实例核查 | | ResolveStep | 将当前待核查/失败/取消叶子步骤核实为SUCCEEDED或FAILED，保存符合固定契约的逻辑结果 | 不重新执行旧任务，不清除实例核查或原停止意图 | +| ContinuePoll | 根据已处理的权威PENDING结果恢复原等待，保留轮次和间隔 | 不重查同轮，不改写原Inbox处置，不解除实例核查 | +| RearmWait | 在原截止内为独立SIGNAL/TIMER建立新等待代次 | 不延长原截止，不丢弃未消费信号，不重跑成功步骤 | | Continue | 在当前证据和步骤状态允许时解除实例核查，保存根接续 | 不解除显式paused，不延长截止，不撤销取消或失败结清 | | RetryStep | 为TASK或POLL建立新的激活和关联任务，继续消耗原任务或原业务轮尝试预算 | 不重跑成功步骤，不改变输入/业务关联，POLL恢复不另起业务轮次 | @@ -27,7 +29,7 @@ ReviewTask还要求`task_id`、`expected_task_revision`、`execution_settled=true`、SUCCEEDED/FAILED结论和非null JSON结果。声明表示全部历史外部执行已结束且不会继续产生效果；租约失效和UNKNOWN本身不能证明这一点。 -ResolveStep和RetryStep都要求固定`step_key`、当前`activation_id`及`expected_step_revision`。前者要求`execution_settled=true`，后者要求`safe_to_retry=true`。Continue要求`review_resolved=true`，服务端仍独立检查当前图、历史任务与来源事件，不把声明作为绕过检查的开关。 +ResolveStep和RetryStep都要求固定`step_key`、当前`activation_id`及`expected_step_revision`。前者要求`execution_settled=true`，后者要求`safe_to_retry=true`。ContinuePoll要求`pending_confirmed=true`，RearmWait要求`wait_verified=true`；这两项还要求StepDecision。Continue要求`review_resolved=true`，服务端仍独立检查当前图、历史任务与来源事件，不把声明作为绕过检查的开关。 同键相同命令返回首次冻结回执；同键不同内容冲突。回执恢复仍检查当前凭据和固定定义引用的全部队列权限。后台新增事实使原实例revision失效，新键携带旧版本会冲突；原键可继续核对已提交决定。每个成功人工操作同时递增revision和control_revision，防止旧暂停/继续命令覆盖人工恢复后的控制状态。 @@ -54,7 +56,7 @@ TASK恢复规则如下。要求没有取消或失败停止意图，实例和原 创建新任务后实例仍处于核查，门控保持关闭。操作者确认其他事项已处理后调用Continue，才允许新任务取得执行授权；原paused仍为true时还需正常RESUME。 -POLL跨激活恢复已实现，保留原业务轮、同轮剩余尝试、查询总截止和该轮原任务截止；成功返回PENDING的轮次不能人工重查。附属最终信号等待沿用原契约/关联/截止并更换代次，详见[查询恢复](17-workflow-poll-recovery.md)。核查后已确认PENDING的显式等待接续仍需补齐；独立SIGNAL/TIMER逻辑结论可核查，等待重建仍待实现。 +POLL跨激活恢复已实现，保留原业务轮、同轮剩余尝试、查询总截止和该轮原任务截止；成功返回PENDING的轮次不能人工重查。附属最终信号等待沿用原契约/关联/截止并更换代次，详见[查询恢复](17-workflow-poll-recovery.md)。核查后已确认PENDING的显式等待接续与独立SIGNAL/TIMER等待重建已实现，见[等待恢复](18-workflow-wait-recovery.md)。 ## 5. 停止与到期 diff --git a/docs/17-workflow-poll-recovery.md b/docs/17-workflow-poll-recovery.md index 3607b61..5fbdd7e 100644 --- a/docs/17-workflow-poll-recovery.md +++ b/docs/17-workflow-poll-recovery.md @@ -1,6 +1,6 @@ # Weave 有界查询的人工恢复 -状态：内部 `RetryStep` 已支持 POLL 新激活，保留跨激活的业务轮次、同轮剩余尝试和绝对截止；附属最终信号等待随新激活换代。核查后已确认 PENDING 的显式等待接续、独立 SIGNAL/TIMER 的等待重建仍待实现，公共实例/恢复 HTTP、独立 SDK 和可运行接入示例继续按[开发计划](08-development-plan.md)交付。验证范围见[阶段报告](reports/phase-5-poll-recovery.md)。 +状态：内部 `RetryStep` 已支持 POLL 新激活，保留跨激活的业务轮次、同轮剩余尝试和绝对截止；附属最终信号等待随新激活换代。[核查后PENDING接续和独立等待重建](18-workflow-wait-recovery.md)已接入内部服务，公共实例/恢复 HTTP、独立 SDK 和可运行接入示例继续按[开发计划](08-development-plan.md)交付。验证范围见[阶段报告](reports/phase-5-poll-recovery.md)。 ## 1. 业务轮次与请求尝试 @@ -15,7 +15,7 @@ | 恢复请求成功返回 PENDING，间隔到达 | 2 | 2 | 3 | 新业务轮的任务截止受原查询总截止截短 | | 第二轮请求失败，已用 1 次，再人工恢复 | 3 | 2 | 2 | 第二轮第一次任务的绝对截止、原查询总截止 | -恢复前核验真实尝试行数、任务计数、原处理器/版本/策略/输入、业务键、互斥键、所有者、连续激活及 `retry_of` 链。每一业务轮的任务总数也不超过原单轮尝试上限，防止反复取消未开始任务而无限制造激活。一次成功查询已返回 PENDING 的轮次不能人工重试；轮次耗尽须核查最终事实或取消。若 UNKNOWN 经 ReviewTask 被核实为成功 PENDING，现有操作不会自动恢复等待；原预算内显式接续后续业务轮的操作仍需实现，不能借同轮 RetryStep 绕过间隔或次数。 +恢复前核验真实尝试行数、任务计数、原处理器/版本/策略/输入、业务键、互斥键、所有者、连续激活及 `retry_of` 链。每一业务轮的任务总数也不超过原单轮尝试上限，防止反复取消未开始任务而无限制造激活。一次成功查询已返回 PENDING 的轮次不能人工重试；轮次耗尽须核查最终事实或取消。若 UNKNOWN 经 ReviewTask 被核实为成功 PENDING，须等待结果消费，再通过[ContinuePoll](18-workflow-wait-recovery.md)显式恢复原等待；不能借同轮 RetryStep 绕过间隔或次数。 ## 2. 恢复前提和事务 @@ -50,7 +50,7 @@ POLL 带最终信号时，恢复创建同契约、同精确关联键、同原截 当前代次仍有 PENDING 信号时，先消费这些事实，恢复拒绝直接丢弃。已有赢家时核对其受理证明与关联后保留历史。最终信号原截止已过时不重新开始等待。 -这些规则仅覆盖 POLL 附属等待，不表示独立 SIGNAL/TIMER 重建已经开放。 +这些规则覆盖 POLL 附属等待；独立 SIGNAL/TIMER 使用[RearmWait](18-workflow-wait-recovery.md)，公共接口仍待交付。 ## 5. 内部调用场景 diff --git a/docs/18-workflow-wait-recovery.md b/docs/18-workflow-wait-recovery.md new file mode 100644 index 0000000..c8b7a89 --- /dev/null +++ b/docs/18-workflow-wait-recovery.md @@ -0,0 +1,55 @@ +# Weave 等待核查与恢复 + +状态：内部服务已实现 `ContinuePoll` 和 `RearmWait`，均使用[人工恢复的权限、版本、证据及原键回执规则](16-workflow-recovery.md)。公共流程 HTTP、独立 SDK、通用可运行场景与生产验收继续按[开发计划](08-development-plan.md)交付；本页不是公共 SDK 调用示例。 + +## 1. 两类操作 + +| 方法 | 适用事实 | 持久化结果 | +| --- | --- | --- | +| ContinuePoll | 当前最后查询任务已确定成功、结果为 PENDING，且对应权威 Inbox 已处理 | 保留当前激活和原业务轮，恢复 WAITING；到原间隔后才允许下一轮 | +| RearmWait | 独立 SIGNAL/TIMER 仍在原截止内，需要重建等待代次 | 关闭旧等待，创建新激活和新等待；保留首次输入、契约、关联键与截止 | + +这两项操作都只针对 NEEDS_REVIEW 实例，不解除实例核查、显式暂停或执行门控。完成必要的其他核查后再调用 Continue；原 paused 为 true 时仍需 RESUME。已有取消或失败停止目标时拒绝恢复等待。 + +共同要求：显式 `workflows:recover`、最新实例 revision、步骤 revision、当前 activation_id、稳定操作键、原因和证据；拒绝未消费或隔离 Inbox，检查历史任务归属及已结束执行。兄弟分支可以继续收集在途结果，但恢复操作不能把其未知事实解释为完成。相同键返回原冻结回执，仍检查当前凭据与队列权限。 + +## 2. 已确认 PENDING 的接续 + +`ContinuePollInput` 在 StepDecision 基础上要求 `pending_confirmed=true`。该声明不能制造查询结果：服务端读取当前最后任务，要求 SUCCEEDED、无未知效果，并找到同版本、同激活的 PROCESSED Inbox。重新核验快照、原事件、任务/步骤/固定队列/尝试归属，结果还必须与当前权威任务结果一致并通过固定 PENDING 映射。 + +典型场景是：请求 UNKNOWN → ReviewTask 核实为 SUCCEEDED / PENDING → 内部结果消费 → ContinuePoll → Continue → 按原间隔开启下一业务轮。ReviewTask 本身不会清除步骤核查；不能通过 RetryStep 重查已经成功返回 PENDING 的同一轮。 + +接续保留： + +- 原激活、已用业务轮数、查询总截止及实例总截止。 +- 原任务、实际尝试、UNKNOWN 证据、原事件和原 Inbox 处置。 +- 下一查询时间为 `min(该权威 PENDING 结果事件时间 + 原 interval_ms, 原查询截止)`。人工核查产生的新结果使用其确认事件时间，不能伪装为更早的外部时间。 +- 不创建任务或消耗新轮次。时钟到期且门控/配额允许后才创建下一轮，该轮获得固定定义的完整单轮尝试策略。 + +已用轮次达到 max_rounds、查询或实例截止已到、原结果不为 PENDING、有未结清执行时拒绝。原请求尝试预算已经耗尽但结果后来被确认为 PENDING，并不消耗额外同轮请求；仍可在剩余业务轮次与查询期限内接续。 + +接续审计保存来源事件/版本、当前激活、轮次、步骤版本、原结清状态、下一查询时间与固定截止。时钟及最终信号运行器接受这条不可变证明，不改写原 Inbox 的 STEP_ALREADY_SETTLED 等历史处置。当前记录与证明不符时停止推进。 + +## 3. POLL 的附属最终信号 + +ContinuePoll 恢复同一个查询激活，因此信号目标代次不变；已拒绝或已消费的旧信号不会重新变为 PENDING。当前等待必须尚未到原截止、没有已选中的最终赢家，且没有当前代次未消费的信号。WAITING 或因先前查询核查而关闭的 CANCELLED 等待可恢复 WAITING，原状态写入管理审计。 + +如果已有最终赢家，不能用“仍在处理”声明覆盖它，应核查冲突事实。真正重查原请求的 RetryStep 则创建新激活并更换信号代次，两条路径见[查询重试约定](17-workflow-poll-recovery.md)。 + +## 4. SIGNAL/TIMER 的新代次 + +`RearmWaitInput` 在 StepDecision 基础上要求 `wait_verified=true`。当前步骤仅允许 WAITING、NEEDS_REVIEW、FAILED 或 CANCELLED；成功步骤不能重新等待。服务端核验完整连续激活历史，最多32个等待激活，包含首次等待；新激活不提供新的时间预算。 + +- 新 sequence = 旧 sequence + 1；首次输入、信号契约、精确关联键和原等待截止不变。 +- 旧 WAITING 激活关闭为 CANCELLED，旧等待也关闭；已有失败/核查激活保留原状态及输出。它们继续由审计和历史关联追踪，不能误解为整个流程取消。 +- 当前代次存在 PENDING 信号时拒绝重建，先处理已有事实。已匹配赢家不能被新代次绕过。 +- 新等待没有赢家；旧代次后续信号被记录为 STALE_ACTIVATION，调用方须读取新代次后发送新信号。 +- TIMER 始终以首次激活时间加固定 delay_ms 计时，重建和人工成功核查都不能提前满足时长或从新创建时间重算。等待或实例截止已到时拒绝重建。 + +新激活、旧等待结清、新等待、检查点、步骤版本、管理审计和原键回执同事务。审计失败全部回滚；本操作不创建 Pulse 任务，不占业务执行许可。恢复后仍等待显式 Continue 解除核查。 + +## 5. 实现与验收边界 + +复用现有激活、等待、查询状态、Inbox 和不可变审计表，没有新增迁移或持久化字段。仍为35个实体、305个带中文注释的直接持久化字段，无外键和级联删除。 + +实现见[等待重建](../internal/storage/mysql/workflow_wait_rearm.go)、[PENDING 接续与审计证明](../internal/storage/mysql/workflow_poll_continue.go)和[恢复服务](../weave/recovery/service.go)。验证范围见[本阶段报告](reports/phase-5-wait-recovery.md)。本机单 MySQL 与多控制进程回归不能替代生产数据库切换、备份恢复、容量及 RPO/RTO 验收。 diff --git a/docs/reports/phase-5-wait-recovery.md b/docs/reports/phase-5-wait-recovery.md new file mode 100644 index 0000000..70ac496 --- /dev/null +++ b/docs/reports/phase-5-wait-recovery.md @@ -0,0 +1,39 @@ +# P5 等待核查与恢复回归 + +日期：2026-09-18。范围为内部 ContinuePoll 和 RearmWait，以及它们与原时钟、最终信号、人工核查和节点接管的衔接。实例公共 HTTP/SDK、完整接入场景与 P6–P7 生产验收继续实施，本报告不作为整个平台交付结论。 + +## 实现与边界 + +[等待恢复约定](../18-workflow-wait-recovery.md)区分两类操作：查询已被核实为成功 PENDING 时，保留当前激活，按权威结果事件时间和原间隔接续；独立 SIGNAL/TIMER 重建等待时，创建新代次并保留首次输入、契约、关联和截止。 + +继续查询需要当前权威任务、已处理 Inbox、原事件和固定结果契约相符，并写入来源事件、版本、轮次、步骤版本和时限的不可变审计证明。运行器核验证明后才能继续，原 Inbox 处置不被改写。等待换代最多32个激活，包含首次等待；TIMER 的自动运行和人工成功核查均使用首次激活时间。所有操作均需明确权限、原因、证据和当前版本；原键返回冻结回执。已到期、停止意图、信号赢家及未消费信号不能通过恢复绕过。 + +新等待、旧等待结清、步骤、检查点、审计及回执同事务；两类恢复均不自动解除实例核查，也不自行取得执行许可。复用已有结构，无新增迁移或持久化字段，保持35个实体、305个直接持久化字段声明、无数据库外键。 + +## 环境与证据 + +本机 darwin/arm64、Go 1.27.1、Docker MySQL 8.4.10。MySQL 均由脚本创建为隔离的一次性测试库。 + +| 检查 | 结果 | +| --- | --- | +| 存储专项 | 等待恢复10组及既有查询恢复、人工核查、时钟和信号回归通过，`-race`，62.023秒 | +| 实际进程专项 | 新增等待恢复与实际 Orbit 进程用例通过，`-race`，64.299秒；其中实际进程48.61秒、等待恢复14.39秒 | +| 基础回归 | `make fmt test vet build`通过，退出0，包含独立SDK和两个业务示例模块 | +| 完整隔离回归 | 整条脚本通过，退出0；存储全量214.576秒，实际流程进程包48.875秒（进程用例47.51秒），Worker SDK 92.595秒，最终SDK进程故障组93.095秒；19个SDK示例及2个独立业务示例全部通过 | +| 模型注释 | Go AST核对305个带db标签的直接字段声明，缺中文注释0 | +| 实际数据库结构 | v22测试库外键0、业务表无COMMENT字段0；迁移工具自身的5个元数据列单独计数，不包含在模型字段数中 | +| 文档静态检查 | 最终957个本地链接、53段JSON通过；`git diff --check`通过 | + +10组专项覆盖：权威 PENDING 来源与原间隔、最终信号优先完成、接续审计失败全部回滚、假 PENDING 和轮次耗尽拒绝、时钟与审计证明不符拒绝、信号换代保留原截止及隔离旧信号、TIMER 使用首次计时、八路同键并发与旧目标拒绝及32代上限、未消费信号不得丢弃、等待重建审计回滚与截止耗尽。 + +实际进程用例确认所有旧控制进程退出后： + +1. UNKNOWN 查询经人工核查确认为成功 PENDING，由两个新节点处理结果；再次全部退出后，显式接续原等待，再由两个全新节点按原间隔只创建下一业务轮。 +2. 独立 SIGNAL 重建为第二代等待，旧截止不变；节点重新启动后接收第二代信号并推进后继任务。 +3. 独立 TIMER 重建为第二代等待，节点重新启动后按第一次的绝对时间到期，后继任务最终完成。 + +该进程用例同时覆盖此前 TASK、POLL 同轮重试、停止结清、早到信号和查询竞争场景，累计22个任务，最终执行许可为0。 + +证据实现见[等待恢复专项](../../internal/storage/mysql/workflow_wait_recovery_integration_test.go)、[实际进程专项](../../internal/storage/mysql/workflow_runtime_process_integration_test.go)及[服务输入校验](../../weave/recovery/service_test.go)。完整复现入口为[隔离 MySQL 脚本](../../scripts/test-mysql.sh)。缺少一次性 DSN 或实际二进制而跳过的用例不计作通过；本机单数据库和多进程验证不替代生产复制、切换、备份恢复、容量或 RPO/RTO 验收。 + +完整回归开始后没有继续修改执行代码，仅补充阶段状态、接口交付顺序及验证记录。回归结束后脚本已清理本轮一次性测试容器。 diff --git a/internal/storage/mysql/workflow_clock.go b/internal/storage/mysql/workflow_clock.go index 513463b..0e6f4fb 100644 --- a/internal/storage/mysql/workflow_clock.go +++ b/internal/storage/mysql/workflow_clock.go @@ -61,10 +61,6 @@ func (r *WorkflowRuntime) advanceClock(ctx context.Context, tx *sql.Tx, g *workf } 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" { node, _ := g.program.Step(row.view.Key) if node.Poll.FinalSignal != nil { @@ -103,7 +99,11 @@ func (r *WorkflowRuntime) advanceClock(ctx context.Context, tx *sql.Tx, g *workf return false, err } node, _ := g.program.Step(row.view.Key) - intended := activated.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond) + origin, err := workflowActivationOrigin(ctx, tx, g, row) + if err != nil { + return false, err + } + intended := origin.started.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") } diff --git a/internal/storage/mysql/workflow_consume.go b/internal/storage/mysql/workflow_consume.go index eff97cc..6de5ce6 100644 --- a/internal/storage/mysql/workflow_consume.go +++ b/internal/storage/mysql/workflow_consume.go @@ -31,6 +31,11 @@ type workflowResult struct { } func readWorkflowResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, now time.Time) (*workflowResult, error) { + return readWorkflowResultState(ctx, tx, g, c, now, "PENDING") +} + +// readWorkflowResultState 复用不可变来源核验；管理恢复只读取已处理快照，不改写原Inbox处置。 +func readWorkflowResultState(ctx context.Context, tx *sql.Tx, g *workflowGraph, c workflowCandidate, now time.Time, required string) (*workflowResult, error) { var state string var due time.Time var raw []byte @@ -42,7 +47,7 @@ func readWorkflowResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, c wor if err != nil { return nil, err } - if state != "PENDING" || due.After(now) { + if state != required || required == "PENDING" && due.After(now) { return nil, nil } if _, err := jsonvalue.Canonical(raw); err != nil { diff --git a/internal/storage/mysql/workflow_poll_clock.go b/internal/storage/mysql/workflow_poll_clock.go index 2388344..89a8494 100644 --- a/internal/storage/mysql/workflow_poll_clock.go +++ b/internal/storage/mysql/workflow_poll_clock.go @@ -159,13 +159,9 @@ func (r *WorkflowRuntime) advancePoll(ctx context.Context, tx *sql.Tx, g *workfl 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 { + if err := verifyWorkflowPollPending(ctx, tx, g, row, p, lastTask, now); 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 diff --git a/internal/storage/mysql/workflow_poll_continue.go b/internal/storage/mysql/workflow_poll_continue.go new file mode 100644 index 0000000..e29b6d2 --- /dev/null +++ b/internal/storage/mysql/workflow_poll_continue.go @@ -0,0 +1,159 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/weave/engine" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" + "git.yuntuo.hk/k2safe/orbit/weave/recovery" +) + +// processedWorkflowPollResult 只接受当前最后任务的已处理权威结果，不把人工声明当作PENDING事实。 +func processedWorkflowPollResult(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, last core.ID, now time.Time) (*workflowResult, error) { + task, err := readTask(ctx, tx, g.view.NamespaceID, last, false) + if err != nil { + return nil, err + } + if task.State != "SUCCEEDED" || task.EffectUncertain { + return nil, fault.Conflict + } + var id core.ID + if err := tx.QueryRowContext(ctx, `SELECT id FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND source_activation_id=? AND source_revision=? AND state='PROCESSED'`, g.view.NamespaceID, g.view.ID, last, row.view.ActivationID, task.Revision).Scan(&id); err != nil { + return nil, err + } + source, err := readWorkflowResultState(ctx, tx, g, workflowCandidate{ns: g.view.NamespaceID, workflow: g.view.ID, id: id}, now, "PROCESSED") + if err != nil { + return nil, err + } + if source == nil || source.task != task.ID || source.activation != *row.view.ActivationID || source.step != row || source.revision != task.Revision || source.snapshot.State != "SUCCEEDED" || source.snapshot.EffectUncertain || !sameWorkflowJSON(source.snapshot.Result, task.Result) { + return nil, errors.New("reviewed poll result proof mismatch") + } + kind, err := g.program.ClassifyPoll(row.view.Key, source.snapshot.Result) + if err != nil || kind != engine.PollPending { + return nil, fault.Conflict + } + return source, nil +} + +func noPendingWorkflowSignals(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, sequence uint32) error { + var pending int + if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND target_step_key=? AND target_activation_sequence=? AND state='PENDING'`, g.view.NamespaceID, g.view.ID, row.view.Key, sequence).Scan(&pending); err != nil { + return err + } + if pending != 0 { + return fault.Conflict + } + return nil +} + +// ContinuePoll 为已核实的PENDING恢复原等待，不重查同轮、不重置尝试，也不自动解除实例核查。 +func (s *WorkflowRecoveryStore) ContinuePoll(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.ContinuePollInput) (recovery.Result, error) { + return s.mutate(ctx, t, id, key, "workflow.poll.continue", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) { + row, err := workflowRecoveryStep(g, in.StepDecision) + if err != nil { + return nil, err + } + node, ok := g.program.Step(row.view.Key) + if !ok || node.Poll == nil { + return nil, fault.Invalid("continue poll requires a POLL step") + } + if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) { + return nil, fault.Conflict + } + if err := workflowStepSettled(ctx, tx, g, row); err != nil { + return nil, err + } + p, last, err := readWorkflowPollBudget(ctx, tx, g, row) + if err != nil { + return nil, err + } + if p.CurrentTaskID != nil || p.UsedRounds >= p.MaxRounds || !now.Before(p.DeadlineAt) { + return nil, fault.Conflict + } + if _, err := readWorkflowPollRoundHistory(ctx, tx, g, row, p, p.UsedRounds); err != nil { + return nil, err + } + source, err := processedWorkflowPollResult(ctx, tx, g, row, last, now) + if err != nil { + return nil, err + } + if !source.occurred.Before(p.DeadlineAt) { + return nil, fault.Conflict + } + next := minimumTime(source.occurred.Add(time.Duration(p.IntervalMS)*time.Millisecond), p.DeadlineAt) + meta := map[string]any{"step_id": row.view.ID, "activation_id": row.view.ActivationID, "previous_state": row.view.State, "previous_output": json.RawMessage(row.output), "previous_resolved_at": p.ResolvedAt, "task_id": last, "source_event_id": source.event, "source_revision": source.revision, "round": p.UsedRounds, "next_poll_at": next, "deadline_at": p.DeadlineAt, "pending_confirmed": true} + if node.Poll.FinalSignal != nil { + w, err := readWorkflowSignalWait(ctx, tx, g, row) + if err != nil { + return nil, err + } + if w.winner != nil || w.state != "WAITING" && w.state != "CANCELLED" || !now.Before(w.deadline) { + return nil, fault.Conflict + } + if err := noPendingWorkflowSignals(ctx, tx, g, row, w.sequence); err != nil { + return nil, err + } + meta["wait_id"], meta["previous_wait_state"] = w.id, w.state + result, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='WAITING',resolved_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state=? AND winner_signal_id IS NULL`, now, t.NamespaceID, w.id, w.state) + if err := exactlyOne(result, err); err != nil { + return nil, err + } + } + result, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET next_poll_at=?,resolved_at=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_task_id IS NULL`, next, now, t.NamespaceID, p.ID, p.Revision) + if err := exactlyOne(result, err); err != nil { + return nil, err + } + if err := saveWorkflowResultStep(ctx, tx, g, row, weave.StepWaiting, nil, now); err != nil { + return nil, err + } + if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, next); err != nil { + return nil, err + } + meta["step_revision"] = row.view.Revision + out.Step = &row.view + return meta, nil + }) +} + +// verifyWorkflowPollPending 保留原Inbox处置；人工接续必须有匹配当前快照的不可变审计证明。 +func verifyWorkflowPollPending(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, p weave.PollState, last core.ID, now time.Time) error { + 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`, g.view.NamespaceID, g.view.ID, last, row.view.ActivationID).Scan(&disposition); err != nil { + return err + } + if disposition == "POLL_PENDING" { + return nil + } + source, err := processedWorkflowPollResult(ctx, tx, g, row, last, now) + if err != nil { + return err + } + var raw []byte + if err := tx.QueryRowContext(ctx, "SELECT `after` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow' AND resource_id=? AND action='workflow.poll.continue' AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.step_id'))=? AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.activation_id'))=? AND JSON_UNQUOTE(JSON_EXTRACT(`after`,'$.task_id'))=? ORDER BY created_at DESC,id DESC LIMIT 1", g.view.NamespaceID, g.view.ID, string(row.view.ID), string(*row.view.ActivationID), string(last)).Scan(&raw); err != nil { + return err + } + var proof struct { + StepID core.ID `json:"step_id"` + ActivationID core.ID `json:"activation_id"` + TaskID core.ID `json:"task_id"` + SourceEventID core.ID `json:"source_event_id"` + SourceRevision uint64 `json:"source_revision"` + StepRevision uint64 `json:"step_revision"` + Round uint32 `json:"round"` + NextPollAt time.Time `json:"next_poll_at"` + DeadlineAt time.Time `json:"deadline_at"` + PendingConfirmed bool `json:"pending_confirmed"` + } + expected := minimumTime(source.occurred.Add(time.Duration(p.IntervalMS)*time.Millisecond), p.DeadlineAt) + if json.Unmarshal(raw, &proof) != nil || proof.StepID != row.view.ID || proof.ActivationID != *row.view.ActivationID || proof.TaskID != last || proof.SourceEventID != source.event || proof.SourceRevision != source.revision || proof.StepRevision != row.view.Revision || proof.Round != p.UsedRounds || !proof.PendingConfirmed || !proof.DeadlineAt.Equal(p.DeadlineAt) || p.NextPollAt == nil || !proof.NextPollAt.Equal(*p.NextPollAt) || !proof.NextPollAt.Equal(expected) { + return errors.New("poll continuation audit proof invalid") + } + return nil +} diff --git a/internal/storage/mysql/workflow_recovery.go b/internal/storage/mysql/workflow_recovery.go index f98d586..c711962 100644 --- a/internal/storage/mysql/workflow_recovery.go +++ b/internal/storage/mysql/workflow_recovery.go @@ -155,6 +155,17 @@ func (s *WorkflowRecoveryStore) ReviewTask(ctx context.Context, t auth.Token, id } func workflowRecoveryStep(g *workflowGraph, in recovery.StepDecision) (*workflowStep, error) { + row, err := workflowRecoveryTarget(g, in) + if err != nil { + return nil, err + } + if row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled { + return nil, fault.Conflict + } + return row, nil +} + +func workflowRecoveryTarget(g *workflowGraph, in recovery.StepDecision) (*workflowStep, error) { row := g.steps[in.StepKey] if row == nil || in.StepKey == workflowRootKey { return nil, fault.NotFound @@ -162,9 +173,6 @@ func workflowRecoveryStep(g *workflowGraph, in recovery.StepDecision) (*workflow if row.view.ActivationID == nil || *row.view.ActivationID != in.ActivationID || row.view.Revision != in.ExpectedStepRevision { return nil, fault.Conflict } - if row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled { - return nil, fault.Conflict - } return row, nil } @@ -217,11 +225,11 @@ func (s *WorkflowRecoveryStore) ResolveStep(ctx context.Context, t auth.Token, i } typ = node.Poll.Task.ResultType case node.Timer != nil: - var activated time.Time - if err := tx.QueryRowContext(ctx, `SELECT created_at FROM weave_activations WHERE namespace_id=? AND id=?`, t.NamespaceID, row.view.ActivationID).Scan(&activated); err != nil { + origin, err := workflowActivationOrigin(ctx, tx, g, row) + if err != nil { return nil, err } - if now.Before(activated.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond)) { + if now.Before(origin.started.Add(time.Duration(node.Timer.DelayMS) * time.Millisecond)) { return nil, fault.Conflict } typ = definition.Type{Kind: "NULL"} diff --git a/internal/storage/mysql/workflow_runtime_process_integration_test.go b/internal/storage/mysql/workflow_runtime_process_integration_test.go index 27e6fb3..0b254b4 100644 --- a/internal/storage/mysql/workflow_runtime_process_integration_test.go +++ b/internal/storage/mysql/workflow_runtime_process_integration_test.go @@ -408,4 +408,89 @@ func TestIntegrationWorkflowRuntimeProcess(t *testing.T) { } t.Log("POLL manual same-round recovery survived all-stop and two fresh controllers; reduced request attempts, original deadline, old activation and subsequent business-round progression preserved") + for _, p := range processes { + stop(p) + } + pending := runtimeStart(t, f, recoveryPollSpec(f.resources.QueueID, true)) + grantWorkflowRecovery(f) + grantWorkflowSignal(f) + ref = retryStart(t, f) + retryFinish(t, f, ref, "UNKNOWN", false) + start("pending-review-a") + start("pending-review-b") + wait(func() bool { return f.count("weave_instances", "id=? AND state='NEEDS_REVIEW'", pending.ID) == 1 }, "query unknown review") + for _, p := range processes { + stop(p) + } + task := f.get(ref.TaskID) + _, err = recovery.New(NewWorkflowRecoveryStore(db)).ReviewTask(ctx, f.token, pending.ID, string(executionID(t)), recovery.ReviewTaskInput{Decision: recoveryDecision(t, f, pending.ID), TaskID: task.ID, ExpectedTaskRevision: task.Revision, ExecutionSettled: true, Outcome: "SUCCEEDED", Result: json.RawMessage(`"PENDING"`)}) + if err != nil { + t.Fatal(err) + } + start("pending-proof-a") + start("pending-proof-b") + wait(func() bool { return f.count("weave_inbox", "workflow_id=? AND state='PENDING'", pending.ID) == 0 }, "reviewed pending evidence consumed") + for _, p := range processes { + stop(p) + } + continueReviewedPoll(t, f, pending.ID) + continueWorkflow(t, f, pending.ID) + start("pending-continue-a") + start("pending-continue-b") + wait(func() bool { + return f.count("weave_poll_states", "workflow_id=? AND used_rounds=2 AND current_task_id IS NOT NULL", pending.ID) == 1 + }, "reviewed pending resumes next round across all-stop") + ref = retryStart(t, f) + finishQueryOnly(t, f, ref, "DONE") + wait(func() bool { return f.count("weave_instances", "id=? AND state='SUCCEEDED'", pending.ID) == 1 }, "continued poll completion") + for _, kind := range []string{"SIGNAL", "TIMER"} { + for _, p := range processes { + stop(p) + } + delay := uint64(30000) + key := "signal" + if kind == "TIMER" { + delay = 3000 + key = "timer" + } + waiting := runtimeStart(t, f, waitingParallelSpec(f.resources.QueueID, kind, delay)) + grantWorkflowRecovery(f) + grantWorkflowSignal(f) + ref = retryStart(t, f) + retryFinish(t, f, ref, "FAILED", false) + start("wait-review-" + kind + "-a") + start("wait-review-" + kind + "-b") + wait(func() bool { return f.count("weave_instances", "id=? AND state='NEEDS_REVIEW'", waiting.ID) == 1 }, "parallel failure leaves waiting sibling") + for _, p := range processes { + stop(p) + } + var due time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, waiting.ID).Scan(&due); err != nil { + t.Fatal(err) + } + rearmWorkflowWait(t, f, waiting.ID, key) + if f.count("weave_waits", "workflow_id=? AND deadline_at=?", waiting.ID, due) != 2 { + t.Fatal("wait deadline reset after all-stop") + } + resolveWorkflowStep(t, f, waiting.ID, "left", "SUCCEEDED", `{"n":9007199254740993}`) + continueWorkflow(t, f, waiting.ID) + start("wait-rearm-" + kind + "-a") + start("wait-rearm-" + kind + "-b") + if kind == "SIGNAL" { + input := signalInput(waiting) + input.TargetActivationSequence = 2 + sendSignal(t, f, waiting, input) + } + wait(func() bool { + return f.count("weave_steps", "workflow_id=? AND `key`='task0' AND state='ACTIVE'", waiting.ID) == 1 + }, "rearmed wait completed and started one successor") + ref = retryStart(t, f) + retryFinish(t, f, ref, "SUCCEEDED", false) + wait(func() bool { return f.count("weave_instances", "id=? AND state='SUCCEEDED'", waiting.ID) == 1 }, "rearmed wait workflow completed") + } + if f.count("pulse_tasks", "1=1") != 22 || f.used() != 0 { + t.Fatal("wait recovery duplicated task or leaked permits") + } + t.Log("reviewed PENDING continuation plus SIGNAL/TIMER new generations survived full controller shutdown and two fresh nodes; original intervals/deadlines and durable proof preserved, exactly 22 tasks overall") + } diff --git a/internal/storage/mysql/workflow_signal_runtime.go b/internal/storage/mysql/workflow_signal_runtime.go index 3d59467..a87062c 100644 --- a/internal/storage/mysql/workflow_signal_runtime.go +++ b/internal/storage/mysql/workflow_signal_runtime.go @@ -98,13 +98,9 @@ func (r *WorkflowRuntime) applyWorkflowSignal(ctx context.Context, tx *sql.Tx, g if row.view.State != weave.StepWaiting || poll.NextPollAt == nil { return false, false, "", errors.New("signal poll continuation missing") } - 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`, g.view.NamespaceID, g.view.ID, last, row.view.ActivationID).Scan(&disposition); err != nil { + if err := verifyWorkflowPollPending(ctx, tx, g, row, poll, last, now); err != nil { return false, false, "", err } - if disposition != "POLL_PENDING" { - return false, false, "", errors.New("signal poll previous round not pending") - } } } v, err := selectedWorkflowSignal(ctx, tx, g, row, w, now) diff --git a/internal/storage/mysql/workflow_wait_rearm.go b/internal/storage/mysql/workflow_wait_rearm.go new file mode 100644 index 0000000..c861bb7 --- /dev/null +++ b/internal/storage/mysql/workflow_wait_rearm.go @@ -0,0 +1,159 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" + "git.yuntuo.hk/k2safe/orbit/weave/recovery" +) + +// RearmWait 仅对独立SIGNAL/TIMER换代，固定首次等待的预算；到期事实不能通过新建等待清除。 +func (s *WorkflowRecoveryStore) RearmWait(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.RearmWaitInput) (recovery.Result, error) { + return s.mutate(ctx, t, id, key, "workflow.wait.rearm", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) { + row, err := workflowRecoveryTarget(g, in.StepDecision) + if err != nil { + return nil, err + } + node, ok := g.program.Step(row.view.Key) + if !ok || node.Signal == nil && node.Timer == nil { + return nil, fault.Invalid("wait rearm requires a SIGNAL or TIMER step") + } + if row.view.State != weave.StepWaiting && row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled { + return nil, fault.Conflict + } + if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) { + return nil, fault.Conflict + } + if err := workflowStepSettled(ctx, tx, g, row); err != nil { + return nil, err + } + origin, err := workflowActivationOrigin(ctx, tx, g, row) + if err != nil { + return nil, err + } + if origin.sequence >= recovery.MaxWaitActivations { + return nil, fault.Conflict + } + if err := verifyWorkflowWaitHistory(ctx, tx, g, row, origin); err != nil { + return nil, err + } + var w workflowSignalWait + if node.Signal != nil { + w, err = readWorkflowSignalWait(ctx, tx, g, row) + if err != nil { + return nil, err + } + if err := noPendingWorkflowSignals(ctx, tx, g, row, w.sequence); err != nil { + return nil, err + } + } else { + var count int + if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, t.NamespaceID, id, row.view.ActivationID).Scan(&count); err != nil { + return nil, err + } + if count != 1 { + return nil, errors.New("timer rearm wait set invalid") + } + var kind string + var name, version *string + var wake *core.ID + if err := tx.QueryRowContext(ctx, `SELECT id,kind,state,deadline_at,signal_name,payload_version,correlation_key,winner_signal_id,wake_task_id FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, t.NamespaceID, id, row.view.ActivationID).Scan(&w.id, &kind, &w.state, &w.deadline, &name, &version, &w.correlation, &w.winner, &wake); err != nil { + return nil, err + } + expectedInput, _ := json.Marshal(map[string]uint64{"delay_ms": node.Timer.DelayMS}) + if kind != "TIMER" || name != nil || version != nil || w.correlation != "" || wake != nil || !sameWorkflowJSON(row.input, expectedInput) || !w.deadline.Equal(minimumTime(origin.started.Add(time.Duration(node.Timer.DelayMS)*time.Millisecond), g.view.DeadlineAt)) { + return nil, errors.New("timer rearm contract mismatch") + } + } + if w.winner != nil || w.state != "WAITING" && w.state != "CANCELLED" || !now.Before(w.deadline) { + return nil, fault.Conflict + } + activation, err := core.NewID() + if err != nil { + return nil, err + } + wait, err := core.NewID() + if err != nil { + return nil, err + } + beforeState := row.view.State + if row.view.State == weave.StepWaiting { + result, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state='CANCELLED',completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state='WAITING'`, now, now, t.NamespaceID, in.ActivationID, id, row.view.ID) + if err := exactlyOne(result, err); err != nil { + return nil, err + } + } + if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='CANCELLED',resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state='WAITING'`, now, now, t.NamespaceID, w.id); err != nil { + return nil, err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,input,created_at,updated_at) VALUES (?,?,?,?,?,'WAITING',?,?,?)`, t.NamespaceID, activation, id, row.view.ID, origin.sequence+1, string(origin.input), now, now); err != nil { + return nil, err + } + var name, version *string + if w.contract != nil { + name = &w.contract.Name + version = &w.contract.PayloadVersion + } + 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 (?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, wait, id, activation, row.view.Kind, []byte(w.correlation), w.deadline, name, version, now, now); err != nil { + return nil, err + } + result, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state='WAITING',current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_activation_id=?`, activation, now, t.NamespaceID, row.view.ID, row.view.Revision, in.ActivationID) + if err := exactlyOne(result, err); err != nil { + return nil, err + } + row.view.State, row.view.ActivationID = weave.StepWaiting, &activation + row.view.Revision++ + row.output = nil + due := w.deadline + if node.Signal != nil { + due = now + } + if err := saveWorkflowClock(ctx, tx, g, row, string(row.view.Kind), row.view.Revision, due); err != nil { + return nil, err + } + if err := emitEvent(ctx, tx, t.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": id, "state": row.view.State, "activation_id": activation}); err != nil { + return nil, err + } + out.Step = &row.view + return map[string]any{"step_id": row.view.ID, "previous_activation_id": in.ActivationID, "activation_id": activation, "sequence": origin.sequence + 1, "previous_state": beforeState, "previous_wait_id": w.id, "previous_wait_state": w.state, "wait_id": wait, "deadline_at": w.deadline, "wait_verified": true}, nil + }) +} + +func verifyWorkflowWaitHistory(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, origin workflowActivationBase) error { + rows, err := tx.QueryContext(ctx, `SELECT id,sequence,state,input,task_id FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? ORDER BY sequence LIMIT 33`, g.view.NamespaceID, g.view.ID, row.view.ID) + if err != nil { + return err + } + defer rows.Close() + var count uint32 + var last core.ID + for rows.Next() { + var id core.ID + var sequence uint32 + var state string + var input []byte + var task *core.ID + if err := rows.Scan(&id, &sequence, &state, &input, &task); err != nil { + return err + } + count++ + if sequence != count || task != nil || !sameWorkflowJSON(input, origin.input) || id != *row.view.ActivationID && state != "NEEDS_REVIEW" && state != "FAILED" && state != "CANCELLED" { + return errors.New("wait activation history invalid") + } + last = id + } + if err := rows.Err(); err != nil { + return err + } + if count != origin.sequence || last != *row.view.ActivationID { + return errors.New("wait activation history incomplete") + } + return nil +} diff --git a/internal/storage/mysql/workflow_wait_recovery_integration_test.go b/internal/storage/mysql/workflow_wait_recovery_integration_test.go new file mode 100644 index 0000000..34dc83a --- /dev/null +++ b/internal/storage/mysql/workflow_wait_recovery_integration_test.go @@ -0,0 +1,305 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "os" + "sync" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/instance" + "git.yuntuo.hk/k2safe/orbit/weave/recovery" +) + +func waitingParallelSpec(queue core.ID, kind string, delay uint64) definition.Specification { + spec := runtimeSpec(queue, 1) + tail := spec.Body.Steps[0] + join := runtimeParallel(spec) + var wait definition.Step + var typ definition.Type + if kind == "SIGNAL" { + wait = signalSpec(queue, false).Body.Steps[0] + wait.Signal.TimeoutMS = delay + typ = spec.InputType + } else { + wait = timerSpec(queue, delay, false).Body.Steps[0] + typ = definition.Type{Kind: "NULL"} + } + join.Parallel.Branches[1].Body = definition.Block{Steps: []definition.Step{wait}, OutputType: typ, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: wait.Key}}} + spec.Body.Steps = []definition.Step{join, tail} + return spec +} +func waitRecoveryFixture(t *testing.T, ctx context.Context, db *sql.DB, kind string, delay uint64) (*executionFixture, instance.View) { + t.Helper() + f := newExecutionFixture(t, ctx, db) + v := runtimeStart(t, f, waitingParallelSpec(f.resources.QueueID, kind, delay)) + grantWorkflowRecovery(f) + grantWorkflowSignal(f) + ref := retryStart(t, f) + retryFinish(t, f, ref, "FAILED", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + return f, v +} +func rearmWorkflowWait(t *testing.T, f *executionFixture, v core.ID, key string) recovery.Result { + t.Helper() + out, err := recovery.New(NewWorkflowRecoveryStore(f.db)).RearmWait(f.ctx, f.token, v, string(executionID(t)), recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v, key), WaitVerified: true}) + if err != nil { + t.Fatal(err) + } + return out +} +func reviewedPendingPoll(t *testing.T, ctx context.Context, db *sql.DB, final bool) (*executionFixture, instance.View, core.ID) { + t.Helper() + f := newClockFixture(t, ctx, db) + v := runtimeStart(t, f, recoveryPollSpec(f.resources.QueueID, final)) + grantWorkflowRecovery(f) + grantWorkflowSignal(f) + ref := retryStart(t, f) + retryFinish(t, f, ref, "UNKNOWN", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + task := f.get(ref.TaskID) + _, err := recovery.New(NewWorkflowRecoveryStore(db)).ReviewTask(ctx, f.token, v.ID, string(executionID(t)), recovery.ReviewTaskInput{Decision: recoveryDecision(t, f, v.ID), TaskID: task.ID, ExpectedTaskRevision: task.Revision, ExecutionSettled: true, Outcome: "SUCCEEDED", Result: json.RawMessage(`"PENDING"`)}) + if err != nil { + t.Fatal(err) + } + runtimeConsume(t, f, newRuntime(t, db), task.ID) + return f, v, task.ID +} +func continueReviewedPoll(t *testing.T, f *executionFixture, v core.ID) recovery.Result { + t.Helper() + out, err := recovery.New(NewWorkflowRecoveryStore(f.db)).ContinuePoll(f.ctx, f.token, v, string(executionID(t)), recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v, "query"), PendingConfirmed: true}) + if err != nil { + t.Fatal(err) + } + return out +} +func TestIntegrationWorkflowWaitRecovery(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) + } + api := recovery.New(NewWorkflowRecoveryStore(db)) + t.Run("reviewed_pending_keeps_proof_and_interval_then_new_business_round", func(t *testing.T) { + f, v, task := reviewedPendingPoll(t, ctx, db, false) + before := recoveryStep(t, f, v.ID, "query") + out := continueReviewedPoll(t, f, v.ID) + if out.Instance.State != "NEEDS_REVIEW" || out.Step.State != "WAITING" || *out.Step.ActivationID != before.ActivationID || f.count("pulse_tasks", "1=1") != 1 || f.count("weave_inbox", "source_task_id=? AND error_code='STEP_ALREADY_SETTLED'", task) != 1 { + t.Fatal("continuation restarted request or changed old source") + } + var next, occurred time.Time + if err := db.QueryRowContext(ctx, `SELECT p.next_poll_at,e.occurred_at FROM weave_poll_states p JOIN pulse_tasks t ON t.namespace_id=p.namespace_id AND t.id=? JOIN weave_inbox i ON i.namespace_id=t.namespace_id AND i.source_task_id=t.id AND i.source_revision=t.revision JOIN orbit_events e ON e.namespace_id=i.namespace_id AND e.id=i.event_id WHERE p.namespace_id=? AND p.workflow_id=?`, task, f.resources.NamespaceID, v.ID).Scan(&next, &occurred); err != nil || !next.Equal(occurred.Add(time.Second)) { + t.Fatal("original interval lost", err) + } + c := clockCandidate(t, f, v.ID, "query") + runClock(t, f, c) + if f.count("pulse_tasks", "1=1") != 1 { + t.Fatal("review gate or interval bypassed") + } + continueWorkflow(t, f, v.ID) + c = clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + runClock(t, f, c) + ref := retryStart(t, f) + if f.get(ref.TaskID).Retry.MaxAttempts != 3 { + t.Fatal("next round inherited old request budget") + } + finishPoll(t, f, ref, "DONE") + runtimeView(t, f, v.ID, "SUCCEEDED") + if f.count("pulse_tasks", "1=1") != 2 || f.count("pulse_attempts", "state='UNKNOWN'") != 1 || f.used() != 0 { + t.Fatal("request duplicated or history lost") + } + }) + t.Run("reviewed_pending_final_signal_still_wins_without_new_round", func(t *testing.T) { + f, v, _ := reviewedPendingPoll(t, ctx, db, true) + continueReviewedPoll(t, f, v.ID) + continueWorkflow(t, f, v.ID) + sendSignal(t, f, v, pollSignalInput(v)) + runClock(t, f, clockCandidate(t, f, v.ID, "query")) + runtimeView(t, f, v.ID, "SUCCEEDED") + if f.count("pulse_tasks", "1=1") != 1 || f.count("weave_waits", "state='SATISFIED' AND winner_signal_id IS NOT NULL") != 1 { + t.Fatal("continued final wait lost signal") + } + }) + t.Run("pending_continuation_audit_failure_is_atomic", func(t *testing.T) { + f, v, _ := reviewedPendingPoll(t, ctx, db, false) + in := recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v.ID, "query"), PendingConfirmed: true} + key := string(executionID(t)) + trigger := "reject_poll_continue_" + string(executionID(t)) + f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled continuation audit failure'") + _, err := api.ContinuePoll(ctx, f.token, v.ID, key, in) + f.exec("DROP TRIGGER " + trigger) + if err == nil || f.count("weave_poll_states", "resolved_at IS NOT NULL AND next_poll_at IS NULL") != 1 || f.count("weave_pending_advances", "1=1") != 0 || recoveryDecision(t, f, v.ID).ExpectedRevision != in.ExpectedRevision { + t.Fatal("partial continuation committed", err) + } + first, err := api.ContinuePoll(ctx, f.token, v.ID, key, in) + if err != nil { + t.Fatal(err) + } + again, err := api.ContinuePoll(ctx, f.token, v.ID, key, in) + if err != nil || !again.Deduplicated || again.AuditID != first.AuditID { + t.Fatal("continuation receipt changed", err) + } + }) + t.Run("continuation_rejects_forged_pending_and_exhausted_rounds", func(t *testing.T) { + f, v, _ := failedPoll(t, ctx, db, false) + if _, err := api.ContinuePoll(ctx, f.token, v.ID, string(executionID(t)), recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v.ID, "query"), PendingConfirmed: true}); fault.Code(err) != "STATE_CONFLICT" { + t.Fatal("failure treated as pending", err) + } + f = newClockFixture(t, ctx, db) + spec := recoveryPollSpec(f.resources.QueueID, false) + spec.Body.Steps[0].Poll.MaxRounds = 1 + v = runtimeStart(t, f, spec) + grantWorkflowRecovery(f) + ref := retryStart(t, f) + finishPoll(t, f, ref, "PENDING") + if _, err := api.ContinuePoll(ctx, f.token, v.ID, string(executionID(t)), recovery.ContinuePollInput{StepDecision: recoveryStep(t, f, v.ID, "query"), PendingConfirmed: true}); fault.Code(err) != "STATE_CONFLICT" || f.count("pulse_tasks", "1=1") != 1 { + t.Fatal("round limit reset", err) + } + }) + t.Run("continuation_clock_rejects_changed_due_without_matching_audit", func(t *testing.T) { + f, v, _ := reviewedPendingPoll(t, ctx, db, false) + continueReviewedPoll(t, f, v.ID) + continueWorkflow(t, f, v.ID) + c := clockCandidate(t, f, v.ID, "query") + waitClockDue(t, f, c) + f.exec(`UPDATE weave_poll_states SET next_poll_at=DATE_SUB(next_poll_at,INTERVAL 1 MICROSECOND) WHERE namespace_id=?`, f.resources.NamespaceID) + _, err := newRuntime(t, db).process(ctx, c, false) + f.exec(`UPDATE weave_poll_states SET next_poll_at=DATE_ADD(next_poll_at,INTERVAL 1 MICROSECOND) WHERE namespace_id=?`, f.resources.NamespaceID) + if err == nil || f.count("pulse_tasks", "1=1") != 1 { + t.Fatal("changed checkpoint bypassed audit proof") + } + }) + t.Run("signal_rearm_changes_generation_without_changing_deadline", func(t *testing.T) { + f, v := waitRecoveryFixture(t, ctx, db, "SIGNAL", 30000) + before := recoveryStep(t, f, v.ID, "signal") + var deadline time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&deadline); err != nil { + t.Fatal(err) + } + out := rearmWorkflowWait(t, f, v.ID, "signal") + if *out.Step.ActivationID == before.ActivationID || f.count("weave_waits", "deadline_at=?", deadline) != 2 || f.count("weave_activations", "id=? AND state='CANCELLED'", before.ActivationID) != 1 { + t.Fatal("wait history or original deadline lost") + } + old := sendSignal(t, f, v, signalInput(v)) + if old.Signal.Disposition != "STALE_ACTIVATION" { + t.Fatal("old wait signal accepted") + } + resolveWorkflowStep(t, f, v.ID, "left", "SUCCEEDED", `{"n":9007199254740993}`) + continueWorkflow(t, f, v.ID) + in := signalInput(v) + in.TargetActivationSequence = 2 + sendSignal(t, f, v, in) + runClock(t, f, clockCandidate(t, f, v.ID, "signal")) + 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_rearm_uses_first_activation_elapsed_time", func(t *testing.T) { + f, v := waitRecoveryFixture(t, ctx, db, "TIMER", 2000) + var original time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&original); err != nil { + t.Fatal(err) + } + time.Sleep(300 * time.Millisecond) + out := rearmWorkflowWait(t, f, v.ID, "timer") + if f.count("weave_waits", "deadline_at=?", original) != 2 { + t.Fatal("timer rearm reset original delay") + } + resolveWorkflowStep(t, f, v.ID, "left", "SUCCEEDED", `{"n":9007199254740993}`) + continueWorkflow(t, f, v.ID) + c := clockCandidate(t, f, v.ID, "timer") + waitClockDue(t, f, c) + runClock(t, f, c) + if f.count("weave_activations", "id=? AND state='SUCCEEDED'", out.Step.ActivationID) != 1 { + t.Fatal("timer expected new activation delay") + } + }) + t.Run("wait_rearm_bound_same_key_race_and_stale_target", func(t *testing.T) { + f, v := waitRecoveryFixture(t, ctx, db, "SIGNAL", 60000) + in := recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "signal"), WaitVerified: true} + key := string(executionID(t)) + var wg sync.WaitGroup + errs := make(chan error, 8) + results := make(chan recovery.Result, 8) + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + out, err := api.RearmWait(ctx, f.token, v.ID, key, in) + errs <- err + results <- out + }() + } + wg.Wait() + close(errs) + close(results) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + fresh := 0 + for out := range results { + if !out.Deduplicated { + fresh++ + } + } + if fresh != 1 { + t.Fatal("duplicate wait generations") + } + stale := in + stale.ExpectedRevision = recoveryDecision(t, f, v.ID).ExpectedRevision + if _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), stale); fault.Code(err) != "STATE_CONFLICT" { + t.Fatal("old activation rearmed latest wait", err) + } + for sequence := uint32(2); sequence < recovery.MaxWaitActivations; sequence++ { + rearmWorkflowWait(t, f, v.ID, "signal") + } + if _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "signal"), WaitVerified: true}); fault.Code(err) != "STATE_CONFLICT" || f.count("weave_waits", "1=1") != int(recovery.MaxWaitActivations) { + t.Fatal("wait rearm unbounded", err) + } + }) + t.Run("pending_signal_cannot_be_discarded_by_rearm", func(t *testing.T) { + f, v := waitRecoveryFixture(t, ctx, db, "SIGNAL", 30000) + sendSignal(t, f, v, signalInput(v)) + _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "signal"), WaitVerified: true}) + if fault.Code(err) != "STATE_CONFLICT" || f.count("weave_signals", "state='PENDING'") != 1 || f.count("weave_waits", "1=1") != 1 { + t.Fatal("pending evidence discarded", err) + } + }) + t.Run("wait_rearm_audit_failure_and_expired_deadline", func(t *testing.T) { + f, v := waitRecoveryFixture(t, ctx, db, "TIMER", 1500) + in := recovery.RearmWaitInput{StepDecision: recoveryStep(t, f, v.ID, "timer"), WaitVerified: true} + trigger := "reject_wait_rearm_" + string(executionID(t)) + f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled rearm audit failure'") + _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), in) + f.exec("DROP TRIGGER " + trigger) + if err == nil || f.count("weave_waits", "state='WAITING'") != 1 || f.count("weave_activations", "id=? AND state='WAITING'", in.ActivationID) != 1 || recoveryDecision(t, f, v.ID).ExpectedRevision != in.ExpectedRevision { + t.Fatal("partial wait rearm committed", err) + } + var due time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID).Scan(&due); err != nil { + t.Fatal(err) + } + time.Sleep(time.Until(due.Add(20 * time.Millisecond))) + if _, err := api.RearmWait(ctx, f.token, v.ID, string(executionID(t)), in); fault.Code(err) != "STATE_CONFLICT" || f.count("weave_waits", "1=1") != 1 { + t.Fatal("expired timer was restarted", err) + } + }) +} diff --git a/weave/recovery/service.go b/weave/recovery/service.go index fb06972..be4e563 100644 --- a/weave/recovery/service.go +++ b/weave/recovery/service.go @@ -19,6 +19,9 @@ import ( const Action = "workflows:recover" +// MaxWaitActivations 包含首次等待，约束人工换代次数，防止无执行尝试的无限重建。 +const MaxWaitActivations uint32 = 32 + // Decision 核查基于整个实例的最新证据版本；后台新事实会使旧决定失效，原键仍可恢复回执。 type Decision struct { ExpectedRevision uint64 `json:"expected_revision,string"` // ExpectedRevision 当前实例版本，不能使用control_revision代替。 @@ -59,6 +62,16 @@ type RetryStepInput struct { SafeToRetry bool `json:"safe_to_retry" required:"true"` // SafeToRetry 确认旧执行已结束，并按原业务身份再次执行安全；不能只依据未收到响应。 } +type ContinuePollInput struct { + StepDecision + PendingConfirmed bool `json:"pending_confirmed" required:"true"` // PendingConfirmed 确认权威查询结果为处理中，按原间隔与剩余业务轮次接续。 +} + +type RearmWaitInput struct { + StepDecision + WaitVerified bool `json:"wait_verified" required:"true"` // WaitVerified 已核验当前等待及旧回执，按原契约/截止开启新等待代次。 +} + type Result struct { Instance instance.View `json:"instance"` // Instance 操作提交时的冻结实例回执；核查步骤不自动解除实例核查。 AuditID core.ID `json:"audit_id"` // AuditID 本次流程管理操作的不可变审计标识。 @@ -69,6 +82,8 @@ type Result struct { } type Store interface { + ContinuePoll(context.Context, auth.Token, core.ID, string, ContinuePollInput) (Result, error) + RearmWait(context.Context, auth.Token, core.ID, string, RearmWaitInput) (Result, error) ReviewTask(context.Context, auth.Token, core.ID, string, ReviewTaskInput) (Result, error) ResolveStep(context.Context, auth.Token, core.ID, string, ResolveStepInput) (Result, error) Continue(context.Context, auth.Token, core.ID, string, ContinueInput) (Result, error) @@ -160,3 +175,28 @@ func (s *Service) RetryStep(ctx context.Context, t auth.Token, id core.ID, key s } return s.store.RetryStep(ctx, t, id, key, in) } + +func (s *Service) ContinuePoll(ctx context.Context, t auth.Token, id core.ID, key string, in ContinuePollInput) (Result, error) { + if err := validateDecision(id, key, in.Decision); err != nil { + return Result{}, err + } + if err := validateStep(in.StepDecision); err != nil { + return Result{}, err + } + if !in.PendingConfirmed { + return Result{}, fault.Invalid("pending_confirmed=true required") + } + return s.store.ContinuePoll(ctx, t, id, key, in) +} +func (s *Service) RearmWait(ctx context.Context, t auth.Token, id core.ID, key string, in RearmWaitInput) (Result, error) { + if err := validateDecision(id, key, in.Decision); err != nil { + return Result{}, err + } + if err := validateStep(in.StepDecision); err != nil { + return Result{}, err + } + if !in.WaitVerified { + return Result{}, fault.Invalid("wait_verified=true required") + } + return s.store.RearmWait(ctx, t, id, key, in) +} diff --git a/weave/recovery/service_test.go b/weave/recovery/service_test.go index ab27b00..2f883ab 100644 --- a/weave/recovery/service_test.go +++ b/weave/recovery/service_test.go @@ -28,6 +28,14 @@ func (s *captureStore) RetryStep(context.Context, auth.Token, core.ID, string, R s.calls++ return Result{}, nil } +func (s *captureStore) ContinuePoll(context.Context, auth.Token, core.ID, string, ContinuePollInput) (Result, error) { + s.calls++ + return Result{}, nil +} +func (s *captureStore) RearmWait(context.Context, auth.Token, core.ID, string, RearmWaitInput) (Result, error) { + s.calls++ + return Result{}, nil +} func TestRecoveryValidation(t *testing.T) { id := core.ID(strings.Repeat("a", 32)) d := Decision{ExpectedRevision: 2, Reason: "核实外部结果", EvidenceRef: "evidence://case/1"} @@ -71,7 +79,19 @@ func TestRecoveryValidation(t *testing.T) { if _, err := s.RetryStep(ctx, token, id, "operation", RetryStepInput{StepDecision: step}); err == nil { t.Fatal("unsafe retry accepted") } - if store.calls != 4 { + if _, err := s.ContinuePoll(ctx, token, id, "operation", ContinuePollInput{StepDecision: step}); err == nil { + t.Fatal("unconfirmed pending accepted") + } + if _, err := s.RearmWait(ctx, token, id, "operation", RearmWaitInput{StepDecision: step}); err == nil { + t.Fatal("unverified wait accepted") + } + if _, err := s.ContinuePoll(ctx, token, id, "operation", ContinuePollInput{StepDecision: step, PendingConfirmed: true}); err != nil { + t.Fatal(err) + } + if _, err := s.RearmWait(ctx, token, id, "operation", RearmWaitInput{StepDecision: step, WaitVerified: true}); err != nil { + t.Fatal(err) + } + if store.calls != 6 { t.Fatal("invalid inputs reached storage", store.calls) } }