diff --git a/AGENTS.md b/AGENTS.md index 575c77c..37c3f29 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,7 +31,10 @@ - P4定义发布使用schema_version=1有限结构,词法作用域禁止跨分支/前向读取,任务策略显式冻结;SDK类型独立。发布按认证→应用额度→操作身份→定义→排序队列持锁;原键与同版本去重仍校验当前权限。定义/版本/revision/审计/回执同事务;同版本不同内容不得覆盖。保留精确数字,校验存储原文与结构摘要;不得把发布通过或DSL支持误称为实例执行已实现。 -- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归(docs/reports/phase-4-definitions.md)。下一步实例启动/固定版本/首批任务同事务、内部事件接续、顺序与组合推进、持久化等待;P4整体和P5–P7仍未完成。 +- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归(docs/reports/phase-4-definitions.md)。实例启动/固定版本/首批任务同事务的内部存储已补齐;继续内部事件消费接续、顺序与组合自动推进、持久化等待及实例公共SDK;P4整体和P5–P7仍未完成。 - 内部结果交付使用workflow_activation所有者并固定激活ID,TASK经activation.task_id、POLL经轮次校验归属;weave_inbox来源及独立快照与Pulse结果同事务,先于外部结果流行锁,不在结果路径反锁实例/步骤/订阅。迁移000017隔离缺快照历史,禁止用当前任务重建旧事件。 - weave/engine仅计算有限决策,完整实例存储运行器尚未接入。Plan最多32个变更,More必须持久化待续;耗尽变更预算仍须遍历其他已活动分支,不能误判在途工作不可达。冻结分支选择,精确数值,不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务,不能把纯快照往返测试当作真实进程恢复。 + +- 实例内部受理已实现,公共实例HTTP/SDK及消费/等待运行器尚未开放。Start按认证→应用额度→原身份→排序队列建立门控/固定版本/全部步骤/首批工作,所有失败同事务回滚。首批推进递增revision至2,原键返回冻结回执且仍核验当前权限,不重算截止。 +- 实例根步骤`$root`/JOIN不属于用户定义,Steps查询排除它;More对应根步骤版本化持久检查点。TASK直接关联激活task_id,POLL通过轮次关联,实际任务截止必须受实例/查询绝对截止限制。Wait名称及契约版本冻结,旧NULL等待须核实,不能盲目解释;新关联512字节、存储兼容历史764字节。 diff --git a/README.md b/README.md index cb82257..f1c9659 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台, 已完成 P0–P3 平台功能阶段验收:35 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过;[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 、[队列资源控制](docs/api/queues.md) 、[应用共享配额](docs/api/application-resources.md) 、[资源键互斥](docs/api/keyed-concurrency.md) 和 [优先级/节点公平派发](docs/api/priority-and-fairness.md) 已接入服务端和 SDK,[可靠结果订阅](docs/api/subscriptions.md) 已实现持久化核心、HTTP/WSS、管理 SDK 与示例,[消费 SDK](docs/sdk/event-consumer.md) 和 [独立 MySQL Inbox 示例](docs/examples/result-inbox.md) 已实现;[批量任务受理](docs/api/batch-tasks.md) 已提供逐项幂等、失败分类与 SDK 示例;[人工核查与重试](docs/api/task-recovery.md) 已提供带证据的管理接口、关联新任务、审计查询及运行示例;[业务键与时间范围查询](docs/api/task-queries.md) 已提供过滤、分页与SDK场景;[任务来源查询](docs/api/task-source.md) 已提供人工重试父任务、计划触发记录与固定版本;[业务事务Outbox](docs/examples/task-outbox.md) 已提供独立业务库、稳定提交身份、发送恢复与CLI;[相对延迟执行](docs/examples/delayed-task.md)已提供数据库计时及原请求恢复;完整范围核对见[P3验收](docs/reports/phase-3.md),流程引擎与生产验证按 [持续开发计划](docs/08-development-plan.md) 继续实现。 -P4已实现[流程定义与不可变版本发布](docs/api/workflow-definitions.md)、[结构版本1](docs/api/workflow-definition-format.md)及[SDK发布例子](docs/examples/workflow-definition.md)。[有限步骤决策器与内部结果交付](docs/10-workflow-engine-core.md)已实现;实例存储推进、持久化等待和恢复继续开发,发布定义不会启动任务。 +P4已实现[流程定义与不可变版本发布](docs/api/workflow-definitions.md)、[结构版本1](docs/api/workflow-definition-format.md)及[SDK发布例子](docs/examples/workflow-definition.md)。[有限步骤决策器与内部结果交付](docs/10-workflow-engine-core.md)已实现;已补[实例受理及首批工作存储](docs/11-workflow-admission.md)。Inbox消费、持久化等待运行器和实例公共API/SDK继续开发,发布定义不会启动任务。 数据库禁止外键,关联一致性通过服务端事务校验,索引、唯一约束和 CHECK 保留;readiness 同时检查迁移及零外键要求。这不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 阶段验收](docs/reports/phase-2.md)、[安全重试](docs/reports/phase-3-safe-retry.md)、[计划调度](docs/reports/phase-3-schedules.md)、[取消与无外键回归](docs/reports/phase-3-cancellation.md) 及 [队列资源回归](docs/reports/phase-3-queue-resources.md)。 @@ -32,6 +32,7 @@ internal/timezone/ 随计划冻结的内置时区数据快照 weave/model/ 定义、实例、步骤、信号和查询轮次 weave/definition/ 有限结构/类型校验与定义发布契约 weave/engine/ 不可变规则的有限步骤决策与精确结果映射 +weave/instance/ 内部实例受理/查询契约,公共运行API继续开发 internal/config/ 配置校验 internal/storage/mysql/ 连接、事务、迁移及集成测试 internal/transport/httpserver/ HTTP 接入 diff --git a/docs/06-workflow-spec.md b/docs/06-workflow-spec.md index b0952fc..da16772 100644 --- a/docs/06-workflow-spec.md +++ b/docs/06-workflow-spec.md @@ -1,6 +1,6 @@ # Weave 流程编排功能与设计 -> 版本 0.3 · 2026-09-18。首版功能范围已确认。定义结构、发布校验与不可变版本已实现,见[定义格式](api/workflow-definition-format.md)和[发布API](api/workflow-definitions.md);已补[步骤决策与内部结果交付](10-workflow-engine-core.md),下述实例存储推进、等待、查询与管理仍待P4–P5实现验收。 +> 版本 0.3 · 2026-09-18。首版功能范围已确认。定义结构、发布校验与不可变版本已实现,见[定义格式](api/workflow-definition-format.md)和[发布API](api/workflow-definitions.md);已补[步骤决策与内部结果交付](10-workflow-engine-core.md),[实例受理与元数据查询存储](11-workflow-admission.md)已实现;事件消费、等待运行器、公共API/SDK与管理仍待P4–P5实现验收。 ## 1. 职责与边界 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index f548d7b..fb8ba94 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -191,3 +191,8 @@ delay_ms是普通/批量提交命令字段,首次事务以同一个数据库 ## 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)。 + + +## 实例受理与等待身份(000018) + +Wait新增SignalName和PayloadVersion两个中文注释字段;correlation_key转为VARBINARY(764),新受理仍限制512字节,兼容历史191个四字节字符而不截断。实例business_key使用utf8mb4_0900_bin保留精确尾部空格;仍无外键和级联。35个实体,含公共Record的295个直接持久化字段声明均有中文注释。内部`$root`步骤作为实例级待续检查点,不属于定义静态节点或公共步骤列表。具体初始事务、预算和未接通的运行边界见[实例受理](11-workflow-admission.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 1d9a3dd..5b1aa02 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -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);已补[有限决策器及内部结果交付](reports/phase-4-engine-core.md);实例存储适配、事件消费推进与持久化等待仍待实现 | +| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容;端到端通用流程 | 定义格式/词法类型校验、不可变发布、独立SDK及六类节点定义例子已实现,已通过[子项验收](reports/phase-4-definitions.md);已补[有限决策器及内部结果交付](reports/phase-4-engine-core.md);[实例受理及首批事务](reports/phase-4-admission.md)已实现;事件消费推进、持久化等待运行器和实例公共API/SDK仍待实现 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算;暂停/开始/回执/超时竞争;恢复保留预算;未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行,报告环境/指标/遗留项;容量、RPO/RTO 按真实拓扑验证 | 待开发 | | P7 首版交付核对 | SDK 兼容与发布准备、完整 API 手册、场景示例、全量回归及交付清单 | 新接入者按文档跑通全部示例;所有首版能力可追溯到实现与测试;报告实际生产边界 | 待开发 | diff --git a/docs/10-workflow-engine-core.md b/docs/10-workflow-engine-core.md index 6213c85..1460f1c 100644 --- a/docs/10-workflow-engine-core.md +++ b/docs/10-workflow-engine-core.md @@ -1,6 +1,6 @@ # Weave 执行决策与内部结果交付 -状态:P4 内部基础已实现。已接通Pulse结果到Weave持久化Inbox,并提供有限步骤决策器;**实例启动、实例存储适配、Inbox消费推进、定时/信号运行器及对应公共SDK尚未实现**。不能把本页能力解释成流程已经可以端到端执行。 +状态:P4 内部基础已实现。已接通Pulse结果到Weave持久化Inbox,并提供有限步骤决策器;[实例受理和首批存储适配](11-workflow-admission.md)已补齐。**Inbox消费推进、定时/信号运行器及实例公共API/SDK尚未实现**。不能把本页能力解释成流程已经可以端到端执行。 ## 1. 为什么结果要独立持久化 @@ -74,7 +74,7 @@ v16及以前的未处理记录缺少冻结来源,升级后统一标为QUARANTI ## 4. 后续存储适配必须满足的条件 -此处是尚待实现的要求,不能按纯单测宣称已完成: +以下是存储运行器的完整要求;第1项初始事务已实现并按[受理报告](reports/phase-4-admission.md)验证,其余消费/运行要求尚未完成,不能按纯单测宣称已完成: 1. 启动时认证、固定不可变版本/输入和策略、校验全部队列、计算实例截止,并在同事务创建实例、门控、首批激活/任务及审计回执。相同身份返回原实例。 2. 消费者领取内部事件后核对命名空间、来源任务版本、激活/轮次及当前步骤版本;消费确认、步骤变化、后续任务或持久化待续记录同事务提交。 @@ -82,6 +82,6 @@ v16及以前的未处理记录缺少冻结来源,升级后统一标为QUARANTI 4. AllowNew来自权威实例状态、控制版本、取消、总截止和门控。程序参数本身不是权限证明。任务deadline不得超过实例或查询节点绝对截止。 5. 数据库时间驱动真实等待与超时,按可靠受理时间裁决信号/结果,原截止和预算在恢复时保持。单轮重试与查询轮次不能互相补充次数。 6. 完成实例前核对所有活动工作结清和未知效果;失败/取消需要停止新派发并收集在途结果。人工恢复通过新激活保留历史。 -7. 增加流程来源查询、实例/步骤/信号/操作HTTP和独立SDK、可运行端到端场景、真实节点强杀和竞争回归。以上均属于P4–P5剩余工作。 +7. 增加流程来源查询、实例/步骤/信号/操作HTTP和独立SDK、可运行端到端场景、真实节点强杀和竞争回归。其中未完成项继续属于P4–P5剩余工作。 -当前测试夹具通过SQL构造流程关联,只验证结果事务边界;它不代表公共实例API或运行器已经存在。纯决策器的JSON往返测试验证可从快照重算,也不等同于真实进程恢复验收。 +本页的结果交付专用测试夹具通过SQL构造流程关联,只验证结果事务边界;后续[受理测试](reports/phase-4-admission.md)已使用真实内部Start创建实例,两者均不代表公共实例API或完整运行器已经存在。纯决策器的JSON往返测试验证可从快照重算,也不等同于真实进程恢复验收。 diff --git a/docs/11-workflow-admission.md b/docs/11-workflow-admission.md new file mode 100644 index 0000000..6ae3462 --- /dev/null +++ b/docs/11-workflow-admission.md @@ -0,0 +1,62 @@ +# Weave 实例受理与首批工作 + +状态:内部Go服务与MySQL存储适配已实现实例Start/Get/Steps、固定版本、首批决策落库和可靠结果交付。**尚未开放实例HTTP路由、启动SDK或命令行入口;后台Inbox消费、待续扫描和等待运行器尚未接通。** Weave相关公开SDK目前仅支持定义发布。不能将首批任务可执行解释成整个流程已能自动完成。 + +## 1. 受理事务 + +内部入口为`weave/instance.Service`,存储由`WorkflowInstanceStore`实现。一个Start事务按以下顺序执行: + +1. 验证凭据和`workflows:start`;取得应用额度排他锁,保留现有认证共享锁。 +2. 以命名空间、`workflow.start`、幂等键冻结完整请求;同键不同定义、版本、输入或业务键冲突。 +3. 读取并验证指定定义和不可变版本的原文摘要。所有被引用队列按ID排序,核对当前范围并直接取得排他锁;有业务任务时还需要`tasks:submit`。 +4. 原回执在用量限制检查前返回。当前权限和固定版本归属仍需有效,重报不重算截止、输入、版本或首批工作。 +5. 校验实例输入和未终结实例上限,解析最多32个初始变更;以数据库时间固定总截止。 +6. 同事务创建门控、实例、全部静态步骤、内部根步骤,以及初始激活、任务/等待/轮次或待续记录,写入事件、审计及受理回执。 + +任何后续写入或额度预占失败都回滚整个图,包括首批中已经创建的任务。队列满额可用原身份重试;同一应用的多个队列共享原Pulse额度。队列暂停允许受理,由既有Worker门控阻止领取/开始。 + +流程推进不延长总截止。普通任务的实际deadline为自身任务时限与实例绝对截止的较早者;POLL任务还受查询节点绝对截止限制。冻结的单次尝试和重试策略不因剩余时限变短而重写,由Pulse在实际截止处终止授权。 + +## 2. 首批节点如何持久化 + +| 节点 | 受理时保存的内容 | +| --- | --- | +| TASK | 当前激活、精确输入、所属Pulse任务、流程门控及`workflow_activation`所有者 | +| BRANCH | 首次选择的case索引;未选中节点分批SKIPPED,不创建业务激活 | +| PARALLEL | 父激活、独立分支初始工作;完成汇合仍需要后续消费推进 | +| SIGNAL | WAITING激活、信号名称/内容版本/精确业务关联、固定截止;不分配Worker | +| TIMER | WAITING激活和数据库时间计算的等待截止;不生成业务任务 | +| POLL | 固定输入、查询状态、首轮记录和唯一当前任务;used_rounds=1;可选最终信号等待与查询截止一致 | + +所有关系由事务按namespace核验,不使用外键。TASK激活的task_id直接指向任务;POLL激活不占用该字段,查询任务通过轮次归属。真实Worker完成首批任务时,上一阶段的原子结果交付会写入该实例的Inbox。 + +空流程可在受理事务内完成并保存输出,完成门控关闭;纯等待流程进入WAITING。包含活动TASK/POLL或尚有待续变更的流程保持RUNNING。第一次持久化决策将实例版本从内部创建时的1推进到2,Start回执通常为revision="2";调用者不得假设起始版本或自行递增版本。 + +## 3. 有限推进和查询 + +定义仍最多128个静态节点,每次Plan最多32个变更。存储额外建立不可由研发定义的`$root`/JOIN步骤,保存实例级根状态及当前激活。预算耗尽时,以根步骤ID、当前步骤版本和schema_version=1检查点写入`weave_pending_advances`,不能只依赖内存唤醒。 + +根步骤是内部记录,不出现在Steps查询中。Steps最多返回128项元数据及同一读取边界的实例revision,包含节点、状态、当前激活和直接任务引用;不把全部步骤输入/输出拼成大响应。读取核验固定定义和当前激活归属,缺失或错误关联不会降级为有效结果。 + +Get返回当前实例视图;Start重报返回受理时冻结的原始视图,两者语义不同。查询需要`workflows:read`以及固定版本涉及的全部队列范围。定义发布权限不会自动授予实例执行或读取权限;当前bootstrap的`--workflow-access`仍仅授予定义权限。 + +## 4. 运行边界和模型演进 + +| 项目 | 当前边界 | +| --- | --- | +| 请求输入 | 最多256 KiB;按固定定义严格类型校验,精确数字、显式null与缺失分别处理 | +| 业务键 | 最多191个Unicode字符,精确保留尾部空格;不提供执行幂等 | +| 活跃实例 | 每个namespace最多10000个RUNNING/WAITING/CANCELLING/NEEDS_REVIEW实例;终态不计入;满额仍可恢复原受理回执 | +| 信号关联 | 新受理最多512字节;数据库兼容旧191字符最长764字节,升级不截断历史 | +| 截止 | 数据库时间生成;任务、查询与等待均不超过实例总截止 | +| 自动运行 | 本阶段尚未实现Inbox消费、待续扫描、信号消费、定时到期处理或后续轮询 | + +迁移000018为Wait增加signal_name和payload_version两个带中文注释的可空字段;NULL仅表示TIMER或尚未核实的历史元数据,未来消费者不得把历史NULL自动解释为有效的新信号等待。correlation_key转为精确字节存储;实例business_key改为utf8mb4_0900_bin。35个实体、含公共Record的295个直接持久化字段声明,全部有中文注释,继续保持零外键。 + +## 5. 下一步接入 + +1. 锁定实例、核验来源快照/当前激活,消费确认与后续步骤/任务/待续记录同事务;处理重复、迟到、旧轮次和未知结果。 +2. 实现有界待续扫描与数据库时间驱动的等待,恢复时保留原截止和次数;开始扫描前补齐暂停/截止/凭据所属应用停用的判定。 +3. 接通实例公共HTTP、独立SDK、来源查询和可运行端到端场景,同时实施真实节点退出与接管回归。 + +本阶段验证见[实例受理回归记录](reports/phase-4-admission.md)。纯受理测试不是流程运行验收,P4–P7仍按[完整计划](08-development-plan.md)推进。 diff --git a/docs/reports/phase-4-admission.md b/docs/reports/phase-4-admission.md new file mode 100644 index 0000000..a164857 --- /dev/null +++ b/docs/reports/phase-4-admission.md @@ -0,0 +1,59 @@ +# P4 进展:实例受理与首批事务 + +日期:2026-09-18。本阶段实现实例内部服务/存储Start/Get/Steps,以及初始最多32个变更的原子落库。未开放实例HTTP/SDK;Inbox消费与等待运行器仍未实现,不能按首批任务执行证明整个流程闭环。 + +## 实现 + +- 显式版本、输入契约、当前队列权限、独立启动/读取权限、命名空间未终结实例上限。 +- 原身份回执恢复;固定版本/门控/全部静态步骤/首批任务或等待/轮次/检查点/事件/审计同事务。 +- 六类节点持久化,任务和POLL截止受父绝对截止限制;内部根步骤支持有限推进检查点。 +- 元数据步骤查询核验固定定义与当前激活,实例查询核验输入及冻结策略。 +- 新迁移000018保留旧等待的完整UTF8关联和尾部空格,补齐信号名称及契约版本;无外键,字段提供中文注释。 + +## 验证状态 + +环境为macOS arm64、Go 1.27.1、隔离Docker MySQL 8.4.10;所有数据库为一次性测试库。 + +| 用例 | 当前证据 | +| --- | --- | +| 并发受理 | 12个同键Start只创建一份实例及任务;精确数字、固定版本、原截止和同键冲突验证通过 | +| 真实首批Worker | 服务真实创建任务,Worker开始并完成后恰好一条内部结果;普通任务取消接口不能绕过流程所有权 | +| 六类节点 | TASK/BRANCH/PARALLEL/SIGNAL/TIMER/POLL初始记录、未选中路径、等待不占Worker、512字节关联、128字符名称、轮次和父截止通过 | +| 事务故障与背压 | 队列只能容纳2个但首批需要3个时全部回滚,原键扩容后成功;步骤、等待、任务和审计注入失败均无半套图或额度泄漏 | +| 权限及读取 | 范围撤销后原键和查询拒绝,跨namespace不可见;损坏当前激活和冻结策略的查询拒绝通过 | +| 有限检查点和升级 | 114个静态步骤的32变更首批边界、根检查点及v17等待完整升级通过;相关存储/升级/零外键定向回归连续3轮通过,共59.628秒 | +| 完整检查 | make fmt/test/vet/build通过;完整隔离脚本退出0,存储、SDK、21个示例及进程故障通过;临时容器已清理 | +| 查询关联补强 | 完整脚本运行期间补充Steps直接任务的所有者、门控及固定队列核对,新增损坏owner/queue用例;最新实例受理组race连续3轮通过,共12.673秒,随后完整make检查再次通过 | + +完整脚本测试组耗时如下(秒);最后的只读实例查询补强以上方专门回归为验证证据,未重新运行无关SDK/进程组: + +| 测试组 | 耗时 | +| --- | ---: | +| MySQL存储全量 | 119.723 | +| Worker连接及运行SDK | 91.108 | +| 计划 / 队列资源SDK | 5.651 / 4.479 | +| 批量受理 / 任务查询SDK | 1.378 / 1.412 | +| 订阅管理 / 事件SDK | 33.434 / 52.879 | +| 人工恢复 / 定义发布SDK | 3.406 / 1.356 | +| 独立Inbox / Outbox | 35.611 / 34.673 | +| 应用额度SDK | 1.521 | +| SDK进程故障 | 93.931 | + +共运行19个SDK例子和独立Inbox/Outbox的2个示例,10个SDK进程故障父用例及两个独立模块的进程故障用例通过。它们验证既有Pulse/定义发布链路,不能充当Weave后续自动运行的证据。 + +复现入口: + +```sh +make fmt test vet build +make integration +``` + +本次通过GO参数选择Go 1.27.1、DOCKER参数选择本机Docker。脚本自行建库并清理,普通测试在数据库环境变量缺失时会跳过集成用例,不能代替实际integration结果。 + +首轮数据库回归发现实例字段均未变化时MySQL返回affected_rows=0,不能据此判断实例丢失。已改为每次持久化决策递增实例revision并校验旧版本,首批回执为revision="2"。同时修正测试定义的POLL任务/节点/最终信号时限不一致;生产发布校验保持严格。 + +直接查询完整回归临时数据库:外键数量为0,业务表字段空COMMENT数量为0。代码扫描核对含公共Record共295个持久化字段声明,中文注释无遗漏;808个本地文档链接、53个JSON块及git diff --check通过。 + +## 尚待完成 + +当前测试中的实例由真实内部受理服务创建,首批任务走真实Pulse执行与内部交付;Get不会隐式消费Inbox。尚无流程后继自动推进、定时/信号/轮询恢复或流程进程强杀验收。公共API、SDK启动例子和完整运行器在后续阶段接入。生产容量、跨节点拓扑、清理和运维继续属于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 ed70fe6..ae07a42 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -64,6 +64,8 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { retryFinish(t, g, finished, "SUCCEEDED", false) assertApplicationCounts(t, f, 4, 2, 1) // 还原后续新增表和字段,得到真实 v9 结构;现有任务内容保持不变。 + f.exec(`ALTER TABLE weave_waits DROP COLUMN signal_name, DROP COLUMN payload_version, MODIFY COLUMN correlation_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_as_cs NOT NULL COMMENT '外部业务操作关联键,防止旧信号误用于新工作'`) + f.exec(`ALTER TABLE weave_instances MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_as_cs NULL COMMENT '业务侧关联键,用于检索追踪,不自动提供业务幂等'`) 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)`) diff --git a/internal/storage/mysql/migrations/000018_workflow_wait_identity.up.sql b/internal/storage/mysql/migrations/000018_workflow_wait_identity.up.sql new file mode 100644 index 0000000..7cbc5b5 --- /dev/null +++ b/internal/storage/mysql/migrations/000018_workflow_wait_identity.up.sql @@ -0,0 +1,7 @@ +-- 冻结等待的信号身份和契约;历史NULL记录必须经固定定义核验,不能默认解释为新等待。 +ALTER TABLE weave_waits + ADD COLUMN signal_name VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NULL COMMENT '冻结信号名称;TIMER及待核实历史记录为空', + ADD COLUMN payload_version VARCHAR(32) CHARACTER SET ascii COLLATE ascii_bin NULL COMMENT '冻结信号内容契约版本;TIMER及待核实历史记录为空', + MODIFY COLUMN correlation_key VARBINARY(764) NOT NULL COMMENT '精确UTF8业务关联且保留尾部空格;新受理最多512字节,兼容历史191字符最多764字节,TIMER为空串'; + +ALTER TABLE weave_instances MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NULL COMMENT '精确业务追踪键,保留尾部空格,不提供业务幂等'; diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index 8e9a960..ebb26db 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -217,6 +217,11 @@ func businessKeyValue(key string) *string { // insertAcceptedTask 要求调用方已持有应用额度和队列锁,并将身份、事件及审计同事务提交。 func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID) (service.TaskView, error) { + return insertAcceptedTaskUntil(ctx, tx, ns, taskID, in, retryOf, nil) +} + +// insertAcceptedTaskUntil 在原任务时限基础上收紧绝对父截止,不重写冻结重试策略。 +func insertAcceptedTaskUntil(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID, parentDeadline *time.Time) (service.TaskView, error) { now, err := dbNow(ctx, tx) if err != nil { return service.TaskView{}, err @@ -227,6 +232,9 @@ func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in available = *in.AvailableAt } deadline := now.Add(time.Duration(in.TaskTimeoutMS) * time.Millisecond) + if parentDeadline != nil { + deadline = minimumTime(deadline, *parentDeadline) + } if available.Before(now.Add(-time.Second)) || !available.Before(deadline) { return service.TaskView{}, fault.Invalid("available_at must be current/future and before the task deadline") } diff --git a/internal/storage/mysql/workflow_apply.go b/internal/storage/mysql/workflow_apply.go new file mode 100644 index 0000000..77e9ecf --- /dev/null +++ b/internal/storage/mysql/workflow_apply.go @@ -0,0 +1,272 @@ +package mysql + +import ( + "context" + "database/sql" + "errors" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/engine" + "git.yuntuo.hk/k2safe/orbit/weave/instance" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +func createWorkflowSteps(ctx context.Context, tx *sql.Tx, g *workflowGraph, now time.Time) error { + create := func(key, kind string) error { + id, err := core.NewID() + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, "INSERT INTO weave_steps(namespace_id,id,workflow_id,`key`,kind,created_at,updated_at) VALUES (?,?,?,?,?,?,?)", g.view.NamespaceID, id, g.view.ID, key, kind, now, now); err != nil { + return err + } + g.steps[key] = &workflowStep{view: instance.StepView{ID: id, Key: key, Kind: weave.NodeKind(kind), State: weave.StepBlocked, Revision: 1}} + return nil + } + var visit func(definition.Block) error + visit = func(b definition.Block) error { + for _, s := range b.Steps { + if err := create(s.Key, s.Kind); err != nil { + return err + } + if s.Branch != nil { + for _, c := range s.Branch.Cases { + if err := visit(c.Body); err != nil { + return err + } + } + if err := visit(*s.Branch.Default); err != nil { + return err + } + } + if s.Parallel != nil { + for _, branch := range s.Parallel.Branches { + if err := visit(branch.Body); err != nil { + return err + } + } + } + } + return nil + } + if err := visit(g.version.Specification.Body); err != nil { + return err + } + if err := create(workflowRootKey, "JOIN"); err != nil { + return err + } + root := g.steps[workflowRootKey] + activation, err := core.NewID() + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,input,created_at,updated_at) VALUES (?,?,?,?,1,'ACTIVE',?,?,?)`, g.view.NamespaceID, activation, g.view.ID, root.view.ID, string(g.view.Input), now, now); err != nil { + return err + } + r, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state='ACTIVE',current_activation_id=? WHERE namespace_id=? AND id=?`, activation, g.view.NamespaceID, root.view.ID) + if err := exactlyOne(r, err); err != nil { + return err + } + root.view.State, root.view.ActivationID, root.input = weave.StepActive, &activation, g.view.Input + return nil +} + +// applyWorkflowDecision 调用方必须已按应用→排序队列→实例持锁,结果和待续状态一并提交。 +// 初始受理的实例行尚未对其他事务可见,同样由应用锁保护,不在函数内重排父级锁。 +func applyWorkflowDecision(ctx context.Context, tx *sql.Tx, g *workflowGraph, d engine.Decision, now time.Time) error { + if len(d.Changes) > 32 { + return errors.New("workflow change budget exceeded") + } + for _, c := range d.Changes { + row, ok := g.steps[c.Key] + if !ok || string(row.view.Kind) != c.Kind { + return errors.New("workflow decision step mismatch") + } + switch c.Operation { + case "ACTIVATE": + if row.view.State != weave.StepBlocked || row.view.ActivationID != nil { + return errors.New("workflow step already activated") + } + if err := activateWorkflowStep(ctx, tx, g, row, c, now); err != nil { + return err + } + case "SKIP": + if row.view.State != weave.StepBlocked || row.view.ActivationID != nil || c.State.Status != weave.StepSkipped { + return errors.New("workflow skip state invalid") + } + case "COMPLETE": + if row.view.ActivationID == nil || (row.view.State != weave.StepActive && row.view.State != weave.StepWaiting) || c.State.Status != weave.StepSucceeded { + return errors.New("workflow completion state invalid") + } + r, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state='SUCCEEDED',output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state IN ('ACTIVE','WAITING')`, string(c.State.Output), now, now, g.view.NamespaceID, row.view.ActivationID, g.view.ID, row.view.ID) + if err := exactlyOne(r, err); err != nil { + return err + } + default: + return errors.New("unsupported workflow decision") + } + r, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state=?,current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND revision=?`, c.State.Status, row.view.ActivationID, now, g.view.NamespaceID, row.view.ID, g.view.ID, row.view.Revision) + if err := exactlyOne(r, err); err != nil { + return err + } + row.view.State, row.input, row.output = c.State.Status, c.State.Input, c.State.Output + row.view.Revision++ + if err := emitEvent(ctx, tx, g.view.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": g.view.ID, "key": c.Key, "state": c.State.Status, "activation_id": row.view.ActivationID}); err != nil { + return err + } + } + root := g.steps[workflowRootKey] + if root == nil || root.view.ActivationID == nil { + return errors.New("workflow root missing") + } + state := weave.InstanceRunning + var completed *time.Time + var output any + rootState := weave.StepActive + if d.Complete { + if d.More { + return errors.New("workflow complete decision has pending work") + } + for key, row := range g.steps { + if key != workflowRootKey && row.view.State != weave.StepSucceeded && row.view.State != weave.StepSkipped { + return errors.New("workflow completion has unfinished steps") + } + } + state, rootState, completed, output = weave.InstanceSucceeded, weave.StepSucceeded, &now, string(d.Output) + } else if !d.More { + waiting, active := false, false + for key, row := range g.steps { + if key == workflowRootKey { + continue + } + if row.view.State == weave.StepWaiting { + waiting = true + } + if row.view.State == weave.StepActive && (row.view.Kind == weave.NodeTask || row.view.Kind == weave.NodePoll) { + active = true + } + } + if waiting && !active { + state = weave.InstanceWaiting + } + } + r, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state=?,output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state='ACTIVE'`, rootState, output, completed, now, g.view.NamespaceID, root.view.ActivationID, g.view.ID, root.view.ID) + if err := exactlyOne(r, err); err != nil { + return err + } + r, err = tx.ExecContext(ctx, `UPDATE weave_steps SET state=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, rootState, now, g.view.NamespaceID, root.view.ID, root.view.Revision) + if err := exactlyOne(r, err); err != nil { + return err + } + root.view.Revision++ + root.view.State = rootState + // 包括首批决策在内,每次持久化推进递增实例版本;不能以未改变字段的affected_rows判断存在性。 + r, err = tx.ExecContext(ctx, `UPDATE weave_instances SET state=?,output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, state, output, completed, now, g.view.NamespaceID, g.view.ID, g.view.Revision) + if err := exactlyOne(r, err); err != nil { + return err + } + if d.Complete { + if _, err := tx.ExecContext(ctx, `UPDATE pulse_execution_gates SET paused=TRUE,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, g.view.NamespaceID, g.gate); err != nil { + return err + } + } + if d.More { + id, err := core.NewID() + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at,created_at,updated_at) VALUES (?,?,?,?,?,'{"schema_version":1,"kind":"ROOT"}',?,?,?) ON DUPLICATE KEY UPDATE expected_step_revision=VALUES(expected_step_revision),checkpoint=VALUES(checkpoint),available_at=VALUES(available_at),revision=revision+1,updated_at=VALUES(updated_at)`, g.view.NamespaceID, id, g.view.ID, root.view.ID, root.view.Revision, now, now, now) + if err != nil { + return err + } + } else { + if _, err := tx.ExecContext(ctx, `DELETE FROM weave_pending_advances WHERE namespace_id=? AND workflow_id=? AND step_id=?`, g.view.NamespaceID, g.view.ID, root.view.ID); err != nil { + return err + } + } + g.view.State = state + g.view.Revision++ + g.view.CompletedAt = completed + g.view.Output = d.Output + return nil +} + +func activateWorkflowStep(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, c engine.Change, now time.Time) error { + id, err := core.NewID() + if err != nil { + return err + } + var taskID *core.ID + if c.Task != nil { + v, err := core.NewID() + if err != nil { + return err + } + taskID = &v + } + var direct *core.ID + if c.Kind == "TASK" { + direct = taskID + } + if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,task_id,input,created_at,updated_at) VALUES (?,?,?,?,1,?,?,?,?,?)`, g.view.NamespaceID, id, g.view.ID, row.view.ID, c.State.Status, direct, string(c.State.Input), now, now); err != nil { + return err + } + row.view.ActivationID = &id + row.view.TaskID = direct + deadline := g.view.DeadlineAt + if c.Kind == "POLL" { + node, ok := g.program.Step(c.Key) + if !ok || node.Poll == nil || taskID == nil { + return errors.New("workflow poll decision incomplete") + } + p := node.Poll + deadline = minimumTime(deadline, now.Add(time.Duration(p.TimeoutMS)*time.Millisecond)) + pollID, err := core.NewID() + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `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,created_at,updated_at) VALUES (?,?,?,?,?,1,?,?,?,?,?,?)`, g.view.NamespaceID, pollID, g.view.ID, id, p.MaxRounds, p.Task.Retry.MaxAttempts, p.IntervalMS, deadline, taskID, now, now); err != nil { + return err + } + roundID, err := core.NewID() + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id,created_at,updated_at) VALUES (?,?,?,?,1,?,?,?)", g.view.NamespaceID, roundID, g.view.ID, id, taskID, now, now); err != nil { + return err + } + } + if c.Wait != nil { + waitID, err := core.NewID() + if err != nil { + return err + } + until := minimumTime(deadline, now.Add(time.Duration(c.Wait.TimeoutMS)*time.Millisecond)) + var name, version any + if c.Wait.Kind != "TIMER" { + name, version = c.Wait.Name, c.Wait.PayloadVersion + } + if _, err := tx.ExecContext(ctx, `INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at,signal_name,payload_version,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, g.view.NamespaceID, waitID, g.view.ID, id, c.Wait.Kind, []byte(c.Wait.Correlation), until, name, version, now, now); err != nil { + return err + } + } + if c.Task != nil { + if taskID == nil { + return errors.New("workflow task identity missing") + } + task, err := insertAcceptedTaskUntil(ctx, tx, g.view.NamespaceID, *taskID, *c.Task, nil, &deadline) + if err != nil { + return err + } + r, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET owner_kind=?,owner_id=?,gate_id=? WHERE namespace_id=? AND id=? AND owner_kind IS NULL`, engine.TaskOwnerKind, id, g.gate, g.view.NamespaceID, *taskID) + if err := exactlyOne(r, err); err != nil { + return err + } + if err := emitEvent(ctx, tx, g.view.NamespaceID, *taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": task.ID, "state": task.State, "workflow_id": g.view.ID, "activation_id": id}); err != nil { + return err + } + } + return nil +} diff --git a/internal/storage/mysql/workflow_instances.go b/internal/storage/mysql/workflow_instances.go new file mode 100644 index 0000000..e7f1c7a --- /dev/null +++ b/internal/storage/mysql/workflow_instances.go @@ -0,0 +1,292 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/engine" + "git.yuntuo.hk/k2safe/orbit/weave/instance" +) + +// WorkflowInstanceStore 以应用额度锁串行受理,在一个事务内建立全部归属及初始工作。 +// 当前为内部存储适配层,公共HTTP和SDK在消费运行器接通后开放。 +type WorkflowInstanceStore struct{ *TaskStore } + +func NewWorkflowInstanceStore(db *sql.DB) *WorkflowInstanceStore { + return &WorkflowInstanceStore{NewTaskStore(db)} +} + +var _ instance.Store = (*WorkflowInstanceStore)(nil) + +// 根步骤是实例级有限推进的检查点;$不属于合法的用户步骤字符,不能与定义冲突。 +const workflowRootKey = "$root" + +type workflowGraph struct { + view instance.View + gate core.ID + version definition.VersionView + program *engine.Program + steps map[string]*workflowStep +} +type workflowStep struct { + view instance.StepView + input json.RawMessage + output json.RawMessage +} +type workflowPolicy struct { + SchemaVersion uint32 `json:"schema_version"` // SchemaVersion 冻结策略快照的结构版本,当前为1。 + VersionID core.ID `json:"version_id"` // VersionID 固定的不可变定义版本,读取时核对实例归属。 + ContentDigest string `json:"content_digest"` // ContentDigest 固定定义的SHA-256摘要,防止策略和规则错配。 + FailurePolicy string `json:"failure_policy"` // FailurePolicy 原定义的失败结清策略,不随新发布版本变化。 + MaxTaskAttempts uint32 `json:"max_task_attempts"` // MaxTaskAttempts 所有路径的静态尝试上界,实际预算由任务与轮次独立保存。 +} + +func readWorkflowInstance(ctx context.Context, tx *sql.Tx, ns, id core.ID, lock bool) (instance.View, core.ID, error) { + var v instance.View + var gate core.ID + q := `SELECT i.id,i.namespace_id,i.definition_id,i.definition_version_id,v.version,i.gate_id,i.state,i.revision,i.paused,i.control_revision,i.input,i.output,COALESCE(i.business_key,''),i.deadline_at,i.cancel_requested_at,i.completed_at,i.created_at,i.updated_at + FROM weave_instances i JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.definition_id=i.definition_id AND v.id=i.definition_version_id + JOIN weave_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id + JOIN pulse_execution_gates g ON g.namespace_id=i.namespace_id AND g.id=i.gate_id + WHERE i.namespace_id=? AND i.id=?` + if lock { + q += ` FOR SHARE` + } + var input, output []byte + err := tx.QueryRowContext(ctx, q, ns, id).Scan(&v.ID, &v.NamespaceID, &v.DefinitionID, &v.DefinitionVersionID, &v.Version, &gate, &v.State, &v.Revision, &v.Paused, &v.ControlRevision, &input, &output, &v.BusinessKey, &v.DeadlineAt, &v.CancelRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt) + v.Input, v.Output = input, output + if errors.Is(err, sql.ErrNoRows) { + err = fault.NotFound + } + return v, gate, err +} +func instanceVersion(ctx context.Context, tx *sql.Tx, v instance.View) (definition.VersionView, error) { + ver, err := readWorkflowVersion(ctx, tx, v.NamespaceID, v.DefinitionID, v.Version) + if err != nil { + return ver, err + } + if ver.ID != v.DefinitionVersionID { + return ver, errors.New("workflow fixed version mismatch") + } + if definition.ValidateData(ver.Specification.InputType, v.Input) != nil { + return ver, errors.New("workflow stored input invalid") + } + if v.State == "SUCCEEDED" && definition.ValidateData(ver.Specification.Body.OutputType, v.Output) != nil { + return ver, errors.New("workflow stored output invalid") + } + var raw []byte + if err := tx.QueryRowContext(ctx, `SELECT policy_snapshot FROM weave_instances WHERE namespace_id=? AND id=?`, v.NamespaceID, v.ID).Scan(&raw); err != nil { + return ver, err + } + var policy workflowPolicy + if json.Unmarshal(raw, &policy) != nil || policy != (workflowPolicy{1, ver.ID, ver.ContentDigest, ver.Specification.FailurePolicy, ver.Analysis.MaxTaskAttempts}) { + return ver, errors.New("workflow frozen policy mismatch") + } + return ver, nil +} +func lockWorkflowQueues(ctx context.Context, tx *sql.Tx, ns core.ID, p auth.Permissions, queues []core.ID) error { + // Checked.Queues始终排序去重。这里直接取排他锁,禁止先共享后升级。 + for _, id := range queues { + if !p.AllowsQueue(id) { + return fault.NotFound + } + if _, _, err := lockQueueAfterApplication(ctx, tx, ns, id); err != nil { + return err + } + } + return nil +} + +func (s *WorkflowInstanceStore) Start(ctx context.Context, t auth.Token, key string, in instance.StartInput) (instance.StartResult, error) { + var out instance.StartResult + err := s.transaction(ctx, t, instance.StartAction, func(tx *sql.Tx, p auth.Permissions) error { + if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil { + return err + } + id, err := core.NewID() + if err != nil { + return err + } + identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "workflow.start", "", key, "workflow", id, in) + if err != nil { + return err + } + if _, err := readWorkflowDefinition(ctx, tx, t.NamespaceID, in.DefinitionID, false); err != nil { + return err + } + version, err := readWorkflowVersion(ctx, tx, t.NamespaceID, in.DefinitionID, in.Version) + if err != nil { + return err + } + if len(version.Analysis.Queues) > 0 && !p.Allows("tasks:submit") { + return fault.Forbidden + } + if err := lockWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues); err != nil { + return err + } + if cached != nil { + if err := json.Unmarshal(cached, &out); err != nil { + return err + } + current, _, err := readWorkflowInstance(ctx, tx, t.NamespaceID, out.Instance.ID, false) + if err != nil { + return err + } + if current.DefinitionVersionID != version.ID || out.Instance.DefinitionVersionID != version.ID || out.Instance.NamespaceID != t.NamespaceID || !current.CreatedAt.Equal(out.Instance.CreatedAt) { + return errors.New("workflow start receipt mismatch") + } + out.Deduplicated = true + return nil + } + if err := definition.ValidateData(version.Specification.InputType, in.Input); err != nil { + return fault.Invalid("workflow input does not match fixed version") + } + var count int + if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM (SELECT id FROM weave_instances WHERE namespace_id=? AND state IN ('RUNNING','WAITING','CANCELLING','NEEDS_REVIEW') LIMIT ?) active_workflows`, t.NamespaceID, instance.MaxUnfinished).Scan(&count); err != nil { + return err + } + if count >= instance.MaxUnfinished { + return fault.New("WORKFLOW_INSTANCE_LIMIT", "namespace unfinished workflow limit reached") + } + program, err := engine.Compile(version.Specification) + if err != nil { + return err + } + decision, err := program.Plan(engine.Snapshot{WorkflowID: id, BusinessKey: in.BusinessKey, Input: in.Input, Steps: map[string]engine.State{}, AllowNew: true}, 32) + if err != nil { + return fault.Invalid("workflow initial step input is invalid") + } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + gate, err := core.NewID() + if err != nil { + return err + } + deadline := now.Add(time.Duration(version.Specification.TimeoutMS) * time.Millisecond) + policy, err := jsonwire.Marshal(workflowPolicy{1, version.ID, version.ContentDigest, version.Specification.FailurePolicy, version.Analysis.MaxTaskAttempts}) + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO pulse_execution_gates(namespace_id,id,created_at,updated_at) VALUES (?,?,?,?)`, t.NamespaceID, gate, now, now); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,business_key,deadline_at,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, id, in.DefinitionID, version.ID, gate, string(in.Input), string(policy), businessKeyValue(in.BusinessKey), deadline, now, now); err != nil { + return err + } + graph := workflowGraph{gate: gate, version: version, program: program, steps: map[string]*workflowStep{}} + graph.view, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false) + if err != nil { + return err + } + if err := createWorkflowSteps(ctx, tx, &graph, now); err != nil { + return err + } + if err := applyWorkflowDecision(ctx, tx, &graph, decision, now); err != nil { + return err + } + out.Instance, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false) + if err != nil { + return err + } + meta := map[string]any{"definition_id": in.DefinitionID, "definition_version_id": version.ID, "state": out.Instance.State, "deadline_at": deadline, "steps": version.Analysis.Steps} + if err := emitEvent(ctx, tx, t.NamespaceID, id, out.Instance.Revision, "workflow", "workflow.started", meta); err != nil { + return err + } + if err := appendAudit(ctx, tx, t, identity, id, "workflow", "workflow.start", "start fixed workflow version", nil, meta); err != nil { + return err + } + return saveIdentity(ctx, tx, t.NamespaceID, identity, out) + }) + return out, err +} + +func (s *WorkflowInstanceStore) Get(ctx context.Context, t auth.Token, id core.ID) (instance.View, error) { + var out instance.View + err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error { + var err error + out, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false) + if err != nil { + return err + } + version, err := instanceVersion(ctx, tx, out) + if err != nil { + return err + } + return authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues) + }) + return out, err +} +func (s *WorkflowInstanceStore) Steps(ctx context.Context, t auth.Token, id core.ID) (instance.StepsView, error) { + out := instance.StepsView{InstanceID: id, Steps: []instance.StepView{}} + err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error { + // 先核对队列,后锁实例;未来运行器遵守应用→队列→实例,读取不得逆序锁队列。 + v, _, err := readWorkflowInstance(ctx, tx, t.NamespaceID, id, false) + if err != nil { + return err + } + version, err := instanceVersion(ctx, tx, v) + if err != nil { + return err + } + if err := authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues); err != nil { + return err + } + var gate core.ID + v, gate, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, true) + if err != nil { + return err + } + out.Revision = v.Revision + program, err := engine.Compile(version.Specification) + if err != nil { + return err + } + rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,a.id,a.state,task.id,task.queue_id FROM weave_steps s LEFT JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id LEFT JOIN pulse_tasks task ON task.namespace_id=s.namespace_id AND task.id=a.task_id AND task.owner_kind='workflow_activation' AND task.owner_id=a.id AND task.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? AND s.`key`<>? ORDER BY s.`key` LIMIT 129", gate, t.NamespaceID, id, workflowRootKey) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var step instance.StepView + var activation *core.ID + var task, queue *core.ID + var state sql.NullString + if err := rows.Scan(&step.ID, &step.Key, &step.Kind, &step.State, &step.Revision, &step.ActivationID, &step.TaskID, &activation, &state, &task, &queue); err != nil { + return err + } + node, ok := program.Step(step.Key) + if !ok || node.Kind != string(step.Kind) { + return errors.New("workflow step definition mismatch") + } + inactive := step.State == "BLOCKED" || step.State == "SKIPPED" + if inactive && step.ActivationID != nil || !inactive && (step.ActivationID == nil || activation == nil || *activation != *step.ActivationID || state.String != string(step.State)) { + return errors.New("workflow current activation mismatch") + } + if !inactive && (step.Kind == "TASK") != (step.TaskID != nil) { + return errors.New("workflow direct task ownership mismatch") + } + if step.TaskID != nil && (task == nil || queue == nil || *task != *step.TaskID || node.Task == nil || *queue != node.Task.QueueID) { + return errors.New("workflow task association invalid") + } + out.Steps = append(out.Steps, step) + } + if err := rows.Err(); err != nil { + return err + } + if len(out.Steps) != int(version.Analysis.Steps) { + return errors.New("workflow step set incomplete") + } + return nil + }) + return out, err +} diff --git a/internal/storage/mysql/workflow_instances_integration_test.go b/internal/storage/mysql/workflow_instances_integration_test.go new file mode 100644 index 0000000..1773540 --- /dev/null +++ b/internal/storage/mysql/workflow_instances_integration_test.go @@ -0,0 +1,378 @@ +package mysql + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "sync" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/instance" +) + +func instanceCount(f *executionFixture, q string, args ...any) int { + f.t.Helper() + var n int + if err := f.db.QueryRowContext(f.ctx, q, args...).Scan(&n); err != nil { + f.t.Fatal(err) + } + return n +} + +func grantWorkflowInstance(t *testing.T, f *executionFixture, queues ...string) { + t.Helper() + if len(queues) == 0 { + queues = []string{"*"} + } + p := auth.Permissions{Actions: []string{instance.StartAction, instance.ReadAction, definition.ReadAction, definition.PublishAction, "tasks:submit", "tasks:read", "tasks:cancel", "workers:connect", "workers:consume"}, Queues: queues} + raw, err := json.Marshal(p) + if err != nil { + t.Fatal(err) + } + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.resources.NamespaceID, f.resources.CredentialID) +} +func publishInstanceSpec(t *testing.T, f *executionFixture, spec definition.Specification) instance.StartInput { + t.Helper() + api := definition.New(NewWorkflowDefinitionStore(f.db)) + c, err := api.Create(f.ctx, f.token, string(executionID(t)), definition.CreateInput{Name: "instance-" + string(executionID(t))}) + if err != nil { + t.Fatal(err) + } + if _, err := api.Publish(f.ctx, f.token, c.Definition.ID, "v1", definition.PublishInput{Version: "v1", Specification: spec, Reason: "verify instance admission"}); err != nil { + t.Fatal(err) + } + return instance.StartInput{DefinitionID: c.Definition.ID, Version: "v1", Input: json.RawMessage(`{"n":9007199254740993}`), BusinessKey: "作业关联 "} +} +func allInitialNodes(queue core.ID) definition.Specification { + spec := workflowTestSpec(queue) + typ := spec.InputType + work := *spec.Body.Steps[0].Task + work.Input = definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "input"}} + signal := definition.Signal{Name: strings.Repeat("s", 128), PayloadVersion: "v1", Correlation: definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`"` + strings.Repeat("x", 511) + ` "`)}, PayloadType: typ, TimeoutMS: 4000} + pollTask := work + pollTask.ResultType = definition.Type{Kind: "STRING", Enum: []string{"PENDING", "DONE"}} + pollSignal := signal + pollSignal.PayloadType = pollTask.ResultType + pollSignal.TimeoutMS = 5000 + poll := definition.Poll{Task: pollTask, MaxRounds: 3, TimeoutMS: 5000, IntervalMS: 1000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}, FinalSignal: &pollSignal} + defaultBlock := definition.Block{Steps: []definition.Step{{Key: "chosen", Kind: "TASK", Task: &work}}, OutputType: typ, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "chosen"}}} + left, right := definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`0`)}, definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`1`)} + branch := definition.Branch{Cases: []definition.BranchCase{{When: definition.Predicate{Op: "EQ", Left: &left, Right: &right}, Body: definition.Block{Steps: []definition.Step{{Key: "skipped", Kind: "TASK", Task: &work}}, OutputType: typ, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "skipped"}}}}}, Default: &defaultBlock} + steps := []definition.Step{{Key: "job", Kind: "TASK", Task: &work}, {Key: "query", Kind: "POLL", Poll: &poll}, {Key: "signal", Kind: "SIGNAL", Signal: &signal}, {Key: "timer", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 5000}}, {Key: "choice", Kind: "BRANCH", Branch: &branch}} + parallel := definition.Parallel{} + out := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{}} + for _, step := range steps { + result := typ + if step.Kind == "POLL" { + result = pollTask.ResultType + } + if step.Kind == "TIMER" { + result = definition.Type{Kind: "NULL"} + } + parallel.Branches = append(parallel.Branches, definition.ParallelBranch{Name: step.Key, Body: definition.Block{Steps: []definition.Step{step}, OutputType: result, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: step.Key}}}}) + out.Fields[step.Key] = result + } + spec.Body = definition.Block{Steps: []definition.Step{{Key: "join", Kind: "PARALLEL", Parallel: ¶llel}}, OutputType: out, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "join"}}} + return spec +} + +func TestIntegrationWorkflowInstanceAdmission(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, stop := context.WithTimeout(context.Background(), 120*time.Second) + defer stop() + 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) + } + api := instance.New(NewWorkflowInstanceStore(db)) + t.Run("concurrent_identity_real_worker_and_fixed_version", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, workflowTestSpec(f.resources.QueueID)) + type result struct { + out instance.StartResult + err error + } + ch := make(chan result, 12) + var wg sync.WaitGroup + for i := 0; i < 12; i++ { + wg.Add(1) + go func() { defer wg.Done(); v, e := api.Start(ctx, f.token, "one-start", in); ch <- result{v, e} }() + } + wg.Wait() + close(ch) + var original instance.View + fresh := 0 + for r := range ch { + if r.err != nil { + t.Fatal(r.err) + } + if !r.out.Deduplicated { + fresh++ + original = r.out.Instance + } + } + if fresh != 1 || original.ID == "" || string(original.Input) != string(in.Input) || original.BusinessKey != in.BusinessKey { + t.Fatal("start identity or precision", fresh, original) + } + if instanceCount(f, `SELECT COUNT(*) FROM weave_instances WHERE namespace_id=?`, f.resources.NamespaceID) != 1 || instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 1 { + t.Fatal("duplicate admission") + } + steps, err := api.Steps(ctx, f.token, original.ID) + if err != nil || len(steps.Steps) != 1 || steps.Steps[0].TaskID == nil { + t.Fatal(steps, err) + } + taskID := *steps.Steps[0].TaskID + if _, err := f.tasks.CancelTask(ctx, f.token, taskID, "direct-cancel", service.CancelInput{ExpectedRevision: 1, Reason: "owned task"}); fault.Code(err) != "STATE_CONFLICT" { + t.Fatal("workflow ownership bypassed", err) + } + ref := retryStart(t, f) + if ref.TaskID != taskID { + t.Fatal("wrong claimed task") + } + retryFinish(t, f, ref, "SUCCEEDED", false) + if instanceCount(f, `SELECT COUNT(*) FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND state='PENDING'`, f.resources.NamespaceID, original.ID, taskID) != 1 { + t.Fatal("real instance result not durably delivered") + } + // 查询和重建存储对象不能隐式消费;下一阶段运行器必须消费已持久化的事件。 + after, err := instance.New(NewWorkflowInstanceStore(db)).Get(ctx, f.token, original.ID) + if err != nil || after.State != "RUNNING" { + t.Fatal(after, err) + } + changed := workflowTestSpec(f.resources.QueueID) + changed.TimeoutMS = 60000 + if _, err := definition.New(NewWorkflowDefinitionStore(db)).Publish(ctx, f.token, in.DefinitionID, "v2", definition.PublishInput{Version: "v2", Specification: changed, Reason: "new release"}); err != nil { + t.Fatal(err) + } + dupe, err := api.Start(ctx, f.token, "one-start", in) + if err != nil || !dupe.Deduplicated || dupe.Instance.ID != original.ID || !dupe.Instance.DeadlineAt.Equal(original.DeadlineAt) || dupe.Instance.DefinitionVersionID != original.DefinitionVersionID { + t.Fatal("replay changed fixed admission", err) + } + in.Input = json.RawMessage(`{"n":9007199254740992}`) + if _, err := api.Start(ctx, f.token, "one-start", in); fault.Code(err) != "IDEMPOTENCY_CONFLICT" { + t.Fatal(err) + } + }) + t.Run("all_node_kinds_freeze_waits_and_poll_deadline", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, allInitialNodes(f.resources.QueueID)) + started, err := api.Start(ctx, f.token, "mixed", in) + if err != nil { + t.Fatal(err) + } + steps, err := api.Steps(ctx, f.token, started.Instance.ID) + if err != nil || len(steps.Steps) != 8 { + t.Fatal(steps, err) + } + if instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 3 || instanceCount(f, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID) != 3 { + t.Fatal("task/wait counts") + } + if instanceCount(f, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=? AND kind IN ('SIGNAL','POLL') AND OCTET_LENGTH(correlation_key)=512 AND OCTET_LENGTH(signal_name)=128 AND payload_version='v1'`, f.resources.NamespaceID) != 2 { + t.Fatal("signal metadata truncated") + } + if instanceCount(f, `SELECT COUNT(*) FROM weave_steps WHERE namespace_id=? AND state='SKIPPED' AND current_activation_id IS NULL`, f.resources.NamespaceID) != 1 { + t.Fatal("unselected path activated") + } + var used, max uint32 + var deadline, taskDeadline, waitDeadline time.Time + var direct *core.ID + if err := db.QueryRowContext(ctx, `SELECT p.used_rounds,p.max_rounds,p.deadline_at,t.deadline_at,w.deadline_at,a.task_id FROM weave_poll_states p JOIN pulse_tasks t ON t.namespace_id=p.namespace_id AND t.id=p.current_task_id JOIN weave_activations a ON a.namespace_id=p.namespace_id AND a.id=p.activation_id JOIN weave_waits w ON w.namespace_id=p.namespace_id AND w.activation_id=p.activation_id WHERE p.namespace_id=?`, f.resources.NamespaceID).Scan(&used, &max, &deadline, &taskDeadline, &waitDeadline, &direct); err != nil { + t.Fatal(err) + } + if used != 1 || max != 3 || direct != nil || !deadline.Equal(taskDeadline) || !deadline.Equal(waitDeadline) || !deadline.Before(started.Instance.DeadlineAt) { + t.Fatal("poll budget or parent deadline not frozen") + } + }) + t.Run("waiting_and_empty_completion", func(t *testing.T) { + for _, empty := range []bool{false, true} { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + spec := workflowTestSpec(f.resources.QueueID) + spec.Body.Steps = nil + spec.Body.Output = definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "input"}} + if !empty { + spec.Body.Steps = []definition.Step{{Key: "wait", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 1000}}} + } + in := publishInstanceSpec(t, f, spec) + v, err := api.Start(ctx, f.token, "start", in) + if err != nil { + t.Fatal(err) + } + if empty { + if v.Instance.State != "SUCCEEDED" || v.Instance.CompletedAt == nil || string(v.Instance.Output) != string(in.Input) { + t.Fatal(v) + } + } else if v.Instance.State != "WAITING" || v.Instance.CompletedAt != nil { + t.Fatal(v) + } + if instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 0 { + t.Fatal("wait allocated worker task") + } + } + }) + t.Run("bounded_initial_decision_persists_root_checkpoint", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + spec := workflowTestSpec(f.resources.QueueID) + work := *spec.Body.Steps[0].Task + left, right := definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`0`)}, definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`1`)} + branch := definition.Branch{Default: &spec.Body} + for i := 0; i < 16; i++ { + body := definition.Block{OutputType: spec.InputType, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "input"}}} + for j := 0; j < 7; j++ { + body.Steps = append(body.Steps, definition.Step{Key: fmt.Sprintf("unused_%d_%d", i, j), Kind: "TASK", Task: &work}) + } + branch.Cases = append(branch.Cases, definition.BranchCase{When: definition.Predicate{Op: "EQ", Left: &left, Right: &right}, Body: body}) + } + // 默认块单独保存,避免构造时让指针指向随后被替换的根块。 + fallback := spec.Body + branch.Default = &fallback + spec.Body = definition.Block{Steps: []definition.Step{{Key: "choose", Kind: "BRANCH", Branch: &branch}}, OutputType: spec.InputType, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "choose"}}} + in := publishInstanceSpec(t, f, spec) + out, err := api.Start(ctx, f.token, "budget", in) + if err != nil { + t.Fatal(err) + } + if instanceCount(f, `SELECT COUNT(*) FROM weave_steps WHERE namespace_id=? AND state='SKIPPED'`, f.resources.NamespaceID) != 31 || instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 0 { + t.Fatal("initial decision exceeded budget") + } + if instanceCount(f, "SELECT COUNT(*) FROM weave_pending_advances p JOIN weave_steps s ON s.namespace_id=p.namespace_id AND s.workflow_id=p.workflow_id AND s.id=p.step_id WHERE p.namespace_id=? AND s.`key`='$root' AND p.expected_step_revision=s.revision", f.resources.NamespaceID) != 1 { + t.Fatal("durable checkpoint missing") + } + steps, err := instance.New(NewWorkflowInstanceStore(db)).Steps(ctx, f.token, out.Instance.ID) + if err != nil || len(steps.Steps) != 114 { + t.Fatal(steps, err) + } + if _, err := api.Start(ctx, f.token, "budget", in); err != nil { + t.Fatal(err) + } + if instanceCount(f, `SELECT COUNT(*) FROM weave_pending_advances WHERE namespace_id=?`, f.resources.NamespaceID) != 1 { + t.Fatal("replay changed checkpoint") + } + }) + t.Run("backpressure_rolls_back_complete_graph", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, allInitialNodes(f.resources.QueueID)) + f.exec(`UPDATE pulse_queues SET max_unfinished_tasks=2 WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.QueueID) + if _, err := api.Start(ctx, f.token, "capacity", in); fault.Code(err) != "QUEUE_FULL" { + t.Fatal(err) + } + for _, table := range []string{"weave_instances", "weave_steps", "weave_activations", "weave_waits", "weave_poll_states", "weave_poll_rounds", "pulse_tasks", "pulse_execution_gates"} { + if instanceCount(f, "SELECT COUNT(*) FROM "+table+" WHERE namespace_id=?", f.resources.NamespaceID) != 0 { + t.Fatal("partial graph", table) + } + } + assertApplicationCounts(t, f, 0, 0, 0) + if instanceCount(f, `SELECT COUNT(*) FROM orbit_idempotency_keys WHERE namespace_id=? AND operation='workflow.start'`, f.resources.NamespaceID) != 0 { + t.Fatal("failed start receipt committed") + } + f.exec(`UPDATE pulse_queues SET max_unfinished_tasks=3 WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.QueueID) + v, err := api.Start(ctx, f.token, "capacity", in) + if err != nil || v.Deduplicated { + t.Fatal(v, err) + } + dupe, err := api.Start(ctx, f.token, "capacity", in) + if err != nil || !dupe.Deduplicated { + t.Fatal("receipt blocked at capacity", err) + } + assertApplicationCounts(t, f, 3, 0, 0) + }) + for _, table := range []string{"weave_steps", "weave_waits", "pulse_tasks", "orbit_audit_events"} { + t.Run("rollback_"+table, func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, allInitialNodes(f.resources.QueueID)) + f.exec("CREATE TRIGGER test_reject_workflow_start BEFORE INSERT ON " + table + " FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected workflow admission failure'") + _, err := api.Start(ctx, f.token, "rollback", in) + f.exec(`DROP TRIGGER test_reject_workflow_start`) + if err == nil { + t.Fatal("injected write accepted") + } + if instanceCount(f, `SELECT COUNT(*) FROM weave_instances WHERE namespace_id=?`, f.resources.NamespaceID) != 0 || instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 0 { + t.Fatal("rollback left resources") + } + assertApplicationCounts(t, f, 0, 0, 0) + if _, err := api.Start(ctx, f.token, "rollback", in); err != nil { + t.Fatal(err) + } + }) + } + t.Run("current_authority_and_foreign_associations", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, workflowTestSpec(f.resources.QueueID)) + v, err := api.Start(ctx, f.token, "authorized", in) + if err != nil { + t.Fatal(err) + } + grantWorkflowInstance(t, f, string(executionID(t))) + for _, run := range []func() error{func() error { _, e := api.Start(ctx, f.token, "authorized", in); return e }, func() error { _, e := api.Get(ctx, f.token, v.Instance.ID); return e }, func() error { _, e := api.Steps(ctx, f.token, v.Instance.ID); return e }} { + if fault.Code(run()) != "NOT_FOUND" { + t.Fatal("scope revoked but data disclosed") + } + } + grantWorkflowInstance(t, f) + in.Input = json.RawMessage(`{"n":"bad"}`) + if _, err := api.Start(ctx, f.token, "invalid", in); fault.Code(err) != "INVALID_ARGUMENT" { + t.Fatal(err) + } + other := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, other) + if _, err := api.Get(ctx, other.token, v.Instance.ID); fault.Code(err) != "NOT_FOUND" { + t.Fatal(err) + } + if _, err := api.Start(ctx, other.token, "cross-namespace", in); fault.Code(err) != "NOT_FOUND" { + t.Fatal(err) + } + }) + t.Run("query_rejects_broken_current_activation_and_policy", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, workflowTestSpec(f.resources.QueueID)) + v, err := api.Start(ctx, f.token, "query", in) + if err != nil { + t.Fatal(err) + } + steps, err := api.Steps(ctx, f.token, v.Instance.ID) + if err != nil { + t.Fatal(err) + } + step := steps.Steps[0] + f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, step.ID) + if _, err := api.Steps(ctx, f.token, v.Instance.ID); err == nil { + t.Fatal("missing activation accepted") + } + f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, step.ActivationID, f.resources.NamespaceID, step.ID) + f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, step.TaskID) + if _, err := api.Steps(ctx, f.token, v.Instance.ID); err == nil { + t.Fatal("wrong task owner accepted") + } + f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, step.ActivationID, f.resources.NamespaceID, step.TaskID) + f.exec(`UPDATE pulse_tasks SET queue_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, step.TaskID) + if _, err := api.Steps(ctx, f.token, v.Instance.ID); err == nil { + t.Fatal("wrong task queue accepted") + } + f.exec(`UPDATE pulse_tasks SET queue_id=? WHERE namespace_id=? AND id=?`, f.resources.QueueID, f.resources.NamespaceID, step.TaskID) + f.exec(`UPDATE weave_instances SET policy_snapshot='{}' WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.Instance.ID) + if _, err := api.Get(ctx, f.token, v.Instance.ID); err == nil { + t.Fatal("missing fixed policy accepted") + } + }) +} diff --git a/internal/storage/mysql/workflow_wait_upgrade_integration_test.go b/internal/storage/mysql/workflow_wait_upgrade_integration_test.go new file mode 100644 index 0000000..6794ff5 --- /dev/null +++ b/internal/storage/mysql/workflow_wait_upgrade_integration_test.go @@ -0,0 +1,106 @@ +package mysql + +import ( + "context" + "os" + "strings" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + driver "github.com/go-sql-driver/mysql" +) + +func TestIntegrationWorkflowWaitUpgrade(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_wait_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 > 17 { + 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) + records := map[core.ID]string{} + for _, value := range []string{"关联é ", strings.Repeat("𠮷", 191)} { + id := executionID(t) + records[id] = value + exec(`INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at) VALUES (?,?,?,?,'SIGNAL',?,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))`, ns, id, workflow, executionID(t), value) + } + exec(`INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,business_key,deadline_at) VALUES (?,?,?,?,?,'{}','{}',?,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))`, ns, workflow, executionID(t), executionID(t), executionID(t), "关联é ") + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if err := CheckSchema(ctx, db); err != nil { + t.Fatal(err) + } + for id, want := range records { + var got, state string + var name, version *string + if err := db.QueryRowContext(ctx, `SELECT correlation_key,state,signal_name,payload_version FROM weave_waits WHERE namespace_id=? AND id=?`, ns, id).Scan(&got, &state, &name, &version); err != nil { + t.Fatal(err) + } + if got != want || state != "WAITING" || name != nil || version != nil { + t.Fatal("legacy wait identity rewritten or truncated") + } + } + var key, collation string + if err := db.QueryRowContext(ctx, `SELECT business_key FROM weave_instances WHERE namespace_id=? AND id=?`, ns, workflow).Scan(&key); err != nil || key != "关联é " { + t.Fatal("business key changed", err) + } + if err := db.QueryRowContext(ctx, `SELECT COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='weave_instances' AND COLUMN_NAME='business_key'`).Scan(&collation); err != nil || collation != "utf8mb4_0900_bin" { + t.Fatal("incorrect key collation", err) + } + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + t.Log("v17 wait identity, 764-byte legacy correlation and trailing spaces preserved; new signal metadata remains unverified NULL; schema has zero foreign keys") +} diff --git a/platform/auth/token.go b/platform/auth/token.go index dc0bfde..291cd21 100644 --- a/platform/auth/token.go +++ b/platform/auth/token.go @@ -58,7 +58,7 @@ type Permissions struct { Queues []string `json:"queues"` } -var supported = map[string]bool{"workflows:definitions:read": true, "workflows:definitions:publish": true, "tasks:review": true, "tasks:retry": true, "tasks:audit:read": true, "events:read": true, "events:manage": true, "events:consume": true, "application:resources:read": true, "application:resources:manage": true, "tasks:submit": true, "tasks:read": true, "tasks:cancel": true, "queues:read": true, "queues:manage": true, "workers:connect": true, "workers:consume": true, "schedules:read": true, "schedules:manage": true} +var supported = map[string]bool{"workflows:start": true, "workflows:read": true, "workflows:definitions:read": true, "workflows:definitions:publish": true, "tasks:review": true, "tasks:retry": true, "tasks:audit:read": true, "events:read": true, "events:manage": true, "events:consume": true, "application:resources:read": true, "application:resources:manage": true, "tasks:submit": true, "tasks:read": true, "tasks:cancel": true, "queues:read": true, "queues:manage": true, "workers:connect": true, "workers:consume": true, "schedules:read": true, "schedules:manage": true} func (p Permissions) Validate() error { if len(p.Actions) == 0 || len(p.Actions) > 32 || len(p.Queues) == 0 || len(p.Queues) > 128 { diff --git a/weave/instance/service.go b/weave/instance/service.go new file mode 100644 index 0000000..2c4a4cd --- /dev/null +++ b/weave/instance/service.go @@ -0,0 +1,113 @@ +// Package instance 定义流程实例的受理与查询边界;执行推进由持久化存储运行器负责。 +package instance + +import ( + "context" + "encoding/json" + "strings" + "time" + "unicode/utf8" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +const StartAction = "workflows:start" +const ReadAction = "workflows:read" +const MaxUnfinished = 10000 + +// StartInput 使用显式版本;业务追踪键不能替代请求幂等键。 +type StartInput struct { + DefinitionID core.ID `json:"definition_id"` // DefinitionID 同命名空间的稳定定义身份。 + Version string `json:"version"` // Version 已发布的固定版本名,不解析latest别名。 + Input json.RawMessage `json:"input" nullable:"true"` // Input 按固定版本验证的输入,显式null与未提供不同。 + BusinessKey string `json:"business_key,omitempty"` // BusinessKey 最多191个Unicode字符,保留尾部空格。 +} + +// View 是实例当前权威视图;StartResult中的View冻结在受理事务,不随重报更新。 +type View struct { + ID core.ID `json:"id"` // ID 实例身份。 + NamespaceID core.ID `json:"namespace_id"` // NamespaceID 授权及数据隔离范围。 + DefinitionID core.ID `json:"definition_id"` // DefinitionID 稳定定义身份。 + DefinitionVersionID core.ID `json:"definition_version_id"` // DefinitionVersionID 固定版本记录身份。 + Version string `json:"version"` // Version 启动时固定的发布版本名。 + State weave.InstanceState `json:"state"` // State 执行状态,受理不代表业务成功。 + Revision uint64 `json:"revision,string"` // Revision 实例并发版本。 + Paused bool `json:"paused"` // Paused 独立暂停标志,不表示在途工作已停止。 + ControlRevision uint64 `json:"control_revision,string"` // ControlRevision 控制操作并发版本。 + Input json.RawMessage `json:"input"` // Input 不变的受理输入,保留数字文本。 + Output json.RawMessage `json:"output,omitempty"` // Output 成功输出;缺失与显式null不同。 + BusinessKey string `json:"business_key"` // BusinessKey 原始追踪键。 + DeadlineAt time.Time `json:"deadline_at"` // DeadlineAt 数据库时间确定的固定总截止。 + CancelRequestedAt *time.Time `json:"cancel_requested_at,omitempty"` // CancelRequestedAt 已保存的取消意图时间。 + CompletedAt *time.Time `json:"completed_at,omitempty"` // CompletedAt 终结时间,非终态为空。 + CreatedAt time.Time `json:"created_at"` // CreatedAt UTC受理时间。 + UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最近状态更新时间。 +} +type StartResult struct { + Instance View `json:"instance"` // Instance 原受理回执的实例快照。 + Deduplicated bool `json:"deduplicated"` // Deduplicated 原键恢复,不新建工作或续期。 +} + +// StepView 只列元数据,避免把最多128个大输入输出聚合为无界响应。 +type StepView struct { + ID core.ID `json:"id"` // ID 实例内逻辑步骤身份。 + Key string `json:"key"` // Key 固定定义步骤名称。 + Kind weave.NodeKind `json:"kind"` // Kind 固定节点类型。 + State weave.StepState `json:"state"` // State 当前步骤状态。 + Revision uint64 `json:"revision,string"` // Revision 步骤并发版本。 + ActivationID *core.ID `json:"activation_id,omitempty"` // ActivationID 当前激活,未激活时为空。 + TaskID *core.ID `json:"task_id,omitempty"` // TaskID TASK的直接任务;POLL任务通过轮次查询。 +} +type StepsView struct { + InstanceID core.ID `json:"instance_id"` // InstanceID 所属实例。 + Revision uint64 `json:"revision,string"` // Revision 与步骤读取一致的实例版本。 + Steps []StepView `json:"steps"` // Steps 按键排序,最多128个,不含内部根检查点。 +} +type Store interface { + Start(context.Context, auth.Token, string, StartInput) (StartResult, error) + Get(context.Context, auth.Token, core.ID) (View, error) + Steps(context.Context, auth.Token, core.ID) (StepsView, error) +} +type Service struct{ store Store } + +func New(store Store) *Service { return &Service{store: store} } +func (s *Service) Start(ctx context.Context, t auth.Token, key string, in StartInput) (StartResult, error) { + if err := service.ValidateKey(key); err != nil { + return StartResult{}, err + } + if in.DefinitionID.Validate() != nil || !versionName(in.Version) || !utf8.ValidString(in.BusinessKey) || utf8.RuneCountInString(in.BusinessKey) > 191 || strings.ContainsRune(in.BusinessKey, 0) { + return StartResult{}, fault.Invalid("valid workflow definition, version and business key required") + } + if err := definition.ValidateData(definition.Type{Kind: "JSON"}, in.Input); err != nil { + return StartResult{}, fault.Invalid("valid workflow input required, maximum 256 KiB") + } + return s.store.Start(ctx, t, key, in) +} +func versionName(v string) bool { + if len(v) == 0 || len(v) > 64 { + return false + } + for i, r := range v { + if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || i > 0 && (r == '_' || r == '.' || r == '-')) { + return false + } + } + return true +} +func (s *Service) Get(ctx context.Context, t auth.Token, id core.ID) (View, error) { + if id.Validate() != nil { + return View{}, fault.Invalid("invalid workflow instance id") + } + return s.store.Get(ctx, t, id) +} +func (s *Service) Steps(ctx context.Context, t auth.Token, id core.ID) (StepsView, error) { + if id.Validate() != nil { + return StepsView{}, fault.Invalid("invalid workflow instance id") + } + return s.store.Steps(ctx, t, id) +} diff --git a/weave/instance/service_test.go b/weave/instance/service_test.go new file mode 100644 index 0000000..0f646f1 --- /dev/null +++ b/weave/instance/service_test.go @@ -0,0 +1,59 @@ +package instance + +import ( + "context" + "encoding/json" + "strings" + "testing" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +type captureStore struct { + calls int + input StartInput +} + +func (s *captureStore) Start(_ context.Context, _ auth.Token, _ string, in StartInput) (StartResult, error) { + s.calls++ + s.input = in + return StartResult{}, nil +} +func (s *captureStore) Get(context.Context, auth.Token, core.ID) (View, error) { + s.calls++ + return View{}, nil +} +func (s *captureStore) Steps(context.Context, auth.Token, core.ID) (StepsView, error) { + s.calls++ + return StepsView{}, nil +} +func TestAdmissionBoundaryPreservesExactInput(t *testing.T) { + store := &captureStore{} + api := New(store) + base := StartInput{DefinitionID: "11111111111111111111111111111111", Version: "v1", Input: json.RawMessage(`{"n":9007199254740993}`), BusinessKey: "关联 "} + for _, raw := range []string{`{"n":9007199254740993}`, `null`} { + in := base + in.Input = json.RawMessage(raw) + if _, err := api.Start(context.Background(), auth.Token{}, "identity", in); err != nil || string(store.input.Input) != raw || store.input.BusinessKey != base.BusinessKey { + t.Fatal("input changed", err) + } + } + for _, change := range []func(*StartInput){ + func(in *StartInput) { in.Input = nil }, func(in *StartInput) { in.Input = json.RawMessage(`{"n":1,"n":2}`) }, + func(in *StartInput) { in.Input = json.RawMessage(`"\ud800"`) }, func(in *StartInput) { in.Input = json.RawMessage(`"` + strings.Repeat("a", 256<<10) + `"`) }, + func(in *StartInput) { in.Version = "../v1" }, func(in *StartInput) { in.Version = strings.Repeat("v", 65) }, + func(in *StartInput) { in.BusinessKey = strings.Repeat("中", 192) }, func(in *StartInput) { in.BusinessKey = "a\x00b" }, + func(in *StartInput) { in.DefinitionID = "bad" }, + } { + in := base + change(&in) + before := store.calls + if _, err := api.Start(context.Background(), auth.Token{}, "identity", in); err == nil || store.calls != before { + t.Fatal("invalid input reached store") + } + } + if _, err := api.Start(context.Background(), auth.Token{}, "", base); err == nil { + t.Fatal("missing identity accepted") + } +} diff --git a/weave/model/wait.go b/weave/model/wait.go index b1367d9..de58751 100644 --- a/weave/model/wait.go +++ b/weave/model/wait.go @@ -19,7 +19,11 @@ type Wait struct { Kind string `db:"kind"` // State 当前状态;只允许定义的状态转换并配合事务校验。 State WaitState `db:"state"` - // CorrelationKey 外部业务操作关联键,防止旧信号误用于新工作。 + // SignalName 激活时冻结的信号名称;TIMER及待核实历史记录为空。 + SignalName *string `db:"signal_name"` + // PayloadVersion 激活时冻结的信号内容契约版本;TIMER及待核实历史记录为空。 + PayloadVersion *string `db:"payload_version"` + // CorrelationKey 精确UTF8业务关联且保留尾部空格;新受理最多512字节,历史兼容764字节,TIMER为空串。 CorrelationKey string `db:"correlation_key"` // DeadlineAt 等待截止时间,用可靠受理时间判断信号是否按时到达。 DeadlineAt time.Time `db:"deadline_at"`