From 0f7ac5203b1ce7da224adfcc7568497fb5f100b2 Mon Sep 17 00:00:00 2001 From: k2safe Date: Fri, 18 Sep 2026 01:11:42 +0800 Subject: [PATCH] feat: add bounded workflow decisions and durable result inbox --- AGENTS.md | 3 + README.md | 3 +- docs/06-workflow-spec.md | 2 +- docs/07-data-model.md | 9 +- docs/08-development-plan.md | 4 +- docs/10-workflow-engine-core.md | 87 ++++ docs/reports/phase-4-engine-core.md | 60 +++ ...tion_resources_upgrade_integration_test.go | 1 + .../000017_workflow_result_inbox.up.sql | 18 + .../mysql/no_foreign_keys_integration_test.go | 8 +- internal/storage/mysql/result_journal.go | 11 +- internal/storage/mysql/workflow_events.go | 80 ++++ .../mysql/workflow_events_integration_test.go | 277 +++++++++++ ...workflow_inbox_upgrade_integration_test.go | 120 +++++ weave/engine/evaluate.go | 208 ++++++++ weave/engine/plan.go | 352 ++++++++++++++ weave/engine/plan_test.go | 449 ++++++++++++++++++ weave/engine/poll.go | 54 +++ weave/engine/program.go | 164 +++++++ weave/model/inbox_event.go | 8 + 20 files changed, 1909 insertions(+), 9 deletions(-) create mode 100644 docs/10-workflow-engine-core.md create mode 100644 docs/reports/phase-4-engine-core.md create mode 100644 internal/storage/mysql/migrations/000017_workflow_result_inbox.up.sql create mode 100644 internal/storage/mysql/workflow_events.go create mode 100644 internal/storage/mysql/workflow_events_integration_test.go create mode 100644 internal/storage/mysql/workflow_inbox_upgrade_integration_test.go create mode 100644 weave/engine/evaluate.go create mode 100644 weave/engine/plan.go create mode 100644 weave/engine/plan_test.go create mode 100644 weave/engine/poll.go create mode 100644 weave/engine/program.go diff --git a/AGENTS.md b/AGENTS.md index ccb5f15..575c77c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,3 +32,6 @@ - P4定义发布使用schema_version=1有限结构，词法作用域禁止跨分支/前向读取，任务策略显式冻结；SDK类型独立。发布按认证→应用额度→操作身份→定义→排序队列持锁；原键与同版本去重仍校验当前权限。定义/版本/revision/审计/回执同事务；同版本不同内容不得覆盖。保留精确数字，校验存储原文与结构摘要；不得把发布通过或DSL支持误称为实例执行已实现。 - P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归（docs/reports/phase-4-definitions.md）。下一步实例启动/固定版本/首批任务同事务、内部事件接续、顺序与组合推进、持久化等待；P4整体和P5–P7仍未完成。 + +- 内部结果交付使用workflow_activation所有者并固定激活ID，TASK经activation.task_id、POLL经轮次校验归属；weave_inbox来源及独立快照与Pulse结果同事务，先于外部结果流行锁，不在结果路径反锁实例/步骤/订阅。迁移000017隔离缺快照历史，禁止用当前任务重建旧事件。 +- weave/engine仅计算有限决策，完整实例存储运行器尚未接入。Plan最多32个变更，More必须持久化待续；耗尽变更预算仍须遍历其他已活动分支，不能误判在途工作不可达。冻结分支选择，精确数值，不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务，不能把纯快照往返测试当作真实进程恢复。 diff --git a/README.md b/README.md index aa7aad7..cb82257 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)。实例执行、持久化等待和恢复继续开发；发布定义不会启动任务。 +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)已实现；实例存储推进、持久化等待和恢复继续开发，发布定义不会启动任务。 数据库禁止外键，关联一致性通过服务端事务校验，索引、唯一约束和 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)。 @@ -31,6 +31,7 @@ pulse/schedule/ Cron/间隔、漏跑、版本及计划接口 internal/timezone/ 随计划冻结的内置时区数据快照 weave/model/ 定义、实例、步骤、信号和查询轮次 weave/definition/ 有限结构/类型校验与定义发布契约 +weave/engine/ 不可变规则的有限步骤决策与精确结果映射 internal/config/ 配置校验 internal/storage/mysql/ 连接、事务、迁移及集成测试 internal/transport/httpserver/ HTTP 接入 diff --git a/docs/06-workflow-spec.md b/docs/06-workflow-spec.md index f83ca19..b0952fc 100644 --- a/docs/06-workflow-spec.md +++ b/docs/06-workflow-spec.md @@ -1,6 +1,6 @@ # Weave 流程编排功能与设计 -> 版本 0.3 · 2026-09-18。首版功能范围已确认。定义结构、发布校验与不可变版本已实现，见[定义格式](api/workflow-definition-format.md)和[发布API](api/workflow-definitions.md)；下述实例推进、等待、查询与管理仍待P4–P5实现验收。 +> 版本 0.3 · 2026-09-18。首版功能范围已确认。定义结构、发布校验与不可变版本已实现，见[定义格式](api/workflow-definition-format.md)和[发布API](api/workflow-definitions.md)；已补[步骤决策与内部结果交付](10-workflow-engine-core.md)，下述实例存储推进、等待、查询与管理仍待P4–P5实现验收。 ## 1. 职责与边界 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 5d826e0..f548d7b 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -1,6 +1,6 @@ # 数据模型与实现基线 -更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现，独立 Inbox 场景已实现，流程编排继续开发。 +更新：2026-09-18。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现，独立 Inbox 场景已实现，流程编排继续开发。 ## 1. 模块与依赖 @@ -75,7 +75,7 @@ | weave | `InboxEvent` | `weave_inbox` | | weave | `PendingAdvance` | `weave_pending_advances` | -`orbit_events` 保存原始状态事件，`orbit_result_events` 使用相同事件 ID 保存队列、提交序号和冻结快照，`orbit_event_streams` 串行分配命名空间内的结果序号；对外投递使用 `orbit_deliveries`，内部流程接续预留 `weave_inbox`。`weave_pending_advances` 保存暂停或分批推进后的续接检查点。 +`orbit_events` 保存原始状态事件，`orbit_result_events` 使用相同事件 ID 保存队列、提交序号和冻结快照，`orbit_event_streams` 串行分配命名空间内的结果序号；对外投递使用 `orbit_deliveries`，内部流程结果已原子写入 `weave_inbox`及其独立冻结快照（实例消费推进尚未实现）。`weave_pending_advances` 保存暂停或分批推进后的续接检查点。 ## 4. 已落实的数据库约束 @@ -186,3 +186,8 @@ HTTP/SDK创建时间范围使用包含下界、不包含上界；v2游标以固 ## 相对延迟的持久化边界 delay_ms是普通/批量提交命令字段，首次事务以同一个数据库时间写入created_at和available_at；不增加表字段、迁移或外键。原时长参与幂等请求摘要，原回执保存初始时间；重报不能重新计时。后续安全重试可以更新当前available_at，首次受理回执及历史尝试仍保留。Go命令字段提供中文说明，见[任务契约](api/tasks.md)和[回归报告](reports/phase-3-delay-recovery.md)。 + + +## P4 内部结果快照 + +000017_workflow_result_inbox为InboxEvent增加SourceTaskID、SourceActivationID、SourceRevision、Snapshot四个直接字段；当前35个实体、293个直接声明持久化字段，均有中文注释，无外键或级联。有效待处理记录要求来源齐全；历史缺少快照的未处理记录隔离为QUARANTINED，保留旧代次而不重建结果。内部快照独立于对外结果日志保留期，和Pulse权威结果同事务提交。见[执行决策及交付边界](10-workflow-engine-core.md)与[回归进展](reports/phase-4-engine-core.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 801fd9c..1d9a3dd 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -1,6 +1,6 @@ # Orbit 持续开发与回归计划 -更新：2026-09-17。此计划对应持续开发目标，由开发过程同步维护。数据库已按用户要求采用无外键设计，迁移 000008 移除历史外键，关联一致性由服务端事务保证。 +更新：2026-09-18。此计划对应持续开发目标，由开发过程同步维护。数据库已按用户要求采用无外键设计，迁移 000008 移除历史外键，关联一致性由服务端事务保证。 ## 1. 最终交付目标 @@ -28,7 +28,7 @@ | 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)；实例推进、事件接续与持久化等待仍待实现 | +| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 定义格式/词法类型校验、不可变发布、独立SDK及六类节点定义例子已实现，已通过[子项验收](reports/phase-4-definitions.md)；已补[有限决策器及内部结果交付](reports/phase-4-engine-core.md)；实例存储适配、事件消费推进与持久化等待仍待实现 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | | P7 首版交付核对 | SDK 兼容与发布准备、完整 API 手册、场景示例、全量回归及交付清单 | 新接入者按文档跑通全部示例；所有首版能力可追溯到实现与测试；报告实际生产边界 | 待开发 | diff --git a/docs/10-workflow-engine-core.md b/docs/10-workflow-engine-core.md new file mode 100644 index 0000000..6213c85 --- /dev/null +++ b/docs/10-workflow-engine-core.md @@ -0,0 +1,87 @@ +# Weave 执行决策与内部结果交付 + +状态：P4 内部基础已实现。已接通Pulse结果到Weave持久化Inbox，并提供有限步骤决策器；**实例启动、实例存储适配、Inbox消费推进、定时/信号运行器及对应公共SDK尚未实现**。不能把本页能力解释成流程已经可以端到端执行。 + +## 1. 为什么结果要独立持久化 + +任务已执行成功之后，流程节点可能立即退出。依赖进程内回调会丢失后继工作；从可变任务当前状态读取历史结果，也可能把后来的核查决定误当成旧事件。 + +当前结果事务顺序如下，任何一步失败均回滚： + +```mermaid +flowchart LR + A[更新Pulse任务和尝试] --> B[结算执行许可与用量] + B --> C[写入原始状态事件] + C --> D[核验流程激活归属] + D --> E[写入Weave Inbox和冻结快照] + E --> F[锁定对外结果流并写入结果日志] + F --> G[写入审计和执行回执] + G --> H[提交事务] +``` + +普通任务和计划任务仍使用原有结果链路，不产生Weave Inbox。流程任务使用`owner_kind=workflow_activation`、`owner_id=激活ID`；TASK直接关联激活task_id，POLL通过轮次记录关联。所有关系均限定同一命名空间，核对步骤、实例、门控、固定定义和版本，不使用外键。 + +已实现的内部交付触发为`task.succeeded`、`task.failed`、`task.cancelled`、`task.needs_review`。任务暂时失败并等待自动重试，不新增流程激活或内部结果事件。相同完成回执重报不产生第二个事件；同一来源任务状态版本还有独立唯一约束。 + +发布后使用旧激活的在途结果仍保留原激活身份；生产者不自行判定是否推进当前激活。未来消费者须识别旧激活/旧轮次并留痕，不能将其用于当前工作。 + +生产者在现有结果快照查询中一起读取所有者和关联标记；普通结果不增加一次单独网络查询。流程关联核验使用只读关联查询，不在已持有任务锁后再锁实例、步骤或对外订阅。内部Inbox写入发生在外部结果流行锁之前，保持既有锁顺序。 + +## 2. 数据模型与升级 + +迁移`000017_workflow_result_inbox`为`weave_inbox`新增四个中文注释字段： + +| 字段 | 含义 | +| --- | --- | +| source_task_id | 来源Pulse任务 | +| source_activation_id | 事件发生时的流程激活 | +| source_revision | 来源任务的状态版本，大于零 | +| snapshot | schema_version=1的独立结果JSON文本，最多512 KiB，保留原始数字精度 | + +snapshot包含任务/队列ID、当时的状态与版本、结果、效果不确定标记和当前尝试ID；命名空间及实例由Inbox行固定。快照不依赖对外`orbit_result_events`的保留时间，也不从后来的任务结果重建。 + +`(namespace_id,event_id)`沿用原去重约束，新增`(namespace_id,source_task_id,source_revision)`唯一键。有效待处理记录必须四项完整且快照为合法JSON；只有迁移前的PROCESSED/QUARANTINED历史记录可以四项均为空。没有外键或级联。 + +v16及以前的未处理记录缺少冻结来源，升级后统一标为QUARANTINED，原租约清空、代次保留，error_code为LEGACY_RESULT_UNVERIFIED；已处理历史保持原状态。迁移不使用当前任务倒推旧结果，不赋予旧记录新的执行权。 + +新字段对应`weave/model/InboxEvent`，Go与SQL均有中文注释。原有扫描索引`(state,available_at,namespace_id,id)`保留，后续消费仍按持久化待处理状态和有限批次驱动。 + +关联丢失或类型异常时拒绝提交权威任务结果，不能悄悄丢掉内部交付后返回成功。此类持续异常需要修复存储关系后原身份核对；本阶段尚未提供自动修复工具。 + +## 3. 纯步骤决策器 + +`weave/engine.Compile`校验并独立拷贝已发布的结构版本1定义。Program可并发读取，调用方修改原定义或取出的步骤副本不会改变它。 + +`Program.Plan(snapshot, limit)`接受事务内读取、已校验归属的当前步骤快照，返回最多1–32个ACTIVATE、COMPLETE或SKIP决策。它不执行HTTP、业务Go函数、数据库写入，也不读取当前时间。绝对截止、许可与执行授权继续由存储及Pulse负责。 + +| 能力 | 已实现的决策行为 | +| --- | --- | +| 顺序 | 仅消费已成功步骤输出；后继不会提前读取结果 | +| 分支 | 首次按条件选择，保存case索引；再次推进复用选择，未选择路径分批标为SKIPPED | +| 并行 | 分支分别使用进入节点前的数据，全部完成才产生组合输出；兄弟结果通过命名汇合读取 | +| TASK | 解析输入与追踪/资源键，按Pulse运行数据边界检查；返回待持久化任务参数，不授予执行权 | +| SIGNAL | 解析并冻结非空业务关联、名称与参数版本，返回等待命令 | +| TIMER | 返回持久化等待命令，不分配业务Worker | +| POLL | 解析固定查询输入及可选回执关联；成功查询结果有独立分类器，PENDING不等于节点成功 | +| 暂停门控 | AllowNew=false时不激活新业务工作，允许整理已有组合结果；不替代数据库门控校验 | +| 有限推进 | More=true时说明本轮变更预算耗尽，必须由适配层保存待续记录；已有活动分支仍继续校验，不能仅安排内存回调 | + +只在已保存的SUCCEEDED投影上读取输出；类型不匹配、未知/失败状态、不可达的执行、已完成组合下仍有未结清子节点、伪造组合输出等都会拒绝整轮决策。不返回可以部分应用的错误结果。 + +引用与对象/数组构造保留JSON原始数值，输出及累计构造有大小上限；数字条件使用精确有理数比较。相邻大整数、长小数和科学计数法不会先转成float64。规则不提供任意脚本、动态时间或随机数据。 + +`ClassifyPoll`仅处理“查询请求成功”的业务结果，返回PENDING、SUCCEEDED、FAILED或NEEDS_REVIEW；未知值/类型错误拒绝。即使快照将POLL标为SUCCEEDED，结果仍在处理中、失败或待核查时也不能推进后继。 + +## 4. 后续存储适配必须满足的条件 + +此处是尚待实现的要求，不能按纯单测宣称已完成： + +1. 启动时认证、固定不可变版本/输入和策略、校验全部队列、计算实例截止，并在同事务创建实例、门控、首批激活/任务及审计回执。相同身份返回原实例。 +2. 消费者领取内部事件后核对命名空间、来源任务版本、激活/轮次及当前步骤版本；消费确认、步骤变化、后续任务或持久化待续记录同事务提交。 +3. Plan返回值只有存储事务提交后才生效。额度不足或事务失败不能留下孤立激活/许可；重启从持久化快照重新计算，不重复创建已有任务。 +4. AllowNew来自权威实例状态、控制版本、取消、总截止和门控。程序参数本身不是权限证明。任务deadline不得超过实例或查询节点绝对截止。 +5. 数据库时间驱动真实等待与超时，按可靠受理时间裁决信号/结果，原截止和预算在恢复时保持。单轮重试与查询轮次不能互相补充次数。 +6. 完成实例前核对所有活动工作结清和未知效果；失败/取消需要停止新派发并收集在途结果。人工恢复通过新激活保留历史。 +7. 增加流程来源查询、实例/步骤/信号/操作HTTP和独立SDK、可运行端到端场景、真实节点强杀和竞争回归。以上均属于P4–P5剩余工作。 + +当前测试夹具通过SQL构造流程关联，只验证结果事务边界；它不代表公共实例API或运行器已经存在。纯决策器的JSON往返测试验证可从快照重算，也不等同于真实进程恢复验收。 diff --git a/docs/reports/phase-4-engine-core.md b/docs/reports/phase-4-engine-core.md new file mode 100644 index 0000000..5c0738b --- /dev/null +++ b/docs/reports/phase-4-engine-core.md @@ -0,0 +1,60 @@ +# P4 进展：步骤决策与内部结果交付 + +日期：2026-09-18。本次完成纯步骤决策器、Pulse结果原子写入Weave Inbox及独立快照迁移；P4整体尚未完成，实例API、消费推进与真实等待运行器仍待开发。设计与边界见[内部基础](../10-workflow-engine-core.md)。 + +## 代码与数据变化 + +- weave/engine提供不可变编译结果、1–32变更预算的顺序/分支/并行决策、任务参数与等待描述、精确数字条件及POLL业务状态分类。不会调用业务处理器或授予执行权。 +- 结果事务复用现有任务快照查询读取关联；内部交付先于对外结果流行锁，任务/尝试/许可/原事件/内部快照/外部日志/审计/回执一起提交或回滚。 +- 新迁移000017为Inbox增加来源任务、激活、状态版本和独立快照4字段，全部中文注释。35个实体、293个直接声明持久化字段，零外键/级联。旧的未验证记录隔离，不按当前任务重建历史。 +- 尚无新的公共运行API或SDK；已有定义发布SDK和其余场景随完整回归验证，执行器不作为业务端可调用的SDK暴露。 + +## 已完成验证 + +环境：macOS arm64、Go 1.27.1、隔离Docker MySQL 8.4.10。所有数据库测试使用临时库、随机秘密与回环端口。 + +| 检查 | 证据 | +| --- | --- | +| make fmt / race test / vet / build | 通过，覆盖服务端、SDK和独立Inbox/Outbox模块 | +| 决策器单测 | 预算1/2/32反复重编译及JSON恢复，不重复产生任务；精确数字、分支固定、并行汇合、暂停、类型/大小/状态拒绝、不可达执行、已完成组合完整性、并发只读、SIGNAL/TIMER/POLL准备与查询分类均通过；新增嵌套并行分支耗尽预算后仍遍历其他在途分支的回归，race测试通过 | +| 初始真实MySQL定向回归 | 修正测试夹具未提交非成功结果error_code后，Schema、结果交付、v16升级连续3轮通过，共22.917秒；此轮尚未加入后续自然租约到期用例 | +| 新结果交付用例 | 四种结果、8个并发同完成回执、暂停期间收集、POLL旧激活归属、内部/外部写入失败完整回滚、缺失owner/门控/激活/定义拒绝；独立快照不依赖外部结果日志；实际数据库租约自然到期后转NEEDS_REVIEW并交付一次，重复恢复不再写入 | +| 升级用例 | 真实v16结构及PENDING/LEASED/PROCESSED/QUARANTINED记录升级，历史状态/代次及空来源检查；缺来源的新PENDING被CHECK拒绝；schema与零外键检查通过 | +| 存储全量及无外键检查 | 重跑后全部存储集成测试通过，116.335秒；v7的70个外键移除至0、323个原索引组件保留，新结果唯一索引另行核对。v9额度回填和v16升级通过；直接查询当前测试库外键为0、业务字段空COMMENT为0 | +| 完整隔离回归 | scripts/test-mysql.sh退出0；19个SDK示例与独立Inbox/Outbox共21个例子执行通过，10个SDK进程故障父用例及Inbox/Outbox各自进程故障通过，临时容器已清理 | +| 模型及文档核对 | 含公共Record的293个直接持久化字段声明均有中文注释；799个本地文档链接、53个JSON块校验通过；git diff --check通过 | + +完整脚本中的测试组结果如下（秒）： + +| 测试组 | 耗时 | +| --- | ---: | +| MySQL存储全量 | 116.335 | +| Worker连接及自动运行SDK | 92.477 | +| 计划 / 队列资源SDK | 5.676 / 4.488 | +| 批量受理 / 任务查询SDK | 1.424 / 1.627 | +| 订阅管理 / 事件SDK | 33.392 / 51.871 | +| 人工恢复 / 定义发布SDK | 3.395 / 1.354 | +| 独立Inbox / Outbox | 34.588 / 34.330 | +| 应用额度SDK | 1.503 | +| SDK真实进程故障 | 93.116 | + +复现入口（Go及Docker需已安装）： + +```sh +make fmt test vet build +make integration +``` + +本次实际通过GO参数选择Go 1.27.1、DOCKER参数选择本机Docker；脚本自行创建隔离数据库，不连接已有业务数据库。数据库环境变量缺失时普通单测会跳过集成用例，不能代替上述真实integration结果。 + +测试中的流程关联由受控SQL夹具创建，不能作为实例受理实现的证据。数据库写入失败通过触发器注入，之后原完成身份重报成功；没有修改接口约束以接受不合法回执。 + +首次完整回归暴露两处旧升级测试未适配v17：v9还原夹具遗漏删除新增Inbox字段，v7索引保留比较未单独核验新增唯一索引。已修正夹具并增加该唯一索引列序/唯一性断言，重跑存储全量通过；没有修改历史迁移或放宽零外键检查。 + +复查另发现决策器在左侧分支耗尽变更预算后，提前返回另一个已有分支，误报其活动子任务不可达。新增测试先复现，再将预算限制保留在新增变更处，继续遍历已有选中路径；恢复后各任务仍只激活一次，最终输出及完整race检查通过。 + +## 尚未实现及后续工作 + +下一步实现实例存储适配与Start/Get/步骤查询：认证、固定版本/输入、实例及门控、首批任务与幂等/审计同事务；之后内部事件消费、有限推进检查点、持久化等待、流程来源与SDK接入。POLL轮次、信号竞争、暂停/取消/核查恢复与预算维护继续按P4–P5交付。 + +本阶段没有真实流程进程重启回归、持久化轮询或信号/定时消费器。P3已有进程故障测试只证明原任务和订阅链路，不能替代Weave运行器验证。增加关联索引读取与快照写入后的吞吐、存储增长、清理及跨版本上线边界继续在P6–P7验证，不声称生产容量和高可用已达标。 diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index 5a45a10..ed70fe6 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -64,6 +64,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { retryFinish(t, g, finished, "SUCCEEDED", false) assertApplicationCounts(t, f, 4, 2, 1) // 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。 + f.exec(`ALTER TABLE weave_inbox DROP CHECK ck_weave_inbox_source, DROP INDEX uq_weave_task_result, DROP COLUMN source_task_id, DROP COLUMN source_activation_id, DROP COLUMN source_revision, DROP COLUMN snapshot`) f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_queue_business_created, DROP INDEX ix_task_queue_business_state_created, MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_as_cs NULL COMMENT '业务侧关联键，用于检索追踪，不自动提供业务幂等'`) f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_retry_not_self, DROP INDEX uq_task_retry_parent, ADD INDEX fk_pulse_tasks_3(namespace_id,retry_of)`) f.exec(`DROP TABLE orbit_application_resources`) diff --git a/internal/storage/mysql/migrations/000017_workflow_result_inbox.up.sql b/internal/storage/mysql/migrations/000017_workflow_result_inbox.up.sql new file mode 100644 index 0000000..1a20411 --- /dev/null +++ b/internal/storage/mysql/migrations/000017_workflow_result_inbox.up.sql @@ -0,0 +1,18 @@ +-- Weave内部结果保存独立快照，避免外部结果日志清理或任务后续核查改变历史输入。 +ALTER TABLE weave_inbox + ADD COLUMN source_task_id BINARY(16) NULL COMMENT '来源Pulse任务标识；仅迁移前无法核实的历史记录允许为空', + ADD COLUMN source_activation_id BINARY(16) NULL COMMENT '事件发生时的Weave激活标识，保留旧激活归属；仅未核实历史记录可空', + ADD COLUMN source_revision BIGINT UNSIGNED NULL COMMENT '来源任务的状态版本，必须大于零；仅未核实历史记录可空', + ADD COLUMN snapshot MEDIUMTEXT NULL COMMENT 'schema_version版本化结果JSON文本，保留精确数字与当时状态；仅未核实历史记录可空'; + +-- 旧记录缺少冻结结果，不能用当前任务状态倒推历史并自动推进。 +UPDATE weave_inbox SET state='QUARANTINED',lease_owner=NULL,lease_until=NULL, + error_code='LEGACY_RESULT_UNVERIFIED',revision=revision+1,updated_at=UTC_TIMESTAMP(6) + WHERE state<>'PROCESSED'; + +ALTER TABLE weave_inbox + ADD UNIQUE KEY uq_weave_task_result (namespace_id,source_task_id,source_revision), + ADD CONSTRAINT ck_weave_inbox_source CHECK ( + (state IN ('PROCESSED','QUARANTINED') AND source_task_id IS NULL AND source_activation_id IS NULL AND source_revision IS NULL AND snapshot IS NULL) + OR (source_task_id IS NOT NULL AND source_activation_id IS NOT NULL AND source_revision IS NOT NULL AND source_revision>0 AND snapshot IS NOT NULL AND OCTET_LENGTH(snapshot)<=524288 AND JSON_VALID(snapshot)) + ); diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index aa95c81..8e893ea 100644 --- a/internal/storage/mysql/no_foreign_keys_integration_test.go +++ b/internal/storage/mysql/no_foreign_keys_integration_test.go @@ -88,8 +88,8 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { } indexes := func() []string { t.Helper() - // v15 将原 retry_of 非唯一索引替换为相同列的唯一索引，下方单独验证其定义。 - rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) + // v15替换retry_of索引、v17增加内部结果唯一索引，下方单独验证定义。 + rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created','uq_weave_task_result') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) if err != nil { t.Fatal(err) } @@ -173,6 +173,10 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { if err := db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(CONCAT(COLUMN_NAME,':',NON_UNIQUE) ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND INDEX_NAME='uq_task_retry_parent'`).Scan(&retryIndex); err != nil || retryIndex != "namespace_id:0,retry_of:0" { t.Fatalf("retry uniqueness replacement=%q err=%v", retryIndex, err) } + var resultIndex string + if err := db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(CONCAT(COLUMN_NAME,':',NON_UNIQUE) ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='weave_inbox' AND INDEX_NAME='uq_weave_task_result'`).Scan(&resultIndex); err != nil || resultIndex != "namespace_id:0,source_task_id:0,source_revision:0" { + t.Fatalf("workflow result uniqueness=%q err=%v", resultIndex, err) + } var collation string if err := db.QueryRowContext(ctx, `SELECT COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND COLUMN_NAME='business_key'`).Scan(&collation); err != nil || collation != "utf8mb4_0900_bin" { t.Fatalf("business key collation=%q err=%v", collation, err) diff --git a/internal/storage/mysql/result_journal.go b/internal/storage/mysql/result_journal.go index e553830..c3adda3 100644 --- a/internal/storage/mysql/result_journal.go +++ b/internal/storage/mysql/result_journal.go @@ -17,8 +17,13 @@ import ( func appendResultEvent(ctx context.Context, tx *sql.Tx, ns, eventID, taskID core.ID, revision uint64, eventType string) error { var snapshot events.TaskResult var raw []byte + var link workflowTaskLink snapshot.SchemaVersion = 1 - err := tx.QueryRowContext(ctx, `SELECT id,queue_id,state,revision,result,effect_uncertain,current_attempt_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, taskID).Scan(&snapshot.TaskID, &snapshot.QueueID, &snapshot.State, &snapshot.Revision, &raw, &snapshot.EffectUncertain, &snapshot.CurrentAttemptID) + err := tx.QueryRowContext(ctx, `SELECT t.id,t.queue_id,t.state,t.revision,t.result,t.effect_uncertain,t.current_attempt_id,t.owner_kind,t.owner_id,t.gate_id, + EXISTS(SELECT 1 FROM weave_activations a WHERE a.namespace_id=t.namespace_id AND a.task_id=t.id) + OR EXISTS(SELECT 1 FROM weave_poll_rounds r WHERE r.namespace_id=t.namespace_id AND r.task_id=t.id) + OR EXISTS(SELECT 1 FROM weave_instances i WHERE i.namespace_id=t.namespace_id AND i.gate_id=t.gate_id) + FROM pulse_tasks t WHERE t.namespace_id=? AND t.id=?`, ns, taskID).Scan(&snapshot.TaskID, &snapshot.QueueID, &snapshot.State, &snapshot.Revision, &raw, &snapshot.EffectUncertain, &snapshot.CurrentAttemptID, &link.Owner, &link.Activation, &link.Gate, &link.Linked) if err != nil { return err } @@ -30,6 +35,10 @@ func appendResultEvent(ctx context.Context, tx *sql.Tx, ns, eventID, taskID core if err != nil { return err } + // 内部待处理记录和结果同事务，外部结果流锁仍最后取得。 + if err := appendWorkflowResult(ctx, tx, ns, eventID, taskID, revision, body, link); err != nil { + return err + } head, _, err := lockEventStream(ctx, tx, ns) if err != nil { return err diff --git a/internal/storage/mysql/workflow_events.go b/internal/storage/mysql/workflow_events.go new file mode 100644 index 0000000..d6853fe --- /dev/null +++ b/internal/storage/mysql/workflow_events.go @@ -0,0 +1,80 @@ +package mysql + +import ( + "context" + "database/sql" + "errors" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/weave/engine" +) + +type workflowTaskLink struct { + Owner sql.NullString + Activation *core.ID + Gate *core.ID + Linked bool +} + +// appendWorkflowResult 在原Pulse结果事务内创建内部交付，必须先于外部结果流行锁。 +// 这里只核验稳定归属并写入队列，不锁实例/激活、不推进流程、不锁对外订阅。 +func appendWorkflowResult(ctx context.Context, tx *sql.Tx, ns, eventID, taskID core.ID, revision uint64, snapshot []byte, link workflowTaskLink) error { + owner, ownerID, gate := link.Owner, link.Activation, link.Gate + if !owner.Valid || owner.String != engine.TaskOwnerKind { + if owner.Valid && owner.String != "schedule" { + return errors.New("unrecognized task owner") + } + // 无外键情况下，丢失owner不能把已有流程关联静默降级成普通任务。 + if link.Linked { + return errors.New("workflow result ownership missing") + } + return nil + } + if ownerID == nil || gate == nil { + return errors.New("workflow result owner or gate missing") + } + var workflow core.ID + var kind string + var activationTask *core.ID + // 即使旧激活不再是current，也要保存内部事件，消费方再核对当前激活并留痕。 + err := tx.QueryRowContext(ctx, `SELECT a.workflow_id,s.kind,a.task_id + FROM weave_activations a + JOIN weave_steps s ON s.namespace_id=a.namespace_id AND s.id=a.step_id AND s.workflow_id=a.workflow_id + JOIN weave_instances i ON i.namespace_id=a.namespace_id AND i.id=a.workflow_id AND i.gate_id=? + JOIN pulse_execution_gates g ON g.namespace_id=i.namespace_id AND g.id=i.gate_id + JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.id=i.definition_version_id AND v.definition_id=i.definition_id + JOIN weave_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id + WHERE a.namespace_id=? AND a.id=?`, gate, ns, ownerID).Scan(&workflow, &kind, &activationTask) + if errors.Is(err, sql.ErrNoRows) { + return errors.New("workflow result associations invalid") + } + if err != nil { + return err + } + switch kind { + case "TASK": + if activationTask == nil || *activationTask != taskID { + return errors.New("workflow activation task mismatch") + } + case "POLL": + if activationTask != nil { + return errors.New("poll activation cannot own a direct task") + } + var round uint32 + err = tx.QueryRowContext(ctx, `SELECT r.round FROM weave_poll_rounds r JOIN weave_poll_states p ON p.namespace_id=r.namespace_id AND p.workflow_id=r.workflow_id AND p.activation_id=r.activation_id WHERE r.namespace_id=? AND r.workflow_id=? AND r.activation_id=? AND r.task_id=? AND r.round BETWEEN 1 AND p.used_rounds`, ns, workflow, ownerID, taskID).Scan(&round) + if errors.Is(err, sql.ErrNoRows) { + return errors.New("workflow poll task association invalid") + } + if err != nil { + return err + } + default: + return errors.New("workflow task owner kind invalid") + } + id, err := core.NewID() + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO weave_inbox(namespace_id,id,workflow_id,event_id,available_at,source_task_id,source_activation_id,source_revision,snapshot) VALUES (?,?,?,?,UTC_TIMESTAMP(6),?,?,?,?)`, ns, id, workflow, eventID, taskID, ownerID, revision, string(snapshot)) + return err +} diff --git a/internal/storage/mysql/workflow_events_integration_test.go b/internal/storage/mysql/workflow_events_integration_test.go new file mode 100644 index 0000000..f022e34 --- /dev/null +++ b/internal/storage/mysql/workflow_events_integration_test.go @@ -0,0 +1,277 @@ +package mysql + +import ( + "context" + "encoding/json" + "os" + "sync" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/platform/events" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/engine" +) + +// 当前尚无实例API；此夹具只为验证Pulse结果交付边界，不能充当已实现的实例运行器。 +type workflowOwnerFixture struct{ workflow, step, activation, gate core.ID } + +func bindWorkflowOwner(t *testing.T, f *executionFixture, task core.ID, poll bool) workflowOwnerFixture { + t.Helper() + ns := f.resources.NamespaceID + permissions := auth.Permissions{Actions: []string{"tasks:read", "tasks:submit", "tasks:cancel", "workers:connect", "workers:consume", definition.ReadAction, definition.PublishAction}, Queues: []string{"*"}} + raw, _ := json.Marshal(permissions) + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), ns, f.resources.CredentialID) + api := definition.New(NewWorkflowDefinitionStore(f.db)) + created, err := api.Create(f.ctx, f.token, string(executionID(t)), definition.CreateInput{Name: "flow-" + string(executionID(t))}) + if err != nil { + t.Fatal(err) + } + spec := workflowTestSpec(f.resources.QueueID) + kind, key := "TASK", "prepare" + if poll { + kind, key = "POLL", "query" + query := *spec.Body.Steps[0].Task + status := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"status": {Kind: "STRING", Enum: []string{"PENDING", "DONE"}}}} + query.ResultType = status + spec.Body = definition.Block{Steps: []definition.Step{{Key: key, Kind: kind, Poll: &definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, StatusPath: []string{"status"}, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}}}}, OutputType: status, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: key}}} + } + version, err := api.Publish(f.ctx, f.token, created.Definition.ID, string(executionID(t)), definition.PublishInput{Version: "v1", Reason: "internal result delivery regression", Specification: spec}) + if err != nil { + t.Fatal(err) + } + out := workflowOwnerFixture{executionID(t), executionID(t), executionID(t), executionID(t)} + f.exec(`INSERT INTO pulse_execution_gates(namespace_id,id) VALUES (?,?)`, ns, out.gate) + f.exec(`INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,deadline_at) VALUES (?,?,?,?,?,'{"n":9007199254740993}','{}',DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))`, ns, out.workflow, created.Definition.ID, version.Version.ID, out.gate) + f.exec("INSERT INTO weave_steps(namespace_id,id,workflow_id,`key`,kind,state) VALUES (?,?,?,?,?,'ACTIVE')", ns, out.step, out.workflow, key, kind) + var direct *core.ID + if !poll { + direct = &task + } + f.exec(`INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,task_id,input) VALUES (?,?,?,?,1,'ACTIVE',?,'{"n":9007199254740993}')`, ns, out.activation, out.workflow, out.step, direct) + f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, out.activation, ns, out.step) + f.exec(`UPDATE pulse_tasks SET owner_kind=?,owner_id=?,gate_id=? WHERE namespace_id=? AND id=?`, engine.TaskOwnerKind, out.activation, out.gate, ns, task) + if poll { + f.exec(`INSERT INTO weave_poll_states(namespace_id,id,workflow_id,activation_id,max_rounds,used_rounds,max_attempts_per_round,interval_ms,deadline_at,current_task_id) VALUES (?,?,?,?,2,1,1,1000,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR),?)`, ns, executionID(t), out.workflow, out.activation, task) + f.exec("INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id) VALUES (?,?,?,?,1,?)", ns, executionID(t), out.workflow, out.activation, task) + } + return out +} +func assertWorkflowInbox(t *testing.T, f *executionFixture, owner workflowOwnerFixture, task core.ID, state string, raw string) { + t.Helper() + var taskID, activationID, workflowID core.ID + var revision uint64 + var snapshot []byte + var status string + if err := f.db.QueryRowContext(f.ctx, `SELECT source_task_id,source_activation_id,workflow_id,source_revision,snapshot,state FROM weave_inbox WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, owner.workflow).Scan(&taskID, &activationID, &workflowID, &revision, &snapshot, &status); err != nil { + t.Fatal(err) + } + var result events.TaskResult + if err := json.Unmarshal(snapshot, &result); err != nil { + t.Fatal(err) + } + if taskID != task || activationID != owner.activation || workflowID != owner.workflow || revision != result.Revision || status != "PENDING" || result.State != state || result.TaskID != task || string(result.Result) != raw { + t.Fatal("internal result snapshot mismatch", result) + } + if f.count("weave_inbox", "workflow_id=?", owner.workflow) != 1 { + t.Fatal("duplicate workflow delivery") + } + var n int + if err := f.db.QueryRowContext(f.ctx, `SELECT COUNT(*) FROM weave_inbox i JOIN orbit_events e ON e.namespace_id=i.namespace_id AND e.id=i.event_id JOIN orbit_result_events r ON r.namespace_id=i.namespace_id AND r.id=i.event_id WHERE i.namespace_id=? AND i.workflow_id=? AND e.aggregate_id=i.source_task_id AND e.aggregate_revision=i.source_revision`, f.resources.NamespaceID, owner.workflow).Scan(&n); err != nil || n != 1 { + t.Fatal("source events not atomic", err) + } +} +func TestIntegrationWorkflowResultDelivery(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 24, 8) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + for _, outcome := range []string{"SUCCEEDED", "FAILED", "UNKNOWN", "CANCELLED"} { + t.Run(outcome, func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + task := f.submit(1) + owner := bindWorkflowOwner(t, f, task.ID, false) + _, ref := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + // 暂停后在途结果仍需持久化；同一完成回执并发不能产生第二条内部事件。 + f.exec(`UPDATE pulse_execution_gates SET paused=TRUE WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, owner.gate) + in := worker.FinishInput{Ref: ref, Outcome: outcome, Result: json.RawMessage(`{"n":9007199254740993}`)} + if outcome != "SUCCEEDED" { + in.ErrorCode = "CONTROLLED_RESULT" + } + if outcome == "CANCELLED" { + f.exec(`UPDATE pulse_tasks SET cancel_requested_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID) + } + message := executionID(t) + var wg sync.WaitGroup + results := make(chan error, 8) + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { defer wg.Done(); _, err := f.a.Finish(ctx, f.token, f.binding, message, in); results <- err }() + } + wg.Wait() + close(results) + for err := range results { + if err != nil { + t.Fatal(err) + } + } + state := outcome + if outcome == "UNKNOWN" { + state = "NEEDS_REVIEW" + } + assertWorkflowInbox(t, f, owner, task.ID, state, string(in.Result)) + // 外部结果日志可独立保留/清理，不能改变内部接续输入。 + f.exec(`DELETE FROM orbit_result_events WHERE namespace_id=?`, f.resources.NamespaceID) + var frozen string + if err := db.QueryRowContext(ctx, `SELECT snapshot FROM weave_inbox WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, owner.workflow).Scan(&frozen); err != nil { + t.Fatal(err) + } + var result events.TaskResult + if json.Unmarshal([]byte(frozen), &result) != nil || string(result.Result) != string(in.Result) { + t.Fatal("internal result depended on external journal") + } + }) + } + t.Run("poll_and_old_activation_keep_original_identity", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + task := f.submit(1) + owner := bindWorkflowOwner(t, f, task.ID, true) + _, err := f.tasks.CancelTask(ctx, f.token, task.ID, "forbidden-direct", service.CancelInput{ExpectedRevision: task.Revision, Reason: "owner controls cancellation"}) + expectWorkerCode(t, err, "STATE_CONFLICT") + _, ref := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + next := executionID(t) + f.exec(`INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,input) VALUES (?,?,?,?,2,'{}')`, f.resources.NamespaceID, next, owner.workflow, owner.step) + f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, next, f.resources.NamespaceID, owner.step) + raw := `{"status":"DONE"}` + if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(raw)}); err != nil { + t.Fatal(err) + } + assertWorkflowInbox(t, f, owner, task.ID, "SUCCEEDED", raw) + }) + + t.Run("natural_lease_expiry_also_delivers", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + task, err := f.tasks.Submit(ctx, f.token, string(executionID(t)), service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":1}`), TaskTimeoutMS: 1500, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 1000}}) + if err != nil { + t.Fatal(err) + } + owner := bindWorkflowOwner(t, f, task.Task.ID, false) + _, ref := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + for f.get(task.Task.ID).State == pulse.TaskRunning { + if _, err := f.storeB.Recover(ctx, 100); err != nil { + t.Fatal(err) + } + select { + case <-ctx.Done(): + t.Fatal(ctx.Err()) + case <-time.After(50 * time.Millisecond): + } + } + assertWorkflowInbox(t, f, owner, task.Task.ID, "NEEDS_REVIEW", "") + if _, err := f.storeB.Recover(ctx, 100); err != nil { + t.Fatal(err) + } + if f.count("weave_inbox", "workflow_id=?", owner.workflow) != 1 { + t.Fatal("recovery duplicated result") + } + }) + + for _, table := range []string{"weave_inbox", "orbit_result_events"} { + t.Run("rollback_"+table, func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + task := f.submit(1) + owner := bindWorkflowOwner(t, f, task.ID, false) + _, ref := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + trigger := "orbit_test_workflow_delivery_failure" + f.exec(`CREATE TRIGGER ` + trigger + ` BEFORE INSERT ON ` + table + ` FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled workflow delivery failure'`) + message := executionID(t) + in := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{"n":9007199254740993}`)} + _, err := f.a.Finish(ctx, f.token, f.binding, message, in) + f.exec(`DROP TRIGGER ` + trigger) + if err == nil || f.get(task.ID).State != pulse.TaskRunning || f.used() != 1 || f.count("weave_inbox", "workflow_id=?", owner.workflow) != 0 || f.count("orbit_events", "aggregate_id=? AND type='task.succeeded'", task.ID) != 0 || f.count("pulse_attempt_receipts", "attempt_id=? AND operation='finish'", ref.AttemptID) != 0 { + t.Fatal("partial result committed", err) + } + out, err := f.a.Finish(ctx, f.token, f.binding, message, in) + if err != nil || out.Deduplicated { + t.Fatal("original retry failed", err) + } + assertWorkflowInbox(t, f, owner, task.ID, "SUCCEEDED", string(in.Result)) + }) + } + for _, broken := range []string{"owner", "gate", "activation", "definition"} { + t.Run("reject_broken_"+broken, func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + task := f.submit(1) + owner := bindWorkflowOwner(t, f, task.ID, false) + _, ref := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil { + t.Fatal(err) + } + var restore func() + switch broken { + case "owner": + f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID) + restore = func() { + f.exec(`UPDATE pulse_tasks SET owner_kind=?,owner_id=? WHERE namespace_id=? AND id=?`, engine.TaskOwnerKind, owner.activation, f.resources.NamespaceID, task.ID) + } + case "gate": + f.exec(`UPDATE pulse_tasks SET gate_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID) + restore = func() { + f.exec(`UPDATE pulse_tasks SET gate_id=? WHERE namespace_id=? AND id=?`, owner.gate, f.resources.NamespaceID, task.ID) + } + case "activation": + f.exec(`UPDATE weave_activations SET task_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, owner.activation) + restore = func() { + f.exec(`UPDATE weave_activations SET task_id=? WHERE namespace_id=? AND id=?`, task.ID, f.resources.NamespaceID, owner.activation) + } + case "definition": + var old core.ID + if err := db.QueryRowContext(ctx, `SELECT definition_version_id FROM weave_instances WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, owner.workflow).Scan(&old); err != nil { + t.Fatal(err) + } + f.exec(`UPDATE weave_instances SET definition_version_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, owner.workflow) + restore = func() { + f.exec(`UPDATE weave_instances SET definition_version_id=? WHERE namespace_id=? AND id=?`, old, f.resources.NamespaceID, owner.workflow) + } + } + message := executionID(t) + in := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{"n":1}`)} + _, err := f.a.Finish(ctx, f.token, f.binding, message, in) + restore() + if err == nil || f.get(task.ID).State != pulse.TaskRunning { + t.Fatal("broken association accepted", err) + } + if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil { + t.Fatal(err) + } + assertWorkflowInbox(t, f, owner, task.ID, "SUCCEEDED", string(in.Result)) + }) + } +} diff --git a/internal/storage/mysql/workflow_inbox_upgrade_integration_test.go b/internal/storage/mysql/workflow_inbox_upgrade_integration_test.go new file mode 100644 index 0000000..0658b6f --- /dev/null +++ b/internal/storage/mysql/workflow_inbox_upgrade_integration_test.go @@ -0,0 +1,120 @@ +package mysql + +import ( + "context" + "database/sql" + "os" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + driver "github.com/go-sql-driver/mysql" +) + +func TestIntegrationWorkflowInboxUpgrade(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + admin, err := Open(ctx, dsn, 2, 1) + if err != nil { + t.Fatal(err) + } + defer admin.Close() + name := "orbit_workflow_upgrade_" + string(executionID(t)) + if _, err := admin.ExecContext(ctx, "CREATE DATABASE `"+name+"`"); err != nil { + t.Fatal(err) + } + defer func() { + cleanup, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + if _, err := admin.ExecContext(cleanup, "DROP DATABASE `"+name+"`"); err != nil { + t.Error(err) + } + }() + cfg, err := driver.ParseDSN(dsn) + if err != nil { + t.Fatal("invalid test DSN") + } + cfg.DBName = name + db, err := Open(ctx, cfg.FormatDSN(), 4, 2) + if err != nil { + t.Fatal(err) + } + defer db.Close() + exec := func(q string, args ...any) { + t.Helper() + if _, err := db.ExecContext(ctx, q, args...); err != nil { + t.Fatal(err) + } + } + exec(`CREATE TABLE orbit_schema_migrations (version BIGINT UNSIGNED PRIMARY KEY,name VARCHAR(191) NOT NULL,checksum CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,dirty BOOLEAN NOT NULL,applied_at DATETIME(6) NULL) ENGINE=InnoDB`) + migrations, err := Migrations() + if err != nil { + t.Fatal(err) + } + for _, m := range migrations { + if m.Version > 16 { + break + } + statements, err := splitSQL(m.SQL) + if err != nil { + t.Fatal(err) + } + for _, q := range statements { + exec(q) + } + exec(`INSERT INTO orbit_schema_migrations(version,name,checksum,dirty,applied_at) VALUES (?,?,?,FALSE,UTC_TIMESTAMP(6))`, m.Version, m.Name, m.Checksum) + } + ns, workflow := executionID(t), executionID(t) + rows := map[string]core.ID{} + for _, state := range []string{"PENDING", "LEASED", "PROCESSED", "QUARANTINED"} { + id := executionID(t) + rows[state] = id + var lease *core.ID + var until *time.Time + if state == "LEASED" { + v, tm := executionID(t), time.Now().UTC().Add(time.Minute) + lease, until = &v, &tm + } + exec(`INSERT INTO weave_inbox(namespace_id,id,workflow_id,event_id,state,available_at,lease_owner,lease_until,lease_epoch) VALUES (?,?,?,?,?,UTC_TIMESTAMP(6),?,?,7)`, ns, id, workflow, executionID(t), state, lease, until) + } + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if err := CheckSchema(ctx, db); err != nil { + t.Fatal(err) + } + for original, id := range rows { + var state string + var source, activation, owner *core.ID + var version *uint64 + var snapshot []byte + var code sql.NullString + var epoch, revision uint64 + if err := db.QueryRowContext(ctx, `SELECT state,source_task_id,source_activation_id,source_revision,snapshot,lease_owner,error_code,lease_epoch,revision FROM weave_inbox WHERE namespace_id=? AND id=?`, ns, id).Scan(&state, &source, &activation, &version, &snapshot, &owner, &code, &epoch, &revision); err != nil { + t.Fatal(err) + } + expected := "QUARANTINED" + expectedRevision := uint64(2) + if original == "PROCESSED" { + expected = "PROCESSED" + expectedRevision = 1 + } + if state != expected || source != nil || activation != nil || version != nil || snapshot != nil || owner != nil || epoch != 7 || revision != expectedRevision { + t.Fatal("legacy record was reinterpreted", state) + } + if original != "PROCESSED" && code.String != "LEGACY_RESULT_UNVERIFIED" { + t.Fatal("missing quarantine reason") + } + } + if _, err := db.ExecContext(ctx, `INSERT INTO weave_inbox(namespace_id,id,workflow_id,event_id,available_at) VALUES (?,?,?,?,UTC_TIMESTAMP(6))`, ns, executionID(t), workflow, executionID(t)); err == nil { + t.Fatal("new unverified pending input accepted") + } + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + t.Log("v16 legacy inputs quarantined without reconstructed results or restored leases; processed history retained; current schema has zero foreign keys") +} diff --git a/weave/engine/evaluate.go b/weave/engine/evaluate.go new file mode 100644 index 0000000..0e4ab51 --- /dev/null +++ b/weave/engine/evaluate.go @@ -0,0 +1,208 @@ +// Package engine 依据已发布规则和持久化快照计算有限推进命令；不调用业务处理器或外部系统。 +package engine + +import ( + "bytes" + "encoding/json" + "errors" + "math/big" + "strings" + + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" + "git.yuntuo.hk/k2safe/orbit/weave/definition" +) + +type values map[string]json.RawMessage + +func copyJSON(raw json.RawMessage) json.RawMessage { return append(json.RawMessage(nil), raw...) } +func copyValues(in values) values { + out := make(values, len(in)+8) + for k, v := range in { + out[k] = v + } + return out +} + +// 表达式来自已验证且由Program持有的不可变定义；每个值仍按运行数据上限检查。 +func resolve(v definition.Value, expected definition.Type, env values) (json.RawMessage, error) { + raw, err := evaluate(v, env, 0) + if err != nil { + return nil, err + } + if err := definition.ValidateData(expected, raw); err != nil { + return nil, err + } + return raw, nil +} +func evaluate(v definition.Value, env values, depth int) (json.RawMessage, error) { + if depth > 8 { + return nil, errors.New("expression depth exceeded") + } + var out json.RawMessage + switch v.Kind { + case "LITERAL": + out = copyJSON(v.Literal) + case "REFERENCE": + if v.Reference == nil { + return nil, errors.New("reference required") + } + raw, ok := env[v.Reference.Source] + if !ok { + return nil, errors.New("reference unavailable") + } + for _, key := range v.Reference.Path { + var object map[string]json.RawMessage + if json.Unmarshal(raw, &object) != nil { + return nil, errors.New("reference is not an object") + } + raw, ok = object[key] + if !ok { + return nil, errors.New("reference field missing") + } + } + out = copyJSON(raw) + case "OBJECT": + object := make(map[string]json.RawMessage, len(v.Fields)) + size := 0 + for k, child := range v.Fields { + raw, err := evaluate(child, env, depth+1) + if err != nil { + return nil, err + } + size += len(raw) + if size > definition.MaxSpecificationBytes { + return nil, errors.New("resolved object exceeds bounds") + } + object[k] = raw + } + raw, err := jsonwire.Marshal(object) + if err != nil { + return nil, err + } + out = raw + case "ARRAY": + array := make([]json.RawMessage, len(v.Items)) + size := 0 + for i, child := range v.Items { + raw, err := evaluate(child, env, depth+1) + if err != nil { + return nil, err + } + size += len(raw) + if size > definition.MaxSpecificationBytes { + return nil, errors.New("resolved array exceeds bounds") + } + array[i] = raw + } + raw, err := jsonwire.Marshal(array) + if err != nil { + return nil, err + } + out = raw + default: + return nil, errors.New("unknown expression") + } + if len(out) == 0 || len(out) > definition.MaxSpecificationBytes { + return nil, errors.New("resolved value exceeds bounds") + } + return out, nil +} +func compare(p definition.Predicate, env values) (bool, error) { + switch p.Op { + case "AND", "OR", "NOT": + for _, child := range p.Arguments { + ok, err := compare(child, env) + if err != nil { + return false, err + } + if p.Op == "NOT" { + return !ok, nil + } + if p.Op == "AND" && !ok { + return false, nil + } + if p.Op == "OR" && ok { + return true, nil + } + } + return p.Op == "AND", nil + } + if p.Left == nil || p.Right == nil { + return false, errors.New("comparison operands missing") + } + left, err := resolve(*p.Left, definition.Type{Kind: "JSON"}, env) + if err != nil { + return false, err + } + right, err := resolve(*p.Right, definition.Type{Kind: "JSON"}, env) + if err != nil { + return false, err + } + decode := func(raw []byte) (any, error) { + var v any + d := json.NewDecoder(bytes.NewReader(raw)) + d.UseNumber() + err := d.Decode(&v) + return v, err + } + a, err := decode(left) + if err != nil { + return false, err + } + b, err := decode(right) + if err != nil { + return false, err + } + var order int + switch v := a.(type) { + case string: + w, ok := b.(string) + if !ok { + return false, errors.New("comparison type mismatch") + } + order = strings.Compare(v, w) + case json.Number: + w, ok := b.(json.Number) + if !ok { + return false, errors.New("comparison type mismatch") + } + x, ok := new(big.Rat).SetString(string(v)) + if !ok { + return false, errors.New("invalid exact number") + } + y, ok := new(big.Rat).SetString(string(w)) + if !ok { + return false, errors.New("invalid exact number") + } + order = x.Cmp(y) + case bool: + w, ok := b.(bool) + if !ok || p.Op != "EQ" && p.Op != "NE" { + return false, errors.New("comparison type mismatch") + } + if v != w { + order = 1 + } + case nil: + if b != nil || p.Op != "EQ" && p.Op != "NE" { + return false, errors.New("comparison type mismatch") + } + default: + return false, errors.New("comparison requires scalars") + } + switch p.Op { + case "EQ": + return order == 0, nil + case "NE": + return order != 0, nil + case "LT": + return order < 0, nil + case "LE": + return order <= 0, nil + case "GT": + return order > 0, nil + case "GE": + return order >= 0, nil + } + return false, errors.New("unknown comparison") +} diff --git a/weave/engine/plan.go b/weave/engine/plan.go new file mode 100644 index 0000000..a6bd157 --- /dev/null +++ b/weave/engine/plan.go @@ -0,0 +1,352 @@ +package engine + +import ( + "bytes" + "encoding/json" + + "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +type planner struct { + program *Program + states map[string]State + decision Decision + limit int + allowNew bool + verifying int + visited map[string]bool +} + +// Plan 只计算本轮最多limit个步骤状态变更（1至32），不修改快照或重新执行已完成业务。 +// 错误时丢弃全部本轮决策，存储适配层须保留原快照并记录核查原因。 +func (p *Program) Plan(snapshot Snapshot, limit int) (Decision, error) { + if limit < 1 || limit > 32 || snapshot.WorkflowID.Validate() != nil || !validBusinessKey(snapshot.BusinessKey) { + return Decision{}, problem("", "INVALID_SNAPSHOT") + } + if err := definition.ValidateData(p.spec.InputType, snapshot.Input); err != nil { + return Decision{}, problem("", "WORKFLOW_INPUT_INVALID") + } + if len(snapshot.Steps) > len(p.nodes) { + return Decision{}, problem("", "INVALID_SNAPSHOT") + } + state := &planner{program: p, states: make(map[string]State, len(p.nodes)), visited: map[string]bool{}, limit: limit, allowNew: snapshot.AllowNew, decision: Decision{Changes: []Change{}}} + for key, row := range snapshot.Steps { + node, ok := p.nodes[key] + if !ok { + return Decision{}, problem(key, "UNKNOWN_STEP") + } + if row.Status == weave.StepFailed || row.Status == weave.StepNeedsReview || row.Status == weave.StepCancelled { + return Decision{}, problem(key, "STEP_REQUIRES_REVIEW") + } + switch row.Status { + case weave.StepBlocked, weave.StepSkipped: + if len(row.Input) != 0 || len(row.Output) != 0 { + return Decision{}, problem(key, "INVALID_STEP_STATE") + } + case weave.StepActive, weave.StepWaiting, weave.StepSucceeded: + if err := definition.ValidateData(definition.Type{Kind: "JSON"}, row.Input); err != nil { + return Decision{}, problem(key, "INVALID_STEP_INPUT") + } + if node.Kind == "TASK" || node.Kind == "POLL" { + task := node.Task + if node.Poll != nil { + task = &node.Poll.Task + } + if null(row.Input) || definition.ValidateData(task.InputType, row.Input) != nil { + return Decision{}, problem(key, "INVALID_STEP_INPUT") + } + } + if row.Status == weave.StepSucceeded { + if err := validateOutput(node, row.Output); err != nil { + return Decision{}, problem(key, "STEP_OUTPUT_INVALID") + } + } else if len(row.Output) != 0 { + return Decision{}, problem(key, "INVALID_STEP_STATE") + } + default: + return Decision{}, problem(key, "INVALID_STEP_STATE") + } + state.states[key] = row + } + workflow, err := jsonwire.Marshal(map[string]string{"id": string(snapshot.WorkflowID), "business_key": snapshot.BusinessKey}) + if err != nil { + return Decision{}, err + } + complete, output, err := state.block(p.spec.Body, values{"input": copyJSON(snapshot.Input), "workflow": workflow}) + if err != nil { + return Decision{}, err + } + for key, row := range snapshot.Steps { + if (row.Status == weave.StepActive || row.Status == weave.StepWaiting || row.Status == weave.StepSucceeded) && !state.visited[key] { + return Decision{}, problem(key, "UNREACHABLE_EXECUTION") + } + } + state.decision.Complete = complete + state.decision.Output = copyJSON(output) + return state.decision, nil +} +func null(raw []byte) bool { return bytes.Equal(bytes.TrimSpace(raw), []byte("null")) } +func outputType(node definition.Step) definition.Type { + switch node.Kind { + case "TASK": + return node.Task.ResultType + case "POLL": + return node.Poll.Task.ResultType + case "SIGNAL": + return node.Signal.PayloadType + case "TIMER": + return definition.Type{Kind: "NULL"} + case "BRANCH": + return node.Branch.Default.OutputType + case "PARALLEL": + t := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{}} + for _, b := range node.Parallel.Branches { + t.Fields[b.Name] = b.Body.OutputType + } + return t + } + return definition.Type{} +} +func validateOutput(node definition.Step, raw json.RawMessage) error { + if (node.Kind == "TASK" || node.Kind == "POLL") && null(raw) { + return problem(node.Key, "STEP_OUTPUT_INVALID") + } + if node.Poll != nil { + outcome, err := classifyPoll(*node.Poll, raw) + if err != nil { + return err + } + if outcome != PollSucceeded { + return problem(node.Key, "POLL_RESULT_NOT_SUCCESS") + } + } + return definition.ValidateData(outputType(node), raw) +} +func (s *planner) add(change Change) bool { + if len(s.decision.Changes) >= s.limit { + s.decision.More = true + return false + } + s.decision.Changes = append(s.decision.Changes, change) + s.states[change.Key] = change.State + return true +} +func (s *planner) block(block definition.Block, parent values) (bool, json.RawMessage, error) { + env := copyValues(parent) + for _, node := range block.Steps { + complete, output, err := s.step(node, env) + if err != nil { + return false, nil, err + } + if !complete { + return false, nil, nil + } + env[node.Key] = output + } + output, err := resolve(block.Output, block.OutputType, env) + if err != nil { + return false, nil, problem("", "BLOCK_OUTPUT_INVALID") + } + return true, output, nil +} +func (s *planner) step(node definition.Step, env values) (bool, json.RawMessage, error) { + s.visited[node.Key] = true + row, ok := s.states[node.Key] + if !ok { + row.Status = weave.StepBlocked + } + completed := row.Status == weave.StepSucceeded + if s.verifying > 0 && !completed { + return false, nil, problem(node.Key, "INCOMPLETE_COMPOSITE_CHILD") + } + if completed { + if node.Kind != "BRANCH" && node.Kind != "PARALLEL" { + return true, copyJSON(row.Output), nil + } + s.verifying++ + defer func() { s.verifying-- }() + } + if row.Status == weave.StepSkipped { + return false, nil, problem(node.Key, "SELECTED_STEP_WAS_SKIPPED") + } + if row.Status == weave.StepBlocked { + if !s.allowNew || s.decision.More { + return false, nil, nil + } + change := Change{Key: node.Key, Kind: node.Kind, Operation: "ACTIVATE", State: State{Status: weave.StepActive}} + switch node.Kind { + case "TASK", "POLL": + task := node.Task + if node.Poll != nil { + task = &node.Poll.Task + } + prepared, err := prepareTask(*task, env) + if err != nil { + return false, nil, problem(node.Key, "TASK_INPUT_INVALID") + } + change.State.Input = copyJSON(prepared.Payload) + change.Task = &prepared + if node.Poll != nil && node.Poll.FinalSignal != nil { + wait, err := prepareSignal(*node.Poll.FinalSignal, "POLL", env) + if err != nil { + return false, nil, problem(node.Key, "SIGNAL_CORRELATION_INVALID") + } + change.Wait = wait + } + case "SIGNAL": + wait, err := prepareSignal(*node.Signal, "SIGNAL", env) + if err != nil { + return false, nil, problem(node.Key, "SIGNAL_CORRELATION_INVALID") + } + change.Wait = wait + change.State.Status = weave.StepWaiting + change.State.Input, _ = jsonwire.Marshal(map[string]string{"name": wait.Name, "payload_version": wait.PayloadVersion, "correlation": wait.Correlation}) + case "TIMER": + change.Wait = &WaitCommand{Kind: "TIMER", TimeoutMS: node.Timer.DelayMS} + change.State.Status = weave.StepWaiting + change.State.Input, _ = jsonwire.Marshal(map[string]uint64{"delay_ms": node.Timer.DelayMS}) + case "BRANCH": + selected := -1 + for i, path := range node.Branch.Cases { + matches, err := compare(path.When, env) + if err != nil { + return false, nil, problem(node.Key, "BRANCH_CONDITION_INVALID") + } + if matches { + selected = i + break + } + } + change.State.Input, _ = jsonwire.Marshal(map[string]int{"case": selected}) + case "PARALLEL": + change.State.Input = json.RawMessage(`{}`) + default: + return false, nil, problem(node.Key, "UNSUPPORTED_NODE") + } + if !s.add(change) { + return false, nil, nil + } + row = change.State + } + if node.Kind != "BRANCH" && node.Kind != "PARALLEL" { + return false, nil, nil + } + var output json.RawMessage + switch node.Kind { + case "BRANCH": + contract := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"case": {Kind: "NUMBER"}}} + var selection struct { + Case int `json:"case"` + } + if definition.ValidateData(contract, row.Input) != nil || json.Unmarshal(row.Input, &selection) != nil || selection.Case < -1 || selection.Case >= len(node.Branch.Cases) { + return false, nil, problem(node.Key, "BRANCH_DECISION_INVALID") + } + // 已冻结的选择不重算；未选择路径分批标记SKIPPED，不创建业务激活。 + for i, path := range node.Branch.Cases { + if i != selection.Case { + if err := s.skip(path.Body); err != nil { + return false, nil, err + } + } + } + body := *node.Branch.Default + if selection.Case >= 0 { + if err := s.skip(*node.Branch.Default); err != nil { + return false, nil, err + } + body = node.Branch.Cases[selection.Case].Body + } + // 预算只阻止新增变更；已有选中路径仍须遍历，避免遗漏并行分支的在途工作。 + done, raw, err := s.block(body, env) + if err != nil { + return false, nil, err + } + if !done { + return false, nil, nil + } + output = raw + case "PARALLEL": + if definition.ValidateData(definition.Type{Kind: "OBJECT"}, row.Input) != nil { + return false, nil, problem(node.Key, "INVALID_PARALLEL_STATE") + } + result := make(map[string]json.RawMessage, len(node.Parallel.Branches)) + all := true + for _, branch := range node.Parallel.Branches { + done, raw, err := s.block(branch.Body, env) + if err != nil { + return false, nil, err + } + if !done { + all = false + } else { + result[branch.Name] = raw + } + } + if !all { + return false, nil, nil + } + var err error + output, err = jsonwire.Marshal(result) + if err != nil { + return false, nil, problem(node.Key, "STEP_OUTPUT_INVALID") + } + } + if validateOutput(node, output) != nil { + return false, nil, problem(node.Key, "STEP_OUTPUT_INVALID") + } + if completed { + original, err := jsonvalue.Canonical(row.Output) + if err != nil { + return false, nil, problem(node.Key, "STEP_OUTPUT_INVALID") + } + current, err := jsonvalue.Canonical(output) + if err != nil || !bytes.Equal(original, current) { + return false, nil, problem(node.Key, "COMPOSITE_OUTPUT_MISMATCH") + } + return true, copyJSON(row.Output), nil + } + row.Status = weave.StepSucceeded + row.Output = copyJSON(output) + if !s.add(Change{Key: node.Key, Kind: node.Kind, Operation: "COMPLETE", State: row}) { + return false, nil, nil + } + return true, output, nil +} +func (s *planner) skip(block definition.Block) error { + for _, node := range block.Steps { + s.visited[node.Key] = true + row, ok := s.states[node.Key] + if ok && row.Status != weave.StepBlocked && row.Status != weave.StepSkipped { + return problem(node.Key, "UNSELECTED_STEP_HAS_EXECUTION") + } + if !ok || row.Status == weave.StepBlocked { + if s.verifying > 0 { + return problem(node.Key, "INCOMPLETE_COMPOSITE_CHILD") + } + if !s.add(Change{Key: node.Key, Kind: node.Kind, Operation: "SKIP", State: State{Status: weave.StepSkipped}}) { + return nil + } + } + if node.Branch != nil { + for _, c := range node.Branch.Cases { + if err := s.skip(c.Body); err != nil { + return err + } + } + if err := s.skip(*node.Branch.Default); err != nil { + return err + } + } + if node.Parallel != nil { + for _, b := range node.Parallel.Branches { + if err := s.skip(b.Body); err != nil { + return err + } + } + } + } + return nil +} diff --git a/weave/engine/plan_test.go b/weave/engine/plan_test.go new file mode 100644 index 0000000..c0e5a77 --- /dev/null +++ b/weave/engine/plan_test.go @@ -0,0 +1,449 @@ +package engine + +import ( + "bytes" + "encoding/json" + "errors" + "reflect" + "strings" + "sync" + "testing" + + core "git.yuntuo.hk/k2safe/orbit/model" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +func val(raw string) definition.Value { + return definition.Value{Kind: "LITERAL", Literal: json.RawMessage(raw)} +} +func ref(source string, path ...string) definition.Value { + return definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: source, Path: path}} +} +func block(typ definition.Type, value definition.Value, steps ...definition.Step) definition.Block { + return definition.Block{OutputType: typ, Output: value, Steps: steps} +} +func task(queue core.ID, input definition.Value) definition.Task { + return definition.Task{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", InputType: definition.Type{Kind: "NUMBER"}, Input: input, ResultType: definition.Type{Kind: "NUMBER"}, TimeoutMS: 5000, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 1000}, RecoverySafety: pulse.RecoveryReconcileOnly} +} +func programSpec() definition.Specification { + number := definition.Type{Kind: "NUMBER"} + left, right := ref("input", "n"), val(`9007199254740992`) + work := task("11111111111111111111111111111111", ref("input", "n")) + a, b := work, work + fallback := block(number, val(`0`), definition.Step{Key: "unused", Kind: "TASK", Task: &work}) + choose := definition.Step{Key: "choice", Kind: "BRANCH", Branch: &definition.Branch{Cases: []definition.BranchCase{{When: definition.Predicate{Op: "GT", Left: &left, Right: &right}, Body: block(number, ref("selected"), definition.Step{Key: "selected", Kind: "TASK", Task: &work})}}, Default: &fallback}} + a.Input = ref("choice") + b.Input = ref("choice") + joined := definition.Step{Key: "outputs", Kind: "PARALLEL", Parallel: &definition.Parallel{Branches: []definition.ParallelBranch{{Name: "first", Body: block(number, ref("a"), definition.Step{Key: "a", Kind: "TASK", Task: &a})}, {Name: "second", Body: block(number, ref("b"), definition.Step{Key: "b", Kind: "TASK", Task: &b})}}}} + out := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"first": number, "second": number}} + return definition.Specification{SchemaVersion: 1, InputType: definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"n": number}}, TimeoutMS: 30000, FailurePolicy: "REVIEW", Body: block(out, ref("outputs"), choose, joined)} +} +func snapshot() Snapshot { + return Snapshot{WorkflowID: "22222222222222222222222222222222", BusinessKey: "operation", Input: json.RawMessage(`{"n":9007199254740993}`), Steps: map[string]State{}, AllowNew: true} +} +func compiled(t *testing.T, spec definition.Specification) *Program { + t.Helper() + p, err := Compile(spec) + if err != nil { + t.Fatal(err) + } + return p +} +func persist(t *testing.T, s *Snapshot, d Decision) { + t.Helper() + for _, c := range d.Changes { + s.Steps[c.Key] = c.State + } + raw, err := json.Marshal(s.Steps) + if err != nil { + t.Fatal(err) + } + s.Steps = nil + if err := json.Unmarshal(raw, &s.Steps); err != nil { + t.Fatal(err) + } +} +func succeed(s *Snapshot, key string, raw json.RawMessage) { + row := s.Steps[key] + row.Status = weave.StepSucceeded + row.Output = copyJSON(raw) + s.Steps[key] = row +} + +func TestBudgetedRestartedBranchParallelPlan(t *testing.T) { + for _, budget := range []int{1, 2, 32} { + t.Run(string(rune('A'+budget)), func(t *testing.T) { + spec := programSpec() + s := snapshot() + started := map[string]int{} + var final Decision + for iteration := 0; iteration < 40; iteration++ { + // 每轮重新编译并从JSON恢复步骤；模拟存储适配层可丢弃全部进程内状态。 + p := compiled(t, spec) + before, _ := json.Marshal(s.Steps) + d, err := p.Plan(s, budget) + if err != nil { + t.Fatal(err) + } + after, _ := json.Marshal(s.Steps) + if !bytes.Equal(before, after) { + t.Fatal("planner mutated caller snapshot") + } + if len(d.Changes) > budget { + t.Fatal("step budget exceeded") + } + persist(t, &s, d) + for _, c := range d.Changes { + if c.Task != nil { + started[c.Key]++ + if string(c.Task.Payload) != `9007199254740993` { + t.Fatal("numeric reference changed", string(c.Task.Payload)) + } + succeed(&s, c.Key, c.Task.Payload) + } + } + if d.Complete { + final = d + break + } + } + if !final.Complete || string(final.Output) != `{"first":9007199254740993,"second":9007199254740993}` || !reflect.DeepEqual(started, map[string]int{"selected": 1, "a": 1, "b": 1}) { + t.Fatal("incomplete or duplicated work", final, started) + } + if s.Steps["unused"].Status != weave.StepSkipped { + t.Fatal("unselected path was not skipped") + } + repeated, err := compiled(t, spec).Plan(s, budget) + if err != nil || !repeated.Complete || len(repeated.Changes) != 0 { + t.Fatal("completed replay changed state", err) + } + }) + } +} +func TestBudgetExhaustionStillVisitsActiveParallelBranch(t *testing.T) { + spec := programSpec() + number := definition.Type{Kind: "NUMBER"} + work := task("11111111111111111111111111111111", ref("input", "n")) + left, right := val(`0`), val(`1`) + fallback := block(number, ref("y"), definition.Step{Key: "y", Kind: "TASK", Task: &work}) + leftChoice := definition.Step{Key: "left_choice", Kind: "BRANCH", Branch: &definition.Branch{ + Cases: []definition.BranchCase{{When: definition.Predicate{Op: "EQ", Left: &left, Right: &right}, Body: block(number, val(`0`))}}, Default: &fallback, + }} + joined := definition.Step{Key: "joined", Kind: "PARALLEL", Parallel: &definition.Parallel{Branches: []definition.ParallelBranch{ + {Name: "first", Body: block(number, ref("left_choice"), definition.Step{Key: "x", Kind: "TASK", Task: &work}, leftChoice)}, + {Name: "second", Body: block(number, ref("choice"), spec.Body.Steps[0])}, + }}} + spec.Body = block(spec.Body.OutputType, ref("joined"), joined) + s := snapshot() + d, err := compiled(t, spec).Plan(s, 32) + if err != nil { + t.Fatal(err) + } + persist(t, &s, d) + succeed(&s, "x", json.RawMessage(`1`)) + // 左侧激活组合后耗尽预算，右侧已有分支及在途任务仍须遍历，不能误报不可达。 + d, err = compiled(t, spec).Plan(s, 1) + if err != nil || !d.More || len(d.Changes) != 1 || d.Changes[0].Key != "left_choice" { + t.Fatal("budget prevented visiting existing work", d, err) + } + persist(t, &s, d) + succeed(&s, "selected", json.RawMessage(`2`)) + for i := 0; i < 12; i++ { + d, err = compiled(t, spec).Plan(s, 1) + if err != nil { + t.Fatal(err) + } + persist(t, &s, d) + for _, change := range d.Changes { + if change.Task != nil { + if change.Key != "y" { + t.Fatal("existing task was activated again", change.Key) + } + succeed(&s, "y", json.RawMessage(`3`)) + } + } + if d.Complete { + if string(d.Output) != `{"first":3,"second":2}` { + t.Fatal("invalid parallel output", string(d.Output)) + } + return + } + } + t.Fatal("bounded parallel plan did not complete") +} + +func TestExactConditionsAndValues(t *testing.T) { + for _, tc := range []struct { + a, b, op string + want bool + }{ + {`9007199254740993`, `9007199254740992`, "GT", true}, + {`0.10000000000000000001`, `0.1`, "GT", true}, + {`1e3`, `1000.000`, "EQ", true}, + {`-1e-308`, `0`, "LT", true}, + {`"a "`, `"a"`, "EQ", false}, {`true`, `false`, "NE", true}, {`null`, `null`, "EQ", true}, + } { + a, b := val(tc.a), val(tc.b) + got, err := compare(definition.Predicate{Op: tc.op, Left: &a, Right: &b}, values{}) + if err != nil || got != tc.want { + t.Fatalf("%s %s %s: %v %v", tc.a, tc.op, tc.b, got, err) + } + } + value := definition.Value{Kind: "OBJECT", Fields: map[string]definition.Value{"n": ref("input", "n"), "array": {Kind: "ARRAY", Items: []definition.Value{val(`1.0000000000000000001`), val(`null`)}}}} + raw, err := resolve(value, definition.Type{Kind: "JSON"}, values{"input": json.RawMessage(`{"n":9007199254740993}`)}) + if err != nil || string(raw) != `{"array":[1.0000000000000000001,null],"n":9007199254740993}` { + t.Fatal(string(raw), err) + } + a, b := val(`true`), val(`true`) + condition := definition.Predicate{Op: "NOT", Arguments: []definition.Predicate{{Op: "EQ", Left: &a, Right: &b}}} + ok, err := compare(condition, nil) + if err != nil || ok { + t.Fatal("logical comparison", err) + } +} +func TestSnapshotAndResultFailuresProduceNoChanges(t *testing.T) { + p := compiled(t, programSpec()) + for _, change := range []func(*Snapshot){ + func(s *Snapshot) { s.Input = json.RawMessage(`{"n":"wrong"}`) }, + func(s *Snapshot) { s.Steps["unknown"] = State{Status: weave.StepBlocked} }, + func(s *Snapshot) { + s.Steps["selected"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`1`), Output: json.RawMessage(`"wrong"`)} + }, + func(s *Snapshot) { s.Steps["selected"] = State{Status: weave.StepNeedsReview} }, + func(s *Snapshot) { + s.Steps["choice"] = State{Status: weave.StepActive, Input: json.RawMessage(`{"case":100}`)} + }, + func(s *Snapshot) { + s.Steps["choice"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`{"case":0}`), Output: json.RawMessage(`1`)} + }, + func(s *Snapshot) { s.Steps["unused"] = State{Status: weave.StepActive, Input: json.RawMessage(`1`)} }, + } { + s := snapshot() + change(&s) + d, err := p.Plan(s, 32) + if err == nil || len(d.Changes) != 0 || d.Complete { + t.Fatal("unsafe snapshot advanced", d, err) + } + } + for _, limit := range []int{0, 33} { + if _, err := p.Plan(snapshot(), limit); err == nil { + t.Fatal("unbounded iteration accepted") + } + } +} +func TestPauseFrozenChoiceAndCompiledIsolation(t *testing.T) { + spec := programSpec() + p := compiled(t, spec) + s := snapshot() + s.AllowNew = false + d, err := p.Plan(s, 32) + if err != nil || len(d.Changes) != 0 || d.Complete { + t.Fatal("pause activated work", d, err) + } + s.AllowNew = true + d, err = p.Plan(s, 1) + if err != nil || len(d.Changes) != 1 { + t.Fatal(err) + } + persist(t, &s, d) + s.AllowNew = false + d, err = p.Plan(s, 32) + if err != nil { + t.Fatal(err) + } + for _, c := range d.Changes { + if c.Operation == "ACTIVATE" { + t.Fatal("pause activated branch child") + } + } + persist(t, &s, d) + // 编译之后修改调用方定义和公开副本均不能改变正在解释的固定版本。 + spec.Body.Steps[0].Branch.Cases[0].Body.Steps[0].Task.Handler = "changed" + copy, ok := p.Step("selected") + if !ok { + t.Fatal("missing metadata") + } + copy.Task.Handler = "also.changed" + s.AllowNew = true + d, err = p.Plan(s, 32) + if err != nil { + t.Fatal(err) + } + found := false + for _, c := range d.Changes { + if c.Task != nil { + found = true + if c.Task.Handler != "object.prepare" { + t.Fatal("compiled program changed") + } + } + } + if !found { + t.Fatal("frozen branch did not resume") + } + // 已保存的默认分支选择即便重新求值会匹配case，也必须继续原选择。 + s = snapshot() + s.Steps["choice"] = State{Status: weave.StepActive, Input: json.RawMessage(`{"case":-1}`)} + d, err = p.Plan(s, 32) + if err != nil { + t.Fatal(err) + } + for _, c := range d.Changes { + if c.Task != nil && c.Key != "unused" { + t.Fatal("branch was reevaluated") + } + } +} +func TestWaitAndPollPreparationAndRuntimeLimits(t *testing.T) { + spec := programSpec() + str := definition.Type{Kind: "STRING"} + status := definition.Type{Kind: "STRING", Enum: []string{"PENDING", "DONE"}} + spec.InputType = str + signal := definition.Signal{Name: "job.final", PayloadVersion: "v1", Correlation: ref("input"), PayloadType: status, TimeoutMS: 10000} + query := task("11111111111111111111111111111111", val(`1`)) + query.ResultType = status + poll := definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}, FinalSignal: &signal} + spec.Body = block(status, ref("poll"), definition.Step{Key: "signal", Kind: "SIGNAL", Signal: &signal}, definition.Step{Key: "timer", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 1000}}, definition.Step{Key: "poll", Kind: "POLL", Poll: &poll}) + p := compiled(t, spec) + s := snapshot() + s.Input = json.RawMessage(`"operation"`) + d, err := p.Plan(s, 32) + if err != nil || len(d.Changes) != 1 || d.Changes[0].Wait.Correlation != "operation" || d.Changes[0].Task != nil { + t.Fatal("signal", d, err) + } + persist(t, &s, d) + again, err := p.Plan(s, 32) + if err != nil || len(again.Changes) != 0 { + t.Fatal("wait restarted", err) + } + succeed(&s, "signal", json.RawMessage(`"DONE"`)) + d, err = p.Plan(s, 32) + if err != nil || d.Changes[0].Wait.TimeoutMS != 1000 { + t.Fatal("timer", err) + } + persist(t, &s, d) + succeed(&s, "timer", json.RawMessage(`null`)) + d, err = p.Plan(s, 32) + if err != nil || d.Changes[0].Task == nil || d.Changes[0].Wait.Kind != "POLL" { + t.Fatal("poll", err) + } + s = snapshot() + s.Input = json.RawMessage(`""`) + if d, err := p.Plan(s, 32); err == nil || len(d.Changes) != 0 { + t.Fatal("empty runtime correlation accepted") + } + s.Input, _ = json.Marshal(strings.Repeat("x", 513)) + if _, err := p.Plan(s, 32); err == nil { + t.Fatal("unbounded correlation accepted") + } + // 同一引用同时用作两个Pulse键，不能因指针相同漏掉一个键。 + shared := ref("key") + q := task("11111111111111111111111111111111", val(`1`)) + q.BusinessKey = &shared + q.ConcurrencyKey = &shared + prepared, err := prepareTask(q, values{"key": json.RawMessage(`"operation"`)}) + if err != nil || prepared.BusinessKey != "operation" || prepared.ConcurrencyKey != "operation" { + t.Fatal(prepared, err) + } +} +func TestParallelCompletionIntegrityAndConcurrentPlanning(t *testing.T) { + spec := programSpec() + p := compiled(t, spec) + s := snapshot() + for i := 0; i < 15; i++ { + d, err := p.Plan(s, 32) + if err != nil { + t.Fatal(err) + } + persist(t, &s, d) + for _, c := range d.Changes { + if c.Task != nil { + succeed(&s, c.Key, c.Task.Payload) + } + } + if d.Complete { + break + } + } + if !reflect.DeepEqual(s.Steps["outputs"].Status, weave.StepSucceeded) { + t.Fatal("not completed") + } + // 完成的组合不能掩盖未结清子步骤或伪造输出。 + broken := snapshot() + for k, v := range s.Steps { + broken.Steps[k] = v + } + row := broken.Steps["a"] + row.Status = weave.StepActive + row.Output = nil + broken.Steps["a"] = row + _, err := p.Plan(broken, 32) + var issue *Problem + if !errors.As(err, &issue) || issue.Code != "INCOMPLETE_COMPOSITE_CHILD" { + t.Fatal("incomplete child hidden", err) + } + broken.Steps["a"] = s.Steps["a"] + row = broken.Steps["outputs"] + row.Output = json.RawMessage(`{"first":0,"second":0}`) + broken.Steps["outputs"] = row + if _, err := p.Plan(broken, 32); err == nil { + t.Fatal("fabricated composite result accepted") + } + var wg sync.WaitGroup + for i := 0; i < 16; i++ { + wg.Add(1) + go func() { + defer wg.Done() + d, err := p.Plan(s, 32) + if err != nil || !d.Complete || len(d.Changes) != 0 { + t.Error("concurrent immutable plan", err) + } + }() + } + wg.Wait() +} + +func TestUnreachableExecutionsAndAggregateOutputLimit(t *testing.T) { + p := compiled(t, programSpec()) + s := snapshot() + s.Steps["outputs"] = State{Status: weave.StepActive, Input: json.RawMessage(`{}`)} + if d, err := p.Plan(s, 32); err == nil || len(d.Changes) != 0 { + t.Fatal("future execution was ignored") + } + large, _ := json.Marshal(strings.Repeat("x", 140000)) + expr := definition.Value{Kind: "OBJECT", Fields: map[string]definition.Value{"a": ref("value"), "b": ref("value")}} + if _, err := resolve(expr, definition.Type{Kind: "JSON"}, values{"value": large}); err == nil { + t.Fatal("aggregate output limit ignored") + } +} + +func TestPollCompletionMustHaveFinalSuccess(t *testing.T) { + spec := programSpec() + query := task("11111111111111111111111111111111", val(`1`)) + query.ResultType = definition.Type{Kind: "STRING", Enum: []string{"WORKING", "DONE", "FAILED", "REVIEW"}} + poll := definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, Pending: []string{"WORKING"}, Succeeded: []string{"DONE"}, Failed: []string{"FAILED"}, Review: []string{"REVIEW"}} + spec.Body = block(query.ResultType, ref("poll"), definition.Step{Key: "poll", Kind: "POLL", Poll: &poll}) + p := compiled(t, spec) + for status, want := range map[string]PollOutcome{"WORKING": PollPending, "DONE": PollSucceeded, "FAILED": PollFailed, "REVIEW": PollNeedsReview} { + raw, _ := json.Marshal(status) + out, err := p.ClassifyPoll("poll", raw) + if err != nil || out != want { + t.Fatal(status, out, err) + } + s := snapshot() + s.Steps["poll"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`1`), Output: raw} + decision, err := p.Plan(s, 32) + if status == "DONE" { + if err != nil || !decision.Complete { + t.Fatal("final success rejected", err) + } + } else if err == nil || len(decision.Changes) != 0 { + t.Fatal("nonfinal poll result advanced") + } + } + if _, err := p.ClassifyPoll("poll", json.RawMessage(`"unexpected"`)); err == nil { + t.Fatal("unknown status accepted") + } +} diff --git a/weave/engine/poll.go b/weave/engine/poll.go new file mode 100644 index 0000000..6eebc7e --- /dev/null +++ b/weave/engine/poll.go @@ -0,0 +1,54 @@ +package engine + +import ( + "encoding/json" + + "git.yuntuo.hk/k2safe/orbit/weave/definition" +) + +// PollOutcome 仅分类一次成功查询的业务结果，不代表查询请求的Pulse执行状态。 +type PollOutcome string + +const ( + PollPending PollOutcome = "PENDING" + PollSucceeded PollOutcome = "SUCCEEDED" + PollFailed PollOutcome = "FAILED" + PollNeedsReview PollOutcome = "NEEDS_REVIEW" +) + +// ClassifyPoll 解释固定定义中的结果映射；轮次/时限/信号竞态仍由持久化运行器决定。 +func (p *Program) ClassifyPoll(key string, raw json.RawMessage) (PollOutcome, error) { + node, ok := p.nodes[key] + if !ok || node.Poll == nil { + return "", problem(key, "NOT_A_POLL_STEP") + } + result, err := classifyPoll(*node.Poll, raw) + if err != nil { + return "", problem(key, "POLL_RESULT_INVALID") + } + return result, nil +} +func classifyPoll(p definition.Poll, raw json.RawMessage) (PollOutcome, error) { + if err := definition.ValidateData(p.Task.ResultType, raw); err != nil { + return "", err + } + status, err := resolve(definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "result", Path: p.StatusPath}}, definition.Type{Kind: "STRING"}, values{"result": raw}) + if err != nil { + return "", err + } + var value string + if err := json.Unmarshal(status, &value); err != nil { + return "", err + } + for _, group := range []struct { + values []string + outcome PollOutcome + }{{p.Pending, PollPending}, {p.Succeeded, PollSucceeded}, {p.Failed, PollFailed}, {p.Review, PollNeedsReview}} { + for _, candidate := range group.values { + if value == candidate { + return group.outcome, nil + } + } + } + return "", problem("", "UNMAPPED_POLL_RESULT") +} diff --git a/weave/engine/program.go b/weave/engine/program.go new file mode 100644 index 0000000..54b24f5 --- /dev/null +++ b/weave/engine/program.go @@ -0,0 +1,164 @@ +package engine + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "unicode/utf8" + + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +// TaskOwnerKind 标识由Weave激活拥有的Pulse任务；TASK及POLL轮次都使用激活ID作为owner_id。 +const TaskOwnerKind = "workflow_activation" + +// Program 持有独立拷贝的已验证规则，可并发读取；不暴露可变的内部映射。 +type Program struct { + spec definition.Specification + nodes map[string]definition.Step +} + +func Compile(spec definition.Specification) (*Program, error) { + if _, err := definition.Check(spec); err != nil { + return nil, err + } + raw, err := jsonwire.Marshal(spec) + if err != nil { + return nil, err + } + p := &Program{nodes: map[string]definition.Step{}} + if err := json.Unmarshal(raw, &p.spec); err != nil { + return nil, err + } + var visit func(definition.Block) + visit = func(b definition.Block) { + for _, s := range b.Steps { + p.nodes[s.Key] = s + if s.Branch != nil { + for _, c := range s.Branch.Cases { + visit(c.Body) + } + visit(*s.Branch.Default) + } + if s.Parallel != nil { + for _, branch := range s.Parallel.Branches { + visit(branch.Body) + } + } + } + } + visit(p.spec.Body) + return p, nil +} + +// Step 返回定义副本，调用方修改不会改变已编译版本。 +func (p *Program) Step(key string) (definition.Step, bool) { + s, ok := p.nodes[key] + if !ok { + return definition.Step{}, false + } + raw, err := jsonwire.Marshal(s) + if err != nil { + return definition.Step{}, false + } + var out definition.Step + if json.Unmarshal(raw, &out) != nil { + return out, false + } + return out, true +} + +// State 是存储适配层核验身份/激活/版本后提供的当前步骤快照，不是直接信任的外部请求。 +type State struct { + Status weave.StepState `json:"status"` // Status 逻辑步骤当前状态，BLOCKED可无激活。 + Input json.RawMessage `json:"input,omitempty"` // Input 激活时冻结的参数或组合决策，重复推进不得重算。 + Output json.RawMessage `json:"output,omitempty"` // Output 仅SUCCEEDED提供，显式null和未提供不同。 +} + +// Snapshot 必须来自同一实例事务内的持久化视图；节点不以内存缓存授予执行权。 +type Snapshot struct { + WorkflowID core.ID // WorkflowID 固定实例标识，加入workflow引用上下文。 + BusinessKey string // BusinessKey 业务追踪键，不代替幂等身份。 + Input json.RawMessage // Input 启动时冻结且已保存的输入。 + Steps map[string]State // Steps 当前激活投影；缺少的静态步骤视为BLOCKED。 + AllowNew bool // AllowNew 已核验暂停、取消、截止时间和门控后是否允许激活新步骤。 +} + +// WaitCommand 只描述需要持久化的等待；存储层用数据库激活时间固定绝对截止。 +type WaitCommand struct { + Kind string // Kind SIGNAL、TIMER或POLL最终回执。 + Name string // Name 业务信号名称；TIMER为空。 + PayloadVersion string // PayloadVersion 信号契约版本；TIMER为空。 + Correlation string // Correlation 激活时冻结的业务操作，非定时等待必须非空。 + TimeoutMS uint64 // TimeoutMS 相对激活时间的等待上界，还受实例绝对截止限制。 +} + +// Change 是一个必须与激活、任务/等待、检查点和审计同事务落地的决策。 +type Change struct { + Key string // Key 对应静态步骤身份。 + Operation string // Operation ACTIVATE、COMPLETE或SKIP。 + Kind string // Kind 原定义节点类型，存储层仍需核对。 + State State // State 变更后的步骤投影。 + Task *service.SubmitInput // Task 已解析和校验的TASK/POLL输入，不包含实际任务ID或执行授权。 + Wait *WaitCommand // Wait 需要持久化的等待信息；TIMER不占业务Worker。 +} + +// Decision 不含数据库副作用；只有存储事务提交后Changes才可用于后续工作。 +type Decision struct { + Changes []Change // Changes 按依赖顺序的有界决策列表。 + Complete bool // Complete 根块所有已选择步骤完成；不能替代存储层的在途结清校验。 + Output json.RawMessage // Output Complete时的根块输出，已按契约验证。 + More bool // More 因本轮步数预算停止，必须保存持久化待续记录。 +} + +// Problem 是可审计的稳定分类，不携带任意业务参数。 +type Problem struct { + Key string + Code string +} + +func (e *Problem) Error() string { return fmt.Sprintf("workflow step %q: %s", e.Key, e.Code) } +func problem(key, code string) error { return &Problem{Key: key, Code: code} } + +// prepareTask 把冻结定义映射为Pulse受理参数；实际截止及额度由存储事务继续限制。 +func prepareTask(t definition.Task, env values) (service.SubmitInput, error) { + raw, err := resolve(t.Input, t.InputType, env) + if err != nil { + return service.SubmitInput{}, err + } + in := service.SubmitInput{QueueID: t.QueueID, Handler: t.Handler, HandlerVersion: t.HandlerVersion, PayloadVersion: t.PayloadVersion, Payload: raw, TaskTimeoutMS: t.TimeoutMS, Retry: t.Retry, RecoverySafety: t.RecoverySafety, Priority: t.Priority} + for _, key := range []struct { + value *definition.Value + target *string + }{{t.BusinessKey, &in.BusinessKey}, {t.ConcurrencyKey, &in.ConcurrencyKey}} { + if key.value != nil { + raw, err := resolve(*key.value, definition.Type{Kind: "STRING"}, env) + if err != nil { + return in, err + } + if err := json.Unmarshal(raw, key.target); err != nil { + return in, err + } + } + } + return service.NormalizeSubmit(in) +} +func prepareSignal(s definition.Signal, kind string, env values) (*WaitCommand, error) { + raw, err := resolve(s.Correlation, definition.Type{Kind: "STRING"}, env) + if err != nil { + return nil, err + } + var correlation string + if json.Unmarshal(raw, &correlation) != nil || correlation == "" || len(correlation) > 512 { + return nil, errors.New("invalid runtime signal correlation") + } + return &WaitCommand{Kind: kind, Name: s.Name, PayloadVersion: s.PayloadVersion, Correlation: correlation, TimeoutMS: s.TimeoutMS}, nil +} +func validBusinessKey(s string) bool { + return utf8.ValidString(s) && utf8.RuneCountInString(s) <= 191 && !strings.ContainsRune(s, 0) +} diff --git a/weave/model/inbox_event.go b/weave/model/inbox_event.go index 42d06eb..e2b46e7 100644 --- a/weave/model/inbox_event.go +++ b/weave/model/inbox_event.go @@ -15,6 +15,14 @@ type InboxEvent struct { WorkflowID core.ID `db:"workflow_id"` // EventID 来源持久化事件标识，用于消费去重。 EventID core.ID `db:"event_id"` + // SourceTaskID 来源Pulse任务；只有迁移前未核实的历史记录可空，不能据此重新推进。 + SourceTaskID *core.ID `db:"source_task_id"` + // SourceActivationID 事件发生时的激活身份，消费时需核对是否仍为当前激活；未核实历史可空。 + SourceActivationID *core.ID `db:"source_activation_id"` + // SourceRevision 来源任务状态版本，大于零，同一任务版本只交付一次；未核实历史可空。 + SourceRevision *uint64 `db:"source_revision"` + // Snapshot 当时的版本化结果JSON文本，独立于外部订阅日志和当前任务；未核实历史为空。 + Snapshot core.JSON `db:"snapshot"` // State 当前状态；只允许定义的状态转换并配合事务校验。 State InboxState `db:"state"` // AvailableAt 最早可被领取或处理的时间，统一 UTC。