From a4d9900e2cabe5b096f1912c58ec5b7a1fe57df4 Mon Sep 17 00:00:00 2001 From: k2safe Date: Fri, 18 Sep 2026 09:27:45 +0800 Subject: [PATCH] feat: expose workflow state events and persist signal transitions --- AGENTS.md | 14 +- README.md | 2 +- docs/07-data-model.md | 8 + docs/08-development-plan.md | 16 +- docs/README.md | 1 + docs/api/README.md | 1 + docs/api/workflow-events.md | 158 +++++++++ docs/api/workflow-history.md | 2 +- docs/api/workflows.md | 2 + docs/examples/README.md | 2 +- docs/examples/workflow.md | 11 +- docs/reports/phase-5-events.md | 58 ++++ docs/sdk/README.md | 2 + ...tion_resources_upgrade_integration_test.go | 1 + .../000025_workflow_event_timeline.up.sql | 15 + .../mysql/no_foreign_keys_integration_test.go | 4 +- internal/storage/mysql/workflow_settlement.go | 4 +- .../storage/mysql/workflow_signal_events.go | 97 ++++++ .../storage/mysql/workflow_signal_runtime.go | 5 +- .../storage/mysql/workflow_signal_source.go | 5 +- internal/storage/mysql/workflow_timeline.go | 266 +++++++++++++++ .../workflow_timeline_integration_test.go | 319 ++++++++++++++++++ ...kflow_timeline_upgrade_integration_test.go | 157 +++++++++ .../transport/httpserver/workflow_events.go | 27 ++ internal/transport/httpserver/workflows.go | 1 + platform/model/event.go | 2 + scripts/test-mysql.sh | 2 +- sdk/go/examples/workflow/main.go | 33 ++ sdk/go/workflow_events.go | 88 +++++ sdk/go/workflow_events_integration_test.go | 80 +++++ sdk/go/workflow_events_test.go | 78 +++++ weave/instance/events.go | 54 +++ weave/instance/service.go | 2 + weave/instance/service_test.go | 9 + 34 files changed, 1499 insertions(+), 27 deletions(-) create mode 100644 docs/api/workflow-events.md create mode 100644 docs/reports/phase-5-events.md create mode 100644 internal/storage/mysql/migrations/000025_workflow_event_timeline.up.sql create mode 100644 internal/storage/mysql/workflow_signal_events.go create mode 100644 internal/storage/mysql/workflow_timeline.go create mode 100644 internal/storage/mysql/workflow_timeline_integration_test.go create mode 100644 internal/storage/mysql/workflow_timeline_upgrade_integration_test.go create mode 100644 internal/transport/httpserver/workflow_events.go create mode 100644 sdk/go/workflow_events.go create mode 100644 sdk/go/workflow_events_integration_test.go create mode 100644 sdk/go/workflow_events_test.go create mode 100644 weave/instance/events.go diff --git a/AGENTS.md b/AGENTS.md index 0faf016..ea7b481 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,16 +52,16 @@ - 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新轮无父任务。不能改指当前激活、递归展开历史或将来源查询加入派发路径；无外键。 @@ -69,3 +69,5 @@ - 历史读取按固定实例及所有队列授权，明确步骤/激活/信号归属，不依赖外键；摘要分页20/100，游标绑定种类和父资源。指定激活详情的Step是当前视图、Activation是历史视图，不自动改代次；POLL轮次限定单个激活，恢复同轮任务不可与下一业务轮混淆。信号列表保留未核验旧身份，正文单条读取；信号审计用显式workflows:audit:read，不能按全局审计ID越权。000023只增加分页索引；全量回归必须保留零外键与旧数据/索引断言。 - 实例列表按固定定义全部队列在SQL中筛选，再校验原定义/摘要、范围、策略及关联；不能用已执行任务推断权限。000024增加不可变范围索引与7个列表索引，无外键；发布与范围同事务，旧DSL常量中的queue_id不纳入范围。游标绑定命名空间/全部筛选/规范化队列权限，跨页不承诺快照，limit只限制返回候选量。先排序锁整页队列，不在实例锁之后追加队列锁。 + +- 状态时间线只读取已存workflow/workflow_step/workflow_signal事件，排除$root；两条路由均显式workflows:audit:read并核对固定版本全部队列及实际子资源。000025派生只读索引不授予权限、不改历史原文、不补造旧信号转换，无外键；新信号状态和事件同事务，批量拒绝32条一批且失败整笔回滚。时间/ID排序与资源revision均不是提交水位；Pulse可靠消费仍走持久订阅。旧写节点停止后迁移，不能承诺混合版本无历史缺口。 diff --git a/README.md b/README.md index f23ea84..d395f21 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， 已完成 P0–P3 平台功能阶段验收：36 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 、[队列资源控制](docs/api/queues.md) 、[应用共享配额](docs/api/application-resources.md) 、[资源键互斥](docs/api/keyed-concurrency.md) 和 [优先级/节点公平派发](docs/api/priority-and-fairness.md) 已接入服务端和 SDK，[可靠结果订阅](docs/api/subscriptions.md) 已实现持久化核心、HTTP/WSS、管理 SDK 与示例，[消费 SDK](docs/sdk/event-consumer.md) 和 [独立 MySQL Inbox 示例](docs/examples/result-inbox.md) 已实现；[批量任务受理](docs/api/batch-tasks.md) 已提供逐项幂等、失败分类与 SDK 示例；[人工核查与重试](docs/api/task-recovery.md) 已提供带证据的管理接口、关联新任务、审计查询及运行示例；[业务键与时间范围查询](docs/api/task-queries.md) 已提供过滤、分页与SDK场景；[任务来源查询](docs/api/task-source.md) 已提供人工重试父任务、计划触发记录与固定版本；[业务事务Outbox](docs/examples/task-outbox.md) 已提供独立业务库、稳定提交身份、发送恢复与CLI；[相对延迟执行](docs/examples/delayed-task.md)已提供数据库计时及原请求恢复；完整范围核对见[P3验收](docs/reports/phase-3.md)，流程引擎与生产验证按 [持续开发计划](docs/08-development-plan.md) 继续实现。 -Weave已提供[定义发布](docs/api/workflow-definitions.md)、[实例启动、步骤详情、暂停/继续/取消、信号和操作审计](docs/api/workflows.md)，以及[六类人工核查恢复](docs/api/workflow-recovery.md)的HTTP与独立Go SDK。持久运行器支持顺序、分支、有限并行、TIMER、SIGNAL及有界POLL，恢复沿用原预算与截止。[并行作业与回执示例](docs/examples/workflow.md)提供完整调用链；[历史激活/轮次/信号及信号审计](docs/api/workflow-history.md)已接入；[实例筛选分页](docs/api/workflow-queries.md)已接入，统一事件时间线范围核对和更多独立恢复示例继续交付；[任务流程来源](docs/api/task-source.md#5-流程固定来源)已接入，P4/P5与生产验收尚未整体完成。 +Weave已提供[定义发布](docs/api/workflow-definitions.md)、[实例启动、步骤详情、暂停/继续/取消、信号和操作审计](docs/api/workflows.md)，以及[六类人工核查恢复](docs/api/workflow-recovery.md)的HTTP与独立Go SDK。持久运行器支持顺序、分支、有限并行、TIMER、SIGNAL及有界POLL，恢复沿用原预算与截止。[并行作业与回执示例](docs/examples/workflow.md)提供完整调用链；[历史激活/轮次/信号及信号审计](docs/api/workflow-history.md)已接入；[实例筛选分页](docs/api/workflow-queries.md)已接入，[状态事件时间线](docs/api/workflow-events.md)已接入，更多独立轮询/恢复示例继续交付；[任务流程来源](docs/api/task-source.md#5-流程固定来源)已接入，P4/P5与生产验收尚未整体完成。 数据库禁止外键，关联一致性通过服务端事务校验，索引、唯一约束和 CHECK 保留；readiness 同时检查迁移及零外键要求。这不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 阶段验收](docs/reports/phase-2.md)、[安全重试](docs/reports/phase-3-safe-retry.md)、[计划调度](docs/reports/phase-3-schedules.md)、[取消与无外键回归](docs/reports/phase-3-cancellation.md) 及 [队列资源回归](docs/reports/phase-3-queue-resources.md)。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 051b419..5d43074 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -237,3 +237,11 @@ PollState新增FirstRound和FirstRoundAttemptsUsed，记录当前激活的首个 VersionScope新增6个中文注释字段：命名空间、定义版本、定义、原摘要、去重队列JSON、索引创建时间；当前36个实体、311个直接持久化字段声明，全部有中文注释。版本范围不可更新，发布与范围同事务，读取摘要还要核验原定义和关联，不以派生索引授予权限。回填仅遍历DSL步骤，忽略常量中相同字段；旧版本及实例原文不修改。 7个新实例复合索引加现有业务键索引覆盖定义/业务键/状态8种组合，按创建时间与ID分页；全部保留无外键和无级联。范围校验、迁移性能边界和接口见[查询手册](api/workflow-queries.md)。 + +## 流程事件派生索引（000025） + +Event.WorkflowID为只读STORED生成列，从原不可变事件派生流程归属，非流程或非法标识为空；应用不得独立写入，也不能将它作为授权依据。索引为(namespace_id,workflow_id,occurred_at,id)。历史原文不变，不补造过去缺失的信号转换；当前36个模型、312个直接持久化字段均有中文注释，无外键及级联。 + +新运行器将PENDING信号的消费、拒绝或过期与原状态事件同事务保存，批量拒绝分32条追加但整笔事务原子提交。公开查询核对固定版本全部队列、步骤/信号归属及历史激活。原始事件按数据库记录时间分页，版本与时间均不构成可靠消费水位。详见[状态时间线接口](api/workflow-events.md)与[验证报告](reports/phase-5-events.md)。 + +升级先停止旧写入节点，应用000025再启新版本，防止混合节点继续遗漏信号转换；生成列和索引涉及历史行扫描与可能的表重建，需预演维护窗口、磁盘和锁等待。严格SQL模式下先校验十六进制字符再转换，异常历史范围保留NULL，不清理原记录。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index def72f1..99654b6 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -28,30 +28,30 @@ | 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与基础可运行场景已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；[状态事件时间线](api/workflow-events.md)已接入；独立轮询/恢复场景及全范围验收继续开发 | +| P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 有界查询轮次/尝试独立预算及原截止恢复已接入内部运行器；最终信号竞争及在途查询结清已接入；内部暂停/继续/取消与失败结清已接入；执行核查、步骤结论、解除核查及TASK新激活已接入；[POLL跨激活恢复](17-workflow-poll-recovery.md)已接入；等待恢复已接入内部服务；公共API/SDK已接入；流程任务来源已接入；历史激活/轮次/信号与信号审计已接入；实例筛选分页已接入；[状态事件时间线](api/workflow-events.md)已接入；独立轮询/恢复场景及全范围验收继续开发 | | 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)已接入，[实例筛选分页](api/workflow-queries.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)已接入，[状态事件时间线](api/workflow-events.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任务流程来源已交付 | 当前实例/步骤/任务版本、证据和显式声明、旧历史保留、有界响应和权限隔离 | +| 核查与追踪 | 六类恢复、激活/轮次/等待/信号历史读取及信号管理审计已接入；状态事件时间线已接入；Pulse任务流程来源已交付 | 当前实例/步骤/任务版本、证据和显式声明、旧历史保留、有界响应和权限隔离 | | 通用接入场景 | 并行汇合、外部回执、有限轮询、暂停核查恢复的可运行SDK示例 | 真实Worker执行、完整进程退出后接续、未知效果保留、全链路回归 | 接口只使用已经落地的内部事务规则；SDK保持独立module，不能引用服务端模型或数据库实现。生产故障及容量结论仍由P6专门验收。 @@ -176,3 +176,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( - [P2 SDK 协议阶段记录](reports/phase-2-sdk-protocol.md) - [P2 自动 Worker 运行器记录](reports/phase-2-worker-runtime.md) - [P2 执行历史与大载荷记录](reports/phase-2-execution-history.md) + +### 状态事件时间线子项 + +已接入流程/步骤/信号状态事件的公共HTTP、独立SDK、原元数据详情与完整运行例子。000025仅派生索引，历史原字段保持；新增信号转换与状态同事务，批量失败整体回滚。当前阶段验证见[时间线报告](reports/phase-5-events.md)，P4/P5全范围核对、独立POLL/人工恢复示例和P6/P7仍未完成。 diff --git a/docs/README.md b/docs/README.md index 18db9a0..5112841 100644 --- a/docs/README.md +++ b/docs/README.md @@ -23,6 +23,7 @@ | [12 Weave结果消费与推进](12-workflow-runtime.md) | 后台消费、根接续、事务与进程恢复，及尚未完成的等待/控制边界 | | [13 Weave定时等待与查询轮次](13-workflow-clock.md) | 原截止、轮次/尝试预算、时钟检查点、升级与全停恢复 | | [流程历史查询](api/workflow-history.md) | 指定历史激活、查询轮次、信号摘要与信号资源审计 | +| [流程状态时间线](api/workflow-events.md) | 原始流程/步骤/信号状态事件、分页和历史详情 | | [接口手册](api/README.md) | 实际接口状态、调用说明与后续接口规划 | | [SDK 与场景](sdk/README.md) | 可运行示例、接入方法及分阶段 SDK 交付 | diff --git a/docs/api/README.md b/docs/api/README.md index b4dc6f7..705ee8f 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -29,6 +29,7 @@ | `/v1/workflow-definitions` 创建、发布与固定版本读取 | P4定义子项已实现，实例另用启动接口 | [定义接口](workflow-definitions.md)、[格式规范](workflow-definition-format.md) | | `/v1/workflows`、步骤详情、控制、信号及操作审计 | 已接入HTTP与独立SDK | [实例接口](workflows.md) | | 历史激活/轮次/信号列表、历史激活详情及信号资源审计 | 已接入有界分页HTTP与独立SDK | [历史查询](workflow-history.md) | +| `/v1/workflows/{id}/events` 及单条事件详情 | 已接入不可变状态事件、审计权限及独立SDK | [状态时间线](workflow-events.md) | | `/v1/workflows/{id}/recovery/*` 六类人工恢复 | 已接入有证据恢复及独立SDK | [恢复接口](workflow-recovery.md) | 未实现路由返回 404。健康检查就绪不表示任务执行、定时调度或流程引擎已经可用。 diff --git a/docs/api/workflow-events.md b/docs/api/workflow-events.md new file mode 100644 index 0000000..cb38070 --- /dev/null +++ b/docs/api/workflow-events.md @@ -0,0 +1,158 @@ +# 流程状态事件时间线 + +HTTP 与独立 Go SDK 已接入。用于查看流程、用户定义步骤与信号已经保存的状态变化；原事件不会被后续恢复覆盖。运行方法见[并行作业与回执示例](../examples/workflow.md)，验证记录见[时间线报告](../reports/phase-5-events.md)。 + +## 1. 路由、权限与范围 + +| 方法与路径 | 权限 | Go SDK | +| --- | --- | --- | +| `GET /v1/workflows/{id}/events` | workflows:audit:read | ListWorkflowEvents(ctx, id, page) | +| `GET /v1/workflows/{id}/events/{event}` | workflows:audit:read | GetWorkflowEvent(ctx, id, event) | + +成功为200，使用 `Authorization: Bearer `，无需请求体或幂等键。id和event均为32位小写十六进制。两条路由都要求显式审计权限，普通workflows:read不包含此权限；开发初始化用 `--workflow-audit-access`。每次读取都核验当前命名空间及实例固定版本引用的全部队列，包括尚未执行的分支。 + +数据库派生的workflow_id只用于缩小查询范围，不能授予权限。服务端继续核对真实实例、步骤、信号及事件引用的历史激活归属；内部`$root`步骤不出现在公开时间线。详情必须同时指定父实例和事件ID，不能用全局事件ID越权。 + +本接口提供逻辑状态事件。任务执行次数、原输入结果、POLL业务轮次、等待预算和人工决定通过下列接口关联查看： + +| 要查看的内容 | 接口 | +| --- | --- | +| 当前步骤、任务引用、等待原因和剩余预算 | [实例与步骤详情](workflows.md) | +| 指定历史激活的输入输出、等待、POLL轮次 | [流程历史](workflow-history.md) | +| Pulse尝试与迟到/矛盾证据 | [执行历史](execution-history.md) | +| 管理操作人、原因、证据及前后摘要 | [流程操作审计](workflows.md#6-操作审计与错误处理)、[信号审计](workflow-history.md) | +| 任务的固定版本、步骤和历史激活来源 | [任务来源](task-source.md#5-流程固定来源) | + +事件元数据只保留当时实际写入的引用，例如activation_id或audit_id；旧事件没有的字段不会用当前资源补造。操作审计与Pulse尝试不复制进本接口，避免将不同记录误认为同一种状态转换。 + +## 2. 摘要分页 + +只接受可选limit和cursor。limit为1–100，省略默认20；SDK的0表示省略。cursor最多1024字节，需进行URL编码。未知、重复、非法参数返回400；详情不接受查询参数。 + +```sh +curl --get "$ORBIT_URL/v1/workflows/$WORKFLOW_ID/events" \ + --header "Authorization: Bearer $ORBIT_TOKEN" \ + --data-urlencode 'limit=20' +``` + +响应示例： + +```json +{ + "request_id": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "data": { + "workflow_id": "11111111111111111111111111111111", + "revision": "12", + "items": [{ + "id": "22222222222222222222222222222222", + "workflow_id": "11111111111111111111111111111111", + "aggregate_kind": "workflow_signal", + "aggregate_id": "33333333333333333333333333333333", + "aggregate_revision": "2", + "type": "workflow.signal.changed", + "state_scope": "SIGNAL", + "state": "CONSUMED", + "occurred_at": "2026-09-18T08:00:00.123456Z" + }] + } +} +``` + +| 字段 | 含义 | +| --- | --- | +| workflow_id | 本页授权的实例 | +| revision | 读取本页时的实例版本，uint64字符串；不是事件消费水位 | +| items | 有界摘要；空列表为[]，不带payload或业务输入结果 | +| next_cursor | 还有更旧记录时返回；没有下一页则省略 | +| items[].id | 原不可变事件ID | +| aggregate_kind / aggregate_id | 来源类型及身份：workflow、workflow_step、workflow_signal | +| aggregate_revision | 来源资源当时的版本，uint64字符串，各资源独立计数 | +| type | 当时保存的事件类型 | +| state_scope / state | WORKFLOW、STEP或SIGNAL，以及该对象当时保存的状态 | +| step_key | 仅步骤事件返回，取自固定定义，不返回内部根步骤 | +| occurred_at | 数据库记录事件的UTC微秒时间，RFC3339；不是外部业务发生时间 | + +按occurred_at、id双键倒序，相同微秒由ID稳定定位。游标绑定命名空间、实例和事件查询种类，不能用于其他实例、操作审计或激活列表。跨页不承诺固定快照，新提交或迟到提交的事件可能落在已经翻过的位置；重新从第一页读取可查看更新。 + +时间戳及ID均不能表达严格因果或提交顺序。例如同一个启动事务中，首步状态可能先于workflow.started写入；跨资源aggregate_revision不能比较大小。定时退避可能改变资源版本而不改变状态，因此状态事件的版本也可能不连续。需要可靠处理Pulse结果时使用[持久订阅](subscriptions.md)，不要将此审计分页当作消息消费协议。 + +## 3. 类型与原事件详情 + +| type | aggregate_kind | state_scope | 记录内容 | +| --- | --- | --- | --- | +| workflow.started | workflow | WORKFLOW | 首次启动、固定定义及版本 | +| workflow.advanced | workflow | WORKFLOW | 持久推进时保存的实例状态 | +| workflow.controlled | workflow | WORKFLOW | 暂停、继续、取消的控制结果 | +| workflow.recovered | workflow | WORKFLOW | 人工恢复决定及当时关联 | +| workflow.blocked | workflow | WORKFLOW | 进入核查等阻塞状态 | +| workflow.stopped | workflow | WORKFLOW | 停止意图结清后的终态 | +| workflow.step.changed | workflow_step | STEP | 步骤当时状态和已记录的激活/原因 | +| workflow.signal.accepted | workflow | SIGNAL | 保存信号受理记录，state可为PENDING、REJECTED或EXPIRED | +| workflow.signal.changed | workflow_signal | SIGNAL | 已有PENDING信号被消费、拒绝或过期 | + +`workflow.signal.accepted`沿用历史格式，aggregate_id和aggregate_revision属于实例；state_scope为SIGNAL，信号ID在payload.signal_id中。受理成功不等于信号被消费，受理时就被拒绝的记录不会再制造一个虚假的PENDING转换。 + +详情data在摘要字段外增加payload，为原事件元数据对象，最大64KiB。例如上述信号消费事件的payload： + +```json +{ + "workflow_id": "11111111111111111111111111111111", + "signal_id": "33333333333333333333333333333333", + "from": "PENDING", + "state": "CONSUMED", + "disposition": "WAIT_MATCHED", + "consumed_wait_id": "44444444444444444444444444444444" +} +``` + +```sh +curl "$ORBIT_URL/v1/workflows/$WORKFLOW_ID/events/$EVENT_ID" \ + --header "Authorization: Bearer $ORBIT_TOKEN" +``` + +信号消费、过期和拒绝，与对应事件在同一数据库事务提交。批量拒绝最多处理单实例既有上限1024条，每32条批量追加事件；任一批失败会回滚整笔转换，重复处理不重复制造事件。该原子性覆盖平台记录，不保证外部系统业务副作用只发生一次。 + +## 4. SDK调用 + +类型见[workflow_events.go](../../sdk/go/workflow_events.go)。返回revision和aggregate_revision自动解析为uint64；Payload为json.RawMessage，保留精确数字，进一步解析未知元数据时用json.Decoder.UseNumber或明确整数类型。 + +```go +cursor := "" +for { + page, err := client.ListWorkflowEvents(ctx, workflowID, + orbit.HistoryPageRequest{Limit: 20, Cursor: cursor}) + if err != nil { return err } + for _, event := range page.Items { + fmt.Printf("%s %s %s %s\n", event.OccurredAt, event.Type, + event.StateScope, event.State) + // 仅在需要展开某条事件时读取原元数据。 + if event.Type == "workflow.signal.changed" { + detail, err := client.GetWorkflowEvent(ctx, workflowID, event.ID) + if err != nil { return err } + fmt.Printf("%s\n", detail.Payload) + } + } + if page.NextCursor == "" { break } + cursor = page.NextCursor +} +``` + +完整[workflow示例](../examples/workflow.md)运行真实Worker，保存外部回执后按每页5条读取时间线，核对CONSUMED事件的原PENDING状态及赢家等待ID。 + +## 5. 错误、升级与验证边界 + +| HTTP / code | 处理 | +| --- | --- | +| 400 / INVALID_ARGUMENT | 修正非法ID、limit、cursor或参数；不要删除父资源或切换查询种类重试 | +| 401 / UNAUTHENTICATED | 检查有效凭据 | +| 403 / PERMISSION_DENIED | 显式授权workflows:audit:read | +| 404 / NOT_FOUND | 实例/事件不存在、父子错配、范围不可访问；不按全局ID降级查找 | +| 503 / UNAVAILABLE | 数据库故障、已选事件归属损坏、未知历史类型或超大元数据；保留现场核查，不能把空页当作成功 | + +只读请求可在原实例与游标下有限重试；数据改变后结果可能不同。取消HTTP只取消本次读取，不控制流程。SDK拒绝父资源不一致、无效摘要、排序不符和超限详情，不自动补造历史。 + +迁移000025新增只读派生列及`(namespace_id,workflow_id,occurred_at,id)`索引，无外键和级联。已有事件原字段不变；无合法流程范围的历史行派生为NULL，不进入流程时间线。新读取仍核验实际归属，但本接口不是全库损坏扫描，无法自动找出被错误范围隐藏的历史行。 + +旧版本没有保存的信号转换不会回填。部署时先停止旧写入节点，完成迁移再启动新节点，否则混合写入期间可能继续缺少转换事件。STORED生成列和索引需要读取历史数据并可能重建表，生产升级须预演锁等待、时长及磁盘空间；不承诺无停机DDL。旧不可变迁移保持原校验和。 + +真实MySQL验证包括升级原文保留、零外键、相同时间分页、授权、跨父资源、消费/拒绝/过期、分批写入失败回滚、重复处理和历史激活恢复；HTTP/SDK及完整例子已接入标准回归。大库索引开销、生产故障域、主从切换与全量运维仍属于P6/P7验收。 diff --git a/docs/api/workflow-history.md b/docs/api/workflow-history.md index e0e7adf..a1e5305 100644 --- a/docs/api/workflow-history.md +++ b/docs/api/workflow-history.md @@ -1,6 +1,6 @@ # 流程激活、查询轮次、信号与信号审计 -状态：HTTP与独立Go SDK已接入；本页查询明确指定的历史激活，不跟随逻辑步骤的当前激活。实例筛选分页、统一状态事件时间线及更多独立恢复场景仍按[开发计划](../08-development-plan.md)核对与交付。测试与范围见[本批报告](../reports/phase-5-history.md)。 +状态：HTTP与独立Go SDK已接入；本页查询明确指定的历史激活，不跟随逻辑步骤的当前激活。[实例筛选分页](workflow-queries.md)和[状态事件时间线](workflow-events.md)已接入，更多独立恢复场景仍按[开发计划](../08-development-plan.md)交付。测试与范围见[本批报告](../reports/phase-5-history.md)。 ## 1. 路由与调用规则 diff --git a/docs/api/workflows.md b/docs/api/workflows.md index f3d7897..ad133a7 100644 --- a/docs/api/workflows.md +++ b/docs/api/workflows.md @@ -153,6 +153,8 @@ name/payload_version/target_step_key必填，分别最多128/32/64个ASCII名称 ## 6. 操作审计与错误处理 +[状态事件时间线](workflow-events.md)另提供流程、步骤和信号的不可变状态变化，以及原事件元数据；同样要求显式审计权限。此处操作审计保留管理决定的操作人、原因、证据及前后摘要，两类记录通过已保存的身份关联，不将当前资源状态反填为历史。 + ListWorkflowOperations接受HistoryPageRequest：limit默认20、范围1–100；cursor最大1024字节，绑定命名空间、流程ID和查询种类，不可跨资源复用。data为 `{items, next_cursor?}`；每项含 id、workflow_id、operation_id、actor_id、action、created_at，按时间及ID倒序。GetWorkflowOperation除这些字段外还返回reason、evidence_ref（可空省略）、before、after JSON摘要。单独读取正文，避免列表聚合大结果。 该接口当前查询resource_kind=workflow的启动、推进、控制和恢复审计；独立任务核查审计通过恢复回执的task_audit_id关联。信号自身的受理状态通过GetWorkflowSignal读取；[信号资源审计及历史激活查询](workflow-history.md)使用独立路由，统一状态事件时间线仍待核对交付。 diff --git a/docs/examples/README.md b/docs/examples/README.md index 9a2d682..82af19b 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -12,7 +12,7 @@ - [完整命令、输入输出与失败处理](workflow.md)、[独立SDK源码](../../sdk/go/examples/workflow/main.go)。 - 两个任务并行处理并汇合，计时等待后接收显式目标代次的信号，再执行后续任务；最终三个任务完成并有界排空。 -- 启动及信号原键重复不会建立新的实例或信号。 +- 启动及信号原键重复不会建立新的实例或信号；分页查看原状态事件，核对信号消费及赢家等待。 ## 已可运行：业务事务与任务提交Outbox diff --git a/docs/examples/workflow.md b/docs/examples/workflow.md index 761eea0..2020e7a 100644 --- a/docs/examples/workflow.md +++ b/docs/examples/workflow.md @@ -13,7 +13,7 @@ chmod 700 .local go run ./cmd/orbit-admin bootstrap \ --application workflow-example --namespace default --environment development \ --queue workflows --workflow-access --workflow-run-access \ - --workflow-signal-access --worker-access \ + --workflow-signal-access --workflow-audit-access --worker-access \ --output .local/workflow.credentials.json go run ./cmd/orbit ``` @@ -31,7 +31,7 @@ go run ./examples/workflow ## 2. 调用链与输出 -CreateWorkflowDefinition → PublishWorkflowVersion(v1) → StartWorkflow → 原键再次StartWorkflow → 自动Worker并行执行 → GetWorkflowStep读取当前信号契约与代次 → SendWorkflowSignal → 原键再次发送 → 后继任务完成 → GetWorkflow → GetWorkflowSteps和GetTaskSource核对三个任务的固定来源 → Worker有界排空。 +CreateWorkflowDefinition → PublishWorkflowVersion(v1) → StartWorkflow → 原键再次StartWorkflow → 自动Worker并行执行 → GetWorkflowStep读取当前信号契约与代次 → SendWorkflowSignal → 原键再次发送 → 后继任务完成 → GetWorkflow → GetWorkflowSteps和GetTaskSource核对三个任务的固定来源 → 读取历史激活与信号 → 分页读取状态事件并核对消费记录 → Worker有界排空。 预期输出中的ID每次运行不同： @@ -39,6 +39,7 @@ CreateWorkflowDefinition → PublishWorkflowVersion(v1) → StartWorkflow → workflow=<32位实例ID> state=SUCCEEDED result={"text":"HELLO ORBIT"} start_deduplicated=true signal_deduplicated=true workflow_sources_verified=3 fixed_version=v1 workflow_history_verified=true signal_state=CONSUMED +workflow_timeline_verified=true events=<事件数> signal_transition=PENDING->CONSUMED worker_drained=true started=3 confirmed=3 ``` @@ -56,4 +57,8 @@ worker_drained=true started=3 confirmed=3 [公共SDK集成测试](../../sdk/go/workflow_instances_integration_test.go)还实际运行：启动响应丢失的原键恢复；暂停/继续、定时等待、信号消费和取消；UNKNOWN执行核查、步骤结论及解除核查；剩余尝试内TASK重试与SIGNAL换代；确认PENDING后沿原查询预算继续。使用`make integration`在一次性环境复现。 -历史激活和信号列表已接入，示例核对指定历史等待的winner与信号CONSUMED状态。实例筛选分页、更多独立业务示例及生产验收仍按[计划](../08-development-plan.md)推进。 +历史激活和信号列表已接入，示例核对指定历史等待的winner与信号CONSUMED状态。[实例筛选分页](workflow-query.md)已接入；更多独立轮询/恢复示例及生产验收仍按[计划](../08-development-plan.md)推进。 + +## 5. 状态事件审计 + +例子使用显式审计权限，按每页5条调用ListWorkflowEvents，再用GetWorkflowEvent展开已消费信号，核对原状态PENDING和历史等待ID。当前成功流程的事件仍保留早先WAITING等状态。事件数量可能随实际推进次数变化，不将数量写死为业务契约；示例会确认必要消费事件存在。字段、分页限制和旧版本历史缺口见[时间线手册](../api/workflow-events.md)。 diff --git a/docs/reports/phase-5-events.md b/docs/reports/phase-5-events.md new file mode 100644 index 0000000..1ad5ccb --- /dev/null +++ b/docs/reports/phase-5-events.md @@ -0,0 +1,58 @@ +# P4/P5 流程状态事件时间线验证记录 + +日期：2026-09-18。本批范围是流程状态事件的分页、详情、HTTP、独立SDK，以及信号消费/拒绝/过期的事务事件。P4/P5全范围核对、独立POLL/人工恢复场景和P6/P7仍未完成。 + +## 实现与接口 + +- `GET /v1/workflows/{id}/events`及`/{event}`，对应ListWorkflowEvents和GetWorkflowEvent。两条路由都要求显式workflows:audit:read，并核对固定版本全部队列。 +- 原始workflow、workflow_step、workflow_signal事件按数据库记录时间与ID倒序，不展现内部根步骤。摘要默认20、最多100条，原元数据详情不超过64KiB。 +- 实例、步骤和信号是独立聚合，revision不能跨资源比较；时间/ID不能推断提交水位或可靠消费进度。旧事件只展示当时保存的字段，不用当前激活改写历史。 +- PENDING信号的新转换与原状态事件同事务提交。批量拒绝每32条插入事件，任一批失败时全部回滚；保留单实例1024条历史信号上限。 +- 000025增加只读派生列和分页索引，不添加外键或级联，不补造历史缺失的信号转换。索引只用于查找，实际资源归属仍由服务端核验。 +- 36个持久化模型、312个直接字段均有中文注释；新增列同步中文COMMENT。 + +接口、字段及SDK片段见[时间线手册](../api/workflow-events.md)。[workflow完整例子](../examples/workflow.md)在真实Worker任务完成后，按每页5条核对信号原状态与赢家等待ID。原激活/轮次、Pulse尝试及操作审计仍通过各自接口关联，不能把逻辑时间线误称为所有底层记录的单一事件流。 + +## 环境与命令 + +macOS arm64、Go 1.27.1、Docker MySQL 8.4.10，一次性隔离数据库；真实Orbit进程、HTTP及WSS Worker。未访问生产数据库。 + +```sh +make fmt test vet build +DOCKER=/usr/local/bin/docker make integration +``` + +本地通过GO参数指定工具链。标准集成入口增加WorkflowEventsSDK，沿用21个SDK例子和2个独立业务模块，原有流程/Worker/订阅进程故障用例继续执行。 + +## 验证记录 + +| 检查 | 结果 | +| --- | --- | +| fmt、race单元回归、vet与构建 | 通过 | +| 流程及步骤真实推进、列表无payload、详情与摘要一致 | 定向MySQL通过 | +| 权限、队列限制、跨实例事件与跨种类游标 | 定向MySQL通过 | +| 消费赢家、竞争者拒绝、重复处理 | 定向MySQL通过，状态与事件不重复 | +| 单条事件写入失败 | 定向MySQL通过，赢家等待/信号/后继全部回滚 | +| 70条待处理信号停止结清，第二批事件写入失败 | 定向MySQL通过，70条信号仍PENDING、已插入批次全部回滚；恢复后70条REJECTED各有事件 | +| 同一SQL批次相同微秒时间的分页 | 定向MySQL通过，每页7条覆盖相同时间组，无漏无重 | +| 过期、错误派生范围及超大元数据 | 定向MySQL通过；不返回损坏候选的部分页 | +| v24→v25升级与重跑 | 通过，旧8个原字段全部相同、精确JSON保留、非法范围NULL、只读生成列不可覆盖、索引列序正确、外键0 | +| 真实HTTP独立SDK | 通过，11个原事件分6页，信号CONSUMED与流程CANCELLED保留 | +| 完整workflow示例 | 通过，3个Worker任务成功并排空，20条事件包含PENDING→CONSUMED和赢家等待 | +| 定义与原五组公共实例SDK回归 | 定向链路通过 | +| 新增旧激活恢复后事件读取与损坏归属拒绝 | 全量存储回归通过，恢复后原事件保持，错误历史归属返回UNAVAILABLE | +| 完整标准集成脚本 | 通过，退出码0；21个SDK例子、2个独立业务模块、流程/Worker/订阅进程故障全部完成 | + +定向存储12.734秒，SDK组26.530秒，其中时间线SDK2.26秒。这些是测试执行耗时，不是吞吐或延迟指标。当前312个模型字段中文注释检查缺失0；Markdown检查1096个本地链接、68个JSON块，错误0。 + +回归修正了两处测试准备错误：尚在等待的错误关联信号会在受理时直接REJECTED，不能作为PENDING批量转换夹具，因此改为前置TIMER尚未结束时收到的早到信号；SDK模拟响应必须提供合法32位request_id。严格SQL模式还揭示UNHEX非法值在生成列DDL中会报错，因此转换前先核验全部字符，真实v24旧库重跑升级通过。修复没有放宽权限或移除归属检查。 + +## 生产边界和剩余工作 + +- 索引只是查询范围，不是授权真相；被非法范围隐藏的历史行不会由单实例分页自动检出，需要数据库写权限保护与专项核查。 +- 旧节点未保存的信号转换不会自动出现。先停止旧写入节点、应用迁移，再由新版接管；未实现无协调混合写节点升级。 +- STORED生成列和索引涉及旧行扫描及可能的表重建。大库DDL时长、空间、锁等待、查询稀疏性及信号批量事务开销仍需P6真实容量验证。 +- 本地进程故障回归不等于跨故障域高可用、MySQL主从切换或生产备份恢复验收。平台事务原子性也不等于外部业务副作用只执行一次。 +- 继续独立轮询/人工恢复示例、P4/P5原范围核对和P6/P7；当前持续目标保持进行中。 + +完整标准集成脚本退出码0：存储组237.549秒，流程进程恢复48.726秒，最后Worker/订阅进程故障组97.407秒。新时间线6组存储场景与v24升级通过；历史v7→当前迁移再次确认70个外键全部移除、323个原索引组成部分和原任务保留。生产容量与高可用边界仍按上述未验证项处理。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index c144b8f..280e5ed 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -208,3 +208,5 @@ P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) 流程任务同样调用`GetTaskSource`，kind为WORKFLOW时读取workflow中的固定版本、步骤、历史激活代次、查询轮次和直接恢复父任务。该代次不表示当前等待目标；发送信号须另读GetWorkflowStep。完整字段及兼容边界见[来源接口](../api/task-source.md#5-流程固定来源)，[workflow示例](../examples/workflow.md)实际核对三个任务来源。 [ListWorkflows及完整查询示例](../api/workflow-queries.md)已接入，固定筛选沿游标分页，覆盖未执行分支的全部队列权限。 + +[ListWorkflowEvents / GetWorkflowEvent](../api/workflow-events.md)按实例读取不可变状态事件，要求显式workflows:audit:read；[完整workflow示例](../examples/workflow.md)核对信号从PENDING到CONSUMED的原事件与赢家等待。游标用于审计翻页，不是可靠订阅水位。 diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index c8eb1ac..3b46388 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -64,6 +64,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { retryFinish(t, g, finished, "SUCCEEDED", false) assertApplicationCounts(t, f, 4, 2, 1) // 还原后续新增表、字段和索引，得到真实 v9 结构；现有任务内容保持不变。 + f.exec(`ALTER TABLE orbit_events DROP INDEX ix_event_workflow_history,DROP COLUMN workflow_id`) 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`) diff --git a/internal/storage/mysql/migrations/000025_workflow_event_timeline.up.sql b/internal/storage/mysql/migrations/000025_workflow_event_timeline.up.sql new file mode 100644 index 0000000..2e01f57 --- /dev/null +++ b/internal/storage/mysql/migrations/000025_workflow_event_timeline.up.sql @@ -0,0 +1,15 @@ +-- 从已有不可变事件派生流程检索范围，不改写历史原文或伪造过去的状态转换。 +ALTER TABLE orbit_events + ADD COLUMN workflow_id BINARY(16) GENERATED ALWAYS AS ( + CASE + WHEN aggregate_kind='workflow' THEN aggregate_id + WHEN aggregate_kind IN ('workflow_step','workflow_signal') + AND JSON_TYPE(JSON_EXTRACT(payload,'$.workflow_id'))='STRING' + AND OCTET_LENGTH(JSON_UNQUOTE(JSON_EXTRACT(payload,'$.workflow_id')))=32 + -- 严格DDL模式下UNHEX非法值会中止迁移，先确认32个字符全部是十六进制。 + AND OCTET_LENGTH(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(LOWER(JSON_UNQUOTE(JSON_EXTRACT(payload,'$.workflow_id'))),'0',''),'1',''),'2',''),'3',''),'4',''),'5',''),'6',''),'7',''),'8',''),'9',''),'a',''),'b',''),'c',''),'d',''),'e',''),'f',''))=0 + THEN UNHEX(JSON_UNQUOTE(JSON_EXTRACT(payload,'$.workflow_id'))) + ELSE NULL + END + ) STORED COMMENT '从不可变事件派生的流程查询标识；无范围或非法历史值为空，读取仍核对父资源，不能作为授权依据', + ADD KEY ix_event_workflow_history(namespace_id,workflow_id,occurred_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 3507aab..af84793 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() - // 排除后续新增/替换索引；原有项在此逐项比较，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`) + // 排除后续新增/替换索引；原有项在此逐项比较，v24/v25新增项由对应Query/Timeline升级测试核对。 + 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_event_workflow_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_settlement.go b/internal/storage/mysql/workflow_settlement.go index 9e50bc7..9f5a7fb 100644 --- a/internal/storage/mysql/workflow_settlement.go +++ b/internal/storage/mysql/workflow_settlement.go @@ -182,7 +182,7 @@ func (r *WorkflowRuntime) settleWorkflow(ctx context.Context, tx *sql.Tx, g *wor } } // 已无平台内有效授权时不逐秒空转；UNKNOWN不证明外部执行已结束，后续核查负责接续。 - if _, err := tx.ExecContext(ctx, `UPDATE weave_signals SET state='REJECTED',disposition='WORKFLOW_STOPPED_REVIEW',revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND state='PENDING'`, now, g.view.NamespaceID, g.view.ID); err != nil { + if err := rejectPendingWorkflowSignals(ctx, tx, g, "", nil, "WORKFLOW_STOPPED_REVIEW", now); err != nil { return false, err } if err := deleteWorkflowClock(ctx, tx, g, g.steps[workflowRootKey]); err != nil { @@ -212,7 +212,7 @@ func (r *WorkflowRuntime) settleWorkflow(ctx context.Context, tx *sql.Tx, g *wor } changed++ } - if _, err := tx.ExecContext(ctx, `UPDATE weave_signals SET state='REJECTED',disposition='WORKFLOW_STOPPED',revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND state='PENDING'`, now, g.view.NamespaceID, g.view.ID); err != nil { + if err := rejectPendingWorkflowSignals(ctx, tx, g, "", nil, "WORKFLOW_STOPPED", now); err != nil { return false, err } next, rootState := weave.InstanceCancelled, weave.StepCancelled diff --git a/internal/storage/mysql/workflow_signal_events.go b/internal/storage/mysql/workflow_signal_events.go new file mode 100644 index 0000000..c5aff35 --- /dev/null +++ b/internal/storage/mysql/workflow_signal_events.go @@ -0,0 +1,97 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "strings" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/weave/signal" +) + +type signalTransition struct { + WorkflowID core.ID `json:"workflow_id"` // WorkflowID 与信号归属同事务保存的流程身份。 + SignalID core.ID `json:"signal_id"` // SignalID 被修改的信号身份，旧记录保留。 + From string `json:"from"` // From 本次转换前的真实状态，不以当前状态推测过去。 + State string `json:"state"` // State 本次转换提交的目标状态。 + Disposition string `json:"disposition"` // Disposition 消费、过期或拒绝的稳定原因。 + ConsumedWaitID *core.ID `json:"consumed_wait_id,omitempty"` // ConsumedWaitID 消费赢家等待，其他转换省略。 +} + +func emitWorkflowSignalTransition(ctx context.Context, tx *sql.Tx, ns, workflow, id core.ID, revision uint64, state, reason string, wait *core.ID) error { + return emitEvent(ctx, tx, ns, id, revision, "workflow_signal", "workflow.signal.changed", signalTransition{workflow, id, "PENDING", state, reason, wait}) +} + +// rejectPendingWorkflowSignals 的filter只由内部固定SQL构造。调用方已持有实例写锁， +// 一次读取最多1025条检测破坏上限的存储；更新和每条事件同事务，按32条批量追加事件。 +func rejectPendingWorkflowSignals(ctx context.Context, tx *sql.Tx, g *workflowGraph, filter string, args []any, reason string, now time.Time) error { + where := ` WHERE s.namespace_id=? AND s.workflow_id=? AND s.state='PENDING'` + filter + params := append([]any{g.view.NamespaceID, g.view.ID}, args...) + rows, err := tx.QueryContext(ctx, `SELECT s.id,s.revision FROM weave_signals s`+where+` ORDER BY s.id LIMIT ? FOR UPDATE`, append(params, signal.MaxPerWorkflow+1)...) + if err != nil { + return err + } + type entry struct { + id core.ID + revision uint64 + } + var items []entry + for rows.Next() { + var v entry + if err = rows.Scan(&v.id, &v.revision); err != nil { + rows.Close() + return err + } + items = append(items, v) + } + err = rows.Err() + rows.Close() + if err != nil { + return err + } + if len(items) > signal.MaxPerWorkflow { + return errors.New("workflow signal limit violated") + } + if len(items) == 0 { + return nil + } + update := append([]any{reason, now}, params...) + result, err := tx.ExecContext(ctx, `UPDATE weave_signals s SET s.state='REJECTED',s.disposition=?,s.revision=s.revision+1,s.updated_at=?`+where, update...) + if err != nil { + return err + } + n, err := result.RowsAffected() + if err != nil { + return err + } + if n != int64(len(items)) { + return errors.New("workflow signal rejection set changed") + } + for start := 0; start < len(items); start += 32 { + end := start + 32 + if end > len(items) { + end = len(items) + } + var values []string + var insert []any + for _, v := range items[start:end] { + event, err := core.NewID() + if err != nil { + return err + } + raw, err := json.Marshal(signalTransition{WorkflowID: g.view.ID, SignalID: v.id, From: "PENDING", State: "REJECTED", Disposition: reason}) + if err != nil { + return err + } + values = append(values, "(?,?,?,?,?,?,?,UTC_TIMESTAMP(6))") + insert = append(insert, g.view.NamespaceID, event, "workflow_signal", v.id, v.revision+1, "workflow.signal.changed", string(raw)) + } + if _, err = tx.ExecContext(ctx, `INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES `+strings.Join(values, ","), insert...); err != nil { + return err + } + } + return nil +} diff --git a/internal/storage/mysql/workflow_signal_runtime.go b/internal/storage/mysql/workflow_signal_runtime.go index a87062c..11b86b1 100644 --- a/internal/storage/mysql/workflow_signal_runtime.go +++ b/internal/storage/mysql/workflow_signal_runtime.go @@ -53,7 +53,7 @@ func (r *WorkflowRuntime) driveWorkflowSignal(ctx context.Context, tx *sql.Tx, g return false, false, "", err } } - if _, err := tx.ExecContext(ctx, `UPDATE weave_signals s JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.workflow_id=s.workflow_id AND a.sequence=s.target_activation_sequence SET s.state='REJECTED',s.disposition=?,s.revision=s.revision+1,s.updated_at=? WHERE s.namespace_id=? AND s.workflow_id=? AND s.target_step_key=? AND a.id=? AND a.step_id=? AND s.state='PENDING'`, string(invalid), now, g.view.NamespaceID, g.view.ID, row.view.Key, row.view.ActivationID, row.view.ID); err != nil { + if err := rejectPendingWorkflowSignals(ctx, tx, g, ` AND s.target_step_key=? AND EXISTS(SELECT 1 FROM weave_activations a WHERE a.namespace_id=s.namespace_id AND a.workflow_id=s.workflow_id AND a.sequence=s.target_activation_sequence AND a.id=? AND a.step_id=?)`, []any{row.view.Key, row.view.ActivationID, row.view.ID}, string(invalid), now); err != nil { return false, false, "", err } if err := deleteWorkflowClock(ctx, tx, g, row); err != nil { @@ -315,5 +315,8 @@ func (r *WorkflowRuntime) expireWorkflowSignal(ctx context.Context, tx *sql.Tx, if err := exactlyOne(res, err); err != nil { return false, err } + if err := emitWorkflowSignalTransition(ctx, tx, c.ns, c.workflow, v.ID, v.Revision+1, "EXPIRED", "UNMATCHED_DEADLINE", nil); err != nil { + return false, err + } return true, r.audit(ctx, tx, g, c.id, "SIGNAL_EXPIRED", now) } diff --git a/internal/storage/mysql/workflow_signal_source.go b/internal/storage/mysql/workflow_signal_source.go index aaa1a69..bee80ac 100644 --- a/internal/storage/mysql/workflow_signal_source.go +++ b/internal/storage/mysql/workflow_signal_source.go @@ -175,9 +175,8 @@ func claimWorkflowSignal(ctx context.Context, tx *sql.Tx, g *workflowGraph, w *w v.Revision++ v.Disposition = "WAIT_MATCHED" v.ConsumedWaitID = &w.id - return nil + return emitWorkflowSignalTransition(ctx, tx, g.view.NamespaceID, g.view.ID, v.ID, v.Revision, string(v.State), v.Disposition, v.ConsumedWaitID) } func rejectOtherWorkflowSignals(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, sequence uint32, reason string, now time.Time) error { - _, err := tx.ExecContext(ctx, `UPDATE weave_signals SET state='REJECTED',disposition=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND workflow_id=? AND target_step_key=? AND target_activation_sequence=? AND state='PENDING'`, reason, now, g.view.NamespaceID, g.view.ID, row.view.Key, sequence) - return err + return rejectPendingWorkflowSignals(ctx, tx, g, ` AND s.target_step_key=? AND s.target_activation_sequence=?`, []any{row.view.Key, sequence}, reason, now) } diff --git a/internal/storage/mysql/workflow_timeline.go b/internal/storage/mysql/workflow_timeline.go new file mode 100644 index 0000000..5feec23 --- /dev/null +++ b/internal/storage/mysql/workflow_timeline.go @@ -0,0 +1,266 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "strings" + + "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/instance" +) + +const workflowEventColumns = `e.id,e.aggregate_kind,e.aggregate_id,e.aggregate_revision,e.type,e.occurred_at,IF(OCTET_LENGTH(CAST(e.payload AS CHAR))<=65536,e.payload,NULL)` +const workflowEventTables = ` FROM orbit_events e FORCE INDEX(ix_event_workflow_history) + LEFT JOIN weave_steps s ON e.aggregate_kind='workflow_step' AND s.namespace_id=e.namespace_id AND s.id=e.aggregate_id AND s.workflow_id=e.workflow_id + WHERE e.namespace_id=? AND e.workflow_id=? AND (e.aggregate_kind<>'workflow_step' OR s.id IS NULL OR s.key<>'$root')` + +type workflowEventParent struct { + view instance.View + steps map[core.ID]instance.StepView +} + +func readWorkflowEventParent(ctx context.Context, tx *sql.Tx, ns, id core.ID, p auth.Permissions) (workflowEventParent, error) { + var out workflowEventParent + view, gate, ver, err := readWorkflowForQuery(ctx, tx, ns, id, p) + if err != nil { + return out, err + } + steps, err := readWorkflowStepViews(ctx, tx, ns, id, gate, ver, "") + if err != nil { + return out, err + } + out.view = view + out.steps = map[core.ID]instance.StepView{} + for _, step := range steps { + out.steps[step.ID] = step + } + return out, nil +} +func scanWorkflowEvent(row scanner) (instance.EventDetail, error) { + var out instance.EventDetail + var raw []byte + err := row.Scan(&out.ID, &out.AggregateKind, &out.AggregateID, &out.AggregateRevision, &out.Type, &out.OccurredAt, &raw) + out.Payload = raw + return out, err +} +func eventIDField(fields map[string]json.RawMessage, name string) (core.ID, error) { + var id core.ID + if json.Unmarshal(fields[name], &id) != nil || id.Validate() != nil { + return "", errors.New("invalid workflow event reference") + } + return id, nil +} +func verifyWorkflowEvent(ctx context.Context, tx *sql.Tx, h workflowEventParent, e *instance.EventDetail) error { + if e.ID.Validate() != nil || e.AggregateID.Validate() != nil || e.AggregateRevision == 0 || len(e.Payload) == 0 { + return errors.New("workflow event metadata invalid or oversized") + } + var fields map[string]json.RawMessage + if json.Unmarshal(e.Payload, &fields) != nil || fields == nil || json.Unmarshal(fields["state"], &e.State) != nil { + return errors.New("workflow event state missing") + } + e.WorkflowID = h.view.ID + switch e.AggregateKind { + case "workflow": + if e.AggregateID != h.view.ID || e.AggregateRevision > h.view.Revision { + return errors.New("workflow event aggregate mismatch") + } + e.StateScope = "WORKFLOW" + switch e.Type { + case "workflow.started": + def, err := eventIDField(fields, "definition_id") + if err != nil || def != h.view.DefinitionID { + return errors.New("workflow event definition mismatch") + } + version, err := eventIDField(fields, "definition_version_id") + if err != nil || version != h.view.DefinitionVersionID { + return errors.New("workflow event fixed version mismatch") + } + case "workflow.advanced", "workflow.controlled", "workflow.recovered", "workflow.blocked", "workflow.stopped": + case "workflow.signal.accepted": + e.StateScope = "SIGNAL" + signal, err := eventIDField(fields, "signal_id") + if err != nil { + return err + } + if err = verifyEventSignal(ctx, tx, h, signal, 0); err != nil { + return err + } + owner, err := eventIDField(fields, "workflow_id") + if err != nil || owner != h.view.ID { + return errors.New("workflow signal event scope mismatch") + } + default: + return errors.New("unsupported workflow event type") + } + case "workflow_step": + step, ok := h.steps[e.AggregateID] + if !ok || e.AggregateRevision > step.Revision || e.Type != "workflow.step.changed" { + return errors.New("workflow step event aggregate mismatch") + } + owner, err := eventIDField(fields, "workflow_id") + if err != nil || owner != h.view.ID { + return errors.New("workflow step event scope mismatch") + } + e.StepKey = step.Key + e.StateScope = "STEP" + if raw, ok := fields["key"]; ok { + var key string + if json.Unmarshal(raw, &key) != nil || key != step.Key { + return errors.New("workflow event step key mismatch") + } + } + if raw, ok := fields["activation_id"]; ok && string(raw) != "null" { + activation, err := eventIDField(fields, "activation_id") + if err != nil { + return err + } + var count int + if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? AND id=?`, h.view.NamespaceID, h.view.ID, step.ID, activation).Scan(&count); err != nil { + return err + } + if count != 1 { + return errors.New("workflow event historical activation mismatch") + } + } + case "workflow_signal": + e.StateScope = "SIGNAL" + if e.Type != "workflow.signal.changed" { + return errors.New("unsupported workflow signal event") + } + owner, err := eventIDField(fields, "workflow_id") + if err != nil || owner != h.view.ID { + return errors.New("workflow signal event scope mismatch") + } + signal, err := eventIDField(fields, "signal_id") + if err != nil || signal != e.AggregateID { + return errors.New("workflow event signal reference mismatch") + } + if err = verifyEventSignal(ctx, tx, h, signal, e.AggregateRevision); err != nil { + return err + } + default: + return errors.New("unsupported workflow event aggregate") + } + valid := false + switch e.StateScope { + case "WORKFLOW": + switch e.State { + case "RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED": + valid = true + } + case "STEP": + switch e.State { + case "BLOCKED", "ACTIVE", "WAITING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED", "SKIPPED": + valid = true + } + case "SIGNAL": + switch e.State { + case "PENDING", "CONSUMED", "EXPIRED", "REJECTED": + valid = true + } + } + if !valid { + return errors.New("workflow event recorded state invalid") + } + return nil +} +func verifyEventSignal(ctx context.Context, tx *sql.Tx, h workflowEventParent, id core.ID, revision uint64) error { + var actual uint64 + err := tx.QueryRowContext(ctx, `SELECT revision FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND id=?`, h.view.NamespaceID, h.view.ID, id).Scan(&actual) + if err != nil { + return err + } + if revision > actual { + return errors.New("workflow signal event version mismatch") + } + return nil +} +func (s *WorkflowInstanceStore) ListEvents(ctx context.Context, t auth.Token, id core.ID, in service.HistoryPageInput) (instance.EventPage, error) { + out := instance.EventPage{WorkflowID: id, Items: []instance.EventSummary{}} + err := s.transaction(ctx, t, instance.AuditAction, func(tx *sql.Tx, p auth.Permissions) error { + h, err := readWorkflowEventParent(ctx, tx, t.NamespaceID, id, p) + if err != nil { + return err + } + out.Revision = h.view.Revision + q := `SELECT ` + workflowEventColumns + workflowEventTables + args := []any{t.NamespaceID, id} + if in.Cursor != "" { + c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, id, "", "workflow-events") + if err != nil { + return err + } + q += ` AND (e.occurred_at in.Limit + if more { + entries = entries[:in.Limit] + } + for _, e := range entries { + out.Items = append(out.Items, e.EventSummary) + } + if more { + last := out.Items[len(out.Items)-1] + out.NextCursor, err = encodeHistoryCursor(historyCursor{Version: 1, Kind: "workflow-events", Namespace: t.NamespaceID, Task: id, ID: last.ID, Created: &last.OccurredAt}) + } + return err + }) + if err != nil { + return instance.EventPage{}, err + } + return out, nil +} +func (s *WorkflowInstanceStore) GetEvent(ctx context.Context, t auth.Token, id, event core.ID) (instance.EventDetail, error) { + var out instance.EventDetail + err := s.transaction(ctx, t, instance.AuditAction, func(tx *sql.Tx, p auth.Permissions) error { + h, err := readWorkflowEventParent(ctx, tx, t.NamespaceID, id, p) + if err != nil { + return err + } + // 明确事件ID后走主键；仍保留流程派生范围和实际归属核验。 + q := `SELECT ` + workflowEventColumns + strings.Replace(workflowEventTables, "FORCE INDEX(ix_event_workflow_history)", "FORCE INDEX(PRIMARY)", 1) + ` AND e.id=?` + out, err = scanWorkflowEvent(tx.QueryRowContext(ctx, q, t.NamespaceID, id, event)) + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + if err != nil { + return err + } + return verifyWorkflowEvent(ctx, tx, h, &out) + }) + if err != nil { + return instance.EventDetail{}, err + } + return out, nil +} diff --git a/internal/storage/mysql/workflow_timeline_integration_test.go b/internal/storage/mysql/workflow_timeline_integration_test.go new file mode 100644 index 0000000..617776a --- /dev/null +++ b/internal/storage/mysql/workflow_timeline_integration_test.go @@ -0,0 +1,319 @@ +package mysql + +import ( + "context" + "encoding/json" + "os" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/weave/definition" + "git.yuntuo.hk/k2safe/orbit/weave/instance" +) + +func grantWorkflowEvents(f *executionFixture) { + f.exec(`UPDATE orbit_credentials SET permissions=JSON_ARRAY_APPEND(permissions,'$.actions','workflows:audit:read') WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) +} +func TestIntegrationWorkflowTimeline(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 16, 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("real_transitions_pagination_scope_and_no_payload_list", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + v := runtimeStart(t, f, runtimeSpec(f.resources.QueueID, 2)) + if _, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{}); fault.Code(err) != "PERMISSION_DENIED" { + t.Fatal("audit permission missing", err) + } + grantWorkflowEvents(f) + ref := retryStart(t, f) + retryFinish(t, f, ref, "SUCCEEDED", false) + runtimeConsume(t, f, newRuntime(t, db), ref.TaskID) + first, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 1}) + if err != nil || len(first.Items) != 1 || first.NextCursor == "" { + t.Fatal(first, err) + } + seen := map[core.ID]bool{} + states := map[string]bool{} + cursor := "" + var last *instance.EventSummary + for n := 0; n < 30; n++ { + page, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 1, Cursor: cursor}) + if err != nil { + t.Fatal(err) + } + raw, _ := json.Marshal(page) + if strings.Contains(string(raw), `"payload"`) || strings.Contains(string(raw), `9007199254740993`) { + t.Fatal("body exposed in list") + } + for _, e := range page.Items { + if seen[e.ID] || e.WorkflowID != v.ID || e.StepKey == workflowRootKey { + t.Fatal("event identity", e) + } + seen[e.ID] = true + states[e.Type+":"+e.State] = true + if last != nil && (e.OccurredAt.After(last.OccurredAt) || e.OccurredAt.Equal(last.OccurredAt) && e.ID >= last.ID) { + t.Fatal("unstable event order") + } + current := e + last = ¤t + detail, err := api.GetEvent(ctx, f.token, v.ID, e.ID) + if err != nil || detail.EventSummary != e || !json.Valid(detail.Payload) { + t.Fatal("event detail mismatch", detail, err) + } + } + if page.NextCursor == "" { + break + } + cursor = page.NextCursor + } + if len(seen) < 5 || !states["workflow.started:RUNNING"] || !states["workflow.step.changed:SUCCEEDED"] || !states["workflow.step.changed:ACTIVE"] { + t.Fatal("missing actual transitions", states) + } + if _, err := api.ListOperations(ctx, f.token, v.ID, service.HistoryPageInput{Cursor: first.NextCursor}); fault.Code(err) != "INVALID_ARGUMENT" { + t.Fatal("cross kind cursor", err) + } + other := runtimeStart(t, f, runtimeSpec(f.resources.QueueID, 1)) + grantWorkflowEvents(f) + if _, err := api.GetEvent(ctx, f.token, other.ID, first.Items[0].ID); fault.Code(err) != "NOT_FOUND" { + t.Fatal("cross workflow event", err) + } + if _, err := api.ListEvents(ctx, f.token, other.ID, service.HistoryPageInput{Cursor: first.NextCursor}); fault.Code(err) != "INVALID_ARGUMENT" { + t.Fatal("cross workflow cursor", err) + } + grantWorkflowInstance(t, f, string(executionID(t))) + grantWorkflowEvents(f) + if _, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{}); fault.Code(err) != "NOT_FOUND" { + t.Fatal("queue bypass", err) + } + }) + t.Run("signal_consumption_and_rejection_atomic_events", func(t *testing.T) { + f, v, _ := signalStart(t, ctx, db, false) + grantWorkflowEvents(f) + first := sendSignal(t, f, v, signalInput(v)) + second := sendSignal(t, f, v, signalInput(v)) + c := clockCandidate(t, f, v.ID, "signal") + f.exec(`CREATE TRIGGER timeline_signal_failure BEFORE INSERT ON orbit_events FOR EACH ROW SET NEW.type=IF(NEW.aggregate_kind='workflow_signal',NULL,NEW.type)`) + _, err := newRuntime(t, db).process(ctx, c, false) + f.exec(`DROP TRIGGER timeline_signal_failure`) + if err == nil || f.count("weave_signals", "state='PENDING'") != 2 || f.count("weave_waits", "winner_signal_id IS NOT NULL") != 0 || f.count("pulse_tasks", "1=1") != 0 || f.count("orbit_events", "aggregate_kind='workflow_signal'") != 0 { + t.Fatal("partial signal transition committed", err) + } + if !runClock(t, f, c) { + t.Fatal("retry failed") + } + page, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 100}) + if err != nil { + t.Fatal(err) + } + changed := map[core.ID]string{} + for _, e := range page.Items { + if e.AggregateKind == "workflow_signal" { + changed[e.AggregateID] = e.State + if e.StateScope != "SIGNAL" { + t.Fatal("wrong state subject") + } + detail, err := api.GetEvent(ctx, f.token, v.ID, e.ID) + if err != nil { + t.Fatal(err) + } + var meta signalTransition + if json.Unmarshal(detail.Payload, &meta) != nil || meta.From != "PENDING" { + t.Fatal("lost previous state") + } + } + } + if changed[first.Signal.ID] != "CONSUMED" || changed[second.Signal.ID] != "REJECTED" || len(changed) != 2 { + t.Fatal("signal transitions", changed) + } + runClock(t, f, c) + if f.count("orbit_events", "aggregate_kind='workflow_signal'") != 2 { + t.Fatal("duplicate transition") + } + }) + t.Run("bulk_stop_preserves_all_signal_events_and_rollback", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + spec := signalSpec(f.resources.QueueID, true) + spec.Body.Steps[0] = definition.Step{Key: "before", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 60000}} + v := runtimeStart(t, f, spec) + grantWorkflowSignal(f) + grantWorkflowControl(f) + grantWorkflowEvents(f) + for n := 0; n < 70; n++ { + in := signalInput(v) + in.CorrelationKey += "unmatched" + if sent := sendSignal(t, f, v, in); sent.Signal.State != "PENDING" { + t.Fatal("fixture must retain early pending signals", sent.Signal.State) + } + } + stopped := applyControl(t, f, runtimeView(t, f, v.ID, "WAITING"), "CANCEL").Instance + var target core.ID + if err := db.QueryRowContext(ctx, `SELECT id FROM weave_signals WHERE namespace_id=? ORDER BY id LIMIT 1 OFFSET 40`, f.resources.NamespaceID).Scan(&target); err != nil { + t.Fatal(err) + } + f.exec(`CREATE TRIGGER timeline_batch_failure BEFORE INSERT ON orbit_events FOR EACH ROW SET NEW.type=IF(NEW.aggregate_kind='workflow_signal' AND NEW.aggregate_id=UNHEX('` + string(target) + `'),NULL,NEW.type)`) + c := clockCandidate(t, f, v.ID, workflowRootKey) + var transitionErr error + // 停止会先结清等待，根接续可能退避；推进到实际批量转换事务再注入失败。 + for n := 0; n < 10; n++ { + f.exec(`UPDATE weave_pending_advances SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, c.ns, c.id) + _, transitionErr = newRuntime(t, db).process(ctx, c, false) + if transitionErr != nil { + break + } + } + f.exec(`DROP TRIGGER timeline_batch_failure`) + if transitionErr == nil || f.count("weave_signals", "state='PENDING'") != 70 || f.count("orbit_events", "aggregate_kind='workflow_signal'") != 0 { + t.Fatal("batch prefix survived rollback", transitionErr, f.count("weave_signals", "state='PENDING'"), f.count("orbit_events", "aggregate_kind='workflow_signal'")) + } + drainStop(t, f, stopped.ID, "CANCELLED") + if f.count("weave_signals", "state='REJECTED'") != 70 || f.count("orbit_events", "aggregate_kind='workflow_signal' AND type='workflow.signal.changed'") != 70 { + t.Fatal("incomplete rejection history") + } + page, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 100}) + if err != nil || len(page.Items) != 100 || page.NextCursor == "" { + t.Fatal("large timeline", len(page.Items), err) + } + for _, e := range page.Items { + if e.StepKey == workflowRootKey { + t.Fatal("internal root exposed") + } + } + // 每批32条事件共享SQL语句时间；小页跨过相同时间，必须按ID无漏无重。 + seen := map[core.ID]bool{} + times := map[time.Time]int{} + cursor := "" + for n := 0; n < 100; n++ { + page, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 7, Cursor: cursor}) + if err != nil { + t.Fatal(err) + } + for _, e := range page.Items { + if seen[e.ID] { + t.Fatal("duplicate tied event", e.ID) + } + seen[e.ID] = true + times[e.OccurredAt]++ + } + if page.NextCursor == "" { + break + } + cursor = page.NextCursor + } + var expected int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*)`+workflowEventTables, f.resources.NamespaceID, v.ID).Scan(&expected); err != nil || len(seen) != expected { + t.Fatal("tied pagination skipped events", len(seen), expected, err) + } + hasTies := false + for _, count := range times { + hasTies = hasTies || count > 7 + } + if !hasTies { + t.Fatal("fixture did not cross timestamp ties") + } + }) + t.Run("expiry_records_original_transition_once", func(t *testing.T) { + f, v, _ := signalStart(t, ctx, db, true) + grantWorkflowEvents(f) + in := signalInput(v) + in.CorrelationKey += "unmatched" + sig := sendSignal(t, f, v, in) + f.exec(`UPDATE weave_signals SET expires_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND),available_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, sig.Signal.ID) + c := workflowCandidate{kind: "SIGNAL", ns: f.resources.NamespaceID, workflow: v.ID, id: sig.Signal.ID, revision: 1} + if _, err := newRuntime(t, db).process(ctx, c, false); err != nil { + t.Fatal(err) + } + page, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 100}) + if err != nil { + t.Fatal(err) + } + count := 0 + for _, e := range page.Items { + if e.AggregateKind == "workflow_signal" && e.AggregateID == sig.Signal.ID && e.State == "EXPIRED" { + count++ + } + } + if count != 1 { + t.Fatal("expiry event missing", count) + } + }) + t.Run("derived_scope_is_not_authority_and_payload_is_bounded", func(t *testing.T) { + f := newExecutionFixture(t, ctx, db) + v := runtimeStart(t, f, runtimeSpec(f.resources.QueueID, 1)) + other := runtimeStart(t, f, runtimeSpec(f.resources.QueueID, 1)) + grantWorkflowEvents(f) + var step core.ID + if err := db.QueryRowContext(ctx, "SELECT id FROM weave_steps WHERE namespace_id=? AND workflow_id=? AND `key`='task0'", f.resources.NamespaceID, other.ID).Scan(&step); err != nil { + t.Fatal(err) + } + forged := executionID(t) + f.exec(`INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,'workflow_step',?,1,'workflow.step.changed',JSON_OBJECT('workflow_id',?,'state','ACTIVE'),UTC_TIMESTAMP(6))`, f.resources.NamespaceID, forged, step, string(v.ID)) + if page, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{}); err == nil || len(page.Items) != 0 { + t.Fatal("forged scope returned partial data") + } + if _, err := api.GetEvent(ctx, f.token, v.ID, forged); err == nil { + t.Fatal("forged parent accepted") + } + huge := executionID(t) + f.exec(`INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,'workflow',?,1,'workflow.blocked',JSON_OBJECT('state','RUNNING','padding',?),UTC_TIMESTAMP(6))`, f.resources.NamespaceID, huge, other.ID, strings.Repeat("x", 65536)) + if _, err := api.GetEvent(ctx, f.token, other.ID, huge); err == nil { + t.Fatal("oversized detail accepted") + } + }) + t.Run("historical_activation_survives_recovery_and_checks_parent", func(t *testing.T) { + f, v := waitRecoveryFixture(t, ctx, db, "TIMER", 30000) + grantWorkflowEvents(f) + old, err := api.Step(ctx, f.token, v.ID, "timer") + if err != nil { + t.Fatal(err) + } + page, err := api.ListEvents(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 100}) + if err != nil { + t.Fatal(err) + } + var saved instance.EventDetail + for _, event := range page.Items { + if event.AggregateKind != "workflow_step" || event.AggregateID != old.Step.ID { + continue + } + detail, err := api.GetEvent(ctx, f.token, v.ID, event.ID) + if err != nil { + t.Fatal(err) + } + var meta struct { + ActivationID core.ID `json:"activation_id"` + } + if json.Unmarshal(detail.Payload, &meta) == nil && meta.ActivationID == *old.Step.ActivationID { + saved = detail + break + } + } + if saved.ID == "" { + t.Fatal("historical activation event missing") + } + rearmWorkflowWait(t, f, v.ID, "timer") + after, err := api.GetEvent(ctx, f.token, v.ID, saved.ID) + if err != nil || after.EventSummary != saved.EventSummary || string(after.Payload) != string(saved.Payload) { + t.Fatal("old event replaced by recovery", err) + } + f.exec(`UPDATE weave_activations SET workflow_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, *old.Step.ActivationID) + if _, err := api.GetEvent(ctx, f.token, v.ID, saved.ID); fault.Code(err) != "UNAVAILABLE" { + t.Fatal("foreign historical activation accepted", err) + } + }) +} diff --git a/internal/storage/mysql/workflow_timeline_upgrade_integration_test.go b/internal/storage/mysql/workflow_timeline_upgrade_integration_test.go new file mode 100644 index 0000000..f79de6d --- /dev/null +++ b/internal/storage/mysql/workflow_timeline_upgrade_integration_test.go @@ -0,0 +1,157 @@ +package mysql + +import ( + "context" + "database/sql" + "os" + "reflect" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + driver "github.com/go-sql-driver/mysql" +) + +// 用真实v24事件验证派生范围回填，历史内容不变，不凭空补造信号转换。 +func TestIntegrationWorkflowTimelineUpgrade(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_timeline_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 > 24 { + 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, owner := executionID(t), executionID(t) + type sample struct { + kind, payload string + aggregate core.ID + want *core.ID + } + cases := []sample{ + {"workflow", `{"state":"WAITING"}`, owner, &owner}, + {"workflow_step", `{"workflow_id":"` + string(owner) + `","state":"ACTIVE","n":9007199254740993}`, executionID(t), &owner}, + {"workflow_signal", `{"workflow_id":"` + string(owner) + `","state":"CONSUMED"}`, executionID(t), &owner}, + {"task", `{"workflow_id":"` + string(owner) + `"}`, executionID(t), nil}, + {"workflow_step", `{"workflow_id":"gggggggggggggggggggggggggggggggg"}`, executionID(t), nil}, + {"workflow_step", `{"workflow_id":"abc"}`, executionID(t), nil}, + {"workflow_step", `{"workflow_id":null}`, executionID(t), nil}, + {"workflow_step", `{"workflow_id":12345678901234567890123456789012}`, executionID(t), nil}, + {"workflow_step", `{"state":"ACTIVE"}`, executionID(t), nil}, + } + ids := make([]core.ID, len(cases)) + for i, v := range cases { + ids[i] = executionID(t) + exec(`INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,?,?,7,'legacy.event',?,'2026-01-02 03:04:05.123456')`, ns, ids[i], v.kind, v.aggregate, v.payload) + } + snapshot := func() [][]sql.NullString { + t.Helper() + rows, err := db.QueryContext(ctx, `SELECT namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at FROM orbit_events ORDER BY namespace_id,id`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out [][]sql.NullString + for rows.Next() { + row := make([]sql.NullString, 8) + args := make([]any, 8) + 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 err := CheckSchema(ctx, db); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(before, snapshot()) { + t.Fatal("timeline migration changed or invented historical events") + } + for i, v := range cases { + var got *core.ID + if err := db.QueryRowContext(ctx, `SELECT workflow_id FROM orbit_events WHERE namespace_id=? AND id=?`, ns, ids[i]).Scan(&got); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(v.want, got) { + t.Fatalf("case %d derived owner mismatch: want %v got %v", i, v.want, got) + } + } + var columns 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='orbit_events' AND INDEX_NAME='ix_event_workflow_history'`).Scan(&columns); err != nil || columns != "namespace_id,workflow_id,occurred_at,id" { + t.Fatal("timeline index mismatch", columns, err) + } + var foreignKeys int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&foreignKeys); err != nil || foreignKeys != 0 { + t.Fatal("foreign keys forbidden", foreignKeys, err) + } + if _, err := db.ExecContext(ctx, `INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at,workflow_id) VALUES (?,?,'workflow',?,1,'forbidden.override','{}',UTC_TIMESTAMP(6),?)`, ns, executionID(t), owner, owner); err == nil { + t.Fatal("generated scope accepted explicit override") + } + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(before, snapshot()) { + t.Fatal("repeated migration changed history") + } + t.Log("v24 -> v25: original fields and precision preserved, invalid scopes remain NULL, index ordered, generated scope protected, foreign keys=0") +} diff --git a/internal/transport/httpserver/workflow_events.go b/internal/transport/httpserver/workflow_events.go new file mode 100644 index 0000000..361efae --- /dev/null +++ b/internal/transport/httpserver/workflow_events.go @@ -0,0 +1,27 @@ +package httpserver + +import ( + "net/http" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/weave/instance" +) + +func registerWorkflowEvents(mux *http.ServeMux, s *instance.Service) { + mux.HandleFunc("GET /v1/workflows/{id}/events", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + in, err := historyQuery(r, true) + if err != nil { + return nil, 0, err + } + out, err := s.ListEvents(r.Context(), t, core.ID(r.PathValue("id")), in) + return out, 200, err + })) + mux.HandleFunc("GET /v1/workflows/{id}/events/{event}", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + if _, err := historyQuery(r, false); err != nil { + return nil, 0, err + } + out, err := s.GetEvent(r.Context(), t, core.ID(r.PathValue("id")), core.ID(r.PathValue("event"))) + return out, 200, err + })) +} diff --git a/internal/transport/httpserver/workflows.go b/internal/transport/httpserver/workflows.go index f379e6f..ea2f30c 100644 --- a/internal/transport/httpserver/workflows.go +++ b/internal/transport/httpserver/workflows.go @@ -17,6 +17,7 @@ func RegisterWorkflows(server *http.Server, instances *instance.Service, control mux := server.Handler.(*http.ServeMux) registerWorkflowHistory(mux, instances, signals) registerWorkflowQuery(mux, instances) + registerWorkflowEvents(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/platform/model/event.go b/platform/model/event.go index 3065487..0796da5 100644 --- a/platform/model/event.go +++ b/platform/model/event.go @@ -11,6 +11,8 @@ import ( type Event struct { // Record 包含命名空间、记录标识、并发版本和 UTC 时间。 core.Record + // WorkflowID 是数据库从不可变事件派生的流程查询索引；不可写，非流程或无有效范围时为空，不能代替归属授权。 + WorkflowID *core.ID `db:"workflow_id"` // AggregateKind 产生事件的聚合类型，例如任务或流程。 AggregateKind string `db:"aggregate_kind"` // AggregateID 产生事件的聚合标识。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index ec30140..41e88ab 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -84,7 +84,7 @@ 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|Query)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|Events)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) diff --git a/sdk/go/examples/workflow/main.go b/sdk/go/examples/workflow/main.go index b89f78d..10519ca 100644 --- a/sdk/go/examples/workflow/main.go +++ b/sdk/go/examples/workflow/main.go @@ -189,6 +189,39 @@ func run() error { return errors.New("signal history missing") } fmt.Println("workflow_history_verified=true signal_state=CONSUMED") + // 时间线用于审计展示，游标只是倒序翻页位置，不作为可靠消息消费水位。 + cursor, eventCount, consumed := "", 0, false + for { + page, err := client.ListWorkflowEvents(ctx, accepted.Instance.ID, orbit.HistoryPageRequest{Limit: 5, Cursor: cursor}) + if err != nil { + return err + } + for _, event := range page.Items { + eventCount++ + if event.AggregateKind == "workflow_signal" && event.AggregateID == sent.Signal.ID && event.State == "CONSUMED" { + detail, err := client.GetWorkflowEvent(ctx, accepted.Instance.ID, event.ID) + if err != nil { + return err + } + var transition struct { + From string `json:"from"` + ConsumedWaitID string `json:"consumed_wait_id"` + } + if json.Unmarshal(detail.Payload, &transition) != nil || transition.From != "PENDING" || transition.ConsumedWaitID != historical.Wait.ID { + return errors.New("signal transition evidence mismatch") + } + consumed = true + } + } + if page.NextCursor == "" { + break + } + cursor = page.NextCursor + } + if !consumed { + return errors.New("consumed signal event missing") + } + fmt.Printf("workflow_timeline_verified=true events=%d signal_transition=PENDING->CONSUMED\n", eventCount) stop() err = <-done diff --git a/sdk/go/workflow_events.go b/sdk/go/workflow_events.go new file mode 100644 index 0000000..dd04079 --- /dev/null +++ b/sdk/go/workflow_events.go @@ -0,0 +1,88 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "time" +) + +// WorkflowEventSummary 只表示已保存的事件，不把当前状态或当前激活反填到历史。 +type WorkflowEventSummary struct { + ID string `json:"id"` // ID 原不可变状态事件标识。 + WorkflowID string `json:"workflow_id"` // WorkflowID 已核验授权的流程归属。 + AggregateKind string `json:"aggregate_kind"` // AggregateKind workflow、workflow_step或workflow_signal。 + AggregateID string `json:"aggregate_id"` // AggregateID 产生事件的实例、步骤或信号身份。 + AggregateRevision uint64 `json:"aggregate_revision,string"` // AggregateRevision 来源资源版本，各资源独立，不是全局序号。 + Type string `json:"type"` // Type 原始事件类型，不按当前状态改写。 + StateScope string `json:"state_scope"` // StateScope WORKFLOW、STEP或SIGNAL，说明State记录的是哪类对象。 + State string `json:"state"` // State 该事件保存的状态，不代表对象当前状态。 + StepKey string `json:"step_key,omitempty"` // StepKey 步骤事件的固定定义名称；其他事件省略。 + OccurredAt time.Time `json:"occurred_at"` // OccurredAt 数据库事件记录时间，UTC微秒精度，不是外部业务发生时间或提交水位。 +} +type WorkflowEventPage struct { + WorkflowID string `json:"workflow_id"` // WorkflowID 本页授权的实例身份。 + Revision uint64 `json:"revision,string"` // Revision 本页读取时实例版本，不能作为信号或步骤事件的消费水位。 + Items []WorkflowEventSummary `json:"items"` // Items 按记录时间和ID倒序的有界摘要，空列表为[]。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 绑定实例与事件查询种类，不承诺跨页快照。 +} +type WorkflowEventDetail struct { + WorkflowEventSummary + Payload json.RawMessage `json:"payload"` // Payload 原不可变事件元数据，最多64KiB；嵌套数字保留原文本，输入/结果另走资源详情。 +} + +// ListWorkflowEvents 读取实例、步骤和信号的不可变状态事件，需要workflows:audit:read。 +func (c *Client) ListWorkflowEvents(ctx context.Context, id string, in HistoryPageRequest) (WorkflowEventPage, error) { + var out WorkflowEventPage + if !validID(id) { + return out, errors.New("invalid workflow id") + } + query, err := historyPageQuery(in) + if err != nil { + return out, err + } + if err = c.call(ctx, "GET", "/v1/workflows/"+id+"/events"+query, "", nil, &out); err != nil { + return WorkflowEventPage{}, err + } + if out.WorkflowID != id || out.Revision == 0 || out.Items == nil || !historyResponseSize(len(out.Items), in) { + return WorkflowEventPage{}, errors.New("invalid workflow event page") + } + for i, e := range out.Items { + if !validWorkflowEvent(e, id) || i > 0 && (e.OccurredAt.After(out.Items[i-1].OccurredAt) || e.OccurredAt.Equal(out.Items[i-1].OccurredAt) && e.ID >= out.Items[i-1].ID) { + return WorkflowEventPage{}, errors.New("invalid workflow event summary or ordering") + } + } + return out, nil +} +func validWorkflowEvent(e WorkflowEventSummary, id string) bool { + if e.WorkflowID != id || !validID(e.ID) || !validID(e.AggregateID) || e.AggregateRevision == 0 || e.Type == "" || e.State == "" || e.OccurredAt.IsZero() { + return false + } + switch e.AggregateKind { + case "workflow": + if e.AggregateID != id { + return false + } + case "workflow_step": + if !workflowVersionName(e.StepKey) { + return false + } + case "workflow_signal": + default: + return false + } + return e.StateScope == "WORKFLOW" || e.StateScope == "STEP" || e.StateScope == "SIGNAL" +} +func (c *Client) GetWorkflowEvent(ctx context.Context, id, event string) (WorkflowEventDetail, error) { + var out WorkflowEventDetail + if !validID(id) || !validID(event) { + return out, errors.New("invalid workflow event identity") + } + if err := c.call(ctx, "GET", "/v1/workflows/"+id+"/events/"+event, "", nil, &out); err != nil { + return WorkflowEventDetail{}, err + } + if out.ID != event || !validWorkflowEvent(out.WorkflowEventSummary, id) || len(out.Payload) == 0 || len(out.Payload) > 65536 || !json.Valid(out.Payload) { + return WorkflowEventDetail{}, errors.New("invalid workflow event detail") + } + return out, nil +} diff --git a/sdk/go/workflow_events_integration_test.go b/sdk/go/workflow_events_integration_test.go new file mode 100644 index 0000000..b2c9916 --- /dev/null +++ b/sdk/go/workflow_events_integration_test.go @@ -0,0 +1,80 @@ +package orbit + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" +) + +func TestIntegrationWorkflowEventsSDK(t *testing.T) { + api, queue, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + operation := workerTestID(t) + spec := workflowSDKSpec(queue, "events."+operation) + spec.Body = WorkflowBlock{Steps: []WorkflowStep{ + {Key: "receipt", Kind: "SIGNAL", Signal: &WorkflowSignal{Name: "done", PayloadVersion: "v1", Correlation: WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"` + operation + `"`)}, PayloadType: spec.InputType, TimeoutMS: 60000}}, + {Key: "delay", Kind: "TIMER", Timer: &WorkflowTimer{DelayMS: 60000}}, + }, OutputType: WorkflowType{Kind: "NULL"}, Output: WorkflowRef("delay")} + v := startWorkflowSDK(t, ctx, api, spec) + sent, err := api.SendWorkflowSignal(ctx, v.ID, operation, SendWorkflowSignalRequest{Name: "done", PayloadVersion: "v1", TargetStepKey: "receipt", TargetActivationSequence: 1, CorrelationKey: operation, Payload: workflowExact, Reason: "verify timeline"}) + if err != nil { + t.Fatal(err) + } + waitWorkflowStepSDK(t, ctx, api, v.ID, "receipt", func(d WorkflowStepDetail) bool { return d.Step.State == "SUCCEEDED" }) + current, err := api.GetWorkflow(ctx, v.ID) + if err != nil { + t.Fatal(err) + } + if _, err = api.ControlWorkflow(ctx, v.ID, operation, WorkflowControlRequest{Operation: "CANCEL", ExpectedControlRevision: current.ControlRevision, Reason: "timeline cleanup"}); err != nil { + t.Fatal(err) + } + waitWorkflowSDK(t, ctx, api, v.ID, "CANCELLED") + seen := map[string]bool{} + types := map[string]bool{} + cursor := "" + pages := 0 + for { + page, err := api.ListWorkflowEvents(ctx, v.ID, HistoryPageRequest{Limit: 2, Cursor: cursor}) + if err != nil { + t.Fatal(err) + } + pages++ + raw, _ := json.Marshal(page) + if strings.Contains(string(raw), `"payload"`) || strings.Contains(string(raw), `9007199254740993`) { + t.Fatal("payload in summaries") + } + for _, e := range page.Items { + if seen[e.ID] || e.WorkflowID != v.ID || e.StepKey == "$root" { + t.Fatal("event identity", e) + } + seen[e.ID] = true + types[e.Type] = true + detail, err := api.GetWorkflowEvent(ctx, v.ID, e.ID) + if err != nil || detail.WorkflowEventSummary != e { + t.Fatal("detail", detail, err) + } + if e.AggregateKind == "workflow_signal" && (e.AggregateID != sent.Signal.ID || e.State != "CONSUMED" || e.StateScope != "SIGNAL") { + t.Fatal("signal transition", e) + } + } + if page.NextCursor == "" { + break + } + cursor = page.NextCursor + if pages > 20 { + t.Fatal("unbounded page repetition") + } + } + for _, kind := range []string{"workflow.started", "workflow.step.changed", "workflow.signal.accepted", "workflow.signal.changed", "workflow.controlled", "workflow.stopped"} { + if !types[kind] { + t.Fatal("event missing", kind, types) + } + } + if len(seen) < 7 || pages < 4 { + t.Fatal("incomplete timeline", len(seen), pages) + } + t.Logf("real HTTP event timeline: %d immutable events over %d pages; signal transition and cancelled workflow retained", len(seen), pages) +} diff --git a/sdk/go/workflow_events_test.go b/sdk/go/workflow_events_test.go new file mode 100644 index 0000000..43062ca --- /dev/null +++ b/sdk/go/workflow_events_test.go @@ -0,0 +1,78 @@ +package orbit + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestWorkflowEventsSDKIdentityPrecisionAndBounds(t *testing.T) { + const id = "11111111111111111111111111111111" + const event = "22222222222222222222222222222222" + var calls atomic.Int32 + var corrupt atomic.Bool + summary := WorkflowEventSummary{ID: event, WorkflowID: id, AggregateKind: "workflow", AggregateID: id, AggregateRevision: 9007199254740993, Type: "workflow.advanced", StateScope: "WORKFLOW", State: "WAITING", OccurredAt: time.Date(2026, 9, 18, 0, 0, 0, 0, time.UTC)} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.Method != "GET" || r.Header.Get("Authorization") != "Bearer token" || r.Header.Get("Idempotency-Key") != "" { + t.Error("read request invalid") + } + item := summary + if corrupt.Load() { + item.WorkflowID = event + } + var data any + if strings.HasSuffix(r.URL.Path, "/events") { + if r.URL.Query().Get("limit") != "1" || r.URL.Query().Get("cursor") != "cursor+/=" { + t.Error("cursor encoding changed") + } + data = WorkflowEventPage{WorkflowID: id, Revision: 9007199254740993, Items: []WorkflowEventSummary{item}} + } else { + data = WorkflowEventDetail{WorkflowEventSummary: item, Payload: json.RawMessage(`{"n":9007199254740993,"state":"WAITING"}`)} + } + json.NewEncoder(w).Encode(map[string]any{"request_id": event, "data": data}) + })) + defer server.Close() + c, err := NewClient(server.URL, Options{Token: "token"}) + if err != nil { + t.Fatal(err) + } + input := HistoryPageRequest{Limit: 1, Cursor: "cursor+/="} + page, err := c.ListWorkflowEvents(context.Background(), id, input) + if err != nil || page.Items[0].AggregateRevision != 9007199254740993 { + t.Fatal(page, err) + } + detail, err := c.GetWorkflowEvent(context.Background(), id, event) + if err != nil || !strings.Contains(string(detail.Payload), "9007199254740993") { + t.Fatal(detail, err) + } + before := calls.Load() + for _, bad := range []string{"", "../wrong", strings.Repeat("A", 32)} { + if _, err := c.ListWorkflowEvents(context.Background(), bad, input); err == nil { + t.Fatal("invalid parent") + } + if _, err := c.GetWorkflowEvent(context.Background(), id, bad); err == nil { + t.Fatal("invalid event") + } + } + for _, bad := range []HistoryPageRequest{{Limit: -1}, {Limit: 101}, {Cursor: strings.Repeat("x", 1025)}} { + if _, err := c.ListWorkflowEvents(context.Background(), id, bad); err == nil { + t.Fatal("invalid page") + } + } + if calls.Load() != before { + t.Fatal("invalid request reached HTTP") + } + corrupt.Store(true) + if _, err := c.ListWorkflowEvents(context.Background(), id, input); err == nil { + t.Fatal("foreign item accepted") + } + if _, err := c.GetWorkflowEvent(context.Background(), id, event); err == nil { + t.Fatal("foreign detail accepted") + } +} diff --git a/weave/instance/events.go b/weave/instance/events.go new file mode 100644 index 0000000..d89e6fa --- /dev/null +++ b/weave/instance/events.go @@ -0,0 +1,54 @@ +package instance + +import ( + "context" + "encoding/json" + "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" +) + +// EventSummary 只表示已保存的事件，不把当前状态或当前激活反填到历史。 +type EventSummary struct { + ID core.ID `json:"id"` // ID 原不可变状态事件标识。 + WorkflowID core.ID `json:"workflow_id"` // WorkflowID 已核验授权的流程归属。 + AggregateKind string `json:"aggregate_kind"` // AggregateKind workflow、workflow_step或workflow_signal。 + AggregateID core.ID `json:"aggregate_id"` // AggregateID 产生事件的实例、步骤或信号身份。 + AggregateRevision uint64 `json:"aggregate_revision,string"` // AggregateRevision 来源资源版本，各资源独立，不是全局序号。 + Type string `json:"type"` // Type 原始事件类型，不按当前状态改写。 + StateScope string `json:"state_scope"` // StateScope WORKFLOW、STEP或SIGNAL，说明State记录的是哪类对象。 + State string `json:"state"` // State 该事件保存的状态，不代表对象当前状态。 + StepKey string `json:"step_key,omitempty"` // StepKey 步骤事件的固定定义名称；其他事件省略。 + OccurredAt time.Time `json:"occurred_at"` // OccurredAt 数据库事件记录时间，UTC微秒精度，不是外部业务发生时间或提交水位。 +} +type EventPage struct { + WorkflowID core.ID `json:"workflow_id"` // WorkflowID 本页授权的实例身份。 + Revision uint64 `json:"revision,string"` // Revision 本页读取时实例版本，不能作为信号或步骤事件的消费水位。 + Items []EventSummary `json:"items"` // Items 按记录时间和ID倒序的有界摘要，空列表为[]。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 绑定实例与事件查询种类，不承诺跨页快照。 +} +type EventDetail struct { + EventSummary + Payload json.RawMessage `json:"payload"` // Payload 原不可变事件元数据，最多64KiB；嵌套数字保留原文本，输入/结果另走资源详情。 +} + +// ListEvents 要求workflows:audit:read，并核对固定版本全部队列；不能按全局事件ID跨资源读取。 +func (s *Service) ListEvents(ctx context.Context, t auth.Token, id core.ID, in service.HistoryPageInput) (EventPage, error) { + if id.Validate() != nil { + return EventPage{}, fault.Invalid("invalid workflow id") + } + in, err := historyPage(in) + if err != nil { + return EventPage{}, err + } + return s.store.ListEvents(ctx, t, id, in) +} +func (s *Service) GetEvent(ctx context.Context, t auth.Token, id, event core.ID) (EventDetail, error) { + if id.Validate() != nil || event.Validate() != nil { + return EventDetail{}, fault.Invalid("invalid workflow event identity") + } + return s.store.GetEvent(ctx, t, id, event) +} diff --git a/weave/instance/service.go b/weave/instance/service.go index 87323f3..c6ca123 100644 --- a/weave/instance/service.go +++ b/weave/instance/service.go @@ -72,6 +72,8 @@ type StepsView struct { Steps []StepView `json:"steps"` // Steps 按键排序，最多128个，不含内部根检查点。 } type Store interface { + ListEvents(context.Context, auth.Token, core.ID, service.HistoryPageInput) (EventPage, error) + GetEvent(context.Context, auth.Token, core.ID, core.ID) (EventDetail, error) 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) diff --git a/weave/instance/service_test.go b/weave/instance/service_test.go index 5ab49be..b62077a 100644 --- a/weave/instance/service_test.go +++ b/weave/instance/service_test.go @@ -89,3 +89,12 @@ func (s *captureStore) List(context.Context, auth.Token, ListInput) (Page, error s.calls++ return Page{}, nil } + +func (s *captureStore) ListEvents(context.Context, auth.Token, core.ID, service.HistoryPageInput) (EventPage, error) { + s.calls++ + return EventPage{}, nil +} +func (s *captureStore) GetEvent(context.Context, auth.Token, core.ID, core.ID) (EventDetail, error) { + s.calls++ + return EventDetail{}, nil +}