diff --git a/AGENTS.md b/AGENTS.md index 410ff45..2c76868 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,10 +31,10 @@ - P4定义发布使用schema_version=1有限结构，词法作用域禁止跨分支/前向读取，任务策略显式冻结；SDK类型独立。发布按认证→应用额度→操作身份→定义→排序队列持锁；原键与同版本去重仍校验当前权限。定义/版本/revision/审计/回执同事务；同版本不同内容不得覆盖。保留精确数字，校验存储原文与结构摘要；不得把发布通过或DSL支持误称为实例执行已实现。 -- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归（docs/reports/phase-4-definitions.md）。实例启动/固定版本/首批任务同事务的内部存储已补齐；内部事件消费、顺序与组合推进及根接续已接入；TIMER与后续POLL时钟已接入，SIGNAL及最终查询竞争已接入，内部控制与停止结清已接入，等待恢复已接入，实例公共API/SDK已接入，流程任务来源已接入，已接入有界历史查询，实例分页已接入，已接入状态事件时间线，独立轮询/恢复示例已接入，继续W06/W09/W14/W15精确回归；P4整体和P5–P7仍未完成。 +- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归（docs/reports/phase-4-definitions.md）。实例启动/固定版本/首批任务同事务的内部存储已补齐；内部事件消费、顺序与组合推进及根接续已接入；TIMER与后续POLL时钟已接入，SIGNAL及最终查询竞争已接入，内部控制与停止结清已接入，等待恢复已接入，实例公共API/SDK已接入，流程任务来源已接入，已接入有界历史查询，实例分页已接入，已接入状态事件时间线，独立轮询/恢复示例已接入，W06/W09/W14/W15已补实际事务竞争和公共SDK全停回归；P4整体和P5–P7仍未完成。 - 内部结果交付使用workflow_activation所有者并固定激活ID，TASK经activation.task_id、POLL经轮次校验归属；weave_inbox来源及独立快照与Pulse结果同事务，先于外部结果流行锁，不在结果路径反锁实例/步骤/订阅。迁移000017隔离缺快照历史，禁止用当前任务重建旧事件。 -- weave/engine仅计算有限决策；实例受理、Inbox消费与根接续已接入，等待和停止结清运行器已接入，内部核查/TASK恢复已接入，等待恢复已接入内部服务，公共API/SDK已接入，流程任务来源已接入，有界历史已接入，实例分页已接入，已接入状态事件时间线，独立轮询/恢复示例已接入，继续W06/W09/W14/W15精确回归。Plan最多32个变更，More必须持久化待续；耗尽变更预算仍须遍历其他已活动分支，不能误判在途工作不可达。冻结分支选择，精确数值，不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务，不能把纯快照往返测试当作真实进程恢复。 +- weave/engine仅计算有限决策；实例受理、Inbox消费与根接续已接入，等待和停止结清运行器已接入，内部核查/TASK恢复已接入，等待恢复已接入内部服务，公共API/SDK已接入，流程任务来源已接入，有界历史已接入，实例分页已接入，已接入状态事件时间线，独立轮询/恢复示例已接入，W06/W09/W14/W15已补实际事务竞争和公共SDK全停回归。Plan最多32个变更，More必须持久化待续；耗尽变更预算仍须遍历其他已活动分支，不能误判在途工作不可达。冻结分支选择，精确数值，不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务，不能把纯快照往返测试当作真实进程恢复。 - 实例内部受理已实现，信号内部运行器已接入，公共实例/信号HTTP及独立SDK已接入；结果消费和根待续扫描已接入后台。Start按认证→应用额度→原身份→排序队列建立门控/固定版本/全部步骤/首批工作，所有失败同事务回滚。首批推进递增revision至2，原键返回冻结回执且仍核验当前权限，不重算截止。 - 实例根步骤`$root`/JOIN不属于用户定义，Steps查询排除它；More对应根步骤版本化持久检查点。TASK直接关联激活task_id，POLL通过轮次关联，实际任务截止必须受实例/查询绝对截止限制。Wait名称及契约版本冻结，旧NULL等待须核实，不能盲目解释；新关联512字节、存储兼容历史764字节。 @@ -72,4 +72,8 @@ - 状态时间线只读取已存workflow/workflow_step/workflow_signal事件，排除$root；两条路由均显式workflows:audit:read并核对固定版本全部队列及实际子资源。000025派生只读索引不授予权限、不改历史原文、不补造旧信号转换，无外键；新信号状态和事件同事务，批量拒绝32条一批且失败整笔回滚。时间/ID排序与资源revision均不是提交水位；Pulse可靠消费仍走持久订阅。旧写节点停止后迁移，不能承诺混合版本无历史缺口。 -- 独立workflow-poll示例验证2轮3次请求与2轮耗尽核查；workflow-recovery实际提交后断线，只提交一次，按证据依次ReviewTask/ResolveStep/Continue，再显式RESUME，旧UNKNOWN和原截止保持。受控内存渠道不是持久生产证据。下一批依docs/reports/phase-4-5.md补W06实际版本后继、W09三方竞争、W14暂停竞争和W15暂停全停；P4/P5仍不整体宣称完成，P6/P7继续保留。 +- 独立workflow-poll示例验证2轮3次请求与2轮耗尽核查；workflow-recovery实际提交后断线，只提交一次，按证据依次ReviewTask/ResolveStep/Continue，再显式RESUME，旧UNKNOWN和原截止保持。受控内存渠道不是持久生产证据。W06实际版本后继、W09三方竞争、W14暂停竞争和W15暂停全停已补，见docs/reports/phase-4-5-races.md；P4/P5仍不整体宣称完成，P6/P7继续保留。 + +- 竞争回归通过performance_schema确认参与者实际进入数据库锁等待，禁止将请求已发出或sleep当作并发证据。W09以原数据库截止自然到达放行；PAUSE以后不得新建任务/尝试或授权，已授权结果仍需收集。公共SDK暂停全停场景须在全部旧节点Wait确认退出后才启动新节点；保留原回执、等待身份、版本、查询轮次与截止。普通SIGNAL竞争不能替代最终查询/回执/下一轮的独立故障边界；COMMIT不明、A04、生产HA与容量等继续按P6/P7验收。 + +- 版本/暂停竞争本批已通过21个真实存储子场景、公共SDK两旧节点全停及两新节点恢复、23个SDK示例/2个业务模块的标准全量回归。证据见docs/reports/phase-4-5-races.md；下一批按docs/08-development-plan.md的P6首批实施顺序先补A04/W01/W13/W16精确故障，不重复将W06/W09/W14/W15列为未做。全平台目标继续保持，不能以本批通过替代观测、凭据轮换、归档/备份、生产拓扑和发布交付。 diff --git a/README.md b/README.md index c30acc8..4413202 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， 已完成 P0–P3 平台功能阶段验收：36 个数据模型、版本化迁移、凭据及范围校验、队列与任务 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) 继续实现。 -Weave已提供[定义发布](docs/api/workflow-definitions.md)、[实例启动、步骤详情、暂停/继续/取消、信号和操作审计](docs/api/workflows.md)，以及[六类人工核查恢复](docs/api/workflow-recovery.md)的HTTP与独立Go SDK。持久运行器支持顺序、分支、有限并行、TIMER、SIGNAL及有界POLL，恢复沿用原预算与截止。[并行作业与回执示例](docs/examples/workflow.md)提供完整调用链；[历史激活/轮次/信号及信号审计](docs/api/workflow-history.md)已接入；[实例筛选分页](docs/api/workflow-queries.md)已接入，[状态事件时间线](docs/api/workflow-events.md)已接入，独立[轮询](docs/examples/workflow-poll.md)与[恢复](docs/examples/workflow-recovery.md)示例已接入，剩余故障见[P4/P5核对](docs/reports/phase-4-5.md)；[任务流程来源](docs/api/task-source.md#5-流程固定来源)已接入，P4/P5与生产验收尚未整体完成。 +Weave已提供[定义发布](docs/api/workflow-definitions.md)、[实例启动、步骤详情、暂停/继续/取消、信号和操作审计](docs/api/workflows.md)，以及[六类人工核查恢复](docs/api/workflow-recovery.md)的HTTP与独立Go SDK。持久运行器支持顺序、分支、有限并行、TIMER、SIGNAL及有界POLL，恢复沿用原预算与截止。[并行作业与回执示例](docs/examples/workflow.md)提供完整调用链；[历史激活/轮次/信号及信号审计](docs/api/workflow-history.md)已接入；[实例筛选分页](docs/api/workflow-queries.md)已接入，[状态事件时间线](docs/api/workflow-events.md)已接入，独立[轮询](docs/examples/workflow-poll.md)与[恢复](docs/examples/workflow-recovery.md)示例已接入，已补[版本/暂停竞争/全停回归](docs/reports/phase-4-5-races.md)，剩余故障见[P4/P5核对](docs/reports/phase-4-5.md)；[任务流程来源](docs/api/task-source.md#5-流程固定来源)已接入，P4/P5与生产验收尚未整体完成。 数据库禁止外键，关联一致性通过服务端事务校验，索引、唯一约束和 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/05-production-acceptance.md b/docs/05-production-acceptance.md index 7a0396e..18cbea4 100644 --- a/docs/05-production-acceptance.md +++ b/docs/05-production-acceptance.md @@ -1,6 +1,6 @@ # Orbit 生产验收与发布门槛 -> 版本 0.2 · 2026-09-17。本文是全平台生产发布门槛。P0–P2 已积累阶段证据，见 [开发计划](08-development-plan.md) 与 [P2 验收](reports/phase-2.md)；P3的逐项功能及故障证据边界见[阶段核对](reports/phase-3.md)，新增[相对延迟与重试全停恢复](reports/phase-3-delay-recovery.md)。未覆盖的场景继续实施，尚无全平台生产验收或容量结论。 +> 版本 0.2 · 2026-09-17。本文是全平台生产发布门槛。P0–P2 已积累阶段证据，见 [开发计划](08-development-plan.md) 与 [P2 验收](reports/phase-2.md)；P3的逐项功能及故障证据边界见[阶段核对](reports/phase-3.md)，新增[相对延迟与重试全停恢复](reports/phase-3-delay-recovery.md)。新增[流程版本、暂停竞争与全停恢复](reports/phase-4-5-races.md)覆盖W06/W09/W14/W15。未覆盖的场景继续实施，尚无全平台生产验收或容量结论。 ## 1. 验收对象与证据 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 37606e9..7ce08c6 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)已接入；等待恢复已接入内部服务；实例公共API/SDK与基础可运行场景已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；[状态事件时间线](api/workflow-events.md)已接入；独立轮询/恢复示例已接入；[精确故障验收](reports/phase-4-5.md)继续开发 | -| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器；最终信号竞争及在途查询结清已接入；内部暂停/继续/取消与失败结清已接入；执行核查、步骤结论、解除核查及TASK新激活已接入；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；等待恢复已接入内部服务；公共API/SDK已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；[状态事件时间线](api/workflow-events.md)已接入；独立轮询/恢复示例已接入；[精确故障验收](reports/phase-4-5.md)继续开发 | +| 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与基础可运行场景已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；[状态事件时间线](api/workflow-events.md)已接入；独立轮询/恢复示例已接入；[版本/暂停竞争/全停回归](reports/phase-4-5-races.md)已补；[剩余故障验收](reports/phase-4-5.md)继续推进 | +| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器；最终信号竞争及在途查询结清已接入；内部暂停/继续/取消与失败结清已接入；执行核查、步骤结论、解除核查及TASK新激活已接入；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；等待恢复已接入内部服务；公共API/SDK已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；[状态事件时间线](api/workflow-events.md)已接入；独立轮询/恢复示例已接入；[版本/暂停竞争/全停回归](reports/phase-4-5-races.md)已补；[剩余故障验收](reports/phase-4-5.md)继续推进 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | | P7 首版交付核对 | SDK 兼容与发布准备、完整 API 手册、场景示例、全量回归及交付清单 | 新接入者按文档跑通全部示例；所有首版能力可追溯到实现与测试；报告实际生产边界 | 待开发 | @@ -42,7 +42,7 @@ 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接续和独立等待重建](18-workflow-wait-recovery.md)也已接入内部服务，保留原预算、轮次、结果和证据。 -4. 实例/当前步骤/信号/控制/六类恢复及流程操作审计HTTP和独立SDK已接入，步骤查询返回实际激活代次；并行执行与信号回执场景已有可运行示例。激活/轮次/信号历史及信号资源审计已接入；实例筛选分页已接入；状态事件时间线已接入；独立轮询及恢复示例已接入，继续补W06/W09/W14/W15精确回归；[Pulse任务流程来源](api/task-source.md#5-流程固定来源)已交付。逐项验证原键恢复、权限、版本冲突、竞争与进程全停。 +4. 实例/当前步骤/信号/控制/六类恢复及流程操作审计HTTP和独立SDK已接入，步骤查询返回实际激活代次；并行执行与信号回执场景已有可运行示例。激活/轮次/信号历史及信号资源审计已接入；实例筛选分页已接入；状态事件时间线已接入；独立轮询及恢复示例已接入，已补[W06/W09/W14/W15专项回归](reports/phase-4-5-races.md)；[Pulse任务流程来源](api/task-source.md#5-流程固定来源)已交付。逐项验证原键恢复、权限、版本冲突、竞争与进程全停。 5. 转入P6/P7的生产故障、容量、运维、发布与全部文档核对。内部存储阶段通过不会提前勾选公共接口或生产验收。 公共接入按以下调用链推进；本批已接入启动/当前详情、控制、信号、六类恢复和流程操作审计，并提供独立SDK及基础示例。任务流程来源和有界历史查询也已接入。实例分页和独立查询示例已接入；状态事件时间线已接入；剩余精确故障场景按[P4/P5核对](reports/phase-4-5.md)继续交付： @@ -56,6 +56,13 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( 接口只使用已经落地的内部事务规则；SDK保持独立module，不能引用服务端模型或数据库实现。生产故障及容量结论仍由P6专门验收。 +### P6 首批实施顺序 + +1. 统一可复现故障工具：补A04“派发已提交但消息尚未送达”、W01/W13事务提交结果不明，以及W16最终查询结果/回执/下一轮竞争。每个切点必须有实际数据库或进程证据，不能用响应错误推断事务是否已提交，也不能用普通SIGNAL竞争替代最终查询场景。 +2. 实现运行观测：受理/授权/完成及流程推进延迟、重试/核查/恢复、待处理与到期积压、连接/发送队列和数据库连接池。明确采集范围与开销；任务、流程、业务键和凭据不得作为无界指标标签。 +3. 补齐凭据签发、轮换、撤销及Worker/订阅安全重绑，保持最小权限、管理审计与未知回执可恢复。 +4. 实现保留/归档与受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 + ### P3 功能范围核对 可靠订阅链路完成后，重新对照 [原功能范围](01-functional-spec.md) 和 [事务衔接设计](04-reliability-and-data.md)，以下原定首版能力已按核对结果补齐；进一步发现的相对延迟与改参数关联示例也已交付。[P3验收](reports/phase-3.md)逐项保留实现证据及尚待P4–P7完成的原范围，不按已有模块反推全平台完成。 diff --git a/docs/reports/phase-4-5-races.md b/docs/reports/phase-4-5-races.md new file mode 100644 index 0000000..db05a70 --- /dev/null +++ b/docs/reports/phase-4-5-races.md @@ -0,0 +1,74 @@ +# P4/P5 版本、暂停竞争与全停恢复回归 + +日期：2026-09-18。对应[原故障矩阵](../05-production-acceptance.md)中的W06、W09、W14和W15。本批新增可复现的数据库竞争和公共SDK进程故障测试，没有修改服务端执行规则、接口、模型或迁移。无外键约束及中文模型注释保持。 + +## 验证方式 + +[真实MySQL测试](../../internal/storage/mysql/workflow_races_integration_test.go)使用一次性数据库，创建独立应用和命名空间。竞争驱动先锁定应用额度行，启动实际服务事务，并通过`performance_schema.data_lock_waits`确认每个参与者都已经进入InnoDB锁等待，再释放锁。测试账户需要读取这些诊断表；生产运行账户无需新增该权限。 + +这种方式证明事务实际重叠，不以goroutine启动、请求发出或固定sleep代替。测试不声称穷尽所有线程调度；两个参与者交换入队次序，三个参与者遍历六种入队次序。合法结果以持久状态、控制回执和数据库时间为准。 + +[公共SDK进程测试](../../sdk/go/workflow_process_integration_test.go)不导入服务端模块或数据库驱动，所有发布、启动、控制、信号、执行和查询走HTTP/WSS。启动两个真实Orbit进程；对两个进程分别发送SIGKILL并Wait确认退出；原TIMER唤醒和POLL下一轮就绪时间均过去后，才启动两个新进程。新节点ID必须彼此不同，也不能复用旧节点ID。 + +## W06：发布新版本后继续执行旧版本 + +v1实例先进入SIGNAL等待，此时没有后继任务。随后给同一定义发布v2，改变任务处理器版本、实际输入和流程总时限。旧实例收到信号后实际领取、开始、完成后继任务，验证仍使用v1处理器、精确数字9007199254740993、原定义版本ID与原截止。 + +再显式启动v2、发送信号并实际完成任务，验证使用v2处理器和新输入7。Worker能力集合属于进程身份，因此为v2能力注册新Worker身份；不能改写旧身份的能力集合。两个实例共两个任务，最终许可释放。验证覆盖后继执行及输出，不只检查启动回执。 + +## W09：信号、自然截止与取消三方竞争 + +SIGNAL等待设置1秒原时限。信号受理、截止消费者和CANCEL三个事务都进入锁等待后，测试查询数据库时间，等其自然到达原截止再放行；不提前修改等待截止。 + +六种入队顺序各有两组：没有及时信号，以及此前已经持久受理一个及时信号。本次竞争的新信号必须在原截止后受理，只能保留为过期或拒绝，不能成为赢家。此前按时受理的信号仍必须被消费并成为唯一等待赢家，不能因为消费延迟或取消结清而丢失。 + +断言包括原截止不变、CANCEL意图保留、无悬挂PENDING信号、无任务执行尝试、无许可遗留、后继至多一个、同键重报返回原信号回执。截止先完成时可能保留NEEDS_REVIEW；取消先完成时可进入CANCELLED。测试保留这两种合法结果，不把取消当作解除已有核查的权限。 + +## W14：暂停与四种实际操作竞争 + +分别让PAUSE与Start、Claim、已确认结果的后继创建、已知失败的安全重试上报竞争，每组交换入队次序。以PAUSE回执的数据库时间核对任务创建、尝试创建和开始授权，禁止暂停后发生这些新工作。 + +- 已领取但未授权的尝试在暂停时Start必须拒绝；解除暂停后使用原尝试，不能再扣一次预算。 +- 已授权的工作保留许可，暂停后仍能上报并收集结果，但不能创建后继；结果结清后释放许可。 +- 重试上报可保留失败事实和原退避，暂停期间即使退避到期也不能领取下一次尝试。 +- 解除暂停后从持久进度完成两个任务，普通场景恰好两个尝试，安全重试场景恰好三个尝试；原流程截止与原任务预算保持。 + +存储测试通过显式调用内部消费者推进。根待续采用已有集成测试辅助函数将扫描就绪时间置为当前时间，业务截止和重试退避不因此改变；完整后台扫描行为由下述真实进程场景验证。 + +## W15：暂停跨全部控制节点退出后恢复 + +并行分支包括一个已开始TASK、一个SIGNAL、一个TIMER和一个两轮POLL，汇合后还有一个TASK。第一轮查询取得PENDING后暂停实例；暂停期间发送信号并完成已授权任务，确认任务结果已提交，然后使两个原节点全部退出。 + +新节点接管后，公共SDK检查： + +- 原任务结果、原信号受理时间/赢家和TIMER到期事实保留；等待ID、激活ID/代次及原截止保持。 +- paused和control_revision保留，汇合后的TASK仍BLOCKED，未创建任务。 +- POLL仍只使用第一轮，保留原下一轮时间、查询截止和2轮/每轮2次预算；停机不补发轮次。 +- 在另一新节点重报原START、PAUSE、SIGNAL，仍返回首次冻结回执。 +- 通过另一新节点RESUME后，第二轮查询DONE，再完成汇合后的TASK。最终按业务键查到恰好四个任务，每个恰好一次尝试；轮次历史恰好两轮，原版本、输出、截止及首任务策略不变。 +- 成功后重报原RESUME仍返回控制版本3，不创建第二次恢复。 + +## 命令及验证记录 + +标准入口： + +```sh +make fmt test vet build +make integration +``` + +`scripts/test-mysql.sh`会创建并清理隔离MySQL；存储组自动执行新的21个子场景。原HTTP服务退出、原Worker/订阅进程测试完成清理后，脚本用独立流程权限凭据单独执行`TestIntegrationWorkflowProcessPauseRecovery`，避免其他控制节点代为推进。 + +环境：macOS arm64、Go 1.27.1、Docker MySQL 8.4.10。初次定向回归发现测试夹具两处调用错误：修改旧Worker身份的能力集合，以及失败上报遗漏必填非null结果；平台按已有契约拒绝，夹具已改为新身份和合法失败结果。修正后存储定向测试退出0（24.062秒），公共SDK全停测试退出0（10.968秒）。基础fmt、race单元、vet、build退出0。 + +随后补强数据库授权时间、精确输入数字、全部任务数和完整轮次历史断言，并纳入标准全量回归。完整`scripts/test-mysql.sh`退出0：存储组249.189秒、原流程进程组48.751秒、Inbox34.678秒、Outbox34.688秒、Worker/订阅进程组95.466秒、新暂停全停组13.797秒；23个SDK示例和2个独立业务模块全部通过。这些数字是测试耗时，不是吞吐或延迟指标。 + +收尾给测试中的“尝试创建或授权时间”SQL条件补上括号，使两个条件都受命名空间限制；该测试修改晚于全量脚本的存储组执行。最终21个存储子场景另在新隔离MySQL、`-race -count=1`下完整重跑，退出0（24.732秒）。服务端、SDK运行代码及迁移均未变化；没有改变生产契约来迎合测试。 + +最终Markdown检查1173个本地链接、68个JSON块，错误0；格式和差异检查通过。全量升级回归继续确认70个历史外键移除、323个原索引组成部分和原任务保留，最终结构外键0。 + +## 阶段边界与后续 + +这四项补强不能代替全平台生产验收。原范围继续保留：W01/W13精确COMMIT结果不明切点、A04派发落库但消息尚未送达、MySQL失联/主从切换、跨故障域容量、热点锁等待、慢消费者及重连风暴、凭据轮换、保留归档、备份恢复和滚动兼容。 + +下一步按[P4/P5原范围核对](phase-4-5.md)汇总阶段证据，再推进P6统一故障注入和运行观测、P7 SDK发布及实际文件转换示例。当前进程强杀验证使用单机数据库，不提供生产HA、RPO/RTO或百万企业容量结论。 diff --git a/docs/reports/phase-4-5.md b/docs/reports/phase-4-5.md index a27ade9..03df483 100644 --- a/docs/reports/phase-4-5.md +++ b/docs/reports/phase-4-5.md @@ -2,6 +2,8 @@ 日期：2026-09-18。本报告按[原流程功能](../06-workflow-spec.md)、[开发计划](../08-development-plan.md)及[生产故障矩阵W01–W17](../05-production-acceptance.md)核对，保留未完成项。功能实现、单个场景通过与整个阶段验收分开记录；P4/P5尚不标为整体完成，P6/P7及持续目标保持进行中。 +新增补强：[版本、暂停竞争与全停恢复回归](phase-4-5-races.md)提供W06/W09/W14/W15的实际事务及公共SDK进程证据。下表随该批更新，原轮询/恢复示例的验证记录保留。 + ## 本批交付 新增两个独立可运行SDK示例，均只导入公开SDK及示例内部辅助包，不访问Orbit数据库或服务端模型： @@ -19,11 +21,11 @@ | --- | --- | --- | | 研发定义发布、类型和资源约束 | [定义API](../api/workflow-definitions.md)、[结构格式](../api/workflow-definition-format.md)、workflow-definition例子 | [定义存储回归](../../internal/storage/mysql/workflow_definitions_integration_test.go)：并发同键/同版本、精确数字、非法/损坏定义、权限撤销、事务失败；[SDK](../../sdk/go/workflow_definitions_integration_test.go)提交后丢回执原键恢复 | | 顺序、条件分支、有限并行汇合 | weave/engine与持久运行器；[workflow例子](../examples/workflow.md)实际并行后汇合 | [决策器测试](../../weave/engine/plan_test.go)精确条件、分支选择冻结与预算；[实例](../../internal/storage/mysql/workflow_instances_integration_test.go)覆盖六种初始节点；[推进](../../internal/storage/mysql/workflow_runtime_integration_test.go)重复消费、并行结果只建一个后继、满额回滚和根待续 | -| 固定版本及输入 | StartWorkflow显式version；GetWorkflowVersion及实例来源 | 实例并发受理固定版本、发布v2后原键回执保留；[实例查询](../../internal/storage/mysql/workflow_query_integration_test.go)继续按v1全部队列授权。运行中发布v2后继续执行后继步骤的专门场景仍需补强 | +| 固定版本及输入 | StartWorkflow显式version；GetWorkflowVersion及实例来源 | 实例并发受理固定版本、发布v2后原键回执保留；[实例查询](../../internal/storage/mysql/workflow_query_integration_test.go)继续按v1全部队列授权。新增版本回归实际完成v1后继和v2新实例，验证不同处理器/输入、原版本与原截止 | | SIGNAL等待、早到和旧代次隔离 | [信号约定](../14-workflow-signals.md)、SendWorkflowSignal及workflow例子 | [受理](../../internal/storage/mysql/workflow_signals_integration_test.go)身份/契约/范围/事务；[消费](../../internal/storage/mysql/workflow_signal_runtime_integration_test.go)并发赢家、真实到期、及时受理晚消费、篡改及最终查询结清 | | TIMER持久等待、原截止 | [时钟约定](../13-workflow-clock.md)、workflow例子 | [时钟存储](../../internal/storage/mysql/workflow_clock_integration_test.go)自然到期、重复消费、实例截止截断；[真实进程](../../internal/storage/mysql/workflow_runtime_process_integration_test.go)全停、两新节点接续 | | 有界查询与独立尝试预算 | [workflow-poll例子](../examples/workflow-poll.md)、POLL定义与历史轮次API | 时钟存储验证同轮重试、下一轮、满额/暂停不耗轮次、截止与在途结果；新例子分别验证成功和耗尽后的实际请求次数 | -| 暂停/继续、取消、分支失败结清 | [公共控制](../api/workflows.md)、[停止规则](../15-workflow-controls.md) | [控制回归](../../internal/storage/mysql/workflow_controls_integration_test.go)原control_revision、原键/并发/撤权、开始与取消竞争、并行FAIL_AFTER_SETTLED、UNKNOWN不终结、超过32步待续；精确暂停同时竞争仍有下面列出的缺口 | +| 暂停/继续、取消、分支失败结清 | [公共控制](../api/workflows.md)、[停止规则](../15-workflow-controls.md) | [控制回归](../../internal/storage/mysql/workflow_controls_integration_test.go)原control_revision、原键/并发/撤权、开始与取消竞争、并行FAIL_AFTER_SETTLED、UNKNOWN不终结、超过32步待续；新增真实锁等待重叠的PAUSE/开始/领取/后继/重试竞争 | | 人工核查、TASK/POLL重试及等待恢复 | [六类恢复API](../api/workflow-recovery.md)、workflow-recovery例子 | [TASK恢复](../../internal/storage/mysql/workflow_recovery_integration_test.go)、[POLL恢复](../../internal/storage/mysql/workflow_poll_recovery_integration_test.go)、[等待恢复](../../internal/storage/mysql/workflow_wait_recovery_integration_test.go)：旧激活、尝试/轮次/原截止、证据、失败回滚、当前版本和停止意图 | | 完整查询与审计入口 | 当前详情、[历史](../api/workflow-history.md)、[实例筛选](../api/workflow-queries.md)、[状态事件](../api/workflow-events.md)、[任务来源](../api/task-source.md) | 对应存储回归验证当前与历史激活区分、分页、归属损坏、全部固定队列权限；[独立SDK](../../sdk/go/workflow_instances_integration_test.go)覆盖六类恢复和读取，时间线另有SDK与批量事件回滚验证 | | 不依赖内存的接续 | MySQL检查点、内部Inbox、额度、门控与同事务事件 | 真实流程进程测试先确认原进程SIGKILL退出，再启动两个新节点；结果、TIMER/POLL、早到/最终信号、停止意图、TASK/POLL人工恢复及等待换代均保留 | @@ -39,17 +41,17 @@ | W03 | 早到、重复、迟到、错误关联及代次的实际存储与SDK回归 | 仍须纳入容量模型中的大量信号积压 | | W04 | TIMER、早到SIGNAL、独立等待恢复均有全停后接续 | 生产长时间等待、保留与清理策略按P6验收 | | W05 | 并行已知失败结清、未知效果保留核查，成功步骤保持 | 业务补偿不属于已确认首版，不冒充平台回滚外部效果 | -| W06 | 不可变版本、原键固定实例、v1范围授权已验证 | 补“v1等待中发布语义不同v2，再由旧实例继续后继”的完整执行断言 | +| W06 | v1等待时发布不同处理器/输入/时限的v2，实际完成旧实例后继及v2新实例，精确输出、固定版本和原截止保持 | 服务端/SDK滚动兼容仍按P7，不能用定义版本隔离替代 | | W07 | 已授权任务取消后收集SUCCEEDED/FAILED/CANCELLED/UNKNOWN，含进程全停 | 大量在途取消、外部系统执行失联仍需生产负载和接入方证明 | | W08 | 同一实例并行结果竞争，后继与汇合仅一次 | 高负载热点实例锁等待在P6测量 | -| W09 | 及时受理后过截止再消费、等待取消、停止前后信号及并发赢家已验证 | 补信号受理、真实截止唤醒、CANCEL三方同时执行的专项回归；顺序用例不能替代 | +| W09 | 信号受理/原截止消费者/CANCEL三个事务均进入MySQL锁等待后放行，六种顺序分别覆盖有/无及时信号；迟到信号不胜出，及时信号保留 | 热点负载、长时间消费积压及数据库切换仍按P6 | | W10 | 同恢复键竞争、旧激活结果、旧信号代次及恢复后全停 | 滚动版本兼容、长期历史保留另按P6/P7 | | W11 | 同轮重试与PENDING新轮次的存储、SDK及实际HTTP例子 | 渠道峰值与大批量同刻唤醒按P6 | | W12 | 最大轮次、截止、原轮尝试耗尽拒绝绕过；新例子保留NEEDS_REVIEW | 不把本地时间预算测试换算成生产延迟指标 | | W13 | 轮次创建失败回滚、PENDING落库及检查点全停恢复 | 精确COMMIT不确定窗口仍按W01/P6补强 | -| W14 | 已暂停拒绝LEASED开始、开始/取消竞争、暂停期间结果收集及原预算 | 补PAUSE与开始/后继创建/领取/重试的实际并发交错；CANCEL竞争不能代替PAUSE竞争 | -| W15 | 暂停期间任务/信号/到期处理，Continue保留paused，原时限不增加 | 补暂停状态跨全节点退出后解除暂停的公共API完整场景 | -| W16 | 最终信号与查询结果按原受理时间裁决，在途PENDING结清、冲突核查、全停保留 | 三方同时触发及热点负载按W09和P6补强 | +| W14 | PAUSE与开始/后继/领取/安全重试的真实事务竞争，双向入队、数据库授权时间、暂停收集结果、恢复任务/尝试数及原截止断言 | 不声称穷尽全部线程交错；高负载锁等待及故障切点继续P6 | +| W15 | 公共SDK暂停后提交任务结果/信号，两原节点全部SIGKILL并Wait退出；TIMER/下一查询轮就绪时间过去后两新节点接管，显式RESUME完成四任务/两轮查询 | 单机数据库进程故障证据；生产数据库高可用和长停机恢复另按P6 | +| W16 | 最终信号与查询结果按原受理时间裁决，在途PENDING结清、冲突核查、全停保留 | 普通SIGNAL三方竞争见W09；最终查询/回执/下一轮同时到达的故障切点及热点负载继续P6，不能拿W09替代 | | W17 | 同版本同内容/不同内容、非法定义、SDK丢发布回执恢复 | 旧SDK/服务版本滚动兼容矩阵按P7发布候选核对 | ## 本批验证记录 @@ -66,7 +68,7 @@ ## 下一步按顺序完成 -1. 补W06的版本变更后实际后继执行、W09三方竞争、W14暂停竞争与W15暂停全停接续，保留可复现的真实数据库/进程断言。 +1. W06/W09/W14/W15已补真实数据库与公共SDK进程断言，并完成标准全量回归，见[专项回归](phase-4-5-races.md)。原四项缺口已关闭，后续按剩余生产故障推进。 2. 据这些结果再决定P4/P5功能阶段是否可验收，不能把本报告中的部分验证改成全项完成。 3. 进入P6统一观测、凭据轮换、历史保留/归档、故障切点工具、容量负载、备份恢复与滚动升级；继续保留A04精确未派达、MySQL切换、跨故障域等原门槛。 4. P7补实际文件转换等仍缺的原场景、SDK版本发布与新接入者全量核对。百万企业支持能力只能根据实际工作负载和目标拓扑验证，当前不作容量承诺。 diff --git a/internal/storage/mysql/workflow_races_integration_test.go b/internal/storage/mysql/workflow_races_integration_test.go new file mode 100644 index 0000000..89e3ef4 --- /dev/null +++ b/internal/storage/mysql/workflow_races_integration_test.go @@ -0,0 +1,390 @@ +package mysql + +import ( + "context" + "encoding/json" + "fmt" + "os" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" + "git.yuntuo.hk/k2safe/orbit/weave/control" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/instance" + "git.yuntuo.hk/k2safe/orbit/weave/signal" +) + +func workflowRaceNumber(raw json.RawMessage, want string) bool { + var data struct { + N json.Number `json:"n"` + } + return json.Unmarshal(raw, &data) == nil && data.N.String() == want +} + +// 先锁住真实应用额度行，再确认所有竞争事务进入InnoDB锁等待。 +// 不用goroutine启动次数或sleep推断事务已重叠；不修改业务时间和状态。 +func competeWorkflow(t *testing.T, f *executionFixture, beforeRelease func(), operations ...func() error) []error { + t.Helper() + guard, err := f.db.BeginTx(f.ctx, nil) + if err != nil { + t.Fatal(err) + } + defer guard.Rollback() + var connection uint64 + if err = guard.QueryRowContext(f.ctx, `SELECT CONNECTION_ID()`).Scan(&connection); err != nil { + t.Fatal(err) + } + if _, err = lockApplicationResources(f.ctx, guard, f.resources.NamespaceID, false); err != nil { + t.Fatal(err) + } + results := make([]chan error, len(operations)) + for i, operation := range operations { + results[i] = make(chan error, 1) + go func(ch chan error, run func() error) { ch <- run() }(results[i], operation) + deadline := time.Now().Add(5 * time.Second) + for { + select { + case early := <-results[i]: + t.Fatalf("operation %d returned before entering the database race: %v", i, early) + default: + } + var waiting int + err = f.db.QueryRowContext(f.ctx, `SELECT COUNT(DISTINCT w.REQUESTING_THREAD_ID) FROM performance_schema.data_lock_waits w JOIN performance_schema.threads b ON b.THREAD_ID=w.BLOCKING_THREAD_ID WHERE b.PROCESSLIST_ID=?`, connection).Scan(&waiting) + if err != nil { + t.Fatal(err) + } + if waiting == i+1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("only %d/%d competing transactions reached the held database lock", waiting, i+1) + } + time.Sleep(5 * time.Millisecond) + } + } + if beforeRelease != nil { + beforeRelease() + } + if err = guard.Commit(); err != nil { + t.Fatal(err) + } + errs := make([]error, len(results)) + for i, ch := range results { + select { + case errs[i] = <-ch: + case <-f.ctx.Done(): + t.Fatal(f.ctx.Err()) + } + } + return errs +} + +func TestIntegrationWorkflowVersionAndRaces(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 32, 8) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + + t.Run("v1_waiting_then_v2_published_old_successor_executes", func(t *testing.T) { + f, v1, _ := signalStart(t, ctx, db, false) + before := clockCandidate(t, f, v1.ID, "signal") + if runClock(t, f, before) || f.count("pulse_tasks", "1=1") != 0 { + t.Fatal("v1 did not wait") + } + changed := signalSpec(f.resources.QueueID, false) + changed.TimeoutMS = 60000 + changed.Body.Steps[1].Task.HandlerVersion = "v2" + changed.Body.Steps[1].Task.Input = definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`{"n":7}`)} + if _, err := definition.New(NewWorkflowDefinitionStore(db)).Publish(ctx, f.token, v1.DefinitionID, "publish-v2", definition.PublishInput{Version: "v2", Specification: changed, Reason: "new successor semantics while v1 waits"}); err != nil { + t.Fatal(err) + } + sendSignal(t, f, v1, signalInput(v1)) + runClock(t, f, clockCandidate(t, f, v1.ID, "signal")) + assignment, ref := f.claim() + if assignment.Task.HandlerVersion != "v1" || !workflowRaceNumber(assignment.Task.Payload, "9007199254740993") { + t.Fatal("v1 successor used v2 handler or input") + } + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + retryFinish(t, f, ref, "SUCCEEDED", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + finished := runtimeView(t, f, v1.ID, "SUCCEEDED") + if finished.DefinitionVersionID != v1.DefinitionVersionID || !finished.DeadlineAt.Equal(v1.DeadlineAt) || !workflowRaceNumber(finished.Output, "9007199254740993") { + t.Fatal("v1 fixed contract changed") + } + second, err := instance.New(NewWorkflowInstanceStore(db)).Start(ctx, f.token, "start-v2", instance.StartInput{DefinitionID: v1.DefinitionID, Version: "v2", Input: v1.Input, BusinessKey: v1.BusinessKey}) + if err != nil { + t.Fatal(err) + } + v2 := second.Instance + sendSignal(t, f, v2, signalInput(v2)) + runClock(t, f, clockCandidate(t, f, v2.ID, "signal")) + f.hello.Capabilities = append(f.hello.Capabilities, worker.Capability{Handler: "object.prepare", HandlerVersion: "v2", PayloadVersion: "v1"}) + // 能力集合属于Worker进程身份，换版本需注册新进程。 + f.hello.InstanceID = executionID(t) + f.binding, _, err = f.a.Connect(ctx, f.token, f.hello) + if err != nil { + t.Fatal(err) + } + assignment, ref = f.claim() + if assignment.Task.HandlerVersion != "v2" || !workflowRaceNumber(assignment.Task.Payload, "7") || v2.DefinitionVersionID == v1.DefinitionVersionID { + t.Fatal("v2 semantics did not take effect for new instance") + } + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{"n":7}`)}); err != nil { + t.Fatal(err) + } + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + newFinished := runtimeView(t, f, v2.ID, "SUCCEEDED") + if !workflowRaceNumber(newFinished.Output, "7") || !newFinished.DeadlineAt.Equal(v2.DeadlineAt) || f.count("pulse_tasks", "1=1") != 2 || f.used() != 0 { + t.Fatal("version executions duplicated or lost output/budget") + } + }) + + for _, kind := range []string{"start", "claim", "successor", "retry"} { + for _, pauseFirst := range []bool{false, true} { + t.Run(fmt.Sprintf("pause_%s_pause_queued_first_%t", kind, pauseFirst), func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + spec := runtimeSpec(f.resources.QueueID, 2) + spec.Body.Steps[0].Task.Retry.MaxAttempts = 2 + spec.Body.Steps[0].Task.Retry.InitialDelayMS = 100 + spec.Body.Steps[0].Task.Retry.MaxDelayMS = 100 + v := runtimeStart(t, f, spec) + grantWorkflowControl(f) + var ref worker.AttemptRef + var claimed *worker.Assignment + var taskDeadline time.Time + if kind == "start" { + _, ref = f.claim() + } + if kind == "successor" || kind == "retry" { + ref = retryStart(t, f) + } + if ref.TaskID != "" { + taskDeadline = f.get(ref.TaskID).DeadlineAt + } + if kind == "successor" { + retryFinish(t, f, ref, "SUCCEEDED", false) + } + var candidate workflowCandidate + if kind == "successor" { + candidate = runtimeInbox(t, f, ref.TaskID) + } + runtime := newRuntime(t, db) + command := executionID(t) + operation := func() error { + var err error + switch kind { + case "start": + _, err = f.a.Start(ctx, f.token, f.binding, command, ref) + case "claim": + claimed, err = f.a.Claim(ctx, f.token, f.binding) + case "successor": + _, err = runtime.process(ctx, candidate, false) + case "retry": + _, err = f.a.Finish(ctx, f.token, f.binding, command, worker.FinishInput{Ref: ref, Outcome: "FAILED", Retryable: true, ErrorCode: "SAFE_RETRY", Result: json.RawMessage(`{"n":9007199254740993}`)}) + } + return err + } + pauseKey := string(executionID(t)) + var pauseResult control.Result + pause := func() error { + var err error + pauseResult, err = control.New(NewWorkflowControlStore(db)).Apply(ctx, f.token, v.ID, pauseKey, control.Input{Operation: "PAUSE", ExpectedControlRevision: v.ControlRevision, Reason: "overlapping database transactions"}) + return err + } + ops := []func() error{operation, pause} + if pauseFirst { + ops[0], ops[1] = ops[1], ops[0] + } + errs := competeWorkflow(t, f, nil, ops...) + operationIndex, controlIndex := 0, 1 + if pauseFirst { + operationIndex, controlIndex = 1, 0 + } + if errs[controlIndex] != nil { + t.Fatal(errs[controlIndex]) + } + if err := errs[operationIndex]; err != nil && !(kind == "start" && fault.Code(err) == "EXECUTION_PAUSED") { + t.Fatal(err) + } + paused := runtimeView(t, f, v.ID, "RUNNING") + if !paused.Paused || paused.ControlRevision != 2 || !paused.DeadlineAt.Equal(v.DeadlineAt) { + t.Fatal("pause/deadline not durable") + } + // 以控制事务冻结的数据库时间校验先后，不用goroutine完成顺序代替授权顺序。 + pauseTime := pauseResult.Instance.UpdatedAt + if f.count("pulse_tasks", "created_at>?", pauseTime) != 0 || f.count("pulse_attempts", "(created_at>? OR authorized_at>?)", pauseTime, pauseTime) != 0 { + t.Fatal("task creation, claim or start authorization occurred after PAUSE committed") + } + if got, err := f.a.Claim(ctx, f.token, f.binding); err != nil || got != nil { + t.Fatal("paused workflow claimed work", err) + } + if kind == "claim" && claimed != nil { + ref = worker.AttemptRef{TaskID: claimed.Task.ID, AttemptID: claimed.AttemptID, Epoch: claimed.Epoch, LeaseSecret: claimed.LeaseSecret} + } + if (kind == "claim" && claimed != nil) || (kind == "start" && f.get(ref.TaskID).State == "LEASED") { + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); fault.Code(err) != "EXECUTION_PAUSED" { + t.Fatal("unstarted paused lease authorized", err) + } + } + if kind == "retry" { + if f.get(ref.TaskID).AttemptCount != 1 || f.used() != 0 || !f.get(ref.TaskID).DeadlineAt.Equal(taskDeadline) { + t.Fatal("retry reset budget or retained permit") + } + // 等待原退避自然到期，暂停时仍不得产生下一次尝试。 + time.Sleep(120 * time.Millisecond) + if got, err := f.a.Claim(ctx, f.token, f.binding); err != nil || got != nil { + t.Fatal("paused retry was claimed", err) + } + } + if kind == "successor" && (f.count("pulse_tasks", "1=1") < 1 || f.count("pulse_tasks", "1=1") > 2) { + t.Fatal("successor duplication") + } + collectedWhilePaused := kind == "start" && f.get(ref.TaskID).State == "RUNNING" + if collectedWhilePaused { + if f.used() != 1 { + t.Fatal("pause released a running permit") + } + retryFinish(t, f, ref, "SUCCEEDED", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + if f.count("pulse_tasks", "1=1") != 1 || f.used() != 0 { + t.Fatal("paused result spawned successor or retained permit") + } + } + t.Logf("overlapping pause/%s; pause queued first=%t; operation error=%v; tasks=%d; running result collected=%t", kind, pauseFirst, errs[operationIndex], f.count("pulse_tasks", "1=1"), collectedWhilePaused) + applyControl(t, f, paused, "RESUME") + if kind == "successor" || collectedWhilePaused { + runtimeAdvance(t, f, newRuntime(t, db), v.ID) + } else { + if kind == "retry" || ref.TaskID == "" { + _, ref = f.claim() + } + if f.get(ref.TaskID).State == "LEASED" { + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + } + retryFinish(t, f, ref, "SUCCEEDED", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + } + last := retryStart(t, f) + retryFinish(t, f, last, "SUCCEEDED", false) + runtimeConsume(t, f, newRuntime(t, db), last.TaskID) + done := runtimeView(t, f, v.ID, "SUCCEEDED") + attempts := 2 + if kind == "retry" { + attempts++ + } + if f.count("pulse_tasks", "1=1") != 2 || f.count("pulse_attempts", "1=1") != attempts || f.used() != 0 || !done.DeadlineAt.Equal(v.DeadlineAt) { + t.Fatal("resume lost work, reset deadline or duplicated attempt") + } + }) + } + } + + // 三个实际事务均已进入锁等待后，等待原截止自然到达，再一起放行。 + // 六种入队次序分别覆盖取消/截止先处理；及时信号与新到信号分开保存。 + for _, timely := range []bool{false, true} { + for _, order := range [][3]int{{0, 1, 2}, {0, 2, 1}, {1, 0, 2}, {1, 2, 0}, {2, 0, 1}, {2, 1, 0}} { + t.Run(fmt.Sprintf("signal_deadline_cancel_timely_%t_order_%v", timely, order), func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + spec := signalSpec(f.resources.QueueID, false) + spec.Body.Steps[0].Signal.TimeoutMS = 1000 + v := runtimeStart(t, f, spec) + grantWorkflowSignal(f) + grantWorkflowControl(f) + var prior core.ID + if timely { + prior = sendSignal(t, f, v, signalInput(v)).Signal.ID + } + clock := clockCandidate(t, f, v.ID, "signal") + var due time.Time + if err := db.QueryRowContext(ctx, `SELECT deadline_at FROM weave_waits WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, v.ID).Scan(&due); err != nil { + t.Fatal(err) + } + api := signal.New(NewWorkflowSignalStore(db)) + runtime := newRuntime(t, db) + key, cancelKey := string(executionID(t)), string(executionID(t)) + var arrived signal.Result + ops := []func() error{ + func() error { + var err error + arrived, err = api.Send(ctx, f.token, v.ID, key, signalInput(v)) + return err + }, + func() error { _, err := runtime.process(ctx, clock, false); return err }, + func() error { + _, err := control.New(NewWorkflowControlStore(db)).Apply(ctx, f.token, v.ID, cancelKey, control.Input{Operation: "CANCEL", ExpectedControlRevision: 1, Reason: "race natural deadline and signal"}) + return err + }, + } + errs := competeWorkflow(t, f, func() { + for { + var reached bool + if err := db.QueryRowContext(ctx, `SELECT UTC_TIMESTAMP(6)>=?`, due).Scan(&reached); err != nil { + t.Fatal(err) + } + if reached { + break + } + time.Sleep(10 * time.Millisecond) + } + }, ops[order[0]], ops[order[1]], ops[order[2]]) + for _, err := range errs { + if err != nil { + t.Fatal(err) + } + } + for i := 0; i < 40 && stopStep(t, f, v.ID); i++ { + } + current, err := instance.New(NewWorkflowInstanceStore(db)).Get(ctx, f.token, v.ID) + if err != nil { + t.Fatal(err) + } + if current.State != "CANCELLED" && current.State != "NEEDS_REVIEW" { + t.Fatal("unsettled triple race", current.State) + } + if current.StopKind == nil || *current.StopKind != "CANCEL" || !current.DeadlineAt.Equal(v.DeadlineAt) || f.used() != 0 { + t.Fatal("lost cancellation, deadline or permits") + } + if f.count("pulse_tasks", "1=1") > 1 || f.count("pulse_attempts", "1=1") != 0 || f.count("weave_signals", "state='PENDING'") != 0 { + t.Fatal("triple race duplicated or stranded work") + } + late, err := api.Get(ctx, f.token, v.ID, arrived.Signal.ID) + if err != nil || late.State == "CONSUMED" || late.State == "PENDING" || late.AcceptedAt.Before(due) { + t.Fatal("late signal overrode winner or time", err) + } + if timely { + original, err := api.Get(ctx, f.token, v.ID, prior) + if err != nil || original.State != "CONSUMED" || !original.AcceptedAt.Before(due) || f.count("weave_waits", "winner_signal_id=?", prior) != 1 { + t.Fatal("timely durable signal lost in cancellation/expiry race", err) + } + } else if f.count("weave_signals", "state='CONSUMED'") != 0 || f.count("pulse_tasks", "1=1") != 0 { + t.Fatal("expired signal advanced workflow") + } + replay, err := api.Send(ctx, f.token, v.ID, key, signalInput(v)) + if err != nil || !replay.Deduplicated || replay.Signal.ID != arrived.Signal.ID || replay.Signal.State != arrived.Signal.State { + t.Fatal("race receipt changed", err) + } + t.Logf("three overlapping transactions; timely=%t order=%v state=%s late=%s", timely, order, current.State, late.Disposition) + }) + } + } +} diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 7be3a6f..6f80bc5 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -119,3 +119,6 @@ kill "$TASK_SERVER_PID" wait "$TASK_SERVER_PID" TASK_SERVER_PID= (cd sdk/go && ORBIT_TEST_APP_CREDENTIALS_FILE="$TASK_TEST_DIR/app.credentials.json" ORBIT_TEST_APP_SECOND_CREDENTIALS_FILE="$TASK_TEST_DIR/app-second.credentials.json" ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" ORBIT_TEST_WORKER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_WORKER_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" test -race -count=1 -run '^TestIntegration(Worker|Event)Process' -v .) + +# 上一组子进程已全部清理；流程组独立凭据验证暂停跨全节点停止的公共SDK链路。 +(cd sdk/go && ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" ORBIT_TEST_WORKER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_WORKER_CREDENTIALS_FILE="$TASK_TEST_DIR/workflow.credentials.json" "$TASK_GO" test -race -count=1 -run '^TestIntegrationWorkflowProcess' -v .) diff --git a/sdk/go/workflow_process_integration_test.go b/sdk/go/workflow_process_integration_test.go new file mode 100644 index 0000000..4ab7a94 --- /dev/null +++ b/sdk/go/workflow_process_integration_test.go @@ -0,0 +1,202 @@ +package orbit + +import ( + "context" + "encoding/json" + "testing" + "time" +) + +// 此测试由脚本在原服务退出后单独运行；所有流程操作只走公开HTTP/WSS SDK。 +// 两旧节点均Wait确认SIGKILL后才等待原时钟到期并启动两新节点。 +func TestIntegrationWorkflowProcessPauseRecovery(t *testing.T) { + a := startOrbitProcess(t) + b := startOrbitProcess(t) + _, queue, token := integrationWorkerClient(t) + api := processClient(t, a, token) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Second) + defer cancel() + handler := "pause.process." + workerTestID(t) + spec := workflowSDKSpec(queue, handler) + businessKey := WorkflowRef("workflow", "business_key") + spec.Body.Steps[0].Task.BusinessKey = &businessKey + initial := *spec.Body.Steps[0].Task + query := initial + query.Handler = handler + ".query" + query.ResultType = WorkflowType{Kind: "STRING", Enum: []string{"PENDING", "DONE"}} + query.TimeoutMS = 20000 + parallel := WorkflowParallel{} + for _, branch := range []struct { + step WorkflowStep + typ WorkflowType + }{ + {WorkflowStep{Key: "initial", Kind: "TASK", Task: &initial}, spec.InputType}, + {WorkflowStep{Key: "signal", Kind: "SIGNAL", Signal: &WorkflowSignal{Name: "external.done", PayloadVersion: "v1", Correlation: WorkflowRef("workflow", "business_key"), PayloadType: spec.InputType, TimeoutMS: 30000}}, spec.InputType}, + {WorkflowStep{Key: "timer", Kind: "TIMER", Timer: &WorkflowTimer{DelayMS: 5000}}, WorkflowType{Kind: "NULL"}}, + {WorkflowStep{Key: "query", Kind: "POLL", Poll: &WorkflowPoll{Task: query, MaxRounds: 2, TimeoutMS: 30000, IntervalMS: 5000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}}}, query.ResultType}, + } { + parallel.Branches = append(parallel.Branches, WorkflowParallelBranch{Name: branch.step.Key, Body: WorkflowBlock{Steps: []WorkflowStep{branch.step}, OutputType: branch.typ, Output: WorkflowRef(branch.step.Key)}}) + } + spec.Body.Steps = append([]WorkflowStep{{Key: "join", Kind: "PARALLEL", Parallel: ¶llel}}, spec.Body.Steps...) + in := publishWorkflowSDK(t, ctx, api, spec) + in.BusinessKey = handler + startKey := workerTestID(t) + started, err := api.StartWorkflow(ctx, startKey, in) + if err != nil { + t.Fatal(err) + } + v := started.Instance + work := workflowWorkerSDK(t, ctx, api, queue, handler) + queries := workflowWorkerSDK(t, ctx, api, queue, query.Handler) + first, err := work.NextAssignment(ctx) + if err != nil { + t.Fatal(err) + } + if _, err = work.Start(ctx, workerTestID(t), first.Attempt()); err != nil { + t.Fatal(err) + } + firstQuery := finishWorkflowSDK(t, ctx, queries, WorkerSucceeded, json.RawMessage(`"PENDING"`)) + pollBefore := waitWorkflowStepSDK(t, ctx, api, v.ID, "query", func(d WorkflowStepDetail) bool { + return d.Poll != nil && d.Poll.UsedRounds == 1 && d.Poll.NextPollAt != nil + }) + timerBefore, err := api.GetWorkflowStep(ctx, v.ID, "timer") + if err != nil || timerBefore.Wait == nil || timerBefore.ActivatedAt == nil || timerBefore.Wait.State != "WAITING" { + t.Fatal("timer not waiting before pause", err) + } + signalBefore, err := api.GetWorkflowStep(ctx, v.ID, "signal") + if err != nil || signalBefore.Wait == nil { + t.Fatal(err) + } + pauseKey := workerTestID(t) + pauseInput := WorkflowControlRequest{Operation: "PAUSE", ExpectedControlRevision: v.ControlRevision, Reason: "persist pause across all controller exits"} + paused, err := api.ControlWorkflow(ctx, v.ID, pauseKey, pauseInput) + if err != nil || !paused.Instance.Paused { + t.Fatal(err) + } + signalKey := workerTestID(t) + signalInput := SendWorkflowSignalRequest{Name: "external.done", PayloadVersion: "v1", TargetStepKey: "signal", TargetActivationSequence: signalBefore.Step.ActivationSequence, CorrelationKey: v.BusinessKey, Payload: workflowExact, Reason: "verified result while paused", EvidenceRef: "proof:paused-process"} + accepted, err := api.SendWorkflowSignal(ctx, v.ID, signalKey, signalInput) + if err != nil { + t.Fatal(err) + } + if _, err = work.Finish(ctx, workerTestID(t), WorkerFinish{Ref: first.Attempt(), Outcome: WorkerSucceeded, Result: workflowExact}); err != nil { + t.Fatal(err) + } + initialTask, err := api.GetTask(ctx, first.Task.ID) + if err != nil || initialTask.State != TaskSucceeded || initialTask.AttemptCount != 1 { + t.Fatal("paused result not committed", err) + } + work.Close() + queries.Close() + a.kill(t) + b.kill(t) + // 以首次激活时间计算原TIMER唤醒，不用进程启动时间重新计时。 + wake := timerBefore.ActivatedAt.Add(5 * time.Second) + if !time.Now().Before(wake) { + t.Fatal("all controllers must stop before the original timer wake") + } + if pollBefore.Poll.NextPollAt.After(wake) { + wake = *pollBefore.Poll.NextPollAt + } + if wait := time.Until(wake.Add(100 * time.Millisecond)); wait > 0 { + select { + case <-time.After(wait): + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + } + c := startOrbitProcess(t) + d := startOrbitProcess(t) + if a.nodeID == b.nodeID || c.nodeID == d.nodeID || c.nodeID == a.nodeID || c.nodeID == b.nodeID || d.nodeID == a.nodeID || d.nodeID == b.nodeID { + t.Fatal("node identity reused") + } + api = processClient(t, c, token) + other := processClient(t, d, token) + for _, entry := range []struct { + key string + before WorkflowStepDetail + }{{"signal", signalBefore}, {"timer", timerBefore}} { + after := waitWorkflowStepSDK(t, ctx, api, v.ID, entry.key, func(d WorkflowStepDetail) bool { return d.Step.State == "SUCCEEDED" }) + if after.Wait == nil || after.Wait.ID != entry.before.Wait.ID || !after.Wait.DeadlineAt.Equal(entry.before.Wait.DeadlineAt) || after.Step.ActivationSequence != entry.before.Step.ActivationSequence || *after.Step.ActivationID != *entry.before.Step.ActivationID { + t.Fatal("wait identity or original deadline changed across pause/restart") + } + } + initialAfter := waitWorkflowStepSDK(t, ctx, api, v.ID, "initial", func(d WorkflowStepDetail) bool { return d.Step.State == "SUCCEEDED" }) + if initialAfter.Step.TaskID == nil || *initialAfter.Step.TaskID != first.Task.ID { + t.Fatal("old result replaced after restart") + } + current, err := other.GetWorkflow(ctx, v.ID) + if err != nil || !current.Paused || current.ControlRevision != paused.Instance.ControlRevision || current.DefinitionVersionID != v.DefinitionVersionID || !current.DeadlineAt.Equal(v.DeadlineAt) { + t.Fatal("paused workflow changed on new nodes", err) + } + tail, err := api.GetWorkflowStep(ctx, v.ID, "work") + if err != nil || tail.Step.State != "BLOCKED" || tail.Step.TaskID != nil { + t.Fatal("paused successor created", err) + } + pollAfter, err := api.GetWorkflowStep(ctx, v.ID, "query") + if err != nil || pollAfter.Poll == nil || pollAfter.Poll.UsedRounds != 1 || pollAfter.Poll.LastTaskID != firstQuery.Task.ID || pollAfter.Poll.CurrentTaskID != nil || pollAfter.Poll.NextPollAt == nil || !pollAfter.Poll.NextPollAt.Equal(*pollBefore.Poll.NextPollAt) || !pollAfter.Poll.DeadlineAt.Equal(pollBefore.Poll.DeadlineAt) || pollAfter.Poll.MaxRounds != 2 || pollAfter.Poll.MaxAttemptsPerRound != 2 { + t.Fatal("paused query used another round or reset original budget", err) + } + received, err := api.GetWorkflowSignal(ctx, v.ID, accepted.Signal.ID) + if err != nil || received.State != "CONSUMED" || !received.AcceptedAt.Equal(accepted.Signal.AcceptedAt) || received.TargetActivationSequence != accepted.Signal.TargetActivationSequence { + t.Fatal("accepted signal lost", err) + } + // 在另一个新节点恢复原PAUSE、SIGNAL和START回执，不能修改当前状态或预算。 + duplicate, err := other.ControlWorkflow(ctx, v.ID, pauseKey, pauseInput) + if err != nil || !duplicate.Deduplicated || duplicate.Instance.ControlRevision != paused.Instance.ControlRevision { + t.Fatal("pause receipt lost", err) + } + signalDuplicate, err := other.SendWorkflowSignal(ctx, v.ID, signalKey, signalInput) + if err != nil || !signalDuplicate.Deduplicated || signalDuplicate.Signal.ID != accepted.Signal.ID || signalDuplicate.Signal.State != accepted.Signal.State { + t.Fatal("signal receipt changed", err) + } + startDuplicate, err := other.StartWorkflow(ctx, startKey, in) + if err != nil || !startDuplicate.Deduplicated || startDuplicate.Instance.ID != v.ID || !startDuplicate.Instance.DeadlineAt.Equal(v.DeadlineAt) { + t.Fatal("start receipt changed", err) + } + resumeKey := workerTestID(t) + resumeInput := WorkflowControlRequest{Operation: "RESUME", ExpectedControlRevision: paused.Instance.ControlRevision, Reason: "resume persisted pause through another node"} + resumed, err := other.ControlWorkflow(ctx, v.ID, resumeKey, resumeInput) + if err != nil || resumed.Instance.Paused || resumed.Instance.ControlRevision != 3 || !resumed.Instance.DeadlineAt.Equal(v.DeadlineAt) { + t.Fatal("resume changed fixed state", err) + } + queries = workflowWorkerSDK(t, ctx, api, queue, query.Handler) + lastQuery := finishWorkflowSDK(t, ctx, queries, WorkerSucceeded, json.RawMessage(`"DONE"`)) + work = workflowWorkerSDK(t, ctx, other, queue, handler) + last := finishWorkflowSDK(t, ctx, work, WorkerSucceeded, workflowExact) + finished := waitWorkflowSDK(t, ctx, api, v.ID, "SUCCEEDED") + if finished.Paused || finished.ControlRevision != 3 || !finished.DeadlineAt.Equal(v.DeadlineAt) || finished.DefinitionVersionID != v.DefinitionVersionID || string(finished.Output) != string(workflowExact) { + t.Fatal("final output or original version/deadline changed") + } + finalPoll, err := api.GetWorkflowStep(ctx, v.ID, "query") + if err != nil || finalPoll.Poll == nil || finalPoll.Poll.UsedRounds != 2 || !finalPoll.Poll.DeadlineAt.Equal(pollBefore.Poll.DeadlineAt) || finalPoll.Poll.CurrentTaskID != nil { + t.Fatal("query did not preserve the two-round budget", err) + } + rounds, err := api.ListWorkflowPollRounds(ctx, v.ID, "query", *finalPoll.Step.ActivationID, HistoryPageRequest{}) + if err != nil || len(rounds.Items) != 2 || rounds.NextCursor != "" || rounds.Items[0].Round != 2 || rounds.Items[0].TaskID != lastQuery.Task.ID || rounds.Items[1].Round != 1 || rounds.Items[1].TaskID != firstQuery.Task.ID { + t.Fatal("query history duplicated or reset after paused downtime", err) + } + tasks, err := api.ListTasks(ctx, ListTasksRequest{QueueID: queue, BusinessKey: handler, Limit: 100}) + if err != nil || len(tasks.Items) != 4 || tasks.NextCursor != "" { + t.Fatal("unexpected tasks in paused workflow", err) + } + for _, assignment := range []WorkerAssignment{first, firstQuery, lastQuery, last} { + task, err := api.GetTask(ctx, assignment.Task.ID) + if err != nil || task.State != TaskSucceeded || task.AttemptCount != 1 { + t.Fatal("duplicated execution after full stop", err) + } + history, err := api.ListTaskAttempts(ctx, task.ID, HistoryPageRequest{}) + if err != nil || len(history.Items) != 1 || history.NextCursor != "" { + t.Fatal("historical attempts lost or duplicated", err) + } + } + unchanged, err := api.GetTask(ctx, initialTask.ID) + if err != nil || !unchanged.DeadlineAt.Equal(initialTask.DeadlineAt) || unchanged.Retry != initialTask.Retry { + t.Fatal("original task policy changed", err) + } + resumeDuplicate, err := api.ControlWorkflow(ctx, v.ID, resumeKey, resumeInput) + if err != nil || !resumeDuplicate.Deduplicated || resumeDuplicate.Instance.ControlRevision != 3 { + t.Fatal("resume receipt changed", err) + } + t.Log("both original nodes SIGKILL confirmed; paused task/signal/timer collected by two fresh nodes; public SDK resume preserves version, receipts, deadlines, two query rounds and four single attempts") +}