From 71c33eae188fe42ba0179bff255b8012fe4cdd3a Mon Sep 17 00:00:00 2001 From: k2safe Date: Sun, 20 Sep 2026 02:56:41 +0800 Subject: [PATCH] feat: add bounded workflow execution measurement --- AGENTS.md | 1 + README.md | 2 +- docs/08-development-plan.md | 2 +- docs/api/README.md | 1 + docs/api/workflow-load.md | 35 ++ docs/release-checklist.md | 2 +- .../phase-6-workflow-load/environment.json | 40 ++ .../data/phase-6-workflow-load/sample-24.json | 1 + docs/reports/phase-6-workflow-load.md | 41 ++ docs/reports/phase-7-regression-status.md | 3 +- scripts/test-mysql.sh | 2 + sdk/go/cmd/orbit-workflow-load/main.go | 153 +++++++ sdk/go/cmd/orbit-workflow-load/measure.go | 424 ++++++++++++++++++ .../cmd/orbit-workflow-load/measure_test.go | 61 +++ sdk/go/cmd/orbit-workflow-load/report.go | 85 ++++ 15 files changed, 849 insertions(+), 4 deletions(-) create mode 100644 docs/api/workflow-load.md create mode 100644 docs/reports/data/phase-6-workflow-load/environment.json create mode 100644 docs/reports/data/phase-6-workflow-load/sample-24.json create mode 100644 docs/reports/phase-6-workflow-load.md create mode 100644 sdk/go/cmd/orbit-workflow-load/main.go create mode 100644 sdk/go/cmd/orbit-workflow-load/measure.go create mode 100644 sdk/go/cmd/orbit-workflow-load/measure_test.go create mode 100644 sdk/go/cmd/orbit-workflow-load/report.go diff --git a/AGENTS.md b/AGENTS.md index f654bd7..acc056e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -87,3 +87,4 @@ - 凭据轮换保留credential_id及订阅/Worker授权绑定，在同一事务替换摘要、固定到期时间、幂等回执和管理审计；新令牌必须重新建立会话，认证路径每次重新核对凭据状态。撤销必须让认证失效并保留原任务、订阅、交付、会话和审计证据，允许在应用或命名空间停用后执行；重复应用同一固定操作幂等返回。管理CLI先把固定操作文件写成0600，再提交；提交不明时保留文件并用原文件重放，不能重新生成秘密。日志、审计摘要、指标和业务表不得保存秘密原文或摘要。不要把轮换同一凭据ID解释成历史令牌仍有效，也不得级联删除订阅、任务或事件。 - `sdk/go/cmd/orbit-load` 只用于隔离环境受理测量：每项使用稳定独立幂等键，固定请求形状和并发上限，按请求超时，不自动重试，不启动Worker或执行外部副作用；输出延迟百分位与有限错误分类，失败必须以非零状态返回。其结果不能替代真实Worker、流程推进、数据库拓扑或生产容量验收。 - `sdk/go/cmd/orbit-exec-load` 独立测量真实Worker执行：每次运行固定专属处理器/业务键/序号/模拟时长，最多一次尝试且无外部副作用；任务成功须核对持久化结果、唯一尝试、时间顺序及单次本地调用。队列额度只读，提交不重试；整体截止和有界排空后输出实际分母，未完成/未知结果必须非零退出。数据库阶段耗时不得混用客户端轮询观察时间，不将隔离样本解释为生产容量。 +- `sdk/go/cmd/orbit-workflow-load` 独立测量真实Weave流程：每次运行创建固定版本的双分支并行定义，启动实例不自动重试，必须核对流程/步骤/任务/尝试/来源/输出及两倍分支调用；队列只读，处理器无外部副作用，截止后保留实际分母并以非零状态退出。隔离样本不代表生产容量、高可用或跨故障域验收。 diff --git a/README.md b/README.md index 189a757..83d5692 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,7 @@ make protocol-check 受理层负载采样工具见 [有界受理负载工具](docs/api/load.md)，只允许在隔离命名空间或一次性数据库中运行；它不自动重试、不启动 Worker，也不提供生产容量承诺。 -真实Worker与任务完成路径测量使用[`orbit-exec-load`](docs/api/execution-load.md)：短/混合/中长模拟任务、持久结果和单次尝试核验、阶段延迟分布；不修改队列额度、不调用外部业务系统。 +真实Worker与任务完成路径测量使用[`orbit-exec-load`](docs/api/execution-load.md)：短/混合/中长模拟任务、持久结果和单次尝试核验、阶段延迟分布；不修改队列额度、不调用外部业务系统。Weave 并行流程链路使用[`orbit-workflow-load`](docs/api/workflow-load.md)，核对固定版本、步骤、来源和最终输出。 SDK 独立发布门禁使用 `make sdk-check`，在 `GOWORK=off` 下检查完整依赖图，并运行 SDK 的竞态测试、vet 和全包构建；规则和接入边界见 [SDK 发布门禁](docs/sdk/release.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 7ced9ff..20fe137 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -62,7 +62,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( 2. 已实现[基础运行观测](api/observability.md)及[有界业务状态与事件采样](reports/phase-6-business-observability.md)：HTTP/WS、数据库连接池、后台批次、固定任务/交付/流程/Inbox状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供，运维人员也可用 `orbit-admin observe snapshot` 获取同一有界单节点快照。仍需端到端外部效果延迟、全局持久吞吐、Inbox从事件产生到推进完成的精确阶段延迟、全局发送队列、多节点聚合及生产采集/通知验证。 3. 已补齐受保护管理环境中的固定凭据操作文件、轮换、撤销、幂等回执和脱敏管理审计；轮换保留授权绑定并要求新令牌重新建立会话，旧 Worker 会话被排空且立即过期，撤销允许在停用范围内执行且不改写历史记录。已用一次性 MySQL 定向验证旧令牌失效、新令牌重绑和零外键；仍需补齐提交不明故障切点、多节点安全重绑演练、凭据最小权限配置核对和生产密钥托管/审计演练。 4. 已补[`orbit-admin retention inventory`](api/retention.md)只读盘点和[阶段报告](reports/phase-6-retention-inventory.md)，能区分过期幂等、确定终态、未知效果、未终结流程、已确认/未确认交付、已消费/未消费Inbox、事件和审计规模；它不执行归档或删除。后续仍需评审各类保留周期、归档格式、受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 -5. 已补[`orbit-load`](api/load.md)有界受理负载工具及24次/1000次隔离样本，见[受理报告](reports/phase-6-load.md)。新增独立[`orbit-exec-load`](api/execution-load.md)，通过真实Worker、任务结果和尝试历史核对，提供短/混合/中长任务的受理、排队、授权执行及完成耗时分布；具体环境、原始JSON和失败分母见[执行测量报告](reports/phase-6-execution-load.md)。标准Go 1.27.1全量仍待复跑，下一步补定时集中触发、Weave流程推进、长期积压和资源曲线；目标数据库拓扑容量仍单独验收。 +5. 已补[`orbit-load`](api/load.md)有界受理负载工具及24次/1000次隔离样本，见[受理报告](reports/phase-6-load.md)。新增独立[`orbit-exec-load`](api/execution-load.md)，通过真实Worker、任务结果和尝试历史核对，提供短/混合/中长任务的受理、排队、授权执行及完成耗时分布；具体环境、原始JSON和失败分母见[执行测量报告](reports/phase-6-execution-load.md)。新增[`orbit-workflow-load`](api/workflow-load.md)，用固定版本的双分支并行流程核对步骤、来源、尝试和输出，见[流程测量报告](reports/phase-6-workflow-load.md)。标准Go 1.27.1全量仍待复跑，下一步补定时集中触发、长期积压和资源曲线；目标数据库拓扑容量仍单独验收。 6. 已补[隔离备份恢复演练](api/backup-restore.md)：源库和恢复库均为一次性 MySQL 容器，逻辑备份恢复后核对迁移账本、核心记录数量、固定索引和零外键；证据见[恢复报告](reports/phase-6-backup-restore.md)。该演练不替代生产备份介质、复制切换、跨版本恢复或 RPO/RTO 验收。 7. 已补[A18 数据库短暂不可用恢复演练](api/db-recovery.md)：readiness 在数据库停止期间返回 503，同卷恢复后返回 200，并用原权限范围继续受理任务；证据见[数据库恢复报告](reports/phase-6-db-recovery.md)。该演练不替代主从切换、复制确认、跨故障域或 RPO/RTO 验收。 diff --git a/docs/api/README.md b/docs/api/README.md index f0f85fc..9e75425 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -25,6 +25,7 @@ | `orbit-admin retention inventory` 保留清单 | 已实现带 UTC 截止时间的只读盘点；不执行归档、删除或权限变更 | [保留清单](retention.md) | | `sdk/go/cmd/orbit-load` 有界受理负载工具 | 已实现固定并发、无自动重试、延迟百分位和错误分类；只用于隔离环境 | [负载工具](load.md) | | `sdk/go/cmd/orbit-exec-load` 有界任务执行测量 | 已实现真实Worker、固定混合时长、最终结果/尝试核验及阶段耗时分布 | [执行测量](execution-load.md) | +| `sdk/go/cmd/orbit-workflow-load` 有界流程执行测量 | 已实现固定版本、双分支并行、真实Worker、步骤/来源/尝试/输出核验及阶段耗时分布 | [流程测量](workflow-load.md) | | `scripts/test-backup-restore.sh` 隔离备份恢复演练 | 已实现一次性 MySQL 逻辑备份、独立恢复、核心计数及零外键核验 | [备份恢复演练](backup-restore.md) | | `scripts/test-db-recovery.sh` MySQL 短暂不可用恢复演练 | 已实现 readiness 503、同卷恢复 200 及恢复后任务受理核验 | [数据库恢复演练](db-recovery.md) | | `/v1/worker/connect` | P2 已实现并通过阶段验收 | [Worker 契约](worker.md)、[Schema](worker-schema.md) | diff --git a/docs/api/workflow-load.md b/docs/api/workflow-load.md new file mode 100644 index 0000000..93cb345 --- /dev/null +++ b/docs/api/workflow-load.md @@ -0,0 +1,35 @@ +# Weave 流程负载测量工具 + +`orbit-workflow-load` 是 SDK 模块中的隔离测量命令，用于验证一条真实的 Weave 流程链路：创建定义、发布不可变 `v1`、启动多个实例、由独立 WSS Worker 执行两个有限并行分支，再核对持久化流程、步骤、任务、尝试、任务来源和最终输出。 + +它只使用 Orbit HTTP/WSS 和 MySQL 服务端已有能力，不连接消息队列、缓存或业务数据库；处理器只回显固定输入，不产生外部业务副作用。工具不会自动重试写请求，响应不明时必须保留原运行身份，通过报告中的分母定位未确认实例。 + +## 使用 + +先为隔离命名空间签发包含 `workflow-access`、`workflow-run-access` 与 `worker-access` 的凭据文件，再执行： + +```sh +ORBIT_URL=https://orbit.example.test \ +ORBIT_CREDENTIALS_FILE=/run/secrets/orbit-load.json \ +go run ./cmd/orbit-workflow-load \ + -requests 100 \ + -concurrency 8 \ + -workers 2 \ + -capacity 4 \ + -run-timeout 5m \ + -poll-interval 250ms +``` + +参数上限是 `requests<=2000`、启动/核验并发 `<=128`、Worker 数 `<=32`、单 Worker 容量 `<=256` 且总容量 `<=1024`。`-timeout` 是单次 HTTP 交换截止，`-run-timeout` 是整次样本截止；两者都不是生产系统的 SLA。 + +工具先只读核对队列，不修改队列额度；定义名、实例幂等键、业务键和处理器名都带一次运行生成的随机前缀。每个实例固定使用输入 `{"run_id":"...","sequence":N}`，流程输出必须同时包含 `left` 和 `right` 两个相同对象。 + +## 报告解释 + +输出为一行 JSON，包含 `attempted`、`accepted`、`verified_success`、`unverified`、`handler_calls`、`worker_started`、`errors` 及 `admission`/`completion` 百分位。`complete=true` 的必要条件是：全部启动调用有唯一回执、全部实例为 `SUCCEEDED`、每个流程恰有两个成功 TASK 步骤、每个任务恰有一次成功尝试、来源固定到同一实例和版本、两倍分支处理器调用全部被确认且没有重复调用。 + +任何预检、发布、启动、读取、步骤、尝试、来源或截止错误都会保留稳定错误分类并以非零状态退出。空分布的百分位是 JSON `null`，不能解释成零延迟。`completion` 是服务端 `created_at` 到 `completed_at` 的持久化耗时；它不是客户端轮询耗时。 + +## 验证边界 + +该工具适合在一次性 MySQL、测试凭据和明确的资源上限下做正确性、故障分母和阶段延迟验证。单机样本不能证明多节点吞吐、跨故障域高可用、备份恢复、RPO/RTO 或面向生产的容量；生产验收还需要目标拓扑、数据规模、连接池、备份介质、监控采集和发布兼容矩阵的独立证据。 diff --git a/docs/release-checklist.md b/docs/release-checklist.md index eec952d..7dbdc08 100644 --- a/docs/release-checklist.md +++ b/docs/release-checklist.md @@ -38,7 +38,7 @@ | 项目 | 必须提供的证据 | 状态 | | --- | --- | --- | -| 受理、派发、流程推进和恢复吞吐 | 固定硬件、数据量、任务/流程混合模型，p50/p95/p99和资源曲线 | 已有受理样本及[真实Worker短/混合/中长执行样本](reports/phase-6-execution-load.md)；定时/流程/积压/资源曲线和目标生产容量未验收 | +| 受理、派发、流程推进和恢复吞吐 | 固定硬件、数据量、任务/流程混合模型，p50/p95/p99和资源曲线 | 已有受理样本、[真实Worker短/混合/中长执行样本](reports/phase-6-execution-load.md)及[Weave双分支流程样本](reports/phase-6-workflow-load.md)；定时/积压/资源曲线和目标生产容量未验收 | | 100万+企业规模推导 | 租户、队列、连接、事件、索引和保存周期的容量模型 | 未验收，不能由租户数直接推导 | | MySQL 复制、切换、RPO/RTO | 目标拓扑上的实际切换、复制确认、恢复点和恢复时间 | 未验收 | | 慢消费者、重连风暴、磁盘和锁等待 | 受控故障负载及积压清空证据 | 未验收 | diff --git a/docs/reports/data/phase-6-workflow-load/environment.json b/docs/reports/data/phase-6-workflow-load/environment.json new file mode 100644 index 0000000..f60392d --- /dev/null +++ b/docs/reports/data/phase-6-workflow-load/environment.json @@ -0,0 +1,40 @@ +{ + "date_local": "2026-09-20", + "timezone": "Asia/Shanghai", + "test_go": "go1.25.14", + "declared_go": "1.27.1", + "temporary_modfile": true, + "mysql": "8.4.10", + "docker_vm_cpus": 12, + "docker_vm_memory_bytes": 12529274880, + "architecture": "arm64", + "orbit_processes": 1, + "mysql_nodes": 1, + "worker_processes": 1, + "worker_connections_normal": 2, + "worker_capacity_each": 4, + "workflow_requests": 24, + "workflow_concurrency": 6, + "queue_max_concurrency": 4, + "queue_rate_per_second": 10, + "poll_interval_ms": 100, + "http_timeout_ms": 5000, + "run_timeout_ms": 90000, + "metrics_enabled": false, + "cpu_memory_limits": "No dedicated limits; shared local Docker VM, other workloads possible", + "fresh_disposable_database": true, + "network": "Go runner shares MySQL container network namespace; loopback HTTP/WSS and MySQL", + "schema_verification": { + "schema_version": 28, + "migration_count": 28, + "foreign_key_count": 0, + "required_index_count": 6, + "complete": true + }, + "source_sha256": { + "sdk/go/cmd/orbit-workflow-load/main.go": "05a160431b26d91ab430392b23f4fcf76bd23dfeb21bd2047fe8372c17e93fe5", + "sdk/go/cmd/orbit-workflow-load/measure.go": "1a31d06d3a9e483aee5aca115f7f1230649e050a0bddae1770659fe2735a99a3", + "sdk/go/cmd/orbit-workflow-load/measure_test.go": "42e710139f515aa339b5684107d00fcb413688e7512fc4c36ebfe56776ad41a1", + "sdk/go/cmd/orbit-workflow-load/report.go": "701a27a3f4f56cf9566fbcf97fba8024ced36b32754f4a5dcdae4646a2a4b24a" + } +} diff --git a/docs/reports/data/phase-6-workflow-load/sample-24.json b/docs/reports/data/phase-6-workflow-load/sample-24.json new file mode 100644 index 0000000..ffa815e --- /dev/null +++ b/docs/reports/data/phase-6-workflow-load/sample-24.json @@ -0,0 +1 @@ +{"run_id":"e4f954ba8507b35fda519f197c301304","go_version":"go1.25.14","goos":"linux","goarch":"arm64","started_at":"2026-09-19T18:55:18.343555301Z","admission_finished_at":"2026-09-19T18:55:18.553077926Z","finished_at":"2026-09-19T18:55:33.351706711Z","requests":24,"concurrency":6,"workers":2,"capacity_per_worker":4,"run_timeout_ms":90000,"poll_interval_ms":100,"http_timeout_ms":5000,"queue":{"max_unfinished_tasks":100000,"unfinished_tasks":"0","id":"6ba1b31c689aa7249f9f41bc873b3a6f","namespace_id":"5dd105847384f7782a1923aa4288508e","name":"workflow","paused":false,"max_concurrency":4,"rate_per_second":10,"revision":"1"},"definition_id":"5650b9f2a6838263ea3222dd879f6540","version_id":"73608bdb26c948a98700d101c1f5e787","attempted":24,"not_submitted":0,"accepted":24,"admission_errors":0,"verified_success":24,"unverified":0,"handler_calls":48,"duplicate_invocations":0,"worker_started":48,"worker_reconnects":0,"worker_renewals_confirmed":0,"admission":{"count":24,"p50_ms":44.767208,"p95_ms":49.584459,"p99_ms":56.462333,"max_ms":56.462333},"completion":{"count":24,"p50_ms":7898.658,"p95_ms":13838.444,"p99_ms":14788.909,"max_ms":14788.909},"errors":{},"complete":true} diff --git a/docs/reports/phase-6-workflow-load.md b/docs/reports/phase-6-workflow-load.md new file mode 100644 index 0000000..449fcaf --- /dev/null +++ b/docs/reports/phase-6-workflow-load.md @@ -0,0 +1,41 @@ +# P6 Weave 流程执行测量 + +验证日期：2026-09-20（UTC 运行时间记录为 2026-09-19 18:51）。这是一次性 MySQL 8.4 容器中的隔离正确性与阶段延迟样本，不是生产容量或高可用验收。 + +## 测量内容 + +`orbit-workflow-load` 通过独立 Go SDK 完成以下链路： + +1. 只读检查队列状态。 +2. 创建一个新的稳定定义并发布不可变 `v1`。 +3. 启动 24 个独立实例，每个实例使用自己的幂等键和业务键。 +4. 每个实例执行一个包含 `left`、`right` 两个静态分支的 `PARALLEL` 节点；两个分支各创建一个 `TASK`。 +5. 两个真实 WSS Worker（每个容量4）回显固定输入。 +6. 对每个成功实例读取步骤、任务、唯一尝试、流程任务来源和最终输出，核对实例固定版本及来源归属。 + +处理器只回显 `run_id` 和 `sequence`，没有外部副作用；启动调用和流程读取都没有 SDK 自动重试。流程定义把任务尝试上限固定为1，响应不明时保留原运行身份。 + +## 实际结果 + +| 项目 | 结果 | +| --- | ---: | +| 计划/实际启动调用 | 24 / 24 | +| 唯一受理实例 | 24 | +| 持久化核验成功 | 24 | +| 未核验实例 | 0 | +| 处理器调用 | 48（每实例两个分支） | +| 重复调用 | 0 | +| Worker 启动任务数 | 48 | +| 错误分类 | 空 | +| 完成标记 | `true` | +| 启动耗时 P50 / P95 / P99 | 44.767208 / 49.584459 / 56.462333 ms | +| 服务端流程完成耗时 P50 / P95 / P99 / Max | 7898.658 / 13838.444 / 14788.909 / 14788.909 ms | + +原始 JSON：[sample-24.json](data/phase-6-workflow-load/sample-24.json)，工具链和源码摘要见[环境清单](data/phase-6-workflow-load/environment.json)。执行工具和字段语义见[流程负载测量手册](../api/workflow-load.md)。 + +## 环境和边界 + +- 负载工具：Go `1.25.14`，Linux `arm64`；请求并发6，Worker 2×4，轮询100ms，单次 HTTP 5s，整次样本90s。 +- 服务端：一次性 Docker MySQL `8.4`，通过当前迁移器初始化；结束时 `orbit-admin verify` 输出 `schema_version=28`、`migration_count=28`、`foreign_key_count=0`、`required_index_count=6`、`complete=true`。 +- 队列为只读测试资源，`max_concurrency=4`、`rate_per_second=10`；队列额度和数据库结构没有因测量修改。 +- 该样本证明流程发布、并行推进、Worker 执行、持久化回执和来源核对在此环境可复现；它不证明多节点吞吐、跨故障域恢复、生产连接池容量、RPO/RTO、备份介质或市场化部署 SLA。 diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index ad76921..3cf79b6 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -8,7 +8,7 @@ | --- | --- | --- | | 服务端单元与竞态 | Docker `golang:1.25`，临时 modfile；`GOWORK=off go test -race ./...` | PASS | | 服务端 vet | Docker `golang:1.25`；`GOWORK=off go vet ./...` | PASS | -| SDK 竞态、vet、构建 | Docker `golang:1.25`；SDK 全 module、全部示例和 `orbit-load` | PASS | +| SDK 竞态、vet、构建 | Docker `golang:1.25`；SDK 全 module、全部示例、`orbit-load`、`orbit-exec-load` 和 `orbit-workflow-load` | PASS | | 当前协议门禁 | `make protocol-check` | PASS，当前 v1 契约同步 | | 外键迁移防回归 | `TestPostRemovalMigrationGuard`；服务端迁移运行时校验 | PASS；000008 以前仅保留兼容清理历史，后续迁移在执行前拒绝引用约束 DDL | | MySQL 存储集成 | MySQL 8.4.10，一次性容器；`go test -race -run Integration -skip 'TestIntegration(WorkflowCommitFaults\|CredentialCommitFaults)$'` | PASS，210.893 秒；包含空库无外键基线、v7升级、Pulse/Weave/API/观测/恢复与W16竞争 | @@ -18,6 +18,7 @@ | SDK 应用配额子进程故障 | 独立双命名空间应用配额凭据；共享网络命名空间；`TestIntegrationWorkerProcessApplicationBackpressure` | PASS，1.069 秒；重启后共享额度、原回执和跨命名空间背压保持 | | Orbit HTTP/metrics 子进程 | 同一共享网络命名空间；`TestIntegrationMetricsProcess`；metrics 启用/禁用各一组 | PASS，3.081 秒；监听隔离、隐私、后台采样和优雅退出均通过 | | 真实Worker执行测量 | `orbit-exec-load`；最终短/混合/中长及截止不足四组，环境及原始JSON见[测量报告](phase-6-execution-load.md) | 96+96+6项全部逐项核验，长任务18次续租；24项截止负例按预期非零退出。SDK全模块竞态/vet/build通过，生产容量未验收 | +| 真实Weave流程测量 | `orbit-workflow-load`；24实例、固定版本双分支并行，原始JSON见[流程测量报告](phase-6-workflow-load.md) | 24/24实例成功，48/48任务尝试与来源、输出逐项核验，重复调用0；生产容量未验收 | ## 当前环境仍未覆盖 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 3239c1c..0fe1399 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -47,6 +47,7 @@ ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" "$TASK_GO" test -race -count=1 - "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-admin" ./cmd/orbit-admin (cd sdk/go && GOWORK=off "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-load" ./cmd/orbit-load) (cd sdk/go && GOWORK=off "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-exec-load" ./cmd/orbit-exec-load) +(cd sdk/go && GOWORK=off "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-workflow-load" ./cmd/orbit-workflow-load) ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-migrate" status ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" verify TASK_SNAPSHOT_JSON=$(ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" observe snapshot) @@ -111,6 +112,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_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/workflow.credentials.json" "$TASK_TEST_DIR/orbit-workflow-load" -requests 6 -concurrency 2 -workers 2 -capacity 2 -run-timeout 30s -poll-interval 100ms) (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) diff --git a/sdk/go/cmd/orbit-workflow-load/main.go b/sdk/go/cmd/orbit-workflow-load/main.go new file mode 100644 index 0000000..2779206 --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/main.go @@ -0,0 +1,153 @@ +// orbit-workflow-load 在隔离环境测量真实流程发布、并行激活、Worker执行和持久化完成链路。 +// 流程只包含两个有限并行分支，处理器回显输入，不访问外部业务系统。 +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +type config struct { + URL, Token, QueueID string + Requests, Concurrency, Workers, Capacity int + Timeout, RunTimeout, PollInterval time.Duration +} + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func run() error { + cfg, err := parseConfig(os.Args[1:]) + if err != nil { + return err + } + client, err := orbit.NewClient(cfg.URL, orbit.Options{Token: cfg.Token, HTTPClient: &http.Client{Timeout: cfg.Timeout}}) + if err != nil { + return errors.New("invalid client configuration") + } + runID, err := orbit.NewWorkerMessageID() + if err != nil { + return errors.New("cannot generate run identity") + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + report := measure(ctx, cfg, client, runID) + if err := json.NewEncoder(os.Stdout).Encode(report); err != nil { + return err + } + if !report.Complete { + return errors.New("workflow sample incomplete; inspect report and original run identities") + } + return nil +} + +func parseConfig(args []string) (config, error) { + cfg := config{URL: os.Getenv("ORBIT_URL"), Token: os.Getenv("ORBIT_TOKEN"), QueueID: os.Getenv("ORBIT_QUEUE_ID"), Requests: 100, Concurrency: 8, Workers: 2, Capacity: 4, Timeout: 5 * time.Second, RunTimeout: 5 * time.Minute, PollInterval: 250 * time.Millisecond} + flags := flag.NewFlagSet("orbit-workflow-load", flag.ContinueOnError) + var credentials string + flags.StringVar(&cfg.URL, "url", cfg.URL, "Orbit HTTP(S) URL") + flags.StringVar(&credentials, "credentials", os.Getenv("ORBIT_CREDENTIALS_FILE"), "private credentials file for an isolated namespace") + flags.StringVar(&cfg.QueueID, "queue", cfg.QueueID, "existing isolated queue; configuration is not changed") + flags.IntVar(&cfg.Requests, "requests", cfg.Requests, "workflow instances, 1..2000") + flags.IntVar(&cfg.Concurrency, "concurrency", cfg.Concurrency, "maximum concurrent start/read requests, 1..128") + flags.IntVar(&cfg.Workers, "workers", cfg.Workers, "independent WSS Workers in this process, 1..32") + flags.IntVar(&cfg.Capacity, "capacity", cfg.Capacity, "concurrent handlers per Worker, 1..256; total at most 1024") + flags.DurationVar(&cfg.Timeout, "timeout", cfg.Timeout, "per HTTP request timeout, 1ms..1m") + flags.DurationVar(&cfg.RunTimeout, "run-timeout", cfg.RunTimeout, "whole sample deadline, 2s..1h, plus at most 5s Worker shutdown") + flags.DurationVar(&cfg.PollInterval, "poll-interval", cfg.PollInterval, "per active verifier polling interval, 100ms..10s") + if err := flags.Parse(args); err != nil { + return config{}, err + } + if flags.NArg() != 0 { + return config{}, errors.New("positional arguments are not supported") + } + if credentials != "" { + f, err := os.Open(credentials) + if err != nil { + return config{}, errors.New("cannot open private credentials file") + } + raw, err := io.ReadAll(io.LimitReader(f, (16<<10)+1)) + _ = f.Close() + var saved struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if err != nil || len(raw) > 16<<10 || json.Unmarshal(raw, &saved) != nil { + return config{}, errors.New("invalid credentials file (maximum 16 KiB)") + } + if cfg.Token == "" { + cfg.Token = saved.Token + } + if cfg.QueueID == "" { + cfg.QueueID = saved.QueueID + } + } + if cfg.URL == "" || cfg.Token == "" || !validID(cfg.QueueID) { + return config{}, errors.New("URL, token and a 32-character lowercase hexadecimal queue ID are required") + } + if cfg.Requests < 1 || cfg.Requests > 2000 || cfg.Concurrency < 1 || cfg.Concurrency > 128 || cfg.Workers < 1 || cfg.Workers > 32 || cfg.Capacity < 1 || cfg.Capacity > 256 || cfg.Workers*cfg.Capacity > 1024 { + return config{}, errors.New("request, concurrency or Worker count is out of bounds") + } + if cfg.Timeout < time.Millisecond || cfg.Timeout > time.Minute || cfg.RunTimeout < 2*time.Second || cfg.RunTimeout > time.Hour || cfg.PollInterval < 100*time.Millisecond || cfg.PollInterval > 10*time.Second { + return config{}, errors.New("timeout or poll interval is out of bounds") + } + if cfg.Concurrency > cfg.Requests { + cfg.Concurrency = cfg.Requests + } + return cfg, nil +} + +func validID(id string) bool { + if len(id) != 32 { + return false + } + for _, c := range id { + if !(c >= '0' && c <= '9' || c >= 'a' && c <= 'f') { + return false + } + } + return true +} + +func classify(err error) string { + if errors.Is(err, context.DeadlineExceeded) { + return "TIMEOUT" + } + if errors.Is(err, context.Canceled) { + return "CANCELLED" + } + var api *orbit.APIError + if errors.As(err, &api) { + switch api.StatusCode { + case 401, 403: + return "AUTH" + case 409: + return "CONFLICT" + case 429: + return "BACKPRESSURE" + case 400, 422: + return "INVALID_REQUEST" + } + return "HTTP" + } + var h *orbit.HTTPError + if errors.As(err, &h) { + return "HTTP" + } + return "TRANSPORT" +} diff --git a/sdk/go/cmd/orbit-workflow-load/measure.go b/sdk/go/cmd/orbit-workflow-load/measure.go new file mode 100644 index 0000000..e8274be --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/measure.go @@ -0,0 +1,424 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "sync" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +// workflowPayload 是两个并行分支共享的固定输入；不含凭据或外部地址。 +type workflowPayload struct { + RunID string `json:"run_id"` // RunID 隔离本次测量，不作为业务幂等保证。 + Sequence int `json:"sequence"` // Sequence 从0开始的实例序号，用于发现错配和重复调用。 +} + +func decodePayload(raw json.RawMessage) (workflowPayload, bool) { + var wire struct { + RunID *string `json:"run_id"` + Sequence *int `json:"sequence"` + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + if decoder.Decode(&wire) != nil || wire.RunID == nil || wire.Sequence == nil { + return workflowPayload{}, false + } + var trailing any + if decoder.Decode(&trailing) != io.EOF { + return workflowPayload{}, false + } + return workflowPayload{RunID: *wire.RunID, Sequence: *wire.Sequence}, true +} + +func handlerName(runID string) string { return "load.workflow." + runID } +func definitionKey(runID string) string { return "orbit-workflow-load-" + runID } +func requestKey(runID string, sequence int) string { + return fmt.Sprintf("orbit-workflow-%s-%06d", runID, sequence) +} +func businessKey(runID string, sequence int) string { + return fmt.Sprintf("orbit-workflow-%s-%06d", runID, sequence) +} + +type workflowAPI interface { + GetQueue(context.Context, string) (orbit.Queue, error) + CreateWorkflowDefinition(context.Context, string, orbit.CreateWorkflowDefinitionRequest) (orbit.WorkflowDefinitionResult, error) + PublishWorkflowVersion(context.Context, string, string, orbit.PublishWorkflowVersionRequest) (orbit.WorkflowVersionResult, error) + StartWorkflow(context.Context, string, orbit.StartWorkflowRequest) (orbit.WorkflowStartResult, error) + GetWorkflow(context.Context, string) (orbit.WorkflowInstance, error) + GetWorkflowSteps(context.Context, string) (orbit.WorkflowSteps, error) + GetTask(context.Context, string) (orbit.Task, error) + ListTaskAttempts(context.Context, string, orbit.HistoryPageRequest) (orbit.AttemptPage, error) + GetTaskSource(context.Context, string) (orbit.TaskSource, error) +} + +type workflowAdmission struct { + sequence int + instance orbit.WorkflowInstance + duration time.Duration + code string +} + +type workflowVerification struct { + sequence int + duration time.Duration + code string +} + +// invocationLog 有界保存每个流程的实际处理器调用，两个并行分支各应调用一次。 +type invocationLog struct { + mu sync.Mutex + calls []int + invalid int +} + +func newInvocationLog(n int) *invocationLog { return &invocationLog{calls: make([]int, n)} } + +func (log *invocationLog) handler(ctx context.Context, cfg config, runID string) orbit.WorkerHandler { + return orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: handlerName(runID), HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(work context.Context, execution orbit.WorkerExecution) (json.RawMessage, error) { + payload, valid := decodePayload(execution.Task.Payload) + if !valid || payload.RunID != runID || payload.Sequence < 0 || payload.Sequence >= cfg.Requests || execution.Task.Handler != handlerName(runID) || execution.Task.HandlerVersion != "v1" || execution.Task.PayloadVersion != "v1" || execution.Task.QueueID != cfg.QueueID { + log.mu.Lock() + log.invalid++ + log.mu.Unlock() + return nil, &orbit.WorkerFailure{Code: "WORKFLOW_INPUT_MISMATCH"} + } + log.mu.Lock() + log.calls[payload.Sequence]++ + log.mu.Unlock() + select { + case <-work.Done(): + return nil, work.Err() + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + return execution.Task.Payload, nil + }} +} + +func workflowSpec(queue, handler string, timeout time.Duration) orbit.WorkflowSpecification { + typ := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"run_id": {Kind: "STRING"}, "sequence": {Kind: "NUMBER"}}} + task := orbit.WorkflowTask{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", InputType: typ, Input: orbit.WorkflowRef("input"), ResultType: typ, TimeoutMS: uint64(timeout.Milliseconds()), Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(timeout.Milliseconds())}, RecoverySafety: orbit.RecoveryReconcileOnly} + parallel := orbit.WorkflowParallel{Branches: []orbit.WorkflowParallelBranch{ + {Name: "left", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "left", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("left")}}, + {Name: "right", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "right", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("right")}}, + }} + output := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"left": typ, "right": typ}} + return orbit.WorkflowSpecification{SchemaVersion: 1, InputType: typ, TimeoutMS: uint64(timeout.Milliseconds()), FailurePolicy: "REVIEW", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "pair", Kind: "PARALLEL", Parallel: ¶llel}}, OutputType: output, Output: orbit.WorkflowRef("pair")}} +} + +func measure(parent context.Context, cfg config, api *orbit.Client, runID string) workflowReport { + return measureClient(parent, cfg, api, runID) +} + +func measureClient(parent context.Context, cfg config, api *orbit.Client, runID string) workflowReport { + r := newReport(cfg, runID) + ctx, cancel := context.WithTimeout(parent, cfg.RunTimeout) + defer cancel() + requestCtx, stop := context.WithTimeout(ctx, cfg.Timeout) + queue, err := api.GetQueue(requestCtx, cfg.QueueID) + stop() + if err != nil { + r.Errors["PREFLIGHT_"+classify(err)]++ + return r.finish() + } + r.Queue = &queue + if queue.ID != cfg.QueueID || queue.Paused || queue.MaxConcurrency == 0 || queue.RatePerSecond == 0 { + r.Errors["QUEUE_UNAVAILABLE"]++ + return r.finish() + } + handler := handlerName(runID) + typicalTimeout := cfg.RunTimeout + if typicalTimeout > 30*time.Second { + typicalTimeout = 30 * time.Second + } + spec := workflowSpec(cfg.QueueID, handler, typicalTimeout) + requestCtx, stop = context.WithTimeout(ctx, cfg.Timeout) + definition, err := api.CreateWorkflowDefinition(requestCtx, definitionKey(runID), orbit.CreateWorkflowDefinitionRequest{Name: definitionKey(runID), Description: "bounded parallel workflow verification"}) + stop() + if err != nil { + r.Errors["CREATE_"+classify(err)]++ + return r.finish() + } + r.DefinitionID = definition.Definition.ID + if !validID(r.DefinitionID) { + r.Errors["INVALID_DEFINITION_RECEIPT"]++ + return r.finish() + } + requestCtx, stop = context.WithTimeout(ctx, cfg.Timeout) + version, err := api.PublishWorkflowVersion(requestCtx, r.DefinitionID, runID, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "bounded parallel workflow verification"}) + stop() + if err != nil { + r.Errors["PUBLISH_"+classify(err)]++ + return r.finish() + } + r.VersionID = version.Version.ID + if !validID(r.VersionID) || version.Version.Version != "v1" || version.Version.DefinitionID != r.DefinitionID { + r.Errors["INVALID_VERSION_RECEIPT"]++ + return r.finish() + } + + log := newInvocationLog(cfg.Requests) + workers := make([]*orbit.Worker, 0, cfg.Workers) + for i := 0; i < cfg.Workers; i++ { + worker, e := orbit.NewWorker(api, orbit.WorkerOptions{QueueID: cfg.QueueID, Capacity: uint32(cfg.Capacity), ShutdownTimeout: 5 * time.Second, Handlers: []orbit.WorkerHandler{log.handler(ctx, cfg, runID)}}) + if e != nil { + r.Errors["WORKER_CONFIG"]++ + return r.finish() + } + workers = append(workers, worker) + } + workerCtx, stopWorkers := context.WithCancel(ctx) + done := make(chan error, len(workers)) + for _, worker := range workers { + go func(w *orbit.Worker) { + e := w.Run(workerCtx) + if workerCtx.Err() == nil { + cancel() + } + done <- e + }(worker) + } + + admissions := submitWorkflows(ctx, cfg, api, runID, r.DefinitionID) + r.Attempted = len(admissions) + r.NotSubmitted = cfg.Requests - r.Attempted + var accepted []workflowAdmission + var admissionDurations []time.Duration + for _, a := range admissions { + admissionDurations = append(admissionDurations, a.duration) + if a.code != "" { + r.AdmissionErrors++ + r.Errors["ADMISSION_"+a.code]++ + continue + } + accepted = append(accepted, a) + } + r.Accepted = len(accepted) + r.Admission = distributionOf(admissionDurations) + r.AdmissionFinishedAt = time.Now().UTC() + verified := verifyWorkflows(ctx, cfg, api, runID, r.DefinitionID, accepted) + var completions []time.Duration + for _, v := range verified { + if v.code != "" { + r.Errors[v.code]++ + continue + } + r.VerifiedSuccess++ + completions = append(completions, v.duration) + } + if ctx.Err() != nil { + r.Errors["RUN_"+classify(ctx.Err())]++ + } + stopWorkers() + for range workers { + if e := <-done; e != nil { + r.Errors["WORKER_STOP"]++ + } + } + for _, worker := range workers { + stats := worker.Stats() + r.WorkerStarted += stats.Started + r.Reconnects += stats.Reconnects + r.Renewals += stats.RenewalsConfirmed + } + log.mu.Lock() + for _, calls := range log.calls { + r.HandlerCalls += calls + if calls > 2 { + r.DuplicateInvocations += calls - 2 + } + if calls != 2 { + r.Errors["HANDLER_CALL_MISMATCH"]++ + } + } + r.Errors["INVALID_HANDLER_INPUT"] += log.invalid + log.mu.Unlock() + r.Completion = distributionOf(completions) + return r.finish() +} + +func submitWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string) []workflowAdmission { + jobs := make(chan int) + out := make(chan workflowAdmission, cfg.Requests) + var wg sync.WaitGroup + for i := 0; i < cfg.Concurrency; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for sequence := range jobs { + if ctx.Err() != nil { + continue + } + payload, _ := json.Marshal(workflowPayload{RunID: runID, Sequence: sequence}) + started := time.Now() + call, stop := context.WithTimeout(ctx, cfg.Timeout) + result, err := api.StartWorkflow(call, requestKey(runID, sequence), orbit.StartWorkflowRequest{DefinitionID: definitionID, Version: "v1", Input: payload, BusinessKey: businessKey(runID, sequence)}) + stop() + a := workflowAdmission{sequence: sequence, instance: result.Instance, duration: time.Since(started)} + if err != nil { + a.code = classify(err) + } else if result.Deduplicated || !validID(result.Instance.ID) || result.Instance.DefinitionID != definitionID || result.Instance.Version != "v1" || result.Instance.BusinessKey != businessKey(runID, sequence) { + a.code = "INVALID_RECEIPT" + } + out <- a + } + }() + } +feed: + for sequence := 0; sequence < cfg.Requests; sequence++ { + select { + case <-ctx.Done(): + break feed + case jobs <- sequence: + } + } + close(jobs) + wg.Wait() + close(out) + results := make([]workflowAdmission, 0, cfg.Requests) + for result := range out { + results = append(results, result) + } + return results +} + +func verifyWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string, admissions []workflowAdmission) []workflowVerification { + jobs := make(chan workflowAdmission, len(admissions)) + for _, admission := range admissions { + jobs <- admission + } + close(jobs) + out := make(chan workflowVerification, len(admissions)) + var wg sync.WaitGroup + for i := 0; i < cfg.Concurrency; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for admission := range jobs { + duration, code := verifyWorkflow(ctx, cfg, api, runID, definitionID, admission) + out <- workflowVerification{sequence: admission.sequence, duration: duration, code: code} + } + }() + } + wg.Wait() + close(out) + results := make([]workflowVerification, 0, len(admissions)) + for result := range out { + results = append(results, result) + } + return results +} + +func verifyWorkflow(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string, admission workflowAdmission) (time.Duration, string) { + for { + if ctx.Err() != nil { + return 0, "VERIFY_" + classify(ctx.Err()) + } + call, stop := context.WithTimeout(ctx, cfg.Timeout) + instance, err := api.GetWorkflow(call, admission.instance.ID) + stop() + if err != nil { + return 0, "READ_" + classify(err) + } + if instance.ID != admission.instance.ID || instance.DefinitionID != definitionID || instance.Version != "v1" || instance.BusinessKey != businessKey(runID, admission.sequence) { + return 0, "WORKFLOW_MISMATCH" + } + switch instance.State { + case "SUCCEEDED": + if instance.CompletedAt == nil || instance.CreatedAt.IsZero() || instance.CompletedAt.Before(instance.CreatedAt) { + return 0, "COMPLETION_MISMATCH" + } + call, stop = context.WithTimeout(ctx, cfg.Timeout) + steps, e := api.GetWorkflowSteps(call, instance.ID) + stop() + if e != nil { + return 0, "STEPS_" + classify(e) + } + if code := verifySteps(ctx, cfg, api, instance, runID, admission.sequence, steps); code != "" { + return 0, code + } + if !matchesOutput(instance.Output, runID, admission.sequence) { + return 0, "OUTPUT_MISMATCH" + } + return instance.CompletedAt.Sub(instance.CreatedAt), "" + case "NEEDS_REVIEW", "FAILED", "CANCELLED": + return 0, "WORKFLOW_" + instance.State + case "RUNNING", "WAITING", "PAUSED", "PENDING": + default: + return 0, "UNKNOWN_WORKFLOW_STATE" + } + timer := time.NewTimer(cfg.PollInterval) + select { + case <-ctx.Done(): + timer.Stop() + return 0, "VERIFY_" + classify(ctx.Err()) + case <-timer.C: + } + } +} + +func verifySteps(ctx context.Context, cfg config, api workflowAPI, instance orbit.WorkflowInstance, runID string, sequence int, steps orbit.WorkflowSteps) string { + var taskSteps []orbit.WorkflowStepStatus + for _, step := range steps.Steps { + if step.Kind == "TASK" { + taskSteps = append(taskSteps, step) + } + } + if len(taskSteps) != 2 { + return "STEP_COUNT_MISMATCH" + } + for _, step := range taskSteps { + if step.State != "SUCCEEDED" || step.TaskID == nil || !validID(*step.TaskID) || step.ActivationSequence != 1 || step.ActivationID == nil { + return "STEP_MISMATCH" + } + call, stop := context.WithTimeout(ctx, cfg.Timeout) + task, err := api.GetTask(call, *step.TaskID) + stop() + if err != nil { + return "TASK_" + classify(err) + } + payload, valid := decodePayload(task.Result) + if task.ID != *step.TaskID || task.State != orbit.TaskSucceeded || task.EffectUncertain || task.AttemptCount != 1 || !valid || payload != (workflowPayload{RunID: runID, Sequence: sequence}) { + return "TASK_MISMATCH" + } + call, stop = context.WithTimeout(ctx, cfg.Timeout) + history, err := api.ListTaskAttempts(call, task.ID, orbit.HistoryPageRequest{Limit: 2}) + stop() + if err != nil { + return "HISTORY_" + classify(err) + } + if len(history.Items) != 1 || history.NextCursor != "" || history.Items[0].TaskID != task.ID || history.Items[0].State != orbit.AttemptSucceeded { + return "ATTEMPT_MISMATCH" + } + call, stop = context.WithTimeout(ctx, cfg.Timeout) + source, err := api.GetTaskSource(call, task.ID) + stop() + if err != nil || source.Kind != "WORKFLOW" || source.Workflow == nil || source.Workflow.WorkflowID != instance.ID || source.Workflow.DefinitionID != instance.DefinitionID || source.Workflow.DefinitionVersionID != instance.DefinitionVersionID || source.Workflow.Version != instance.Version || source.Workflow.StepID != step.ID || source.Workflow.ActivationID != *step.ActivationID || source.Workflow.ActivationSequence != 1 { + return "SOURCE_MISMATCH" + } + } + return "" +} + +func matchesOutput(raw json.RawMessage, runID string, sequence int) bool { + var wire struct { + Left workflowPayload `json:"left"` + Right workflowPayload `json:"right"` + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + if decoder.Decode(&wire) != nil { + return false + } + var trailing any + if decoder.Decode(&trailing) != io.EOF { + return false + } + return wire.Left == (workflowPayload{RunID: runID, Sequence: sequence}) && wire.Right == wire.Left +} diff --git a/sdk/go/cmd/orbit-workflow-load/measure_test.go b/sdk/go/cmd/orbit-workflow-load/measure_test.go new file mode 100644 index 0000000..c084b2f --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/measure_test.go @@ -0,0 +1,61 @@ +package main + +import ( + "encoding/json" + "reflect" + "testing" + "time" +) + +func TestDecodePayloadRejectsMissingNullUnknownAndTrailing(t *testing.T) { + want := workflowPayload{RunID: "run", Sequence: 3} + for _, raw := range []string{ + `{"run_id":"run","sequence":3}`, + `{"run_id":"run","sequence":3} `, + } { + got, ok := decodePayload(json.RawMessage(raw)) + if !ok || !reflect.DeepEqual(got, want) { + t.Fatalf("decode %s: %#v %t", raw, got, ok) + } + } + for _, raw := range []string{ + `{"run_id":"run"}`, + `{"run_id":null,"sequence":3}`, + `{"run_id":"run","sequence":null}`, + `{"run_id":"run","sequence":3,"extra":true}`, + `{"run_id":"run","sequence":3}{"again":true}`, + } { + if _, ok := decodePayload(json.RawMessage(raw)); ok { + t.Fatalf("accepted invalid payload %s", raw) + } + } +} + +func TestWorkflowSpecHasFiniteParallelBranchesAndFixedRetry(t *testing.T) { + spec := workflowSpec("0123456789abcdef0123456789abcdef", "load.workflow.run", 5*time.Second) + if len(spec.Body.Steps) != 1 || spec.Body.Steps[0].Kind != "PARALLEL" || spec.Body.Steps[0].Parallel == nil || len(spec.Body.Steps[0].Parallel.Branches) != 2 { + t.Fatal("parallel shape changed") + } + for _, branch := range spec.Body.Steps[0].Parallel.Branches { + if len(branch.Body.Steps) != 1 || branch.Body.Steps[0].Task == nil || branch.Body.Steps[0].Task.Retry.MaxAttempts != 1 || branch.Body.Steps[0].Task.RecoverySafety != "RECONCILE_ONLY" { + t.Fatalf("unsafe branch %#v", branch) + } + } +} + +func TestOutputVerificationRequiresBothBranches(t *testing.T) { + runID := "run" + good := `{"left":{"run_id":"run","sequence":3},"right":{"run_id":"run","sequence":3}}` + if !matchesOutput(json.RawMessage(good), runID, 3) { + t.Fatal("valid output rejected") + } + for _, raw := range []string{ + `{"left":{"run_id":"run","sequence":3}}`, + `{"left":{"run_id":"run","sequence":3},"right":{"run_id":"other","sequence":3}}`, + `{"left":{"run_id":"run","sequence":3},"right":{"run_id":"run","sequence":3},"extra":true}`, + } { + if matchesOutput(json.RawMessage(raw), runID, 3) { + t.Fatalf("accepted invalid output %s", raw) + } + } +} diff --git a/sdk/go/cmd/orbit-workflow-load/report.go b/sdk/go/cmd/orbit-workflow-load/report.go new file mode 100644 index 0000000..b964378 --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/report.go @@ -0,0 +1,85 @@ +package main + +import ( + "math" + "runtime" + "sort" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +// distribution 只统计完成核验的实例；空集合的百分位保持null。 +type distribution struct { + Count int `json:"count"` // Count 纳入本项统计的样本数。 + P50 *float64 `json:"p50_ms"` // P50 最近秩中位数，毫秒。 + P95 *float64 `json:"p95_ms"` // P95 最近秩95百分位，毫秒。 + P99 *float64 `json:"p99_ms"` // P99 最近秩99百分位，毫秒。 + Max *float64 `json:"max_ms"` // Max 纳入样本的最大毫秒数。 +} + +// workflowReport 是一次有界流程测量结果，不是容量承诺或生产高可用验收。 +type workflowReport struct { + RunID string `json:"run_id"` // RunID 本次定义、实例和处理器身份的隔离前缀。 + GoVersion string `json:"go_version"` // GoVersion 实际编译工具链版本。 + GOOS string `json:"goos"` // GOOS 负载进程操作系统。 + GOARCH string `json:"goarch"` // GOARCH 负载进程架构。 + StartedAt time.Time `json:"started_at"` // StartedAt 从预检开始的本机时间。 + AdmissionFinishedAt time.Time `json:"admission_finished_at"` // AdmissionFinishedAt 实例受理组完成时间。 + FinishedAt time.Time `json:"finished_at"` // FinishedAt 核验及Worker收尾完成时间。 + Requests int `json:"requests"` // Requests 计划启动的流程实例数。 + Concurrency int `json:"concurrency"` // Concurrency 启动和核验请求的最大并发。 + Workers int `json:"workers"` // Workers 本进程独立WSS Worker数。 + Capacity int `json:"capacity_per_worker"` // Capacity 每个Worker的执行槽上限。 + RunTimeoutMS int64 `json:"run_timeout_ms"` // RunTimeoutMS 整次测量的截止时间。 + PollIntervalMS int64 `json:"poll_interval_ms"` // PollIntervalMS 流程状态轮询间隔。 + HTTPTimeoutMS int64 `json:"http_timeout_ms"` // HTTPTimeoutMS 单次HTTP交换截止时间。 + Queue *orbit.Queue `json:"queue,omitempty"` // Queue 预检时的队列快照，不修改配置。 + DefinitionID string `json:"definition_id,omitempty"` // DefinitionID 本次新建的稳定定义身份。 + VersionID string `json:"version_id,omitempty"` // VersionID 本次发布的不可变版本身份。 + Attempted int `json:"attempted"` // Attempted 实际发起的启动调用数。 + NotSubmitted int `json:"not_submitted"` // NotSubmitted 截止前未发起的实例数。 + Accepted int `json:"accepted"` // Accepted 收到并核验唯一启动回执的实例数。 + AdmissionErrors int `json:"admission_errors"` // AdmissionErrors 明确拒绝或回执不合法的实例数。 + VerifiedSuccess int `json:"verified_success"` // VerifiedSuccess 流程、步骤、任务、来源和结果均核验通过数。 + Unverified int `json:"unverified"` // Unverified 已受理但没有完成全部核验的实例数。 + HandlerCalls int `json:"handler_calls"` // HandlerCalls 有效处理器调用总数，两个分支各一次。 + DuplicateInvocations int `json:"duplicate_invocations"` // DuplicateInvocations 超过两个分支预期的额外调用数。 + WorkerStarted uint64 `json:"worker_started"` // WorkerStarted SDK本进程领取并启动的任务数。 + Reconnects uint64 `json:"worker_reconnects"` // Reconnects SDK重连次数。 + Renewals uint64 `json:"worker_renewals_confirmed"` // Renewals SDK确认的租约续期次数。 + Admission distribution `json:"admission"` // Admission 启动请求本机往返耗时。 + Completion distribution `json:"completion"` // Completion 数据库实例创建至完成的服务端耗时。 + Errors map[string]int `json:"errors"` // Errors 稳定错误分类，不写入凭据、SQL或请求正文。 + Complete bool `json:"complete"` // Complete 仅全部实例和两倍处理器调用逐项通过核验时为真。 +} + +func newReport(cfg config, runID string) workflowReport { + return workflowReport{RunID: runID, GoVersion: runtime.Version(), GOOS: runtime.GOOS, GOARCH: runtime.GOARCH, StartedAt: time.Now().UTC(), Requests: cfg.Requests, NotSubmitted: cfg.Requests, Concurrency: cfg.Concurrency, Workers: cfg.Workers, Capacity: cfg.Capacity, RunTimeoutMS: cfg.RunTimeout.Milliseconds(), PollIntervalMS: cfg.PollInterval.Milliseconds(), HTTPTimeoutMS: cfg.Timeout.Milliseconds(), Errors: map[string]int{}} +} + +func (r workflowReport) finish() workflowReport { + for code, count := range r.Errors { + if count == 0 { + delete(r.Errors, code) + } + } + r.FinishedAt = time.Now().UTC() + r.Unverified = r.Accepted - r.VerifiedSuccess + r.Complete = r.Attempted == r.Requests && r.NotSubmitted == 0 && r.AdmissionErrors == 0 && r.Accepted == r.Requests && r.VerifiedSuccess == r.Requests && r.HandlerCalls == 2*r.Requests && r.WorkerStarted == uint64(2*r.Requests) && r.DuplicateInvocations == 0 && len(r.Errors) == 0 + return r +} + +func distributionOf(values []time.Duration) distribution { + d := distribution{Count: len(values)} + if len(values) == 0 { + return d + } + sort.Slice(values, func(i, j int) bool { return values[i] < values[j] }) + percentile := func(q float64) *float64 { + v := float64(values[int(math.Ceil(float64(len(values))*q))-1]) / float64(time.Millisecond) + return &v + } + d.P50, d.P95, d.P99, d.Max = percentile(.5), percentile(.95), percentile(.99), percentile(1) + return d +}