diff --git a/AGENTS.md b/AGENTS.md index 54c093c..0faf016 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,17 +31,17 @@ - P4定义发布使用schema_version=1有限结构，词法作用域禁止跨分支/前向读取，任务策略显式冻结；SDK类型独立。发布按认证→应用额度→操作身份→定义→排序队列持锁；原键与同版本去重仍校验当前权限。定义/版本/revision/审计/回执同事务；同版本不同内容不得覆盖。保留精确数字，校验存储原文与结构摘要；不得把发布通过或DSL支持误称为实例执行已实现。 -- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归（docs/reports/phase-4-definitions.md）。实例启动/固定版本/首批任务同事务的内部存储已补齐；内部事件消费、顺序与组合推进及根接续已接入；TIMER与后续POLL时钟已接入，SIGNAL及最终查询竞争已接入，内部控制与停止结清已接入，等待恢复已接入，实例公共API/SDK已接入，流程任务来源已接入，已接入有界历史查询，继续实例分页和统一事件时间线范围核对；P4整体和P5–P7仍未完成。 +- P4定义子项已通过真实MySQL、HTTP丢回执、独立SDK和全量回归（docs/reports/phase-4-definitions.md）。实例启动/固定版本/首批任务同事务的内部存储已补齐；内部事件消费、顺序与组合推进及根接续已接入；TIMER与后续POLL时钟已接入，SIGNAL及最终查询竞争已接入，内部控制与停止结清已接入，等待恢复已接入，实例公共API/SDK已接入，流程任务来源已接入，已接入有界历史查询，实例分页已接入，继续统一事件时间线范围核对；P4整体和P5–P7仍未完成。 - 内部结果交付使用workflow_activation所有者并固定激活ID，TASK经activation.task_id、POLL经轮次校验归属；weave_inbox来源及独立快照与Pulse结果同事务，先于外部结果流行锁，不在结果路径反锁实例/步骤/订阅。迁移000017隔离缺快照历史，禁止用当前任务重建旧事件。 -- weave/engine仅计算有限决策；实例受理、Inbox消费与根接续已接入，等待和停止结清运行器已接入，内部核查/TASK恢复已接入，等待恢复已接入内部服务，公共API/SDK已接入，流程任务来源已接入，有界历史已接入，继续实例分页和事件时间线范围核对。Plan最多32个变更，More必须持久化待续；耗尽变更预算仍须遍历其他已活动分支，不能误判在途工作不可达。冻结分支选择，精确数值，不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务，不能把纯快照往返测试当作真实进程恢复。 +- weave/engine仅计算有限决策；实例受理、Inbox消费与根接续已接入，等待和停止结清运行器已接入，内部核查/TASK恢复已接入，等待恢复已接入内部服务，公共API/SDK已接入，流程任务来源已接入，有界历史已接入，实例分页已接入，继续事件时间线范围核对。Plan最多32个变更，More必须持久化待续；耗尽变更预算仍须遍历其他已活动分支，不能误判在途工作不可达。冻结分支选择，精确数值，不以PENDING查询结果完成POLL。后续消费确认/当前激活校验/新任务或检查点必须同事务，不能把纯快照往返测试当作真实进程恢复。 - 实例内部受理已实现，信号内部运行器已接入，公共实例/信号HTTP及独立SDK已接入；结果消费和根待续扫描已接入后台。Start按认证→应用额度→原身份→排序队列建立门控/固定版本/全部步骤/首批工作，所有失败同事务回滚。首批推进递增revision至2，原键返回冻结回执且仍核验当前权限，不重算截止。 - 实例根步骤`$root`/JOIN不属于用户定义，Steps查询排除它；More对应根步骤版本化持久检查点。TASK直接关联激活task_id，POLL通过轮次关联，实际任务截止必须受实例/查询绝对截止限制。Wait名称及契约版本冻结，旧NULL等待须核实，不能盲目解释；新关联512字节、存储兼容历史764字节。 - 结果消费按身份→应用额度→排序队列→实例→门控→步骤/Inbox持锁，禁止先锁Inbox再锁父资源。来源使用独立冻结快照，旧激活/低版本留痕；数据库故障不归类为业务失败。消费确认、步骤、后继或根待续、事件及审计同事务。 - 后继满额用SAVEPOINT回滚整轮新工作，保留已接受结果和待续；恢复必须重读持久图。多次根推进使用独立系统审计操作身份；迟到证据也递增实例历史版本，禁止复用已有workflow.advanced唯一身份。 -- 本轮消费运行器和真实Orbit强杀/双节点接管已验证，见docs/reports/phase-4-runtime.md。TIMER到期与POLL后续轮次已接入，见docs/13-workflow-clock.md；SIGNAL及最终信号竞争已接入（docs/14-workflow-signals.md）；内部FAIL_AFTER_SETTLED分支结清与控制已接入（docs/15-workflow-controls.md）；公共控制API和实例SDK已接入，流程任务来源已接入，实例列表、统一事件时间线范围核对与更多场景仍须完成，不得按部分自动运行宣称P4/P5完成。 +- 本轮消费运行器和真实Orbit强杀/双节点接管已验证，见docs/reports/phase-4-runtime.md。TIMER到期与POLL后续轮次已接入，见docs/13-workflow-clock.md；SIGNAL及最终信号竞争已接入（docs/14-workflow-signals.md）；内部FAIL_AFTER_SETTLED分支结清与控制已接入（docs/15-workflow-controls.md）；公共控制API和实例SDK已接入，流程任务来源已接入，实例列表已接入，统一事件时间线范围核对与更多场景仍须完成，不得按部分自动运行宣称P4/P5完成。 - TIMER/POLL复用weave_pending_advances及到期索引，时钟固定激活ID；迁移000019只回填原等待/查询时钟，不改截止和预算。后续POLL只由已确认POLL_PENDING开启，首轮精确输入/资源键冻结，每轮Pulse尝试独立计数；满额/暂停不消耗轮数。截止遇到已落库结果先等待Inbox，仍在途则关闭门控但保留ACTIVE和当前任务，许可由Pulse结清；不得由时钟假定外部请求已结束。 @@ -52,18 +52,20 @@ - StopKind/StopRequestedAt冻结首次CANCEL或FAILURE目标，后来的用户取消不覆盖原失败目标。结清复用ROOT，每事务最多一个Pulse取消、32项步骤收尾；结果流锁取得后不能再锁另一任务。终态前校验全部历史任务的双向归属、未结束尝试和未释放许可，UNKNOWN不得终结。迁移000021恢复旧停止意图，不添加外键。 - 人工核查使用显式workflows:recover、当前实例revision、步骤/任务版本及激活身份，必须有原因和证据；执行结论、步骤结论、解除核查分别确认。新决定拒绝PENDING/QUARANTINED Inbox，原键回执仍核验权限。核查回执/额度/双审计/结果流同事务，结果流锁最后取得。 -- TASK人工重试保留首次绝对截止和精确输入，以新激活/关联任务继续消耗原max_attempts；激活数也受该上限限制，不能无限制造零尝试任务。成功步骤、未知执行、原停止意图或预算耗尽均不得重试。Continue不解除paused、重置预算或撤销停止。POLL跨激活恢复已实现（docs/17-workflow-poll-recovery.md），等待恢复已接入内部服务，公共HTTP/SDK已接入，流程任务来源已接入，实例列表、统一事件时间线范围核对与更多场景仍需补齐，不能据此宣称P5完成。 +- TASK人工重试保留首次绝对截止和精确输入，以新激活/关联任务继续消耗原max_attempts；激活数也受该上限限制，不能无限制造零尝试任务。成功步骤、未知执行、原停止意图或预算耗尽均不得重试。Continue不解除paused、重置预算或撤销停止。POLL跨激活恢复已实现（docs/17-workflow-poll-recovery.md），等待恢复已接入内部服务，公共HTTP/SDK已接入，流程任务来源已接入，实例列表已接入，统一事件时间线范围核对与更多场景仍需补齐，不能据此宣称P5完成。 - POLL人工恢复以新激活重查同一业务轮，FirstRound/FirstRoundAttemptsUsed保留原轮次和已用尝试；实际尝试行、retry_of链和冻结参数必须相符。同轮任务数也受原max_attempts限制，PENDING不得人工重查绕过轮次上限。查询截止按首次激活、请求截止按该轮首任务，不允许重计时；下一业务轮才获得完整单轮策略。 -- POLL恢复的最终信号等待保留原契约/关联/截止并更换代次，旧winner和已消费信号只留历史；存在当前代次PENDING信号时拒绝重建。迁移000022无外键，新增2个中文注释字段。公共HTTP/SDK已接入，有界历史查询已接入，实例列表、事件时间线范围核对与生产验收继续推进。 +- POLL恢复的最终信号等待保留原契约/关联/截止并更换代次，旧winner和已消费信号只留历史；存在当前代次PENDING信号时拒绝重建。迁移000022无外键，新增2个中文注释字段。公共HTTP/SDK已接入，有界历史查询已接入，实例列表已接入，事件时间线范围核对与生产验收继续推进。 - 核查将UNKNOWN请求确认成成功PENDING时，当前RetryStep仍拒绝重查同轮；ContinuePoll显式恢复原等待，必须核验当前SUCCEEDED/PENDING任务、已处理Inbox及不可变来源。新审计固定来源事件/版本、原间隔与截止，时钟/最终信号核验证明后才推进；不改写旧Inbox处置，ReviewTask不自动解除核查。 - RearmWait仅重建独立SIGNAL/TIMER，显式wait_verified及最新步骤/激活/实例版本，连续激活上限32。旧WAITING激活/等待关闭，新代次保留首次输入、契约、关联及截止；PENDING信号、已有赢家、停止意图及过期均拒绝。TIMER运行和成功核查按首次激活计时，恢复后仍须Continue；无新持久化字段或外键。 -- 公共流程API包括实例启动/读取、最多128项步骤元数据、单步当前输入输出/预算/等待、控制、信号、六类恢复和流程资源操作审计。当前激活代次同时进入Steps、详情和新恢复回执；旧冻结回执缺字段时只能另读当前详情，不改写历史回执。Pulse流程来源已交付；历史激活/轮次/等待/信号及信号审计已接入；实例列表、统一状态事件时间线范围核对与更多独立示例仍未完成。 +- 公共流程API包括实例启动/读取、最多128项步骤元数据、单步当前输入输出/预算/等待、控制、信号、六类恢复和流程资源操作审计。当前激活代次同时进入Steps、详情和新恢复回执；旧冻结回执缺字段时只能另读当前详情，不改写历史回执。Pulse流程来源已交付；历史激活/轮次/等待/信号及信号审计已接入；实例列表已接入，统一状态事件时间线范围核对与更多独立示例仍未完成。 - 恢复请求的匿名嵌入结构按扁平精确JSON字段解析，继承字段也拒绝别名/重复/未知字段，保留显式null例外与必填确认。bootstrap先验证权限集合；工作流执行/控制/信号/恢复/审计分别显式授予，旧workflow-access仅定义发布。 - 任务来源WORKFLOW只返回已授权任务的固定流程/版本/步骤/历史激活及POLL轮次；tasks:read不授予完整流程和其他队列权限。核对正反向关联、固定定义及门控，恢复父任务须为前一激活且POLL同轮。TASK恢复的retry_of位于workflow对象内，后续POLL新轮无父任务。不能改指当前激活、递归展开历史或将来源查询加入派发路径；无外键。 - 历史读取按固定实例及所有队列授权，明确步骤/激活/信号归属，不依赖外键；摘要分页20/100，游标绑定种类和父资源。指定激活详情的Step是当前视图、Activation是历史视图，不自动改代次；POLL轮次限定单个激活，恢复同轮任务不可与下一业务轮混淆。信号列表保留未核验旧身份，正文单条读取；信号审计用显式workflows:audit:read，不能按全局审计ID越权。000023只增加分页索引；全量回归必须保留零外键与旧数据/索引断言。 + +- 实例列表按固定定义全部队列在SQL中筛选，再校验原定义/摘要、范围、策略及关联；不能用已执行任务推断权限。000024增加不可变范围索引与7个列表索引，无外键；发布与范围同事务，旧DSL常量中的queue_id不纳入范围。游标绑定命名空间/全部筛选/规范化队列权限，跨页不承诺快照，limit只限制返回候选量。先排序锁整页队列，不在实例锁之后追加队列锁。 diff --git a/README.md b/README.md index e9b3c29..f23ea84 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,9 @@ 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) 继续实现。 +已完成 P0–P3 平台功能阶段验收：36 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 、[队列资源控制](docs/api/queues.md) 、[应用共享配额](docs/api/application-resources.md) 、[资源键互斥](docs/api/keyed-concurrency.md) 和 [优先级/节点公平派发](docs/api/priority-and-fairness.md) 已接入服务端和 SDK，[可靠结果订阅](docs/api/subscriptions.md) 已实现持久化核心、HTTP/WSS、管理 SDK 与示例，[消费 SDK](docs/sdk/event-consumer.md) 和 [独立 MySQL Inbox 示例](docs/examples/result-inbox.md) 已实现；[批量任务受理](docs/api/batch-tasks.md) 已提供逐项幂等、失败分类与 SDK 示例；[人工核查与重试](docs/api/task-recovery.md) 已提供带证据的管理接口、关联新任务、审计查询及运行示例；[业务键与时间范围查询](docs/api/task-queries.md) 已提供过滤、分页与SDK场景；[任务来源查询](docs/api/task-source.md) 已提供人工重试父任务、计划触发记录与固定版本；[业务事务Outbox](docs/examples/task-outbox.md) 已提供独立业务库、稳定提交身份、发送恢复与CLI；[相对延迟执行](docs/examples/delayed-task.md)已提供数据库计时及原请求恢复；完整范围核对见[P3验收](docs/reports/phase-3.md)，流程引擎与生产验证按 [持续开发计划](docs/08-development-plan.md) 继续实现。 -Weave已提供[定义发布](docs/api/workflow-definitions.md)、[实例启动、步骤详情、暂停/继续/取消、信号和操作审计](docs/api/workflows.md)，以及[六类人工核查恢复](docs/api/workflow-recovery.md)的HTTP与独立Go SDK。持久运行器支持顺序、分支、有限并行、TIMER、SIGNAL及有界POLL，恢复沿用原预算与截止。[并行作业与回执示例](docs/examples/workflow.md)提供完整调用链；[历史激活/轮次/信号及信号审计](docs/api/workflow-history.md)已接入；实例筛选分页和更多独立恢复示例继续交付；[任务流程来源](docs/api/task-source.md#5-流程固定来源)已接入，P4/P5与生产验收尚未整体完成。 +Weave已提供[定义发布](docs/api/workflow-definitions.md)、[实例启动、步骤详情、暂停/继续/取消、信号和操作审计](docs/api/workflows.md)，以及[六类人工核查恢复](docs/api/workflow-recovery.md)的HTTP与独立Go SDK。持久运行器支持顺序、分支、有限并行、TIMER、SIGNAL及有界POLL，恢复沿用原预算与截止。[并行作业与回执示例](docs/examples/workflow.md)提供完整调用链；[历史激活/轮次/信号及信号审计](docs/api/workflow-history.md)已接入；[实例筛选分页](docs/api/workflow-queries.md)已接入，统一事件时间线范围核对和更多独立恢复示例继续交付；[任务流程来源](docs/api/task-source.md#5-流程固定来源)已接入，P4/P5与生产验收尚未整体完成。 数据库禁止外键，关联一致性通过服务端事务校验，索引、唯一约束和 CHECK 保留；readiness 同时检查迁移及零外键要求。这不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 阶段验收](docs/reports/phase-2.md)、[安全重试](docs/reports/phase-3-safe-retry.md)、[计划调度](docs/reports/phase-3-schedules.md)、[取消与无外键回归](docs/reports/phase-3-cancellation.md) 及 [队列资源回归](docs/reports/phase-3-queue-resources.md)。 @@ -89,3 +89,5 @@ make integration - [设计文档总览](docs/README.md) Git 远端：`https://git.yuntuo.hk/k2safe/orbit.git`。当前在本地开发分支推进，SDK 尚未发布版本。 + +[流程实例筛选分页](docs/api/workflow-queries.md)已提供独立SDK和[完整示例](docs/examples/workflow-query.md)，按固定版本全部队列鉴权，数据库保持零外键。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index c543321..051b419 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -35,7 +35,7 @@ ## 3. 实体清单 -当前共 35 个实体：初始迁移包含 30 个，000003 增加进程会话头和迟到证据，000010 增加应用共享额度，000013 增加结果流与不可变结果快照。迁移自身另有 `orbit_schema_migrations` 记录。 +当前共 36 个实体：初始迁移包含 30 个，000003 增加进程会话头和迟到证据，000010 增加应用共享额度，000013 增加结果流与不可变结果快照。000024 增加固定流程版本的查询范围。迁移自身另有 `orbit_schema_migrations` 记录。 | 模块 | Go 模型 | 数据表 | | --- | --- | --- | @@ -65,6 +65,7 @@ | pulse | `Occurrence` | `pulse_schedule_occurrences` | | weave | `Definition` | `weave_definitions` | | weave | `DefinitionVersion` | `weave_definition_versions` | +| weave | `VersionScope` | `weave_version_scopes` | | weave | `Instance` | `weave_instances` | | weave | `Step` | `weave_steps` | | weave | `Activation` | `weave_activations` | @@ -230,3 +231,9 @@ PollState新增FirstRound和FirstRoundAttemptsUsed，记录当前激活的首个 ## 流程历史查询索引 000023_workflow_signal_history只为weave_signals增加(namespace_id,workflow_id,accepted_at,id)分页索引，保留全部原信号数据和状态。仍为35个持久化实体/305个直接字段声明，无新增字段或外键；HTTP/SDK的摘要和详情字段均有中文注释。[历史查询](api/workflow-history.md)明确区分当前逻辑步骤与指定历史激活，不以当前代次替换旧来源。基础测试禁止000008之后再次引入数据库引用约束，真实MySQL升级及readiness验证外键为0。 + +## 固定版本查询范围与实例分页（000024） + +VersionScope新增6个中文注释字段：命名空间、定义版本、定义、原摘要、去重队列JSON、索引创建时间；当前36个实体、311个直接持久化字段声明，全部有中文注释。版本范围不可更新，发布与范围同事务，读取摘要还要核验原定义和关联，不以派生索引授予权限。回填仅遍历DSL步骤，忽略常量中相同字段；旧版本及实例原文不修改。 + +7个新实例复合索引加现有业务键索引覆盖定义/业务键/状态8种组合，按创建时间与ID分页；全部保留无外键和无级联。范围校验、迁移性能边界和接口见[查询手册](api/workflow-queries.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 46beeeb..def72f1 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -28,28 +28,28 @@ | P1 任务受理闭环 | 应用凭据与作用域权限、队列、幂等提交、查询、待执行取消、受理事件；SDK 客户端 | API/SDK 创建查询闭环；重复键与内容冲突、响应未确认重报、跨命名空间拒绝、参数边界 | 已完成并通过阶段回归（见 P1 报告；真实断路/提交 ACK 丢失注入留后续故障阶段） | | P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复；Worker SDK | 真 Worker 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 已完成并通过阶段回归：服务端、自动 SDK、历史查询、完整 Schema 与真实进程强杀恢复；测试拓扑和剩余生产故障矩阵见 P2 验收报告 | | P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 已完成平台功能阶段验收：[原范围逐项核对](reports/phase-3.md)、[本轮完整回归](reports/phase-3-delay-recovery.md)。补齐相对延迟、改参数关联示例和四种重试全停恢复场景；20个可运行示例全部通过。全平台生产门槛及SDK首版交付剩余项仍按P6–P7推进 | -| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 定义格式/词法类型校验、不可变发布、独立SDK及六类节点定义例子已实现，已通过[子项验收](reports/phase-4-definitions.md)；已补[有限决策器及内部结果交付](reports/phase-4-engine-core.md)；[实例受理及首批事务](reports/phase-4-admission.md)已实现；[结果消费和根接续](reports/phase-4-runtime.md)已接入后台，顺序/分支/并行及真实进程恢复已验证；[TIMER与后续POLL轮次](13-workflow-clock.md)已接入；[信号消费和最终查询竞争](14-workflow-signals.md)已接入内部运行器；[内部控制与停止结清](15-workflow-controls.md)已接入；[内部核查与TASK新激活](16-workflow-recovery.md)已接入；等待恢复已接入内部服务；实例公共API/SDK与基础可运行场景已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例列表、统一事件时间线范围核对与更多场景继续开发 | -| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器；最终信号竞争及在途查询结清已接入；内部暂停/继续/取消与失败结清已接入；执行核查、步骤结论、解除核查及TASK新激活已接入；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；等待恢复已接入内部服务；公共API/SDK已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例列表、统一事件时间线范围核对与更多场景继续开发 | +| P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 定义格式/词法类型校验、不可变发布、独立SDK及六类节点定义例子已实现，已通过[子项验收](reports/phase-4-definitions.md)；已补[有限决策器及内部结果交付](reports/phase-4-engine-core.md)；[实例受理及首批事务](reports/phase-4-admission.md)已实现；[结果消费和根接续](reports/phase-4-runtime.md)已接入后台，顺序/分支/并行及真实进程恢复已验证；[TIMER与后续POLL轮次](13-workflow-clock.md)已接入；[信号消费和最终查询竞争](14-workflow-signals.md)已接入内部运行器；[内部控制与停止结清](15-workflow-controls.md)已接入；[内部核查与TASK新激活](16-workflow-recovery.md)已接入；等待恢复已接入内部服务；实例公共API/SDK与基础可运行场景已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；统一事件时间线范围核对与更多场景继续开发 | +| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器；最终信号竞争及在途查询结清已接入；内部暂停/继续/取消与失败结清已接入；执行核查、步骤结论、解除核查及TASK新激活已接入；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；等待恢复已接入内部服务；公共API/SDK已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；统一事件时间线范围核对与更多场景继续开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | | P7 首版交付核对 | SDK 兼容与发布准备、完整 API 手册、场景示例、全量回归及交付清单 | 新接入者按文档跑通全部示例；所有首版能力可追溯到实现与测试；报告实际生产边界 | 待开发 | SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最后才开始补文档。 -P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告](reports/phase-4-clock.md)。外部信号与最终查询竞争见[信号报告](reports/phase-4-signals.md)。公共实例、控制、信号、六类恢复和操作审计已接入HTTP与独立SDK，见[公共接入报告](reports/phase-5-public-api.md)。流程任务来源及[历史查询](api/workflow-history.md)已接入，实例列表、统一事件时间线及更多恢复示例仍须核对交付。 +P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告](reports/phase-4-clock.md)。外部信号与最终查询竞争见[信号报告](reports/phase-4-signals.md)。公共实例、控制、信号、六类恢复和操作审计已接入HTTP与独立SDK，见[公共接入报告](reports/phase-5-public-api.md)。流程任务来源及[历史查询](api/workflow-history.md)已接入，[实例筛选分页](api/workflow-queries.md)已接入，统一事件时间线及更多恢复示例仍须核对交付。 ### P4/P5 下一步实施顺序 1. 已实现FAIL_AFTER_SETTLED与显式取消的内部分支结清，见[控制报告](reports/phase-5-controls.md)：持久取消意图、复用Pulse取消、收集已发出查询/任务结果、等待全部已激活工作结清后才进入失败或取消终态。效果未知仍保留核查。 2. 已实现实例暂停/继续/取消内部操作，以control_revision、操作幂等身份、明确原因及审计保护控制决定；暂停不延长原截止，继续不覆盖核查或取消状态。 3. 已实现内部执行核查、步骤结论、解除核查及TASK新激活和旧激活隔离，见[核查报告](reports/phase-5-recovery.md)。POLL跨激活恢复已实现（见[查询恢复](17-workflow-poll-recovery.md)）；[核查后PENDING接续和独立等待重建](18-workflow-wait-recovery.md)也已接入内部服务，保留原预算、轮次、结果和证据。 -4. 实例/当前步骤/信号/控制/六类恢复及流程操作审计HTTP和独立SDK已接入，步骤查询返回实际激活代次；并行执行与信号回执场景已有可运行示例。激活/轮次/信号历史及信号资源审计已接入；继续补齐实例筛选分页、核对统一状态事件时间线，并提供独立轮询及恢复示例；[Pulse任务流程来源](api/task-source.md#5-流程固定来源)已交付。逐项验证原键恢复、权限、版本冲突、竞争与进程全停。 +4. 实例/当前步骤/信号/控制/六类恢复及流程操作审计HTTP和独立SDK已接入，步骤查询返回实际激活代次；并行执行与信号回执场景已有可运行示例。激活/轮次/信号历史及信号资源审计已接入；实例筛选分页已接入；继续核对统一状态事件时间线，并提供独立轮询及恢复示例；[Pulse任务流程来源](api/task-source.md#5-流程固定来源)已交付。逐项验证原键恢复、权限、版本冲突、竞争与进程全停。 5. 转入P6/P7的生产故障、容量、运维、发布与全部文档核对。内部存储阶段通过不会提前勾选公共接口或生产验收。 -公共接入按以下调用链推进；本批已接入启动/当前详情、控制、信号、六类恢复和流程操作审计，并提供独立SDK及基础示例。任务流程来源和有界历史查询也已接入。以下实例分页、统一事件时间线范围核对和更多独立场景继续交付： +公共接入按以下调用链推进；本批已接入启动/当前详情、控制、信号、六类恢复和流程操作审计，并提供独立SDK及基础示例。任务流程来源和有界历史查询也已接入。实例分页和独立查询示例已接入；以下统一事件时间线范围核对和更多独立场景继续交付： | 调用链 | 需要补齐的内容 | 验收重点 | | --- | --- | --- | -| 启动与读取 | 显式固定版本启动、当前实例、步骤及当前激活代次；已补有界历史读取，后续补筛选分页 | 原键丢回执恢复、当前权限、读取快照一致性、未激活步骤语义 | +| 启动与读取 | 显式固定版本启动、当前实例、步骤及当前激活代次；已补有界历史读取及实例筛选分页 | 原键丢回执恢复、当前权限、读取快照一致性、未激活步骤语义 | | 控制与信号 | 暂停/继续/取消、固定步骤和代次的信号提交及原受理回执 | control_revision与通用revision区分、迟到/旧代次、暂停期间结果收集 | | 核查与追踪 | 六类恢复、激活/轮次/等待/信号历史读取及信号管理审计已接入；统一状态事件时间线待核对；Pulse任务流程来源已交付 | 当前实例/步骤/任务版本、证据和显式声明、旧历史保留、有界响应和权限隔离 | | 通用接入场景 | 并行汇合、外部回执、有限轮询、暂停核查恢复的可运行SDK示例 | 真实Worker执行、完整进程退出后接续、未知效果保留、全链路回归 | @@ -130,6 +130,10 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( ## 7. 文档入口 +- [实例筛选分页与SDK](api/workflow-queries.md) +- [实例查询示例](examples/workflow-query.md) +- [实例查询验证记录](reports/phase-5-queries.md) + - [流程历史查询与SDK](api/workflow-history.md) - [历史查询验证记录](reports/phase-5-history.md) - [流程实例公共API](api/workflows.md) diff --git a/docs/README.md b/docs/README.md index 7649950..18db9a0 100644 --- a/docs/README.md +++ b/docs/README.md @@ -89,3 +89,5 @@ Orbit 不内置特定行业规则；Pulse 管理执行，Weave 管理步骤依 2026-09-17，文档更新至 0.2：用户确认研发定义发布流程及 Weave 首版功能，明确纳入有界重复查询和暂停/继续等生产管理能力。技术细则与参数继续按建议或待定标注，未将功能确认等同于实现或验收通过。 持续开发目标已建立。当前实现和验收进度以 [开发计划](08-development-plan.md)、[P0 报告](reports/phase-0.md)、[P1 报告](reports/phase-1.md) 与 [P2 阶段验收](reports/phase-2.md) 、[P3 安全重试报告](reports/phase-3-safe-retry.md) 、[P3 计划调度报告](reports/phase-3-schedules.md) 和 [P3 完整范围验收](reports/phase-3.md) 为准；早期设计稿中的“未实现”描述保留其编写阶段语境。 + +[实例查询接口与SDK](api/workflow-queries.md)已接入，固定筛选沿游标分页，覆盖未执行分支的全部队列权限。 diff --git a/docs/api/README.md b/docs/api/README.md index eb9c113..b4dc6f7 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -49,3 +49,5 @@ P1 业务接口已遵守下述约定，后续接口继续沿用： ## 每个正式接口的文档模板 实现时必须补齐：用途与权限、HTTP 方法和路径、请求头、字段类型/必填/默认值/限制、完整请求与响应、错误码、幂等与并发规则、超时/重试/取消语义、SDK 对应方法、可运行示例、兼容版本和已覆盖的回归场景。 + +[流程实例筛选分页与权限](workflow-queries.md)已接入，固定筛选沿游标分页，覆盖未执行分支的全部队列权限。 diff --git a/docs/api/workflow-queries.md b/docs/api/workflow-queries.md new file mode 100644 index 0000000..41c667c --- /dev/null +++ b/docs/api/workflow-queries.md @@ -0,0 +1,114 @@ +# 流程实例筛选与分页 + +状态：已实现HTTP和独立Go SDK。查询返回有限元数据，不聚合业务输入输出。单个实例的完整视图和步骤详情见[实例接口](workflows.md)，历史激活、轮次与信号见[历史查询](workflow-history.md)。 + +## 接口与权限 + +`GET /v1/workflows`，要求显式 `workflows:read`。命名空间取自认证凭据，不接受查询参数覆盖；返回的每个实例必须满足其**启动时固定定义版本的全部队列权限**，包含尚未执行的分支、后续步骤和POLL队列。只拥有某个已生成任务的权限不代表可以读整个流程。 + +没有权限覆盖的实例在数据库中被过滤。指定定义/版本但当前范围没有匹配项，返回空列表；该结果不能用于断言资源不存在。只有SIGNAL/TIMER且不引用队列的版本仍需要 `workflows:read`，队列集合为空。不额外要求 `tasks:read` 或定义发布权限。 + +| 参数 | 语义 | +| --- | --- | +| `definition_id` | 可选，32位小写十六进制定义ID；不传则查询命名空间内可读实例 | +| `version` | 可选，必须同时指定definition_id；1–64位发布版本名，首位字母/数字，后续可含`_ . -` | +| `state` | 可选：RUNNING、WAITING、CANCELLING、NEEDS_REVIEW、SUCCEEDED、FAILED、CANCELLED | +| `paused` | 可选，只接受`true`或`false`；与state独立，false不等于不筛选 | +| `business_key` | 可选，非空且最多191个Unicode字符，无NUL；精确匹配，保留大小写、换行和尾部空格，URL必须正确编码 | +| `created_from` | 创建时间下界，包含；RFC3339，最多微秒精度 | +| `created_before` | 创建时间上界，不包含；与下界同时提供时必须晚于下界 | +| `limit` | 默认20，1–100 | +| `cursor` | 上一页next_cursor，最多1024字节 | + +时间规范化到UTC，UTC年份2000–9998；拒绝微秒以下精度。拒绝未知、重复、空值参数和无效URL编码。业务键空字符串在SDK中表示不筛选；HTTP显式`business_key=`被拒绝，首版不提供“只查无业务键”条件。 + +请求示意： + +```http +GET /v1/workflows?definition_id=11111111111111111111111111111111&version=v1&state=WAITING&paused=false&business_key=object-001%20&limit=20 +Authorization: Bearer +``` + +响应信封使用 `request_id` 与 `data`： + +```json +{ + "request_id": "example-request", + "data": { + "items": [{ + "id": "22222222222222222222222222222222", + "namespace_id": "33333333333333333333333333333333", + "definition_id": "11111111111111111111111111111111", + "definition_version_id": "44444444444444444444444444444444", + "version": "v1", + "state": "WAITING", + "revision": "2", + "paused": false, + "control_revision": "1", + "business_key": "object-001 ", + "deadline_at": "2026-09-18T01:00:00Z", + "created_at": "2026-09-18T00:00:00Z", + "updated_at": "2026-09-18T00:00:00Z" + }] + } +} +``` + +摘要另可包含 `cancel_requested_at`、`stop_kind`、`stop_requested_at`、`completed_at`。版本数值使用十进制字符串；state和paused可以独立变化。没有 `input`、`output`、定义正文或策略快照，读取正文使用 `GetWorkflow`。空列表为 `items: []`；没有更多结果时省略 `next_cursor`。 + +## 翻页一致性 + +按 `created_at DESC, id DESC` 排序，两个字段共同定位，时间相同也不会因只按时间定位而漏项。每页最多返回100条，数据库读取最多limit+1个候选用于判断下一页。队列范围在LIMIT前筛选，不全量加载后在Go中裁剪。 + +游标绑定命名空间、全部筛选和规范化队列权限范围，不绑定页长。后续页可调整limit，不能删掉筛选重试；不同权限范围必须重新从第一页查询。队列权限顺序变化不影响游标。游标不授予权限，每页仍重新认证并核验版本归属和原定义。 + +分页不提供跨页数据库快照。期间新建的较新实例不会出现在已有游标后，状态或paused变化可能令实例进入或离开筛选。需要长期同步应另外设计有提交水位的事件消费，不能拿列表游标代替结果订阅确认。 + +## SDK + +```go +paused := false +query := orbit.ListWorkflowsRequest{ + DefinitionID: definitionID, + Version: "v1", + State: "WAITING", + Paused: &paused, + BusinessKey: "object-001 ", + CreatedFrom: &from, + CreatedBefore: &before, + Limit: 20, +} +for { + page, err := client.ListWorkflows(ctx, query) + if err != nil { + return err // 保留筛选；不能降级成全量查询 + } + for _, item := range page.Items { + fmt.Println(item.ID, item.State, item.ControlRevision) + } + if page.NextCursor == "" { + break + } + query.Cursor = page.NextCursor +} +``` + +这是函数内片段，完整初始化、发布、创建、分页和取消清理见[可运行示例](../examples/workflow-query.md)。SDK采用独立 `WorkflowSummary`，不引用服务端模型。 + +## 存储与错误边界 + +迁移000024增加 `weave_version_scopes`，保存每个不可变版本的完整去重队列集合和原内容摘要。发布版本与范围索引同事务，升级从原DSL节点结构回填，输入/输出常量中的同名queue_id不计入范围。模型与SQL字段均有中文注释，**没有外键或级联约束**。 + +索引只协助SQL过滤。返回前仍严格验证原定义及其摘要、队列集合、固定版本/定义/门控归属和冻结策略；数据损坏时整页失败，不返回部分已读数据，也不把缺失范围解释为无需队列权限。范围中错误增加的队列可能使合法记录被隐藏，应通过数据库运维修复损坏，不能以放宽权限绕过。列表不验证实例的完整输入/输出正文；这些数据由单实例详情与运行器按固定契约核验。 + +八种定义/业务键/状态组合使用固定复合索引；version与paused作为附加筛选。授权过滤稀疏或paused选择性低时，扫描行数可能高于返回行数，limit不是扫描上限。P6须评估队列权限分布、数据保留量、索引写放大和查询超时预算，当前回归不构成百万租户容量结论。 + +| 情况 | 返回 | +| --- | --- | +| 参数、游标、筛选或权限指纹不匹配 | 400 / INVALID_ARGUMENT | +| 无workflows:read | 403 / PERMISSION_DENIED | +| 当前筛选无可读记录 | 200 / 空items | +| 关联或范围损坏 | 拒绝返回该页；可能为404或内部错误，保留原请求定位数据，禁止自动删条件重试 | +| 数据库暂不可用 | 保留原筛选及游标按有限退避重试，不解释为无匹配记录 | + +本迁移按维护窗口执行：先停止旧版本的流程发布写入，完成范围回填，再启动新版服务开放发布。回填后仍由旧节点发布会生成缺少范围索引的版本，新查询会拒绝返回，不能靠重试修复；不支持新旧发布写节点无协调并行升级。旧服务没有此GET集合接口，不支持时SDK返回服务端错误，不静默降级。验证记录见[本阶段报告](../reports/phase-5-queries.md)。 diff --git a/docs/api/workflows.md b/docs/api/workflows.md index 117a4de..f3d7897 100644 --- a/docs/api/workflows.md +++ b/docs/api/workflows.md @@ -169,3 +169,5 @@ ListWorkflowOperations接受HistoryPageRequest：limit默认20、范围1–100 | 503 UNAVAILABLE / 网络异常 | 提交结果可能不明，保持原键和原命令核对；SDK不自动换键 | 已实现路由之外的路径为404。回归证据和剩余交付范围见[公共接入报告](../reports/phase-5-public-api.md)。 + +实例集合读取 `GET /v1/workflows` 已接入筛选分页，参数、权限及SDK见[查询手册](workflow-queries.md)。 diff --git a/docs/examples/workflow-query.md b/docs/examples/workflow-query.md new file mode 100644 index 0000000..4a5b7e1 --- /dev/null +++ b/docs/examples/workflow-query.md @@ -0,0 +1,48 @@ +# 按业务关联分页查找流程 + +场景：一个对象关联多次处理流程，管理端需要按固定定义版本、精确业务关联及时间范围查询等待中的实例，再进入详情处理。示例同时证明业务键末尾有无空格属于不同关联。 + +源码：[workflow-query/main.go](../../sdk/go/examples/workflow-query/main.go)。服务端入口：[筛选接口](../api/workflow-queries.md)。示例发布一个SIGNAL等待流程，创建三个实例，其中两个使用相同业务键、第三个去掉尾部空格；每页1项读取匹配的两个实例，最后将三个实例取消并等待CANCELLED。 + +## 启动 + +需要已完成迁移的Orbit、MySQL 8.4+与Go 1.27.1。从仓库根目录使用专用开发数据库运行： + +```sh +go run ./cmd/orbit-migrate up +mkdir -p .local +chmod 700 .local +go run ./cmd/orbit-admin bootstrap \ + --application query-example --namespace default --environment development \ + --queue work --workflow-access --workflow-run-access --workflow-control-access \ + --output .local/workflow-query.credentials.json +go run ./cmd/orbit +``` + +另开终端： + +```sh +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ +ORBIT_CREDENTIALS_FILE=../../.local/workflow-query.credentials.json \ +go run ./examples/workflow-query +``` + +查询本身只要求workflows:read。此完整示例额外需要研发创建/发布、启动与控制权限，由上述bootstrap选项授予。SIGNAL等待不用Worker，也不模拟外部业务成功；正常结束通过显式取消结清。 + +预期输出： + +```text +workflow_query_verified=true matched=2 pages=2 exact_trailing_space=true +workflow_query_cleanup=CANCELLED count=3 +``` + +每次独立运行生成新的操作身份与定义，单次运行内发布/启动/控制分别保留稳定操作身份。输入为JSON null，输出契约为外部回执布尔值，本例不提交回执。流程120秒总时限保持有限。 + +## 错误与清理 + +无授权、发布失败、筛选丢失、重复项、取消失败均非零退出。输出不含凭据。网络失败不自动创建新操作或换筛选；正式业务应先持久保存请求、定义ID和原操作键，再按[实例接口](../api/workflows.md)恢复回执。 + +正常结束保留定义、版本、已取消实例和审计。若中途退出，已有等待保留原绝对截止，不直接删除运行数据；查询当前实例，按最新control_revision发起有原因的取消，存在未知效果则进入核查流程。示例不会更改已有实例，也不清空数据库；连续运行会累积定义，达到每命名空间100个定义的上限应使用新的隔离演示命名空间，不删除生产历史规避限制。 + +`make integration`在一次性数据库中启动真实Orbit，验证SDK与本例，并清理该次容器。跨页不是快照，实际管理界面应允许刷新第一页，并保留后续页的全部查询条件。 diff --git a/docs/reports/phase-5-queries.md b/docs/reports/phase-5-queries.md new file mode 100644 index 0000000..d5b90b3 --- /dev/null +++ b/docs/reports/phase-5-queries.md @@ -0,0 +1,53 @@ +# P4/P5 实例筛选分页验证记录 + +日期：2026-09-18。范围为 `GET /v1/workflows`、独立SDK `ListWorkflows`、固定版本查询范围迁移和完整分页示例。P4/P5整体与P6/P7仍未完成；下一项是统一状态事件时间线范围核对、独立轮询/恢复场景和阶段验收。 + +## 实现与边界 + +- 认证命名空间内按定义、固定版本、状态、paused、精确业务键、创建时间筛选；默认20、最多100条摘要，不读取实例的大输入输出。 +- SQL在LIMIT前按固定版本全部队列过滤，包含未执行分支。范围不是授权真相，返回前仍校验原定义/摘要、队列、固定版本/定义/门控归属及策略。 +- 游标绑定命名空间、全部规范化筛选和当前队列范围；同一时间戳用ID定位。拒绝改筛选、权限范围变化、重复或歧义JSON，不承诺跨页快照。 +- 000024只新增派生范围表、不可更新触发器及7个复合索引；回填遍历DSL步骤块，保留原版本/实例。最终表结构外键为0，无级联。当前36个模型、311个直接持久化字段有中文注释。 +- HTTP严格解析参数；SDK与服务端保持独立module。独立示例发布等待流程、验证精确分页、取消结清，不需Worker，也不模拟业务成功。 + +接口：[查询手册](../api/workflow-queries.md)。示例：[运行说明](../examples/workflow-query.md)及[源码](../../sdk/go/examples/workflow-query/main.go)。 + +## 环境与命令 + +macOS arm64、Go 1.27.1、Docker MySQL 8.4.10，一次性隔离数据库。服务端编译为真实进程，SDK通过真实HTTP访问；原有Worker、恢复及订阅链路继续通过WSS回归。未使用生产数据库。 + +```sh +make fmt test vet build +DOCKER=/usr/local/bin/docker make integration +``` + +本地工具链经Makefile的GO变量显式选择。隔离脚本只清理本次数据库容器与临时普通文件，不删除业务目录。完整脚本保留全部既有回归，增加实例查询SDK及第21个SDK示例。 + +## 已验证 + +| 验证 | 结果 | +| --- | --- | +| fmt、race单元回归、vet、服务端及独立业务示例构建 | 通过 | +| 真实MySQL升级与定义发布基础：v7外键、v9额度旧库、模型列映射 | 通过，旧70条外键→0，323个原索引组成部分保留 | +| 实例查询4组：8种索引组合、相同时间分页、时间边界、精确键、paused、游标及跨命名空间 | 通过 | +| 全固定版本队列鉴权 | 通过，后续任务尚未生成时仍受第二队列权限约束；发布v2不改变v1实例范围 | +| 无队列版本、缺失/缩小范围、策略损坏及动作权限 | 通过，拒绝损坏数据并不返回部分页 | +| v23→v24升级 | 通过，嵌套PARALLEL/BRANCH/TASK/POLL范围正确、重复去重、常量伪queue_id排除、原版本全部字段保持、7个新增索引列序正确、外键0 | +| 独立SDK真实HTTP | 通过，精确键/时间/版本/paused=false/游标/非法查询及三个实例取消结清 | +| 定义与五组原流程SDK回归、原workflow示例 | 通过 | +| workflow-query独立例子 | 通过，matched=2、pages=2、三个实例CANCELLED | +| 完整标准集成脚本 | 通过，退出码0；21个SDK示例、2个独立业务模块、流程与Worker/订阅进程故障回归全部完成 | + +定向存储回归11.951秒，SDK组27.361秒，其中查询SDK2.92秒。上述数字仅是本次测试执行耗时，不是性能指标。持久化模型中文注释校验311项，缺失0；Markdown检查1067个本地链接、66个JSON块，错误0。 + +全量回归中存储测试222.089秒、流程进程恢复49.000秒、流程SDK组26.665秒（查询SDK2.87秒）、最终Worker/订阅进程故障组95.296秒。独立Inbox/Outbox分别35.641秒/33.862秒。以上均通过，日志未出现失败。 + +本轮定向回归最初发现三处测试夹具问题：第二队列缺少必填配额、共享Go指针同时改了多个节点导致预期队列数错误、拒绝权限的预期错误码写错。已修复夹具并完整重跑定向链路；升级逻辑没有通过放宽范围或移除校验使测试通过。 + +## 未验证与交付余项 + +- 没有百万租户/高选择性权限稀疏查询压测；limit限制返回候选，不保证数据库扫描行数。每页最多验证101个固定版本、队列统一排序校验；大定义、队列集合与7个新增索引的读写开销须纳入P6容量测试。 +- 没有真实生产主从切换、跨故障域高可用或生产备份恢复结论；本地既有进程强杀回归不能替代这些验收。 +- 迁移需先停止旧版本的流程发布写入，回填后由新版接管；未实现新旧发布写节点无协调并行升级。迁移全量回填既有版本，并为实例增加索引；大库需测量DDL时长、额外磁盘和维护窗口。历史定义结构损坏可能导致迁移失败，应先核查数据，不能跳过迁移或修改历史校验记录。 +- 派生范围被错误增大可能隐藏合法结果；必须保护数据库写权限并按原定义核查修复，不通过放宽业务权限绕过。关联一致性仍由代码事务保证，不引入外键。 +- 原定统一事件时间线范围核对、更多独立场景、P4/P5全范围审计和P6/P7持续推进。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index f3e1f65..c144b8f 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -206,3 +206,5 @@ P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) 已提供独立WorkflowSpecification类型及CreateWorkflowDefinition、GetWorkflowDefinition、PublishWorkflowVersion、GetWorkflowVersion四个HTTP方法。业务侧研发显式发布，服务端冻结版本；实例执行使用[StartWorkflow等方法](../api/workflows.md)，完整运行见[示例](../examples/workflow.md)。见[接口与字段](../api/workflow-definitions.md)、[结构版本1](../api/workflow-definition-format.md)、[完整可运行场景](../examples/workflow-definition.md)。 流程任务同样调用`GetTaskSource`，kind为WORKFLOW时读取workflow中的固定版本、步骤、历史激活代次、查询轮次和直接恢复父任务。该代次不表示当前等待目标；发送信号须另读GetWorkflowStep。完整字段及兼容边界见[来源接口](../api/task-source.md#5-流程固定来源)，[workflow示例](../examples/workflow.md)实际核对三个任务来源。 + +[ListWorkflows及完整查询示例](../api/workflow-queries.md)已接入，固定筛选沿游标分页，覆盖未执行分支的全部队列权限。 diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index f6cacb1..c8eb1ac 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(`DROP TABLE weave_version_scopes`) + f.exec(`ALTER TABLE weave_instances DROP INDEX ix_workflow_created,DROP INDEX ix_workflow_state_created,DROP INDEX ix_workflow_business_state_created,DROP INDEX ix_workflow_definition_created,DROP INDEX ix_workflow_definition_state_created,DROP INDEX ix_workflow_definition_business_created,DROP INDEX ix_workflow_definition_business_state_created`) f.exec(`ALTER TABLE weave_signals DROP INDEX ix_signal_history`) f.exec(`ALTER TABLE weave_poll_states DROP CHECK ck_poll_recovery_budget, DROP COLUMN first_round, DROP COLUMN first_round_attempts_used`) f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_workflow_owner`) diff --git a/internal/storage/mysql/integration_test.go b/internal/storage/mysql/integration_test.go index 9b7da40..f9508be 100644 --- a/internal/storage/mysql/integration_test.go +++ b/internal/storage/mysql/integration_test.go @@ -209,6 +209,7 @@ func checkModelColumns(t *testing.T, ctx context.Context, db *sql.DB) { pulse.Occurrence{}, weave.Definition{}, weave.DefinitionVersion{}, + weave.VersionScope{}, weave.Instance{}, weave.Step{}, weave.Activation{}, diff --git a/internal/storage/mysql/migrations/000024_workflow_query_scopes.up.sql b/internal/storage/mysql/migrations/000024_workflow_query_scopes.up.sql new file mode 100644 index 0000000..dc96f99 --- /dev/null +++ b/internal/storage/mysql/migrations/000024_workflow_query_scopes.up.sql @@ -0,0 +1,57 @@ +-- 固定版本查询范围与列表索引；禁止外键，不改写版本、实例或任务历史。 +CREATE TABLE weave_version_scopes ( + namespace_id BINARY(16) NOT NULL COMMENT '命名空间，参与版本身份及查询授权', + definition_version_id BINARY(16) NOT NULL COMMENT '固定定义版本标识，范围包含所有未执行分支', + definition_id BINARY(16) NOT NULL COMMENT '所属定义标识，服务端核对其与版本及实例的归属', + content_digest BINARY(32) NOT NULL COMMENT '原不可变定义摘要，防止跨版本使用范围', + queue_ids JSON NOT NULL COMMENT '去重队列标识数组，空数组表示无任务队列；只用于查询筛选，返回前仍需核验原定义', + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '查询索引建立时间，UTC；不能替代原定义发布时间', + PRIMARY KEY(namespace_id,definition_version_id), + KEY ix_version_scope_definition(namespace_id,definition_id,definition_version_id), + CONSTRAINT ck_version_scope_queues CHECK (JSON_SCHEMA_VALID('{"type":"array","maxItems":128,"uniqueItems":true,"items":{"type":"string","pattern":"^[0-9a-f]{32}$"}}',queue_ids)) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='不可变流程版本的查询队列范围，供实例列表在数据库内筛选当前权限'; + +-- 只遍历定义结构中的步骤块，不将输入常量中的task/queue_id误当成流程权限。 +-- 合法结构最多4层组合嵌套；此处多保留一层用于识别异常数据，公共读取仍严格校验原定义。 +INSERT INTO weave_version_scopes(namespace_id,definition_version_id,definition_id,content_digest,queue_ids) +WITH RECURSIVE nodes AS ( + SELECT v.namespace_id,v.id AS version_id,s.node,0 AS depth + FROM weave_definition_versions v + JOIN JSON_TABLE(v.specification,'$.body.steps[*]' COLUMNS(node JSON PATH '$')) s + UNION ALL + SELECT n.namespace_id,n.version_id,s.node,n.depth+1 + FROM nodes n + JOIN JSON_TABLE( + CASE JSON_UNQUOTE(JSON_EXTRACT(n.node,'$.kind')) + WHEN 'BRANCH' THEN JSON_MERGE_PRESERVE(COALESCE(JSON_EXTRACT(n.node,'$.branch.cases[*].body.steps[*]'),JSON_ARRAY()),COALESCE(JSON_EXTRACT(n.node,'$.branch.default.steps[*]'),JSON_ARRAY())) + WHEN 'PARALLEL' THEN COALESCE(JSON_EXTRACT(n.node,'$.parallel.branches[*].body.steps[*]'),JSON_ARRAY()) + ELSE JSON_ARRAY() + END,'$[*]' COLUMNS(node JSON PATH '$')) s + WHERE n.depth<5 +), distinct_queues AS ( + SELECT DISTINCT namespace_id,version_id, + CASE JSON_UNQUOTE(JSON_EXTRACT(node,'$.kind')) + WHEN 'TASK' THEN JSON_UNQUOTE(JSON_EXTRACT(node,'$.task.queue_id')) + WHEN 'POLL' THEN JSON_UNQUOTE(JSON_EXTRACT(node,'$.poll.task.queue_id')) + END AS queue_id + FROM nodes + WHERE JSON_UNQUOTE(JSON_EXTRACT(node,'$.kind')) IN ('TASK','POLL') +), scopes AS ( + SELECT namespace_id,version_id,JSON_ARRAYAGG(queue_id) AS queue_ids + FROM distinct_queues GROUP BY namespace_id,version_id +) +SELECT v.namespace_id,v.id,v.definition_id,v.content_digest,COALESCE(s.queue_ids,JSON_ARRAY()) +FROM weave_definition_versions v LEFT JOIN scopes s ON s.namespace_id=v.namespace_id AND s.version_id=v.id; + +CREATE TRIGGER trg_version_scope_immutable BEFORE UPDATE ON weave_version_scopes + FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Workflow query scopes are immutable'; + +-- 定义、业务键及状态都是可选等值筛选；创建时间和ID作为稳定分页位置。 +ALTER TABLE weave_instances + ADD KEY ix_workflow_created(namespace_id,created_at,id), + ADD KEY ix_workflow_state_created(namespace_id,state,created_at,id), + ADD KEY ix_workflow_business_state_created(namespace_id,business_key,state,created_at,id), + ADD KEY ix_workflow_definition_created(namespace_id,definition_id,created_at,id), + ADD KEY ix_workflow_definition_state_created(namespace_id,definition_id,state,created_at,id), + ADD KEY ix_workflow_definition_business_created(namespace_id,definition_id,business_key,created_at,id), + ADD KEY ix_workflow_definition_business_state_created(namespace_id,definition_id,business_key,state,created_at,id); diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index acef890..3507aab 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索引、v17增加内部结果唯一索引、v20增加信号索引，下方单独验证定义。 - 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','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner','ix_signal_history') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) + // 排除后续新增/替换索引；原有项在此逐项比较，v24新增项由WorkflowQueryUpgrade核对。 + 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','weave_version_scopes') 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','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner','ix_signal_history','ix_workflow_created','ix_workflow_state_created','ix_workflow_business_state_created','ix_workflow_definition_created','ix_workflow_definition_state_created','ix_workflow_definition_business_created','ix_workflow_definition_business_state_created') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) if err != nil { t.Fatal(err) } diff --git a/internal/storage/mysql/workflow_definitions.go b/internal/storage/mysql/workflow_definitions.go index e10f3d7..3af4192 100644 --- a/internal/storage/mysql/workflow_definitions.go +++ b/internal/storage/mysql/workflow_definitions.go @@ -225,6 +225,14 @@ func (s *WorkflowDefinitionStore) Publish(ctx context.Context, t auth.Token, id if err != nil { return err } + + scope, err := json.Marshal(append([]core.ID{}, checked.Queues...)) + if err != nil { + return err + } + if _, err = tx.ExecContext(ctx, `INSERT INTO weave_version_scopes(namespace_id,definition_version_id,definition_id,content_digest,queue_ids,created_at) VALUES (?,?,?,?,?,?)`, t.NamespaceID, versionID, id, checked.Digest, string(scope), now); err != nil { + return err + } if _, err := tx.ExecContext(ctx, `UPDATE weave_definitions SET revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, t.NamespaceID, id); err != nil { return err } diff --git a/internal/storage/mysql/workflow_query.go b/internal/storage/mysql/workflow_query.go new file mode 100644 index 0000000..84d9336 --- /dev/null +++ b/internal/storage/mysql/workflow_query.go @@ -0,0 +1,257 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "slices" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/instance" +) + +// 查询游标只表示位置，不授予权限；改变权限范围必须重新查询。 +type workflowCursor struct { + Version int `json:"v"` + Namespace core.ID `json:"n"` + Created time.Time `json:"t"` + ID core.ID `json:"i"` + Filter string `json:"f"` +} + +func workflowQueryQueues(p auth.Permissions) []string { + if slices.Contains(p.Queues, "*") { + return []string{"*"} + } + queues := slices.Clone(p.Queues) + slices.Sort(queues) + return slices.Compact(queues) +} +func workflowFilterDigest(in instance.ListInput, queues []string) (string, error) { + // 页长和位置不属于筛选；nil暂停与false有不同语义。 + raw, err := requestDigest(struct { + Domain string + Definition core.ID + Version string + State string + Paused *bool + Business string + From, Before *time.Time + Queues []string + }{"orbit.workflow-query.v1", in.DefinitionID, in.Version, string(in.State), in.Paused, in.BusinessKey, in.CreatedFrom, in.CreatedBefore, queues}) + return hex.EncodeToString(raw[:]), err +} +func decodeWorkflowCursor(value string, ns core.ID, in instance.ListInput, filter string) (workflowCursor, error) { + var c workflowCursor + invalid := fault.Invalid("cursor does not match this workflow query or queue permissions") + raw, err := base64.RawURLEncoding.Strict().DecodeString(value) + if err != nil || len(raw) > 768 { + return c, invalid + } + if _, err = jsonvalue.Canonical(raw); err != nil { + return c, invalid + } + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil || len(fields) != 5 { + return c, invalid + } + for _, k := range []string{"v", "n", "t", "i", "f"} { + v, ok := fields[k] + if !ok || string(v) == "null" { + return c, invalid + } + } + if json.Unmarshal(raw, &c) != nil || c.Version != 1 || c.Namespace != ns || c.Filter != filter || c.ID.Validate() != nil || c.Created.Year() < 2000 || c.Created.Year() > 9998 || c.Created.Nanosecond()%1000 != 0 { + return c, invalid + } + if in.CreatedFrom != nil && c.Created.Before(*in.CreatedFrom) || in.CreatedBefore != nil && !c.Created.Before(*in.CreatedBefore) { + return c, invalid + } + return c, nil +} + +// 所有可选组合均使用固定索引名，不拼接用户输入；队列筛选在LIMIT之前完成。 +func workflowListSQL(ns core.ID, in instance.ListInput, queues []string, filter string) (string, []any, error) { + index := "ix_workflow" + if in.DefinitionID != "" { + index += "_definition" + } + if in.BusinessKey != "" { + index += "_business" + } + if in.State != "" { + index += "_state" + } + index += "_created" + if in.DefinitionID == "" && in.BusinessKey != "" && in.State == "" { + index = "ix_workflow_business" + } + q := `SELECT i.id,i.namespace_id,i.definition_id,i.definition_version_id,COALESCE(v.version,''),i.state,i.revision,i.paused,i.control_revision,COALESCE(i.business_key,''),i.deadline_at,i.cancel_requested_at,i.stop_kind,i.stop_requested_at,i.completed_at,i.created_at,i.updated_at, + i.policy_snapshot,s.queue_ids,COALESCE(s.definition_id=i.definition_id AND s.content_digest=v.content_digest AND v.definition_id=i.definition_id AND d.id IS NOT NULL AND g.id IS NOT NULL,FALSE) + FROM weave_instances i FORCE INDEX (` + index + `) + LEFT JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.id=i.definition_version_id + LEFT JOIN weave_version_scopes s ON s.namespace_id=i.namespace_id AND s.definition_version_id=i.definition_version_id + LEFT JOIN weave_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id + LEFT JOIN pulse_execution_gates g ON g.namespace_id=i.namespace_id AND g.id=i.gate_id + WHERE i.namespace_id=?` + args := []any{ns} + if in.DefinitionID != "" { + q += " AND i.definition_id=?" + args = append(args, in.DefinitionID) + } + if in.BusinessKey != "" { + q += " AND i.business_key=?" + args = append(args, in.BusinessKey) + } + if in.State != "" { + q += " AND i.state=?" + args = append(args, in.State) + } + if in.Version != "" { + q += " AND v.version=?" + args = append(args, in.Version) + } + if in.Paused != nil { + q += " AND i.paused=?" + args = append(args, *in.Paused) + } + if in.CreatedFrom != nil { + q += " AND i.created_at>=?" + args = append(args, *in.CreatedFrom) + } + if in.CreatedBefore != nil { + q += " AND i.created_ati.definition_id OR s.content_digest<>v.content_digest OR JSON_CONTAINS(?,s.queue_ids))` + args = append(args, string(allowed)) + } + q += " ORDER BY i.created_at DESC,i.id DESC LIMIT ?" + args = append(args, in.Limit+1) + return q, args, nil +} + +type workflowListRow struct { + view instance.Summary + policy, queues []byte + related bool +} + +func (s *WorkflowInstanceStore) List(ctx context.Context, t auth.Token, in instance.ListInput) (instance.Page, error) { + out := instance.Page{Items: []instance.Summary{}} + err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error { + queues := workflowQueryQueues(p) + filter, err := workflowFilterDigest(in, queues) + if err != nil { + return err + } + q, args, err := workflowListSQL(t.NamespaceID, in, queues, filter) + if err != nil { + return err + } + rows, err := tx.QueryContext(ctx, q, args...) + if err != nil { + return err + } + candidates := make([]workflowListRow, 0, in.Limit+1) + for rows.Next() { + var r workflowListRow + v := &r.view + if err = rows.Scan(&v.ID, &v.NamespaceID, &v.DefinitionID, &v.DefinitionVersionID, &v.Version, &v.State, &v.Revision, &v.Paused, &v.ControlRevision, &v.BusinessKey, &v.DeadlineAt, &v.CancelRequestedAt, &v.StopKind, &v.StopRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt, &r.policy, &r.queues, &r.related); err != nil { + rows.Close() + return err + } + candidates = append(candidates, r) + } + err = rows.Err() + rows.Close() + if err != nil { + return err + } + // 最多limit+1个候选，按固定版本缓存严格原文验证，不读取每个实例的大输入输出。 + versions := map[core.ID]definition.VersionView{} + allQueues := []core.ID{} + for _, r := range candidates { + if !r.related { + return errors.New("workflow query relation or scope mismatch") + } + v := r.view + ver, ok := versions[v.DefinitionVersionID] + if !ok { + ver, err = readWorkflowVersion(ctx, tx, t.NamespaceID, v.DefinitionID, v.Version) + if err != nil { + return err + } + versions[v.DefinitionVersionID] = ver + allQueues = append(allQueues, ver.Analysis.Queues...) + } + if ver.ID != v.DefinitionVersionID || ver.DefinitionID != v.DefinitionID || ver.Version != v.Version { + return errors.New("workflow query fixed version mismatch") + } + var scope []core.ID + if json.Unmarshal(r.queues, &scope) != nil || scope == nil { + return errors.New("workflow query scope invalid") + } + slices.Sort(scope) + if !slices.Equal(scope, ver.Analysis.Queues) { + return errors.New("workflow query scope does not match definition") + } + for _, id := range scope { + if !p.AllowsQueue(id) { + return fault.NotFound + } + } + var policy workflowPolicy + if json.Unmarshal(r.policy, &policy) != nil || policy != (workflowPolicy{1, ver.ID, ver.ContentDigest, ver.Specification.FailurePolicy, ver.Analysis.MaxTaskAttempts}) { + return errors.New("workflow frozen policy mismatch") + } + } + // 在任何实例锁之前，整页涉及的队列统一排序；只读列表不锁实例，也不再次读取变化中的状态。 + slices.Sort(allQueues) + if err = authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, slices.Compact(allQueues)); err != nil { + return err + } + more := len(candidates) > in.Limit + if more { + candidates = candidates[:in.Limit] + } + for _, r := range candidates { + out.Items = append(out.Items, r.view) + } + if more { + last := out.Items[len(out.Items)-1] + raw, err := json.Marshal(workflowCursor{1, t.NamespaceID, last.CreatedAt, last.ID, filter}) + if err != nil { + return err + } + out.NextCursor = base64.RawURLEncoding.EncodeToString(raw) + } + return nil + }) + if err != nil { + return instance.Page{}, err + } + return out, nil +} diff --git a/internal/storage/mysql/workflow_query_integration_test.go b/internal/storage/mysql/workflow_query_integration_test.go new file mode 100644 index 0000000..40cf891 --- /dev/null +++ b/internal/storage/mysql/workflow_query_integration_test.go @@ -0,0 +1,228 @@ +package mysql + +import ( + "context" + "encoding/json" + "os" + "slices" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/instance" +) + +func TestIntegrationWorkflowQuery(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, 12, 4) + 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("filters_ties_cursor_and_metadata", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, runtimeSpec(f.resources.QueueID, 1)) + in.BusinessKey = "查询 +/%\n " + var ids []core.ID + for n := 0; n < 4; n++ { + v, err := api.Start(ctx, f.token, string(executionID(t)), in) + if err != nil { + t.Fatal(err) + } + ids = append(ids, v.Instance.ID) + } + stamp := time.Date(2026, 1, 2, 3, 4, 5, 123456000, time.UTC) + before := stamp.Add(time.Microsecond) + f.exec(`UPDATE weave_instances SET created_at=?,updated_at=? WHERE namespace_id=?`, stamp, stamp, f.resources.NamespaceID) + slices.Sort(ids) + slices.Reverse(ids) + // 覆盖全部8个固定索引组合；同时间戳分页不能重复或遗漏。 + for mask := 0; mask < 8; mask++ { + filter := instance.ListInput{Limit: 1, CreatedFrom: &stamp, CreatedBefore: &before} + if mask&1 != 0 { + filter.DefinitionID = in.DefinitionID + } + if mask&2 != 0 { + filter.BusinessKey = in.BusinessKey + } + if mask&4 != 0 { + filter.State = "RUNNING" + } + var got []core.ID + for n := 0; n < 5; n++ { + page, err := api.List(ctx, f.token, filter) + if err != nil { + t.Fatal(mask, err) + } + for _, v := range page.Items { + got = append(got, v.ID) + } + if page.NextCursor == "" { + break + } + filter.Cursor = page.NextCursor + } + if !slices.Equal(got, ids) { + t.Fatal("page order", mask, got, ids) + } + } + base := instance.ListInput{DefinitionID: in.DefinitionID, Version: "v1", BusinessKey: in.BusinessKey, CreatedFrom: &stamp, CreatedBefore: &before, Limit: 1} + first, err := api.List(ctx, f.token, base) + if err != nil || first.NextCursor == "" { + t.Fatal(first, err) + } + raw, _ := json.Marshal(first) + for _, name := range []string{`"input"`, `"output"`, `"policy_snapshot"`, `"specification"`} { + if strings.Contains(string(raw), name) { + t.Fatal("large private payload leaked", name) + } + } + base.Cursor = first.NextCursor + for name, change := range map[string]func(*instance.ListInput){"version": func(v *instance.ListInput) { v.Version = "v2" }, "definition": func(v *instance.ListInput) { v.DefinitionID = executionID(t) }, "business": func(v *instance.ListInput) { v.BusinessKey = strings.TrimSpace(in.BusinessKey) }, "state": func(v *instance.ListInput) { v.State = "WAITING" }, "paused": func(v *instance.ListInput) { b := false; v.Paused = &b }, "from": func(v *instance.ListInput) { z := stamp.Add(-time.Second); v.CreatedFrom = &z }, "before": func(v *instance.ListInput) { z := before.Add(time.Second); v.CreatedBefore = &z }} { + changed := base + change(&changed) + if _, err := api.List(ctx, f.token, changed); fault.Code(err) != "INVALID_ARGUMENT" { + t.Fatal(name, err) + } + } + changed := base + changed.Limit = 2 + if page, err := api.List(ctx, f.token, changed); err != nil || len(page.Items) != 2 { + t.Fatal("page size may change", page, err) + } + base.Cursor = "" + base.BusinessKey = strings.TrimSpace(in.BusinessKey) + if page, err := api.List(ctx, f.token, base); err != nil || len(page.Items) != 0 { + t.Fatal("business key trimmed", page, err) + } + base.BusinessKey = in.BusinessKey + base.CreatedFrom = &before + base.CreatedBefore = nil + if page, err := api.List(ctx, f.token, base); err != nil || len(page.Items) != 0 { + t.Fatal("lower inclusive bound", page, err) + } + base.CreatedFrom = nil + base.CreatedBefore = &stamp + if page, err := api.List(ctx, f.token, base); err != nil || len(page.Items) != 0 { + t.Fatal("upper exclusive bound", page, err) + } + f.exec(`UPDATE weave_instances SET paused=TRUE WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, ids[0]) + paused := true + if page, err := api.List(ctx, f.token, instance.ListInput{Paused: &paused}); err != nil || len(page.Items) != 1 || page.Items[0].ID != ids[0] { + t.Fatal("paused filter", page, err) + } + other := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, other) + if page, err := api.List(ctx, other.token, instance.ListInput{}); err != nil || len(page.Items) != 0 { + t.Fatal("namespace leak", page, err) + } + if _, err := api.List(ctx, other.token, instance.ListInput{Cursor: first.NextCursor}); fault.Code(err) != "INVALID_ARGUMENT" { + t.Fatal("namespace cursor", err) + } + grantWorkflowInstance(t, f, string(f.resources.QueueID)) + if _, err := api.List(ctx, f.token, instance.ListInput{Cursor: first.NextCursor}); fault.Code(err) != "INVALID_ARGUMENT" { + t.Fatal("changed scope cursor", err) + } + }) + t.Run("all_frozen_queues_even_before_their_tasks_exist", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + q2 := executionID(t) + f.exec(`INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,policy) VALUES (?,?,'other',8,100,'{}')`, f.resources.NamespaceID, q2) + spec := runtimeSpec(f.resources.QueueID, 2) + spec.Body.Steps[1].Task.QueueID = q2 + hiddenIn := publishInstanceSpec(t, f, spec) + hidden, err := api.Start(ctx, f.token, "two-queues", hiddenIn) + if err != nil { + t.Fatal(err) + } + visibleIn := publishInstanceSpec(t, f, runtimeSpec(f.resources.QueueID, 1)) + visible, err := api.Start(ctx, f.token, "one-queue", visibleIn) + if err != nil { + t.Fatal(err) + } + // 切换定义新版本到可读队列，旧实例仍按v1全范围鉴权。 + _, err = definition.New(NewWorkflowDefinitionStore(db)).Publish(ctx, f.token, hiddenIn.DefinitionID, "v2", definition.PublishInput{Version: "v2", Specification: runtimeSpec(f.resources.QueueID, 1), Reason: "new scoped version"}) + if err != nil { + t.Fatal(err) + } + grantWorkflowInstance(t, f, string(f.resources.QueueID)) + page, err := api.List(ctx, f.token, instance.ListInput{Limit: 1}) + if err != nil || len(page.Items) != 1 || page.Items[0].ID != visible.Instance.ID || page.NextCursor != "" { + t.Fatal("scope filtering", page, err) + } + if instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=? AND queue_id=?`, f.resources.NamespaceID, q2) != 0 { + t.Fatal("test requires unexecuted queue") + } + grantWorkflowInstance(t, f, string(q2), string(f.resources.QueueID)) + page, err = api.List(ctx, f.token, instance.ListInput{Limit: 1}) + if err != nil || page.NextCursor == "" { + t.Fatal(page, err) + } + grantWorkflowInstance(t, f, string(f.resources.QueueID), string(q2)) + if next, err := api.List(ctx, f.token, instance.ListInput{Limit: 1, Cursor: page.NextCursor}); err != nil || len(next.Items) != 1 || next.Items[0].ID != hidden.Instance.ID { + t.Fatal("queue ordering changed fingerprint", next, err) + } + // 只有workflows:read就足够查询，tasks:read和定义发布权限不会被隐式要求。 + f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY('workflows:read')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + if page, err := api.List(ctx, f.token, instance.ListInput{}); err != nil || len(page.Items) != 2 { + t.Fatal(page, err) + } + f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY('tasks:read')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + if _, err := api.List(ctx, f.token, instance.ListInput{}); fault.Code(err) != "PERMISSION_DENIED" { + t.Fatal("action bypass", err) + } + }) + t.Run("no_queue_workflow_and_corrupt_scope_fail_closed", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, timerSpec(f.resources.QueueID, 5000, false)) + v, err := api.Start(ctx, f.token, "timer", in) + if err != nil { + t.Fatal(err) + } + grantWorkflowInstance(t, f, string(executionID(t))) + if page, err := api.List(ctx, f.token, instance.ListInput{}); err != nil || len(page.Items) != 1 { + t.Fatal("empty queue set", page, err) + } + if _, err := db.ExecContext(ctx, `UPDATE weave_version_scopes SET queue_ids=JSON_ARRAY(?) WHERE namespace_id=?`, string(executionID(t)), f.resources.NamespaceID); err == nil { + t.Fatal("scope mutable") + } + f.exec(`DELETE FROM weave_version_scopes WHERE namespace_id=?`, f.resources.NamespaceID) + if page, err := api.List(ctx, f.token, instance.ListInput{}); err == nil || len(page.Items) != 0 { + t.Fatal("missing scope granted access", page, err) + } + f.exec(`INSERT INTO weave_version_scopes(namespace_id,definition_version_id,definition_id,content_digest,queue_ids) SELECT namespace_id,id,definition_id,content_digest,JSON_ARRAY() FROM weave_definition_versions WHERE namespace_id=?`, f.resources.NamespaceID) + f.exec(`UPDATE weave_instances SET policy_snapshot=JSON_SET(policy_snapshot,'$.max_task_attempts',999) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.Instance.ID) + if page, err := api.List(ctx, f.token, instance.ListInput{}); err == nil || len(page.Items) != 0 { + t.Fatal("policy mismatch", page, err) + } + }) + t.Run("underreported_scope_does_not_grant_queues", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + grantWorkflowInstance(t, f) + in := publishInstanceSpec(t, f, runtimeSpec(f.resources.QueueID, 1)) + if _, err := api.Start(ctx, f.token, "task", in); err != nil { + t.Fatal(err) + } + f.exec(`DELETE FROM weave_version_scopes WHERE namespace_id=?`, f.resources.NamespaceID) + f.exec(`INSERT INTO weave_version_scopes(namespace_id,definition_version_id,definition_id,content_digest,queue_ids) SELECT namespace_id,id,definition_id,content_digest,JSON_ARRAY() FROM weave_definition_versions WHERE namespace_id=?`, f.resources.NamespaceID) + grantWorkflowInstance(t, f, string(executionID(t))) + if page, err := api.List(ctx, f.token, instance.ListInput{}); err == nil || len(page.Items) != 0 { + t.Fatal("scope bypass", page, err) + } + }) +} diff --git a/internal/storage/mysql/workflow_query_test.go b/internal/storage/mysql/workflow_query_test.go new file mode 100644 index 0000000..7435779 --- /dev/null +++ b/internal/storage/mysql/workflow_query_test.go @@ -0,0 +1,27 @@ +package mysql + +import ( + "encoding/base64" + "encoding/json" + "strings" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/weave/instance" +) + +func TestWorkflowQueryCursorRejectsAmbiguousJSON(t *testing.T) { + ns := core.ID("11111111111111111111111111111111") + at := time.Date(2026, 9, 18, 0, 0, 0, 0, time.UTC) + raw, _ := json.Marshal(workflowCursor{1, ns, at, ns, "digest"}) + encode := func(s string) string { return base64.RawURLEncoding.EncodeToString([]byte(s)) } + if _, err := decodeWorkflowCursor(encode(string(raw)), ns, instance.ListInput{}, "digest"); err != nil { + t.Fatal(err) + } + for _, bad := range []string{`null`, `[]`, strings.Replace(string(raw), `"v":1`, `"v":1,"v":1`, 1), strings.Replace(string(raw), `"v":1`, `"V":1`, 1), strings.Replace(string(raw), `"v":1`, `"v":null`, 1), strings.Replace(string(raw), `"v":1`, `"v":2`, 1), strings.Replace(string(raw), `"f":"digest"`, `"f":"other"`, 1), strings.Replace(string(raw), `"v":1`, `"v":1,"x":0`, 1)} { + if _, err := decodeWorkflowCursor(encode(bad), ns, instance.ListInput{}, "digest"); err == nil { + t.Fatal("accepted ambiguous cursor", bad) + } + } +} diff --git a/internal/storage/mysql/workflow_query_upgrade_integration_test.go b/internal/storage/mysql/workflow_query_upgrade_integration_test.go new file mode 100644 index 0000000..72f1403 --- /dev/null +++ b/internal/storage/mysql/workflow_query_upgrade_integration_test.go @@ -0,0 +1,183 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "slices" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "os" + "reflect" + "testing" + "time" + + driver "github.com/go-sql-driver/mysql" +) + +// 用真实v23结构验证范围回填，只读取DSL节点，不采集常量中的伪queue_id。 +func TestIntegrationWorkflowQueryUpgrade(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_query_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 > 23 { + 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, id, def := executionID(t), executionID(t), executionID(t) + q1, q2, q3, decoy := executionID(t), executionID(t), executionID(t), executionID(t) + spec := allInitialNodes(q1) + // 多层PARALLEL/BRANCH的未选分支、默认分支、POLL各自使用队列；重复值只保存一次。 + branches := spec.Body.Steps[0].Parallel.Branches + branches[1].Body.Steps[0].Poll.Task.QueueID = q2 + nestedTask := *branches[4].Body.Steps[0].Branch.Cases[0].Body.Steps[0].Task + nestedTask.QueueID = q3 + branches[4].Body.Steps[0].Branch.Cases[0].Body.Steps[0].Task = &nestedTask + // 无队列TIMER版本中，合法JSON输出常量故意含伪TASK/POLL和queue_id。 + timer := timerSpec(q1, 5000, false) + timer.Body.OutputType = definition.Type{Kind: "JSON"} + timer.Body.Output = definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`{"kind":"TASK","task":{"queue_id":"` + string(decoy) + `"},"body":{"steps":[{"kind":"POLL","poll":{"task":{"queue_id":"` + string(decoy) + `"}}}]}}`)} + specs := []definition.Specification{spec, timer} + ids := []core.ID{id, executionID(t)} + for i, spec := range specs { + checked, err := definition.Check(spec) + if err != nil { + t.Fatal(err) + } + raw, err := json.Marshal(spec) + if err != nil { + t.Fatal(err) + } + exec(`INSERT INTO weave_definition_versions(namespace_id,id,definition_id,version,schema_version,specification,content_digest,published_by,created_at,updated_at) VALUES (?,?,?, ?,1,?,?,?,'2026-01-02 03:04:05.123456','2026-01-02 03:04:05.123456')`, ns, ids[i], def, string(ids[i]), string(raw), checked.Digest, string(executionID(t))) + } + snapshot := func() [][]sql.NullString { + t.Helper() + rows, err := db.QueryContext(ctx, `SELECT * FROM weave_definition_versions WHERE namespace_id=? ORDER BY id`, ns) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + columns, err := rows.Columns() + if err != nil { + t.Fatal(err) + } + var out [][]sql.NullString + for rows.Next() { + row := make([]sql.NullString, len(columns)) + args := make([]any, len(columns)) + for i := range row { + args[i] = &row[i] + } + if err = rows.Scan(args...); err != nil { + t.Fatal(err) + } + out = append(out, row) + } + if err = rows.Err(); err != nil { + t.Fatal(err) + } + return out + } + before := snapshot() + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(before, snapshot()) { + t.Fatal("scope backfill changed original immutable versions") + } + for i, id := range ids { + var raw []byte + if err = db.QueryRowContext(ctx, `SELECT queue_ids FROM weave_version_scopes WHERE namespace_id=? AND definition_version_id=? AND definition_id=?`, ns, id, def).Scan(&raw); err != nil { + t.Fatal(err) + } + var queues []core.ID + if json.Unmarshal(raw, &queues) != nil || queues == nil { + t.Fatal("invalid scope", string(raw)) + } + slices.Sort(queues) + expected := []core.ID{} + if i == 0 { + expected = []core.ID{q1, q2, q3} + slices.Sort(expected) + } + if !slices.Equal(queues, expected) { + t.Fatal("incorrect DSL queue set", queues, expected) + } + } + for index, columns := range map[string]string{ + "ix_workflow_created": "namespace_id,created_at,id", + "ix_workflow_state_created": "namespace_id,state,created_at,id", + "ix_workflow_business_state_created": "namespace_id,business_key,state,created_at,id", + "ix_workflow_definition_created": "namespace_id,definition_id,created_at,id", + "ix_workflow_definition_state_created": "namespace_id,definition_id,state,created_at,id", + "ix_workflow_definition_business_created": "namespace_id,definition_id,business_key,created_at,id", + "ix_workflow_definition_business_state_created": "namespace_id,definition_id,business_key,state,created_at,id", + } { + var actual string + if err = db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='weave_instances' AND INDEX_NAME=?`, index).Scan(&actual); err != nil || actual != columns { + t.Fatal(index, actual, err) + } + } + if err = CheckSchema(ctx, db); err != nil { + t.Fatal(err) + } + var fks int + if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fks); err != nil || fks != 0 { + t.Fatal("foreign keys", fks, err) + } + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + t.Log("v23 upgraded; nested TASK/POLL queues match original DSL, literal decoys excluded, original versions unchanged, seven pagination indexes installed, foreign keys=0") +} diff --git a/internal/transport/httpserver/workflow_query.go b/internal/transport/httpserver/workflow_query.go new file mode 100644 index 0000000..4db57b5 --- /dev/null +++ b/internal/transport/httpserver/workflow_query.go @@ -0,0 +1,75 @@ +package httpserver + +import ( + "net/http" + "net/url" + "strconv" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/weave/instance" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +func workflowListQuery(r *http.Request) (instance.ListInput, error) { + var in instance.ListInput + q, err := url.ParseQuery(r.URL.RawQuery) + if err != nil { + return in, fault.Invalid("invalid query encoding") + } + for k, v := range q { + if len(v) != 1 { + return in, fault.Invalid("duplicate query parameter") + } + switch k { + case "definition_id", "version", "state", "paused", "business_key", "created_from", "created_before", "limit", "cursor": + default: + return in, fault.Invalid("unknown query parameter") + } + if v[0] == "" { + return in, fault.Invalid("empty query parameter") + } + } + in.DefinitionID = core.ID(q.Get("definition_id")) + in.Version = q.Get("version") + in.State = weave.InstanceState(q.Get("state")) + in.BusinessKey = q.Get("business_key") + in.Cursor = q.Get("cursor") + if q.Has("limit") { + in.Limit, err = strconv.Atoi(q.Get("limit")) + if err != nil || in.Limit <= 0 { + return in, fault.Invalid("invalid limit") + } + } + if q.Has("paused") { + v := q.Get("paused") + if v != "true" && v != "false" { + return in, fault.Invalid("paused must be true or false") + } + b := v == "true" + in.Paused = &b + } + for name, field := range map[string]**time.Time{"created_from": &in.CreatedFrom, "created_before": &in.CreatedBefore} { + if !q.Has(name) { + continue + } + v, err := time.Parse(time.RFC3339Nano, q.Get(name)) + if err != nil { + return in, fault.Invalid("invalid creation time bound") + } + *field = &v + } + return in, nil +} +func registerWorkflowQuery(mux *http.ServeMux, s *instance.Service) { + mux.HandleFunc("GET /v1/workflows", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + in, err := workflowListQuery(r) + if err != nil { + return nil, 0, err + } + out, err := s.List(r.Context(), t, in) + return out, 200, err + })) +} diff --git a/internal/transport/httpserver/workflow_query_test.go b/internal/transport/httpserver/workflow_query_test.go new file mode 100644 index 0000000..7f4fa61 --- /dev/null +++ b/internal/transport/httpserver/workflow_query_test.go @@ -0,0 +1,21 @@ +package httpserver + +import ( + "net/http/httptest" + "net/url" + "testing" +) + +func TestWorkflowListQueryStrictFilters(t *testing.T) { + for _, query := range []string{"limit=0", "limit=", "limit=x", "state=", "state=RUNNING&state=WAITING", "paused=0", "paused=True", "unknown=x", "business_key=", "created_from=tomorrow", "version=%zz", "state=WAITING;limit=2"} { + r := httptest.NewRequest("GET", "/v1/workflows?"+query, nil) + if _, err := workflowListQuery(r); err == nil { + t.Fatal("accepted", query) + } + } + q := url.Values{"business_key": {"关联 +/%\n "}, "paused": {"false"}, "created_from": {"2026-09-18T08:00:00.123456+08:00"}, "limit": {"2"}} + in, err := workflowListQuery(httptest.NewRequest("GET", "/v1/workflows?"+q.Encode(), nil)) + if err != nil || in.BusinessKey != q.Get("business_key") || in.Paused == nil || *in.Paused || in.Limit != 2 || in.CreatedFrom.UTC().Hour() != 0 { + t.Fatal(in, err) + } +} diff --git a/internal/transport/httpserver/workflows.go b/internal/transport/httpserver/workflows.go index 705c744..f379e6f 100644 --- a/internal/transport/httpserver/workflows.go +++ b/internal/transport/httpserver/workflows.go @@ -16,6 +16,7 @@ import ( func RegisterWorkflows(server *http.Server, instances *instance.Service, controls *control.Service, signals *signal.Service, reviews *recovery.Service) { mux := server.Handler.(*http.ServeMux) registerWorkflowHistory(mux, instances, signals) + registerWorkflowQuery(mux, instances) mux.HandleFunc("POST /v1/workflows", wrap(func(r *http.Request, t auth.Token) (any, int, error) { if _, err := historyQuery(r, false); err != nil { return nil, 0, err diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index bed1f00..ec30140 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -84,9 +84,10 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - # 流程发布、执行、控制、信号、核查和审计权限逐项显式签发。 ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap --application sdk-workflow --namespace default --environment test --queue workflow --workflow-access --worker-access --workflow-run-access --workflow-control-access --workflow-signal-access --workflow-recovery-access --workflow-audit-access --output "$TASK_TEST_DIR/workflow.credentials.json" -(cd sdk/go && ORBIT_TEST_WORKER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_WORKER_CREDENTIALS_FILE="$TASK_TEST_DIR/workflow.credentials.json" "$TASK_GO" test -race -count=1 -run '^TestIntegrationWorkflow(Definition|Instances)SDK$' -v .) +(cd sdk/go && ORBIT_TEST_WORKER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_WORKER_CREDENTIALS_FILE="$TASK_TEST_DIR/workflow.credentials.json" "$TASK_GO" test -race -count=1 -run '^TestIntegrationWorkflow(Definition|Instances|Query)SDK$' -v .) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/workflow.credentials.json" "$TASK_GO" run ./examples/workflow-definition) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/workflow.credentials.json" "$TASK_GO" run ./examples/workflow) +(cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/workflow.credentials.json" "$TASK_GO" run ./examples/workflow-query) # 接收方业务数据独立建库；数据库驱动仅属于此示例模块，不加入核心SDK。 "$TASK_DOCKER" exec "$TASK_TEST_NAME" sh -c 'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot --protocol=tcp -e "CREATE DATABASE orbit_receiver_test CHARACTER SET utf8mb4 COLLATE utf8mb4_bin"' diff --git a/sdk/go/examples/workflow-query/main.go b/sdk/go/examples/workflow-query/main.go new file mode 100644 index 0000000..c3408ef --- /dev/null +++ b/sdk/go/examples/workflow-query/main.go @@ -0,0 +1,116 @@ +// workflow-query 发布有限等待流程，按精确业务键和固定版本分页，并按控制版本结清示例实例。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} +func run() error { + raw, err := os.ReadFile(os.Getenv("ORBIT_CREDENTIALS_FILE")) + if err != nil { + return err + } + var credentials struct { + Token string `json:"token"` + } + if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" { + return errors.New("invalid private credentials file") + } + client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: credentials.Token}) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + operation, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + spec := orbit.WorkflowSpecification{SchemaVersion: 1, InputType: orbit.WorkflowType{Kind: "NULL"}, TimeoutMS: 120000, FailurePolicy: "REVIEW", Body: orbit.WorkflowBlock{ + Steps: []orbit.WorkflowStep{{Key: "receipt", Kind: "SIGNAL", Signal: &orbit.WorkflowSignal{Name: "object.ready", PayloadVersion: "v1", Correlation: orbit.WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"` + operation + `"`)}, PayloadType: orbit.WorkflowType{Kind: "BOOLEAN"}, TimeoutMS: 120000}}}, + OutputType: orbit.WorkflowType{Kind: "BOOLEAN"}, Output: orbit.WorkflowRef("receipt"), + }} + d, err := client.CreateWorkflowDefinition(ctx, operation, orbit.CreateWorkflowDefinitionRequest{Name: "query." + operation}) + if err != nil { + return err + } + if _, err = client.PublishWorkflowVersion(ctx, d.Definition.ID, operation, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "publish query example"}); err != nil { + return err + } + key := "对象 /+% é " + operation + " " + var instances []orbit.WorkflowInstance + for i, business := range []string{key, key, key[:len(key)-1]} { + out, err := client.StartWorkflow(ctx, fmt.Sprintf("%s-%d", operation, i), orbit.StartWorkflowRequest{DefinitionID: d.Definition.ID, Version: "v1", Input: json.RawMessage(`null`), BusinessKey: business}) + if err != nil { + return err + } + instances = append(instances, out.Instance) + } + from, before := instances[0].CreatedAt, instances[2].CreatedAt.Add(time.Microsecond) + query := orbit.ListWorkflowsRequest{DefinitionID: d.Definition.ID, Version: "v1", BusinessKey: key, State: "WAITING", CreatedFrom: &from, CreatedBefore: &before, Limit: 1} + seen := map[string]bool{} + pages := 0 + for { + page, err := client.ListWorkflows(ctx, query) + if err != nil { + return err + } + pages++ + for _, v := range page.Items { + if v.BusinessKey != key || v.DefinitionID != d.Definition.ID || v.Version != "v1" || seen[v.ID] { + return errors.New("workflow query changed filters or repeated an item") + } + seen[v.ID] = true + } + if page.NextCursor == "" { + break + } + query.Cursor = page.NextCursor + } + if len(seen) != 2 || pages != 2 || !seen[instances[0].ID] || !seen[instances[1].ID] { + return errors.New("expected exactly two matching workflow instances") + } + fmt.Printf("workflow_query_verified=true matched=%d pages=%d exact_trailing_space=true\n", len(seen), pages) + // 取消记录保留，固定定义版本和审计也保留；失败时不要直接删除运行记录。 + for _, v := range instances { + current, err := client.GetWorkflow(ctx, v.ID) + if err != nil { + return err + } + if _, err = client.ControlWorkflow(ctx, v.ID, operation+"-cleanup", orbit.WorkflowControlRequest{Operation: "CANCEL", ExpectedControlRevision: current.ControlRevision, Reason: "query example cleanup"}); err != nil { + return err + } + for { + current, err = client.GetWorkflow(ctx, v.ID) + if err != nil { + return err + } + if current.State == "CANCELLED" { + break + } + if current.State == "NEEDS_REVIEW" { + return errors.New("cleanup requires review") + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(50 * time.Millisecond): + } + } + } + fmt.Println("workflow_query_cleanup=CANCELLED count=3") + return nil +} diff --git a/sdk/go/workflow_query.go b/sdk/go/workflow_query.go new file mode 100644 index 0000000..8b4bc4b --- /dev/null +++ b/sdk/go/workflow_query.go @@ -0,0 +1,97 @@ +package orbit + +import ( + "context" + "errors" + "net/url" + "strconv" + "strings" + "time" + "unicode/utf8" +) + +// ListWorkflowsRequest 在认证命名空间内筛选实例；所有队列权限必须覆盖整个固定定义版本。 +type ListWorkflowsRequest struct { + DefinitionID string // DefinitionID 可选定义身份，未传入时跨已授权定义查询。 + Version string // Version 可选固定发布版本名，仅在指定定义时使用。 + State string // State 可选执行状态，与暂停标志独立。 + Paused *bool // Paused nil不筛选；false明确筛选未暂停实例。 + BusinessKey string // BusinessKey 非空时精确匹配，包含尾部空格；空表示不筛选。 + CreatedFrom *time.Time // CreatedFrom 可选创建时间下界，包含，UTC微秒精度。 + CreatedBefore *time.Time // CreatedBefore 可选创建时间上界，不包含。 + Limit int // Limit 默认20，上限100。 + Cursor string // Cursor 服务端返回的位置，绑定命名空间、全部筛选及队列权限范围。 +} + +// Summary 仅返回实例元数据，不聚合业务输入输出、定义或策略快照。 +type WorkflowSummary struct { + ID string `json:"id"` // ID 实例身份。 + NamespaceID string `json:"namespace_id"` // NamespaceID 授权及数据隔离范围。 + DefinitionID string `json:"definition_id"` // DefinitionID 稳定定义身份。 + DefinitionVersionID string `json:"definition_version_id"` // DefinitionVersionID 固定版本记录身份。 + Version string `json:"version"` // Version 启动时固定的发布版本名。 + State string `json:"state"` // State 执行状态，受理不代表业务成功。 + Revision uint64 `json:"revision,string"` // Revision 实例并发版本。 + Paused bool `json:"paused"` // Paused 独立暂停标志，不表示在途工作已停止。 + ControlRevision uint64 `json:"control_revision,string"` // ControlRevision 控制操作并发版本。 + BusinessKey string `json:"business_key"` // BusinessKey 原始追踪键。 + DeadlineAt time.Time `json:"deadline_at"` // DeadlineAt 数据库时间确定的固定总截止。 + CancelRequestedAt *time.Time `json:"cancel_requested_at,omitempty"` // CancelRequestedAt 已保存的取消意图时间。 + StopKind *string `json:"stop_kind,omitempty"` // StopKind 首次停止原因CANCEL或FAILURE；核查不会清除。 + StopRequestedAt *time.Time `json:"stop_requested_at,omitempty"` // StopRequestedAt 首次停止意图时间，后续操作不会延长。 + CompletedAt *time.Time `json:"completed_at,omitempty"` // CompletedAt 终结时间，非终态为空。 + CreatedAt time.Time `json:"created_at"` // CreatedAt UTC受理时间。 + UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最近状态更新时间。 +} + +type WorkflowPage struct { + Items []WorkflowSummary `json:"items"` // Items 按创建时间及ID倒序的授权摘要；空列表为[]。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 沿相同筛选与权限继续使用，空表示本次无下一页。 +} + +// ListWorkflows 查询当前凭据可读取的流程摘要；分页筛选和队列权限必须保持一致。 +func (c *Client) ListWorkflows(ctx context.Context, in ListWorkflowsRequest) (WorkflowPage, error) { + var out WorkflowPage + if in.Limit < 0 || in.Limit > 100 || len(in.Cursor) > 1024 || in.DefinitionID != "" && !validID(in.DefinitionID) || in.Version != "" && (in.DefinitionID == "" || !workflowVersionName(in.Version)) { + return out, errors.New("invalid workflow query definition, version or page") + } + switch in.State { + case "", "RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED": + default: + return out, errors.New("invalid workflow state") + } + if !utf8.ValidString(in.BusinessKey) || utf8.RuneCountInString(in.BusinessKey) > 191 || strings.ContainsRune(in.BusinessKey, 0) { + return out, errors.New("invalid workflow business key filter") + } + q := url.Values{} + for name, v := range map[string]string{"definition_id": in.DefinitionID, "version": in.Version, "state": in.State, "business_key": in.BusinessKey, "cursor": in.Cursor} { + if v != "" { + q.Set(name, v) + } + } + if in.Paused != nil { + q.Set("paused", strconv.FormatBool(*in.Paused)) + } + if in.Limit != 0 { + q.Set("limit", strconv.Itoa(in.Limit)) + } + for name, field := range map[string]*time.Time{"created_from": in.CreatedFrom, "created_before": in.CreatedBefore} { + if field == nil { + continue + } + v := field.UTC() + if v.Year() < 2000 || v.Year() > 9998 || v.Nanosecond()%1000 != 0 { + return out, errors.New("invalid workflow creation time bound") + } + q.Set(name, v.Format(time.RFC3339Nano)) + } + if in.CreatedFrom != nil && in.CreatedBefore != nil && !in.CreatedFrom.Before(*in.CreatedBefore) { + return out, errors.New("created_from must be before created_before") + } + path := "/v1/workflows" + if len(q) > 0 { + path += "?" + q.Encode() + } + err := c.call(ctx, "GET", path, "", nil, &out) + return out, err +} diff --git a/sdk/go/workflow_query_integration_test.go b/sdk/go/workflow_query_integration_test.go new file mode 100644 index 0000000..18613fb --- /dev/null +++ b/sdk/go/workflow_query_integration_test.go @@ -0,0 +1,84 @@ +package orbit + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "testing" + "time" +) + +func TestIntegrationWorkflowQuerySDK(t *testing.T) { + api, queue, token := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + spec := workflowSDKSpec(queue, "query."+workerTestID(t)) + // SIGNAL保持稳定WAITING，不启动Worker或等待短时任务截止。 + spec.Body = WorkflowBlock{Steps: []WorkflowStep{{Key: "receipt", Kind: "SIGNAL", Signal: &WorkflowSignal{Name: "done", PayloadVersion: "v1", Correlation: WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"query"`)}, PayloadType: spec.InputType, TimeoutMS: 60000}}}, OutputType: spec.InputType, Output: WorkflowRef("receipt")} + in := publishWorkflowSDK(t, ctx, api, spec) + in.BusinessKey = "作业 /+%\n " + workerTestID(t) + " " + var started []WorkflowInstance + for _, key := range []string{in.BusinessKey, in.BusinessKey, strings.TrimSpace(in.BusinessKey)} { + in.BusinessKey = key + out, err := api.StartWorkflow(ctx, workerTestID(t), in) + if err != nil { + t.Fatal(err) + } + started = append(started, out.Instance) + } + paused, err := api.ControlWorkflow(ctx, started[0].ID, workerTestID(t), WorkflowControlRequest{Operation: "PAUSE", ExpectedControlRevision: started[0].ControlRevision, Reason: "verify paused query"}) + if err != nil || !paused.Instance.Paused { + t.Fatal(paused, err) + } + from, before := started[0].CreatedAt, started[2].CreatedAt.Add(time.Microsecond) + query := ListWorkflowsRequest{DefinitionID: in.DefinitionID, Version: "v1", State: "WAITING", BusinessKey: started[0].BusinessKey, CreatedFrom: &from, CreatedBefore: &before, Limit: 1} + first, err := api.ListWorkflows(ctx, query) + if err != nil || len(first.Items) != 1 || first.Items[0].ID != started[1].ID || first.NextCursor == "" { + t.Fatal(first, err) + } + query.Cursor = first.NextCursor + last, err := api.ListWorkflows(ctx, query) + if err != nil || len(last.Items) != 1 || last.Items[0].ID != started[0].ID || !last.Items[0].Paused || last.NextCursor != "" { + t.Fatal(last, err) + } + query.BusinessKey = strings.TrimSpace(query.BusinessKey) + _, err = api.ListWorkflows(ctx, query) + workflowAPIError(t, err, "INVALID_ARGUMENT") + query.Cursor = "" + if page, err := api.ListWorkflows(ctx, query); err != nil || len(page.Items) != 1 || page.Items[0].ID != started[2].ID { + t.Fatal("exact business key", page, err) + } + b := false + query.BusinessKey = started[0].BusinessKey + query.Paused = &b + if page, err := api.ListWorkflows(ctx, query); err != nil || len(page.Items) != 1 || page.Items[0].ID != started[1].ID { + t.Fatal("false paused filter", page, err) + } + for _, suffix := range []string{"?state=WAITING&state=RUNNING", "?paused=1", "?business_key=", "?unknown=x", "?version=v1", "?limit=101"} { + req, err := http.NewRequestWithContext(ctx, "GET", api.baseURL+"/v1/workflows"+suffix, nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer "+token) + res, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + res.Body.Close() + if res.StatusCode != 400 { + t.Fatal("invalid raw query", suffix, res.StatusCode) + } + } + for _, v := range started { + current, err := api.GetWorkflow(ctx, v.ID) + if err != nil { + t.Fatal(err) + } + if _, err = api.ControlWorkflow(ctx, v.ID, workerTestID(t), WorkflowControlRequest{Operation: "CANCEL", ExpectedControlRevision: current.ControlRevision, Reason: "query regression cleanup"}); err != nil { + t.Fatal(err) + } + waitWorkflowSDK(t, ctx, api, v.ID, "CANCELLED") + } + t.Log("real HTTP: exact key, time bounds, fixed version, paused false, filtered cursor, malformed query rejection and cancellation cleanup") +} diff --git a/sdk/go/workflow_query_test.go b/sdk/go/workflow_query_test.go new file mode 100644 index 0000000..f1c1600 --- /dev/null +++ b/sdk/go/workflow_query_test.go @@ -0,0 +1,48 @@ +package orbit + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestWorkflowQueryEncodingAndBounds(t *testing.T) { + const id = "11111111111111111111111111111111" + from := time.Date(2026, 9, 17, 8, 0, 0, 123456000, time.FixedZone("sample", 8*3600)) + before := from.Add(time.Hour) + key := "业务 /+% é " + paused := false + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + q := r.URL.Query() + if r.Method != "GET" || r.URL.Path != "/v1/workflows" || r.Header.Get("Idempotency-Key") != "" || r.Header.Get("Authorization") != "Bearer example-token" || q.Get("paused") != "false" || q.Get("definition_id") != id || q.Get("version") != "v1" || q.Get("business_key") != key || q.Get("created_from") != "2026-09-17T00:00:00.123456Z" || q.Get("created_before") != "2026-09-17T01:00:00.123456Z" || q.Get("cursor") != "opaque+cursor/=" { + t.Error("query lost meaning", r.URL.String()) + } + json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": WorkflowPage{Items: []WorkflowSummary{{ID: id, Revision: 9007199254740993, BusinessKey: key}}}}) + })) + defer server.Close() + c, _ := NewClient(server.URL, Options{Token: "example-token"}) + base := ListWorkflowsRequest{Paused: &paused, DefinitionID: id, Version: "v1", State: "WAITING", BusinessKey: key, CreatedFrom: &from, CreatedBefore: &before, Limit: 2, Cursor: "opaque+cursor/="} + page, err := c.ListWorkflows(context.Background(), base) + if err != nil || len(page.Items) != 1 || page.Items[0].BusinessKey != key || page.Items[0].Revision != 9007199254740993 { + t.Fatal(page, err) + } + for name, change := range map[string]func(*ListWorkflowsRequest){"definition": func(v *ListWorkflowsRequest) { v.DefinitionID = "../queue" }, "version without definition": func(v *ListWorkflowsRequest) { v.DefinitionID = "" }, "version": func(v *ListWorkflowsRequest) { v.Version = "../v1" }, "state": func(v *ListWorkflowsRequest) { v.State = "unknown" }, "limit": func(v *ListWorkflowsRequest) { v.Limit = 101 }, "cursor": func(v *ListWorkflowsRequest) { v.Cursor = strings.Repeat("a", 1025) }, "utf8": func(v *ListWorkflowsRequest) { v.BusinessKey = "\xff" }, "key": func(v *ListWorkflowsRequest) { v.BusinessKey = strings.Repeat("字", 192) }, "nul": func(v *ListWorkflowsRequest) { v.BusinessKey = "x\x00" }, "equal": func(v *ListWorkflowsRequest) { v.CreatedBefore = &from }, "nanosecond": func(v *ListWorkflowsRequest) { z := from.Add(time.Nanosecond); v.CreatedFrom = &z }, "year": func(v *ListWorkflowsRequest) { z := time.Time{}; v.CreatedFrom = &z }} { + t.Run(name, func(t *testing.T) { + in := base + change(&in) + if _, err := c.ListWorkflows(context.Background(), in); err == nil { + t.Fatal("invalid query accepted") + } + }) + } + if calls.Load() != 1 { + t.Fatal("invalid query reached HTTP") + } +} diff --git a/weave/instance/query.go b/weave/instance/query.go new file mode 100644 index 0000000..f4d3f44 --- /dev/null +++ b/weave/instance/query.go @@ -0,0 +1,83 @@ +package instance + +import ( + "context" + "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" + weave "git.yuntuo.hk/k2safe/orbit/weave/model" +) + +// ListInput 在认证命名空间内筛选实例；所有队列权限必须覆盖整个固定定义版本。 +type ListInput struct { + DefinitionID core.ID // DefinitionID 可选定义身份，未传入时跨已授权定义查询。 + Version string // Version 可选固定发布版本名，仅在指定定义时使用。 + State weave.InstanceState // State 可选执行状态，与暂停标志独立。 + Paused *bool // Paused nil不筛选；false明确筛选未暂停实例。 + BusinessKey string // BusinessKey 非空时精确匹配，包含尾部空格；空表示不筛选。 + CreatedFrom *time.Time // CreatedFrom 可选创建时间下界，包含，UTC微秒精度。 + CreatedBefore *time.Time // CreatedBefore 可选创建时间上界，不包含。 + Limit int // Limit 默认20，上限100。 + Cursor string // Cursor 服务端返回的位置，绑定命名空间、全部筛选及队列权限范围。 +} + +// Summary 仅返回实例元数据，不聚合业务输入输出、定义或策略快照。 +type Summary 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 控制操作并发版本。 + BusinessKey string `json:"business_key"` // BusinessKey 原始追踪键。 + DeadlineAt time.Time `json:"deadline_at"` // DeadlineAt 数据库时间确定的固定总截止。 + CancelRequestedAt *time.Time `json:"cancel_requested_at,omitempty"` // CancelRequestedAt 已保存的取消意图时间。 + StopKind *string `json:"stop_kind,omitempty"` // StopKind 首次停止原因CANCEL或FAILURE；核查不会清除。 + StopRequestedAt *time.Time `json:"stop_requested_at,omitempty"` // StopRequestedAt 首次停止意图时间，后续操作不会延长。 + CompletedAt *time.Time `json:"completed_at,omitempty"` // CompletedAt 终结时间，非终态为空。 + CreatedAt time.Time `json:"created_at"` // CreatedAt UTC受理时间。 + UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最近状态更新时间。 +} + +type Page struct { + Items []Summary `json:"items"` // Items 按创建时间及ID倒序的授权摘要；空列表为[]。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 沿相同筛选与权限继续使用，空表示本次无下一页。 +} + +func (s *Service) List(ctx context.Context, t auth.Token, in ListInput) (Page, error) { + if in.Limit == 0 { + in.Limit = 20 + } + if in.Limit < 1 || in.Limit > 100 || len(in.Cursor) > 1024 || in.DefinitionID != "" && in.DefinitionID.Validate() != nil || in.Version != "" && (in.DefinitionID == "" || !versionName(in.Version)) { + return Page{}, fault.Invalid("invalid workflow query definition, version or page") + } + switch in.State { + case "", "RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED": + default: + return Page{}, fault.Invalid("invalid workflow state") + } + if !utf8.ValidString(in.BusinessKey) || utf8.RuneCountInString(in.BusinessKey) > 191 || strings.ContainsRune(in.BusinessKey, 0) { + return Page{}, fault.Invalid("invalid workflow business key filter") + } + for _, f := range []**time.Time{&in.CreatedFrom, &in.CreatedBefore} { + if *f == nil { + continue + } + v := (*f).UTC() + if v.Year() < 2000 || v.Year() > 9998 || v.Nanosecond()%1000 != 0 { + return Page{}, fault.Invalid("invalid workflow creation time bound") + } + *f = &v + } + if in.CreatedFrom != nil && in.CreatedBefore != nil && !in.CreatedFrom.Before(*in.CreatedBefore) { + return Page{}, fault.Invalid("created_from must be before created_before") + } + return s.store.List(ctx, t, in) +} diff --git a/weave/instance/query_test.go b/weave/instance/query_test.go new file mode 100644 index 0000000..e474e02 --- /dev/null +++ b/weave/instance/query_test.go @@ -0,0 +1,27 @@ +package instance + +import ( + "context" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +func TestListBoundary(t *testing.T) { + store := &captureStore{} + api := New(store) + if _, err := api.List(context.Background(), auth.Token{}, ListInput{}); err != nil { + t.Fatal(err) + } + now := time.Now().UTC().Truncate(time.Microsecond) + for _, in := range []ListInput{{Limit: -1}, {Limit: 101}, {Cursor: strings.Repeat("x", 1025)}, {DefinitionID: "bad"}, {Version: "v1"}, {DefinitionID: "11111111111111111111111111111111", Version: "../v1"}, {State: "STOPPED"}, {BusinessKey: "\xff"}, {BusinessKey: "nul\x00"}, {BusinessKey: strings.Repeat("中", 192)}, {CreatedFrom: &now, CreatedBefore: &now}} { + if _, err := api.List(context.Background(), auth.Token{}, in); err == nil { + t.Fatal("invalid query accepted", in) + } + } + if store.calls != 1 { + t.Fatal("invalid query reached store") + } +} diff --git a/weave/instance/service.go b/weave/instance/service.go index e069d06..87323f3 100644 --- a/weave/instance/service.go +++ b/weave/instance/service.go @@ -72,6 +72,7 @@ type StepsView struct { Steps []StepView `json:"steps"` // Steps 按键排序，最多128个，不含内部根检查点。 } type Store interface { + List(context.Context, auth.Token, ListInput) (Page, error) ListActivations(context.Context, auth.Token, core.ID, string, service.HistoryPageInput) (ActivationPage, error) GetActivation(context.Context, auth.Token, core.ID, string, core.ID) (ActivationDetail, error) ListPollRounds(context.Context, auth.Token, core.ID, string, core.ID, service.HistoryPageInput) (PollRoundPage, error) diff --git a/weave/instance/service_test.go b/weave/instance/service_test.go index a72a2e7..5ab49be 100644 --- a/weave/instance/service_test.go +++ b/weave/instance/service_test.go @@ -84,3 +84,8 @@ func (s *captureStore) ListPollRounds(context.Context, auth.Token, core.ID, stri s.calls++ return PollRoundPage{}, nil } + +func (s *captureStore) List(context.Context, auth.Token, ListInput) (Page, error) { + s.calls++ + return Page{}, nil +} diff --git a/weave/model/version_scope.go b/weave/model/version_scope.go new file mode 100644 index 0000000..5f56817 --- /dev/null +++ b/weave/model/version_scope.go @@ -0,0 +1,20 @@ +package model + +import ( + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" +) + +// VersionScope 是固定流程版本的队列查询索引；不能替代原定义校验或授予执行权限。 +// 对应weave_version_scopes，不使用外键；发布与本索引在同一事务提交。 +type VersionScope struct { + NamespaceID core.ID `db:"namespace_id"` // NamespaceID 命名空间，与版本共同构成唯一身份及授权边界。 + DefinitionVersionID core.ID `db:"definition_version_id"` // DefinitionVersionID 固定定义版本标识，包含未执行分支的全部队列。 + DefinitionID core.ID `db:"definition_id"` // DefinitionID 所属定义，查询必须与版本和实例共同核对。 + ContentDigest core.Digest `db:"content_digest"` // ContentDigest 原不可变定义摘要，不能用其他版本的范围授权。 + QueueIDs core.JSON `db:"queue_ids"` // QueueIDs 去重的32位队列标识JSON数组，最多128项；空数组表示流程不执行任务，顺序无语义。 + CreatedAt time.Time `db:"created_at"` // CreatedAt 索引建立时间，UTC；不改变原版本发布时间或实例截止。 +} + +func (VersionScope) TableName() string { return "weave_version_scopes" }