From 8c9c6062efde0c43bda3b730e5390a239bdf6b91 Mon Sep 17 00:00:00 2001 From: k2safe Date: Sun, 20 Sep 2026 14:10:35 +0800 Subject: [PATCH] fix: bind workflow samples to distinct task execution proofs --- AGENTS.md | 2 + docs/08-development-plan.md | 2 +- docs/api/workflow-load.md | 14 +- .../phase-6-workflow-load/deadline-24.json | 206 ++++++ .../phase-6-workflow-load/environment.json | 96 ++- .../phase-6-workflow-load/sample-100.json | 659 ++++++++++++++++++ .../data/phase-6-workflow-load/sample-24.json | 204 +++++- .../data/phase-6-workflow-load/schema.json | 7 + .../data/phase-6-workflow-load/sdk-check.txt | 30 + docs/reports/phase-6-workflow-load.md | 64 +- docs/reports/phase-7-regression-status.md | 2 +- sdk/go/cmd/orbit-workflow-load/handler.go | 87 +++ sdk/go/cmd/orbit-workflow-load/measure.go | 195 ++---- sdk/go/cmd/orbit-workflow-load/payload.go | 64 ++ sdk/go/cmd/orbit-workflow-load/report.go | 79 ++- sdk/go/cmd/orbit-workflow-load/verify.go | 90 +++ sdk/go/cmd/orbit-workflow-load/verify_test.go | 210 ++++++ 17 files changed, 1783 insertions(+), 228 deletions(-) create mode 100644 docs/reports/data/phase-6-workflow-load/deadline-24.json create mode 100644 docs/reports/data/phase-6-workflow-load/sample-100.json create mode 100644 docs/reports/data/phase-6-workflow-load/schema.json create mode 100644 docs/reports/data/phase-6-workflow-load/sdk-check.txt create mode 100644 sdk/go/cmd/orbit-workflow-load/handler.go create mode 100644 sdk/go/cmd/orbit-workflow-load/payload.go create mode 100644 sdk/go/cmd/orbit-workflow-load/verify.go create mode 100644 sdk/go/cmd/orbit-workflow-load/verify_test.go diff --git a/AGENTS.md b/AGENTS.md index c91bfd6..63bebc9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -90,3 +90,5 @@ - `sdk/go/cmd/orbit-workflow-load` 独立测量真实Weave流程：每次运行创建固定版本的双分支并行定义，启动实例不自动重试，必须核对流程/步骤/任务/尝试/来源/输出及两倍分支调用；队列只读，处理器无外部副作用，截止后保留实际分母并以非零状态退出。隔离样本不代表生产容量、高可用或跨故障域验收。 - `sdk/go/cmd/orbit-schedule-load` 使用同时起止的有限FIXED_RATE计划测量集中触发；固定版本/时区快照、触发时间槽、任务/尝试/来源、本地调用逐项核对。配置总数、已观察、已触发和已核验分开计数；截止保留原操作身份并有界暂停，不取消已生成任务；暂停只在明确版本冲突后换键，传输不明不得新键重试。样本不代表长期积压容量或生产HA。 + +- 流程测量核验必须绑定本次发布的不可变版本ID及完整规范，步骤精确为pair/left/right且任务/激活唯一；按任务和尝试关联本地调用，不能用每实例总次数抵充缺失分支。零序号也要求精确必填字段，拒绝重复键/别名/null；报告保留逐请求原键与ID和未确认调用分母。源码更新后重新构建并运行真实样本，原二进制结果不得配上更新后的源码摘要。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 8cecfa1..072888b 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)。新增[`orbit-workflow-load`](api/workflow-load.md)，用固定版本的双分支并行流程核对步骤、来源、尝试和输出，见[流程测量报告](reports/phase-6-workflow-load.md)。新增[`orbit-schedule-load`](api/schedule-load.md)，已完成24/128次定时集中触发与24次计划量的截止负例，保存实际分母、唯一任务/尝试、来源与暂停收尾证据，见[定时测量报告](reports/phase-6-schedule-load.md)。标准Go 1.27.1全量仍待复跑，下一步补长期积压和资源曲线；目标数据库拓扑容量仍单独验收。 +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)，用本次发布版本的双分支并行流程，逐个核对不同任务、唯一尝试、来源、精确输出与本地调用；24/100实例及24实例截止负例已重新构建回归并绑定源码/二进制哈希，见[流程测量报告](reports/phase-6-workflow-load.md)。新增[`orbit-schedule-load`](api/schedule-load.md)，已完成24/128次定时集中触发与24次计划量的截止负例，保存实际分母、唯一任务/尝试、来源与暂停收尾证据，见[定时测量报告](reports/phase-6-schedule-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/workflow-load.md b/docs/api/workflow-load.md index 93cb345..ce956f0 100644 --- a/docs/api/workflow-load.md +++ b/docs/api/workflow-load.md @@ -20,16 +20,24 @@ go run ./cmd/orbit-workflow-load \ -poll-interval 250ms ``` -参数上限是 `requests<=2000`、启动/核验并发 `<=128`、Worker 数 `<=32`、单 Worker 容量 `<=256` 且总容量 `<=1024`。`-timeout` 是单次 HTTP 交换截止，`-run-timeout` 是整次样本截止；两者都不是生产系统的 SLA。 +参数上限是 `requests<=2000`、启动/核验并发 `<=128`、Worker 数 `<=32`、单 Worker 容量 `<=256` 且总容量 `<=1024`。`-timeout` 是单次 HTTP 交换截止，`-run-timeout` 是整次样本截止；整次测量另有最多5秒 Worker 收尾。发布的流程、任务及尝试时限使用 `run-timeout`，不再隐式缩短为30秒；任务实际截止还受实例首次绝对截止限制。这些限制都不是生产系统的 SLA。 -工具先只读核对队列，不修改队列额度；定义名、实例幂等键、业务键和处理器名都带一次运行生成的随机前缀。每个实例固定使用输入 `{"run_id":"...","sequence":N}`，流程输出必须同时包含 `left` 和 `right` 两个相同对象。 +工具先只读核对队列，不修改队列额度；创建及发布回执必须匹配命名空间、定义名称和完整规范，实例必须绑定本次发布回执的不可变版本 ID。定义名、实例幂等键、业务键和处理器名都带一次运行生成的随机前缀。每个实例固定使用输入 `{"run_id":"...","sequence":N}`，流程输出必须同时包含 `left` 和 `right` 两个相同对象。 ## 报告解释 -输出为一行 JSON，包含 `attempted`、`accepted`、`verified_success`、`unverified`、`handler_calls`、`worker_started`、`errors` 及 `admission`/`completion` 百分位。`complete=true` 的必要条件是：全部启动调用有唯一回执、全部实例为 `SUCCEEDED`、每个流程恰有两个成功 TASK 步骤、每个任务恰有一次成功尝试、来源固定到同一实例和版本、两倍分支处理器调用全部被确认且没有重复调用。 +输出为一行 JSON，包含 `attempted`、`accepted`、`verified_success`、`unverified`、`handler_calls`、`unconfirmed_invocations`、`worker_started`、`instances`、`errors` 及 `admission`/`completion` 百分位。`complete=true` 的必要条件是：全部启动调用有唯一回执、全部实例为 `SUCCEEDED`、步骤必须精确为 `pair:PARALLEL`、`left:TASK`、`right:TASK`，全部成功且各自步骤/激活身份唯一；两个分支对应两个不同任务，各有一次成功尝试，来源精确匹配实例、发布版本、步骤键和激活。每个任务必须关联到本地同一尝试的一次调用，不能用同一分支调用两次抵充另一个分支。 + +输入/输出按精确字段集合校验，拒绝重复键、大小写别名、缺字段、null和尾随JSON；实例序号0也不能通过缺失 `sequence` 字段来冒充。任务命名空间、队列、处理器、策略、输入/结果和各阶段时间顺序全部参与核验。 任何预检、发布、启动、读取、步骤、尝试、来源或截止错误都会保留稳定错误分类并以非零状态退出。空分布的百分位是 JSON `null`，不能解释成零延迟。`completion` 是服务端 `created_at` 到 `completed_at` 的持久化耗时；它不是客户端轮询耗时。 +## 原身份与失败核查 + +`instances` 保留每个实际发起的启动请求序号、幂等键、返回实例 ID、受理错误及持久核验标记；回执不明时 ID 可以为空，不代表未提交。定义创建键/名称为 `orbit-workflow-load-{run_id}`，发布键为 `run_id`，版本名为 `v1`。启动键和业务键都是 `orbit-workflow-{run_id}-{sequence六位十进制补零}`，输入为该 `run_id` 和数值 `sequence`。按原目标、原键及原请求核对，不能重跑整个命令来恢复旧请求。 + +`verified_success` 只计完成全部持久核验的实例。`duplicate_invocations` 按同一任务的额外调用计数；`unconfirmed_invocations` 是不能对应到已核验分支任务及尝试的调用数，部分实例尚未完成时可以非零。它们不能混算为重复执行，也不能据此宣称外部效果被撤回。本工具无外部副作用，超时只停止本地 Worker 并有界收尾，不自动取消、删除或重置服务器上的实例；保留原实例和任务以供查询。 + ## 验证边界 该工具适合在一次性 MySQL、测试凭据和明确的资源上限下做正确性、故障分母和阶段延迟验证。单机样本不能证明多节点吞吐、跨故障域高可用、备份恢复、RPO/RTO 或面向生产的容量；生产验收还需要目标拓扑、数据规模、连接池、备份介质、监控采集和发布兼容矩阵的独立证据。 diff --git a/docs/reports/data/phase-6-workflow-load/deadline-24.json b/docs/reports/data/phase-6-workflow-load/deadline-24.json new file mode 100644 index 0000000..b6b7cac --- /dev/null +++ b/docs/reports/data/phase-6-workflow-load/deadline-24.json @@ -0,0 +1,206 @@ +{ + "instances": [ + { + "sequence": 0, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000000", + "id": "60456999954cbecabe6c6e2b829c4109", + "verified": true + }, + { + "sequence": 1, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000001", + "id": "4d7ed85688d258af04264712d6e09af1", + "verified": true + }, + { + "sequence": 2, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000002", + "id": "a07c11ca04f99b64f5ed191f0a2a3b9a", + "verified": false + }, + { + "sequence": 4, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000004", + "id": "00b467fedfe4f48f984df640b2f1ba92", + "verified": false + }, + { + "sequence": 5, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000005", + "id": "512c824d4a9b2b31d132000f4b8c5326", + "verified": false + }, + { + "sequence": 3, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000003", + "id": "ca2a1bb2240396310909f7c08a33af2b", + "verified": false + }, + { + "sequence": 6, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000006", + "id": "4716c4bc65f0be505404bc5e039c2f5f", + "verified": false + }, + { + "sequence": 7, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000007", + "id": "7101e648f6371e02853b2d8e53ae6a65", + "verified": false + }, + { + "sequence": 8, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000008", + "id": "15bcbc090518a5b9ac21dbc25d999040", + "verified": false + }, + { + "sequence": 9, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000009", + "id": "2095245e3af6bbcc56bce343f6e6a677", + "verified": false + }, + { + "sequence": 10, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000010", + "id": "c31127024409f4eb0e7b74bc53783d49", + "verified": false + }, + { + "sequence": 11, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000011", + "id": "c0007eff019e9c4bcbe93fa933d853e5", + "verified": false + }, + { + "sequence": 12, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000012", + "id": "7392523c302f89815842035d1dfa24d0", + "verified": false + }, + { + "sequence": 13, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000013", + "id": "d0af6cd085eb5eb9ff34b62b6d510cd3", + "verified": false + }, + { + "sequence": 14, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000014", + "id": "e5d8a137cec5e7ffcd36968b3c29f5f6", + "verified": false + }, + { + "sequence": 15, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000015", + "id": "8fcf3199c93c75790ceac6d7aa1fd56a", + "verified": false + }, + { + "sequence": 16, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000016", + "id": "1622c4503f4ac3a2735ca84b58b83f4a", + "verified": false + }, + { + "sequence": 17, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000017", + "id": "0aa3b630678edd39d5e1185830aaf1c3", + "verified": false + }, + { + "sequence": 18, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000018", + "id": "780a8c7764e1a32385e0ae531a4eca93", + "verified": false + }, + { + "sequence": 19, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000019", + "id": "3c999281787197c996c7a5cff214535a", + "verified": false + }, + { + "sequence": 20, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000020", + "id": "7961d625beff2df200cc442b302be01b", + "verified": false + }, + { + "sequence": 21, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000021", + "id": "d74cdcc7e2cb12dff9b95de5617f8fca", + "verified": false + }, + { + "sequence": 22, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000022", + "id": "ace8372971f3dfc5a05711b024bde1fc", + "verified": false + }, + { + "sequence": 23, + "key": "orbit-workflow-755e7b5f1948bbb1eede5f4dd1798c59-000023", + "id": "3e41c46ca759b872679f0414e9c528fe", + "verified": false + } + ], + "unconfirmed_invocations": 0, + "run_id": "755e7b5f1948bbb1eede5f4dd1798c59", + "go_version": "go1.25.14", + "goos": "linux", + "goarch": "arm64", + "started_at": "2026-09-20T06:06:51.806845424Z", + "admission_finished_at": "2026-09-20T06:06:52.378917966Z", + "finished_at": "2026-09-20T06:06:53.812021259Z", + "requests": 24, + "concurrency": 6, + "workers": 2, + "capacity_per_worker": 2, + "run_timeout_ms": 2000, + "poll_interval_ms": 100, + "http_timeout_ms": 5000, + "queue": { + "max_unfinished_tasks": 100000, + "unfinished_tasks": "0", + "id": "e1126a23e1d796dd823a0e5ea35664b2", + "namespace_id": "eb6180c00667e59b73f49f91ff9a209d", + "name": "workflow", + "paused": false, + "max_concurrency": 4, + "rate_per_second": 10, + "revision": "1" + }, + "definition_id": "79a61cb9ac4fe19f778c91747a7164fc", + "version_id": "9a30e07678660165208633a35b754262", + "attempted": 24, + "not_submitted": 0, + "accepted": 24, + "admission_errors": 0, + "verified_success": 2, + "unverified": 22, + "handler_calls": 4, + "duplicate_invocations": 0, + "worker_started": 4, + "worker_reconnects": 0, + "worker_renewals_confirmed": 0, + "admission": { + "count": 24, + "p50_ms": 127.582959, + "p95_ms": 200.476584, + "p99_ms": 205.800541, + "max_ms": 205.800541 + }, + "completion": { + "count": 2, + "p50_ms": 1829.687, + "p95_ms": 1830.745, + "p99_ms": 1830.745, + "max_ms": 1830.745 + }, + "errors": { + "RUN_TIMEOUT": 1, + "VERIFY_TIMEOUT": 22 + }, + "complete": false +} diff --git a/docs/reports/data/phase-6-workflow-load/environment.json b/docs/reports/data/phase-6-workflow-load/environment.json index f60392d..817b187 100644 --- a/docs/reports/data/phase-6-workflow-load/environment.json +++ b/docs/reports/data/phase-6-workflow-load/environment.json @@ -1,29 +1,82 @@ { "date_local": "2026-09-20", "timezone": "Asia/Shanghai", - "test_go": "go1.25.14", + "source_base_commit": "2973459", "declared_go": "1.27.1", + "test_go": "go1.25.14", "temporary_modfile": true, "mysql": "8.4.10", + "architecture": "arm64", "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, + "worker_connections": 2, "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", + "network": "Go runner shares disposable MySQL container network namespace; loopback HTTP/WS/MySQL; no host port", + "dedicated_cpu_memory_limits": false, "fresh_disposable_database": true, - "network": "Go runner shares MySQL container network namespace; loopback HTTP/WSS and MySQL", + "cleanup": "Test-created containers stopped and removed; no existing databases used", + "sdk_checks": [ + "go test -race ./...", + "go vet ./...", + "go build ./..." + ], + "modfile_note": "All Go commands use GOWORK=off and temporary sdk.mod/root.mod; repository go.mod unchanged", + "source_to_binary": "All samples used binary rebuilt from the listed final source hashes; no source edits after build", + "supersedes": "Earlier 24-workflow sample lacked a reliable final-source-to-binary evidence chain; this rebuilt rerun replaces that acceptance evidence", + "sample_commands": [ + [ + "-requests", + "24", + "-concurrency", + "6", + "-workers", + "2", + "-capacity", + "4", + "-run-timeout", + "90s", + "-poll-interval", + "100ms" + ], + [ + "-requests", + "100", + "-concurrency", + "8", + "-workers", + "2", + "-capacity", + "4", + "-run-timeout", + "90s", + "-poll-interval", + "100ms" + ], + [ + "-requests", + "24", + "-concurrency", + "6", + "-workers", + "2", + "-capacity", + "2", + "-run-timeout", + "2s", + "-poll-interval", + "100ms" + ] + ], + "exit_codes": [ + 0, + 0, + 1 + ], "schema_verification": { "schema_version": 28, "migration_count": 28, @@ -32,9 +85,26 @@ "complete": true }, "source_sha256": { + "sdk/go/cmd/orbit-workflow-load/handler.go": "d6fac3129fde1e7ca8191241300afe0442ea0e4cbbacc134e6fee59dd2fabd45", "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.go": "15c0cab5ef45391224a03bc4d9b2bf581329c338a43ac4e18f9d4b4ac42b4eb9", "sdk/go/cmd/orbit-workflow-load/measure_test.go": "42e710139f515aa339b5684107d00fcb413688e7512fc4c36ebfe56776ad41a1", - "sdk/go/cmd/orbit-workflow-load/report.go": "701a27a3f4f56cf9566fbcf97fba8024ced36b32754f4a5dcdae4646a2a4b24a" + "sdk/go/cmd/orbit-workflow-load/payload.go": "ca92ea5222b31386858d3b5866633afe5872585284b08a9281a55818a26144b0", + "sdk/go/cmd/orbit-workflow-load/report.go": "89c7447bc35814211fa5d09828ff4e1b81b64ea7ad4638e7586e04cd996b81f5", + "sdk/go/cmd/orbit-workflow-load/verify.go": "57e1d467e8473d806d2593b595fc7ae49930a96e390f806d13365390018982a4", + "sdk/go/cmd/orbit-workflow-load/verify_test.go": "e90e3b7fcade1f12fc3a57a0b82e07659086ffcb2918837ba7ec01902446bde3" + }, + "binary_sha256": { + "orbit": "c98f11c3303de68a5c75c2930d6a866d89e7642131a6d28e08bc1999fcfda56a", + "orbit-admin": "b4567e438d37ad0ed7877d45af77e3d6e310edb478d55fc30c446c0cce346733", + "orbit-migrate": "d19877d8d405d5d5214e6fc4e0733425780a1803cc23c57b8023555d6e87a8f2", + "orbit-workflow-load": "a54b975532c1ff036ef99995bde7b444ad06d7d8a878525f0e586a1215deb421" + }, + "artifacts_sha256": { + "deadline-24.json": "4e07ac787b6aecd9c4deab7d079885de0fcc57ed4802fa57ef5c2371a06e36b3", + "sample-100.json": "26951841ed166d0289662d414c022f7f7623088c3315bfa9d4d0b25bb83a5d24", + "sample-24.json": "9955ae4bda74b09e89124bb6390a3b04ef896b2abda99e3b33ecce621f80bccd", + "schema.json": "9ee118dd4306cddde44582b8d3983f2cd25331148d993672fd70cca86dc06d0a", + "sdk-check.txt": "fd92736744e4e689768a3f63a1771153c9346a0108a610e59439d7ec11b140cc" } } diff --git a/docs/reports/data/phase-6-workflow-load/sample-100.json b/docs/reports/data/phase-6-workflow-load/sample-100.json new file mode 100644 index 0000000..f56d398 --- /dev/null +++ b/docs/reports/data/phase-6-workflow-load/sample-100.json @@ -0,0 +1,659 @@ +{ + "instances": [ + { + "sequence": 1, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000001", + "id": "1eb2c7a6c287df1b60a147b70b5416fb", + "verified": true + }, + { + "sequence": 4, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000004", + "id": "6e6858af3720ca8ed8fb1752c378873e", + "verified": true + }, + { + "sequence": 0, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000000", + "id": "76e45cb8f6ae3525fe9083bd4dbaf623", + "verified": true + }, + { + "sequence": 2, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000002", + "id": "f1d8c9d3486bc4a869121672d18fac65", + "verified": true + }, + { + "sequence": 3, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000003", + "id": "bc53ebfd1cb0e2a46e6991c5db840eea", + "verified": true + }, + { + "sequence": 7, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000007", + "id": "d1364cadabf5d9c86484042cbd5f577f", + "verified": true + }, + { + "sequence": 6, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000006", + "id": "48b12d5b9db856f2ebd703b036e52f91", + "verified": true + }, + { + "sequence": 5, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000005", + "id": "49921d1c998bb8dd5d2f3353b0a82641", + "verified": true + }, + { + "sequence": 8, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000008", + "id": "785903d75d2c53dff46028a6d5b09fec", + "verified": true + }, + { + "sequence": 9, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000009", + "id": "5b171a3ce6c1132d9e7864add8b7d2b3", + "verified": true + }, + { + "sequence": 10, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000010", + "id": "549be5bcf533649d0c9fdfff91830e63", + "verified": true + }, + { + "sequence": 11, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000011", + "id": "349aaef2481fcc56febbc6b20ff7ba65", + "verified": true + }, + { + "sequence": 12, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000012", + "id": "b53281b299ca5f9803dbeff5ffaa34f2", + "verified": true + }, + { + "sequence": 13, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000013", + "id": "9da2baaa870f3f250a54fd6449e45361", + "verified": true + }, + { + "sequence": 14, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000014", + "id": "1431ada272026c80bb99a65c5b09553f", + "verified": true + }, + { + "sequence": 15, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000015", + "id": "4bbaef63c0658b0eafe1a8d940a7adf5", + "verified": true + }, + { + "sequence": 16, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000016", + "id": "370d5992369746817c61dadfb0c8e10d", + "verified": true + }, + { + "sequence": 17, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000017", + "id": "bf92cb0af7dc1b9a488ed9c1ee58316a", + "verified": true + }, + { + "sequence": 18, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000018", + "id": "32b515a82746c824a97a3db31bc86178", + "verified": true + }, + { + "sequence": 19, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000019", + "id": "5476b0f24173b1cfb10960be686f017e", + "verified": true + }, + { + "sequence": 20, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000020", + "id": "5c3a7ac0ca66ac132767f16bf16929f5", + "verified": true + }, + { + "sequence": 21, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000021", + "id": "3d8b953aa47c4f7067910cde4c6d3d34", + "verified": true + }, + { + "sequence": 22, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000022", + "id": "d94a07e7564e3c274127405888b80956", + "verified": true + }, + { + "sequence": 23, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000023", + "id": "b5543de83172c74bbea0957125f5fdb1", + "verified": true + }, + { + "sequence": 24, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000024", + "id": "7aac6ae3a45a4a8a9a733f19c5bd202c", + "verified": true + }, + { + "sequence": 25, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000025", + "id": "c9e86d3898c56b31870a8971ce2d694d", + "verified": true + }, + { + "sequence": 26, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000026", + "id": "56062b048c836351f49e6e01c59c6bb8", + "verified": true + }, + { + "sequence": 27, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000027", + "id": "d1f4296b8c65a2c198336a66ae29716d", + "verified": true + }, + { + "sequence": 28, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000028", + "id": "a545e4c0b9ffb5e5ec4d788778eca602", + "verified": true + }, + { + "sequence": 29, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000029", + "id": "3f2c1c3df9f55e4bb39a1e206e6b5f61", + "verified": true + }, + { + "sequence": 30, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000030", + "id": "6f52c8c925eb061cb72521a04e6d08ba", + "verified": true + }, + { + "sequence": 31, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000031", + "id": "61a9eb0de17906f753cc1f7c49b8ecbb", + "verified": true + }, + { + "sequence": 32, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000032", + "id": "8ff107274a65d19ff01f9654d1c0de8e", + "verified": true + }, + { + "sequence": 33, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000033", + "id": "f3ab7d9769aa7104eca18ad164f64a41", + "verified": true + }, + { + "sequence": 34, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000034", + "id": "98c163ac29191f968545140ad83ad01b", + "verified": true + }, + { + "sequence": 35, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000035", + "id": "70177548fbe3458371e13c6d00912d91", + "verified": true + }, + { + "sequence": 36, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000036", + "id": "d16f46788814ccc5f288b44b27266b08", + "verified": true + }, + { + "sequence": 37, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000037", + "id": "c8f26c0c0619079f9c136687a25b330d", + "verified": true + }, + { + "sequence": 38, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000038", + "id": "0dfe0b8e3dab98c0f5631cf006d72504", + "verified": true + }, + { + "sequence": 39, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000039", + "id": "8e635499b64a07fdfa871b843f220457", + "verified": true + }, + { + "sequence": 40, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000040", + "id": "eda6270631fec7d192a2b091229df374", + "verified": true + }, + { + "sequence": 41, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000041", + "id": "8799eb39abbf965da0433f0ec5555f25", + "verified": true + }, + { + "sequence": 42, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000042", + "id": "2c2e3147489bd0fb4436c8319ba9fbe0", + "verified": true + }, + { + "sequence": 43, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000043", + "id": "4735ac0590283a891b283c6f35b712c8", + "verified": true + }, + { + "sequence": 44, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000044", + "id": "667eb39653dc004b2beda8e0fa818346", + "verified": true + }, + { + "sequence": 45, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000045", + "id": "b6a011ef72c9af38818f7b7cbcaac4e7", + "verified": true + }, + { + "sequence": 46, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000046", + "id": "97bb897778c8a96131b95d194ab9cf5b", + "verified": true + }, + { + "sequence": 47, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000047", + "id": "56ec5ab109b4bdfa61ac583d39b90501", + "verified": true + }, + { + "sequence": 48, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000048", + "id": "dbac6411d44bd948b61648e79d53fccd", + "verified": true + }, + { + "sequence": 49, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000049", + "id": "83c1e457f1811b67ca848b97370dd348", + "verified": true + }, + { + "sequence": 50, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000050", + "id": "33ebffdec1ffd9aa20376d2d74b11783", + "verified": true + }, + { + "sequence": 51, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000051", + "id": "aec15fd65b80e897f89b925b0f905b9f", + "verified": true + }, + { + "sequence": 52, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000052", + "id": "b730c1ca05daaad24090e90370abf10f", + "verified": true + }, + { + "sequence": 53, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000053", + "id": "92da8a4dd3e171f72cd79c956d13304a", + "verified": true + }, + { + "sequence": 54, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000054", + "id": "a82f75b299d9ad0a912cc955ea552b38", + "verified": true + }, + { + "sequence": 55, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000055", + "id": "4e5088f8d220d8e1b1d428886fa1120f", + "verified": true + }, + { + "sequence": 56, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000056", + "id": "f941bb7f0107a3470ca01b083341d780", + "verified": true + }, + { + "sequence": 57, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000057", + "id": "2119b0a7e83873665988fd0887a02745", + "verified": true + }, + { + "sequence": 58, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000058", + "id": "4ac9e2c8992caabc3278be91b6133a02", + "verified": true + }, + { + "sequence": 59, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000059", + "id": "30c3beb2ff80acd3567c346eeae7b3ba", + "verified": true + }, + { + "sequence": 60, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000060", + "id": "3b08cf0dab20611494d14447d3d90a9a", + "verified": true + }, + { + "sequence": 61, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000061", + "id": "00bbc11b77f6e72257c6965da21bdbd5", + "verified": true + }, + { + "sequence": 62, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000062", + "id": "0d17b7a8892676b39534621fb7097199", + "verified": true + }, + { + "sequence": 63, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000063", + "id": "e1a4d8e2b7cf175304e487f9f399890f", + "verified": true + }, + { + "sequence": 64, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000064", + "id": "374107f404ef9bc0571d1e8ced2c8757", + "verified": true + }, + { + "sequence": 65, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000065", + "id": "d98169107fbe26ec88c50c35d8a23b8a", + "verified": true + }, + { + "sequence": 66, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000066", + "id": "e6ed223d20bee01d419b4a2a55e5c2b9", + "verified": true + }, + { + "sequence": 67, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000067", + "id": "a776780f76bc9fe590119d837ac00879", + "verified": true + }, + { + "sequence": 68, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000068", + "id": "9640b06bfb135db329facb35ce9d9755", + "verified": true + }, + { + "sequence": 69, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000069", + "id": "fedcee9b86d4984dd8c6f19fdca10205", + "verified": true + }, + { + "sequence": 70, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000070", + "id": "1addb5d44702caf623c822e51ce60537", + "verified": true + }, + { + "sequence": 71, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000071", + "id": "7e1390b40485274120de017523f464ed", + "verified": true + }, + { + "sequence": 72, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000072", + "id": "880c7642ca635518f2780f7a760c7169", + "verified": true + }, + { + "sequence": 73, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000073", + "id": "9b9fd0710402a219269bccff68c4c71d", + "verified": true + }, + { + "sequence": 74, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000074", + "id": "4f4837e6a9a360a93b3543e6109963f9", + "verified": true + }, + { + "sequence": 75, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000075", + "id": "47d464f570e39e4c46c3d5e9e19e1047", + "verified": true + }, + { + "sequence": 76, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000076", + "id": "1a18e4e72cb05b9f7438c2d390acd801", + "verified": true + }, + { + "sequence": 77, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000077", + "id": "8143729c8b861067770f32b19af4ec9b", + "verified": true + }, + { + "sequence": 78, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000078", + "id": "819a3757c79c76608367d3ed98a26836", + "verified": true + }, + { + "sequence": 79, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000079", + "id": "88876b96e82def84d673caf98e74bb41", + "verified": true + }, + { + "sequence": 80, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000080", + "id": "0dabf9c0470c43f9229144fa6aec8ac9", + "verified": true + }, + { + "sequence": 81, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000081", + "id": "ee7481e0cfa1cc9f6f5e899f30a06ab1", + "verified": true + }, + { + "sequence": 82, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000082", + "id": "d836266313b92583a40b0ce00a692066", + "verified": true + }, + { + "sequence": 83, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000083", + "id": "97023baae274e7f2ddf905fb4f5e5855", + "verified": true + }, + { + "sequence": 84, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000084", + "id": "44235febbfacebe1cb9d0d19690dda4e", + "verified": true + }, + { + "sequence": 85, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000085", + "id": "c438fd6887dbe35182b47b432becec47", + "verified": true + }, + { + "sequence": 86, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000086", + "id": "c84af42dbd7662e74a8f4eb10783545a", + "verified": true + }, + { + "sequence": 87, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000087", + "id": "d1d0f7485e3e746d737a916f7443de60", + "verified": true + }, + { + "sequence": 88, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000088", + "id": "5f90c67dbcc1f10b63589c0b032c8e90", + "verified": true + }, + { + "sequence": 89, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000089", + "id": "ae78d6a1de189f835fd95755ac9af7a8", + "verified": true + }, + { + "sequence": 90, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000090", + "id": "a6bc331cbe2df16ae824eddd207c0e80", + "verified": true + }, + { + "sequence": 91, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000091", + "id": "467b681912598d6a392822268527e58a", + "verified": true + }, + { + "sequence": 92, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000092", + "id": "e1f7d777c052bb9dbc328fd8273f0ef9", + "verified": true + }, + { + "sequence": 93, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000093", + "id": "d91b9cf421cda602de7f8306ad7f75af", + "verified": true + }, + { + "sequence": 94, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000094", + "id": "24f943ca8d943c2958b80abd6c409cc7", + "verified": true + }, + { + "sequence": 95, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000095", + "id": "e9c1004b98e79e356fe5a68f54fe453a", + "verified": true + }, + { + "sequence": 96, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000096", + "id": "743164ece016598aa9a4926fbdcf8a83", + "verified": true + }, + { + "sequence": 97, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000097", + "id": "4d3a293e47db402d664920e0c4cf0756", + "verified": true + }, + { + "sequence": 98, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000098", + "id": "0d0f3d948e57963e823a0881f392e067", + "verified": true + }, + { + "sequence": 99, + "key": "orbit-workflow-391d8e38471d15a152c81c5fe058ce72-000099", + "id": "4190ec0b3157d8c5c8ed9257df4bfc90", + "verified": true + } + ], + "unconfirmed_invocations": 0, + "run_id": "391d8e38471d15a152c81c5fe058ce72", + "go_version": "go1.25.14", + "goos": "linux", + "goarch": "arm64", + "started_at": "2026-09-20T06:06:08.784936918Z", + "admission_finished_at": "2026-09-20T06:06:09.611845377Z", + "finished_at": "2026-09-20T06:06:51.788045466Z", + "requests": 100, + "concurrency": 8, + "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": "e1126a23e1d796dd823a0e5ea35664b2", + "namespace_id": "eb6180c00667e59b73f49f91ff9a209d", + "name": "workflow", + "paused": false, + "max_concurrency": 4, + "rate_per_second": 10, + "revision": "1" + }, + "definition_id": "45b332119ed40a6108594f375e1810a2", + "version_id": "9a4ac139f53eeedb86ed07c3864297ac", + "attempted": 100, + "not_submitted": 0, + "accepted": 100, + "admission_errors": 0, + "verified_success": 100, + "unverified": 0, + "handler_calls": 200, + "duplicate_invocations": 0, + "worker_started": 200, + "worker_reconnects": 0, + "worker_renewals_confirmed": 0, + "admission": { + "count": 100, + "p50_ms": 65.155667, + "p95_ms": 70.221667, + "p99_ms": 71.856958, + "max_ms": 71.906791 + }, + "completion": { + "count": 100, + "p50_ms": 23486.79, + "p95_ms": 39126.668, + "p99_ms": 42091.211, + "max_ms": 42101.309 + }, + "errors": {}, + "complete": true +} diff --git a/docs/reports/data/phase-6-workflow-load/sample-24.json b/docs/reports/data/phase-6-workflow-load/sample-24.json index ffa815e..c6bbc08 100644 --- a/docs/reports/data/phase-6-workflow-load/sample-24.json +++ b/docs/reports/data/phase-6-workflow-load/sample-24.json @@ -1 +1,203 @@ -{"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} +{ + "instances": [ + { + "sequence": 5, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000005", + "id": "178bdd5a704b2a299728955cece6d08e", + "verified": true + }, + { + "sequence": 0, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000000", + "id": "702e49ff76e9d7643433f3a030b02bee", + "verified": true + }, + { + "sequence": 3, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000003", + "id": "ef3037d49f81fcf8102e47df21c51ed1", + "verified": true + }, + { + "sequence": 1, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000001", + "id": "51fb7a4837d1d9293a5dc056aa49e61e", + "verified": true + }, + { + "sequence": 2, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000002", + "id": "b6b7c0d8269800edc9c3a96a88bcbc84", + "verified": true + }, + { + "sequence": 4, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000004", + "id": "6f1314d449c1c3641b6ba17e03893df9", + "verified": true + }, + { + "sequence": 6, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000006", + "id": "b5e72386d42675dee14d3fe9924f9b39", + "verified": true + }, + { + "sequence": 7, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000007", + "id": "a7bdcae9ae4e7bbde3c3acae0328e51f", + "verified": true + }, + { + "sequence": 8, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000008", + "id": "d43133413e7c9d539f20c9c60db5c6f1", + "verified": true + }, + { + "sequence": 9, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000009", + "id": "8c98c3abc0b6a69b8415ad83bc0793e1", + "verified": true + }, + { + "sequence": 10, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000010", + "id": "d8de79ac4ebfa35181c822ed943151a0", + "verified": true + }, + { + "sequence": 11, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000011", + "id": "29e8cfdec4fc050f1e5105e221a6f19f", + "verified": true + }, + { + "sequence": 12, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000012", + "id": "fd8a19493e2df464529607facd7bf85c", + "verified": true + }, + { + "sequence": 13, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000013", + "id": "2d8aeeb18783b90071c2142510523047", + "verified": true + }, + { + "sequence": 14, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000014", + "id": "20683fc19d30086cc828de9e316e2757", + "verified": true + }, + { + "sequence": 15, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000015", + "id": "40c19926a3f7daa6f70f95fd1d781631", + "verified": true + }, + { + "sequence": 16, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000016", + "id": "1db01b95d081710d4904b5652897a3ff", + "verified": true + }, + { + "sequence": 17, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000017", + "id": "129000f4cbedbf49eed7fe383a8831e8", + "verified": true + }, + { + "sequence": 18, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000018", + "id": "6ea8824907aca473fa981d96303b4c56", + "verified": true + }, + { + "sequence": 19, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000019", + "id": "04fccd5a552649179332c1043c13733d", + "verified": true + }, + { + "sequence": 20, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000020", + "id": "763d83694d2892c396dd9efb0e8bb247", + "verified": true + }, + { + "sequence": 21, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000021", + "id": "1aa04eaf26886a007386b159b774a229", + "verified": true + }, + { + "sequence": 22, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000022", + "id": "162f4958d1e76880c1cbb82199561a1d", + "verified": true + }, + { + "sequence": 23, + "key": "orbit-workflow-642d8722371804cbc829b2a1f6b845e9-000023", + "id": "75d09f113500de4fa1ef6160cb387566", + "verified": true + } + ], + "unconfirmed_invocations": 0, + "run_id": "642d8722371804cbc829b2a1f6b845e9", + "go_version": "go1.25.14", + "goos": "linux", + "goarch": "arm64", + "started_at": "2026-09-20T06:05:55.662781301Z", + "admission_finished_at": "2026-09-20T06:05:55.887174051Z", + "finished_at": "2026-09-20T06:06:08.776731835Z", + "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": "e1126a23e1d796dd823a0e5ea35664b2", + "namespace_id": "eb6180c00667e59b73f49f91ff9a209d", + "name": "workflow", + "paused": false, + "max_concurrency": 4, + "rate_per_second": 10, + "revision": "1" + }, + "definition_id": "7bb77cce989caacf82ee86723a37dffb", + "version_id": "df9eaf25179f41fc141d4305d34694e8", + "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": 47.693917, + "p95_ms": 56.058792, + "p99_ms": 64.539583, + "max_ms": 64.539583 + }, + "completion": { + "count": 24, + "p50_ms": 7874.268, + "p95_ms": 12805.768, + "p99_ms": 12816.559, + "max_ms": 12816.559 + }, + "errors": {}, + "complete": true +} diff --git a/docs/reports/data/phase-6-workflow-load/schema.json b/docs/reports/data/phase-6-workflow-load/schema.json new file mode 100644 index 0000000..718bdc7 --- /dev/null +++ b/docs/reports/data/phase-6-workflow-load/schema.json @@ -0,0 +1,7 @@ +{ + "schema_version": 28, + "migration_count": 28, + "foreign_key_count": 0, + "required_index_count": 6, + "complete": true +} diff --git a/docs/reports/data/phase-6-workflow-load/sdk-check.txt b/docs/reports/data/phase-6-workflow-load/sdk-check.txt new file mode 100644 index 0000000..286e946 --- /dev/null +++ b/docs/reports/data/phase-6-workflow-load/sdk-check.txt @@ -0,0 +1,30 @@ +ok git.yuntuo.hk/k2safe/orbit/sdk/go (cached) +ok git.yuntuo.hk/k2safe/orbit/sdk/go/cmd/orbit-exec-load (cached) +ok git.yuntuo.hk/k2safe/orbit/sdk/go/cmd/orbit-load (cached) +ok git.yuntuo.hk/k2safe/orbit/sdk/go/cmd/orbit-schedule-load (cached) +ok git.yuntuo.hk/k2safe/orbit/sdk/go/cmd/orbit-workflow-load (cached) +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/application-resources [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/batch [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/cancellation [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/delayed-task [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/event-protocol [no test files] +ok git.yuntuo.hk/k2safe/orbit/sdk/go/examples/file-conversion (cached) +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/health [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/history [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/internal/flowdemo [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/keyed-concurrency [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/priority [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/recovery [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/resources [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/retry [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/schedules [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/subscriptions [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/task-query [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/tasks [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/worker [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/worker-protocol [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/workflow [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/workflow-definition [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/workflow-poll [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/workflow-query [no test files] +? git.yuntuo.hk/k2safe/orbit/sdk/go/examples/workflow-recovery [no test files] diff --git a/docs/reports/phase-6-workflow-load.md b/docs/reports/phase-6-workflow-load.md index 449fcaf..26bc1bd 100644 --- a/docs/reports/phase-6-workflow-load.md +++ b/docs/reports/phase-6-workflow-load.md @@ -1,41 +1,47 @@ # P6 Weave 流程执行测量 -验证日期：2026-09-20（UTC 运行时间记录为 2026-09-19 18:51）。这是一次性 MySQL 8.4 容器中的隔离正确性与阶段延迟样本，不是生产容量或高可用验收。 +验证日期：2026-09-20，最终运行 UTC 06:05–06:06。一次性 MySQL 8.4.10 中的隔离正确性与阶段耗时样本，P6/P7 仍进行中，不作为生产容量或高可用验收。 -## 测量内容 +## 本轮修正 -`orbit-workflow-load` 通过独立 Go SDK 完成以下链路： +原测量按每个实例的处理器总调用数核对两个分支，对重复分支、任务身份和本次发布版本的约束不足；输出中序号0还可能由缺失字段被解码为零值。本轮改为精确链路核对： -1. 只读检查队列状态。 -2. 创建一个新的稳定定义并发布不可变 `v1`。 -3. 启动 24 个独立实例，每个实例使用自己的幂等键和业务键。 -4. 每个实例执行一个包含 `left`、`right` 两个静态分支的 `PARALLEL` 节点；两个分支各创建一个 `TASK`。 -5. 两个真实 WSS Worker（每个容量4）回显固定输入。 -6. 对每个成功实例读取步骤、任务、唯一尝试、流程任务来源和最终输出，核对实例固定版本及来源归属。 +- 发布回执必须匹配命名空间、定义身份、完整规范；实例绑定本次发布的不可变版本ID，输入和首次绝对截止不变。 +- 步骤精确为 `pair:PARALLEL`、`left:TASK`、`right:TASK`，全部成功，步骤和激活身份唯一；左右分支必须关联不同任务。 +- 任务的队列、命名空间、处理器、策略、输入/结果、固定截止与时间顺序一致，每任务恰有一个成功尝试；来源匹配该实例、版本、步骤键和激活。 +- 每个已核验任务对应本地同一尝试的一次调用。一个分支执行两次不能抵充另一个分支；尚不能关联的调用单独报告。 +- 精确JSON字段集合拒绝缺字段、null、重复键、大小写别名和尾随JSON，序号0也必须明确提供。 +- 发布的流程及任务时限使用 `run-timeout`，移除原工具中未公开的30秒截断；报告增加每个启动请求的原键、实例ID、受理错误及核验标记。 -处理器只回显 `run_id` 和 `sequence`，没有外部副作用；启动调用和流程读取都没有 SDK 自动重试。流程定义把任务尝试上限固定为1，响应不明时保留原运行身份。 +此前24实例记录没有形成最终源码与所运行二进制的可靠对应，不能用于证明本轮实现。当前原始样本已由**重新构建的二进制**运行并替换旧证据；旧内容可从Git历史追溯。本轮环境清单同时保存源码、实际二进制及原始结果哈希，构建后没有修改这些源码。 ## 实际结果 -| 项目 | 结果 | -| --- | ---: | -| 计划/实际启动调用 | 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 | +工具通过独立 SDK 创建定义、发布固定 `v1`、启动流程，用2个真实 Worker回显固定输入，随后逐项读取流程、步骤、任务、尝试与来源。任务最多一次尝试、`RECONCILE_ONLY`，处理器无外部副作用。任何写请求都不由工具自动重试，队列额度只读。 -原始 JSON：[sample-24.json](data/phase-6-workflow-load/sample-24.json)，工具链和源码摘要见[环境清单](data/phase-6-workflow-load/environment.json)。执行工具和字段语义见[流程负载测量手册](../api/workflow-load.md)。 +| 样本 | 请求/受理 | 持久核验成功/未核验 | 本地调用 | 重复/未确认调用 | 完整/退出码 | +| --- | ---: | ---: | ---: | ---: | --- | +| 24实例，90秒截止 | 24/24 | 24/0 | 48 | 0/0 | true / 0 | +| 100实例，90秒截止 | 100/100 | 100/0 | 200 | 0/0 | true / 0 | +| 24实例，2秒截止 | 24/24 | 2/22 | 4 | 0/0 | false / 1 | -## 环境和边界 +原始结果：[24实例](data/phase-6-workflow-load/sample-24.json)、[100实例](data/phase-6-workflow-load/sample-100.json)、[截止负例](data/phase-6-workflow-load/deadline-24.json)。两组正常样本错误集合为空；负例报告 `RUN_TIMEOUT=1`、`VERIFY_TIMEOUT=22`，22个实例保留原身份且未算成功。截止后只停止本地 Worker，不取消服务器实例；该快照不证明未核验工作已撤销或最终失败。 -- 负载工具：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。 +| 正常样本 | 受理往返P50/P95/P99 | 数据库实例创建至完成P50/P95/P99 | +| --- | --- | --- | +| 24实例 | 47.694 / 56.059 / 64.540ms | 7874.268 / 12805.768 / 12816.559ms | +| 100实例 | 65.156 / 70.222 / 71.857ms | 23486.790 / 39126.668 / 42091.211ms | + +100实例样本中最长流程持久耗时42101.309ms，超过旧工具的30秒隐藏上限，在此次90秒配置下仍正确完成。分布只统计完整核验成功实例，不混入客户端轮询等待；该观察不能推导目标生产吞吐。 + +## 环境、回归及复现 + +单节点 Orbit、一次性 MySQL 8.4.10，1个负载进程中2个Worker。队列并发4、每秒授权10，工具不改额度。24/100正常组请求并发分别6/8，每Worker容量4；负例并发6、每Worker容量2。每次HTTP截止5秒，轮询100毫秒。指标关闭，容器共享本机Docker VM，无独占CPU/内存限制；Go进程与MySQL共享网络命名空间，仅回环通信，无宿主机发布端口。 + +实际工具链为 **Go 1.25.14 Linux arm64**，通过临时modfile使用；仓库声明的Go 1.27.1未改。SDK全模块 `go test -race ./...`、`go vet ./...`、`go build ./...` 通过，见[检查日志](data/phase-6-workflow-load/sdk-check.txt)。新增回归覆盖：发布版本替换、精确步骤/激活/任务身份、错误任务策略/队列/截止、缺失/额外尝试、错误来源和时间顺序、并发本地调用、同一分支重复不能抵充第二分支，以及序号0的严格字段解析。 + +真实样本后 `orbit-admin verify` 确认schema version28、28条迁移记录、6个必需索引及**外键0**，见[结构核验](data/phase-6-workflow-load/schema.json)。本轮没有修改数据库模型或迁移。 + +精确参数、工具链、资源和源码/二进制/结果哈希见[环境清单](data/phase-6-workflow-load/environment.json)。按[调用手册](../api/workflow-load.md)准备一次性数据库和带 `workflow-access`、`workflow-run-access`、`worker-access` 的隔离凭据，使用当前源码重新构建工具，再运行环境清单中的三组参数；退出码应为0、0、1。标准 `scripts/test-mysql.sh` 保留6实例回归入口，本轮未重跑整份集成脚本。 + +本批不覆盖目标Go 1.27.1标准全量、跨标签SDK/服务端兼容矩阵、长期积压和资源曲线、多节点吞吐、跨故障域恢复、生产连接池容量、RPO/RTO或发布审批；这些门槛继续按P6/P7完成。 diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index cbb4be2..b90854e 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -18,7 +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；生产容量未验收 | +| 真实Weave流程测量 | `orbit-workflow-load`；24/100实例、固定发布版本双分支并行及24实例截止负例，见[流程测量报告](phase-6-workflow-load.md) | 124个正常实例、248个不同任务/唯一尝试和本地调用逐项关联；负例2成功/22未核验，非零退出。重新构建并保存源码/二进制哈希，替换旧证据；生产容量未验收 | | 真实定时集中触发测量 | `orbit-schedule-load`；4×6及32×4触发，2×12截止负例，见[定时测量报告](phase-6-schedule-load.md) | 正常152项逐项核验；负例18个未观察点、4个未核验任务，非零退出；38个计划均暂停，实际库外键0。SDK全模块竞态/vet/build通过；生产容量未验收 | diff --git a/sdk/go/cmd/orbit-workflow-load/handler.go b/sdk/go/cmd/orbit-workflow-load/handler.go new file mode 100644 index 0000000..214514d --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/handler.go @@ -0,0 +1,87 @@ +package main + +import ( + "context" + "encoding/json" + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "sync" +) + +// invocation 固定首次本地调用及其尝试身份。 +type invocation struct { + sequence int // sequence 输入中的实例序号。 + attemptID string // attemptID 首次调用的尝试身份。 + calls int // calls 同一任务实际调用次数。 +} + +// invocationLog 按任务身份记录，不能用同一分支的两次调用抵充两个分支。 +type invocationLog struct { + mu sync.Mutex // mu 保护并发Worker回调。 + limit int // limit 最多保存两倍计划实例数的任务身份。 + calls map[string]*invocation // calls 固定身份到首次尝试和计数。 + invalid int // invalid 不符合输入约束的调用。 + overflow int // overflow 超过身份预算的调用。 +} + +func newInvocationLog(n int) *invocationLog { + return &invocationLog{limit: 2 * n, calls: map[string]*invocation{}} +} +func (l *invocationLog) handler(run context.Context, c config, runID string) orbit.WorkerHandler { + return orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: handlerName(runID), HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) { + p, ok := decodePayload(e.Task.Payload) + if !ok || p.Sequence < 0 || p.Sequence >= c.Requests || !matchesTask(e.Task, c, runID, p.Sequence) || !validID(e.AttemptID) || e.Epoch != 1 || e.ConcurrencyToken != 0 { + l.mu.Lock() + l.invalid++ + l.mu.Unlock() + return nil, &orbit.WorkerFailure{Code: "WORKFLOW_INPUT_MISMATCH"} + } + l.mu.Lock() + if previous := l.calls[e.Task.ID]; previous != nil { + previous.calls++ + l.mu.Unlock() + return nil, &orbit.WorkerFailure{Code: "DUPLICATE_INVOCATION"} + } + if len(l.calls) >= l.limit { + l.overflow++ + l.mu.Unlock() + return nil, &orbit.WorkerFailure{Code: "INVOCATION_LIMIT"} + } + l.calls[e.Task.ID] = &invocation{sequence: p.Sequence, attemptID: e.AttemptID, calls: 1} + l.mu.Unlock() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-run.Done(): + return nil, run.Err() + default: + } + return e.Task.Payload, nil + }} +} +func (l *invocationLog) summarize(r *workflowReport, proofs []verifiedTask) { + l.mu.Lock() + defer l.mu.Unlock() + known := map[string]verifiedTask{} + for _, p := range proofs { + if _, exists := known[p.taskID]; exists { + r.Errors["DUPLICATE_VERIFIED_TASK"]++ + } + known[p.taskID] = p + call := l.calls[p.taskID] + if call == nil || call.sequence != p.sequence || call.attemptID != p.attemptID { + r.Errors["LOCAL_PROOF_MISMATCH"]++ + } + } + for id, call := range l.calls { + r.HandlerCalls += call.calls + if call.calls > 1 { + r.DuplicateInvocations += call.calls - 1 + } + p, ok := known[id] + if !ok || p.sequence != call.sequence || p.attemptID != call.attemptID { + r.UnconfirmedInvocations += call.calls + } + } + r.Errors["INVALID_HANDLER_INPUT"] += l.invalid + r.Errors["HANDLER_IDENTITY_LIMIT"] += l.overflow +} diff --git a/sdk/go/cmd/orbit-workflow-load/measure.go b/sdk/go/cmd/orbit-workflow-load/measure.go index e8274be..3943e94 100644 --- a/sdk/go/cmd/orbit-workflow-load/measure.go +++ b/sdk/go/cmd/orbit-workflow-load/measure.go @@ -5,7 +5,6 @@ import ( "context" "encoding/json" "fmt" - "io" "sync" "time" @@ -18,23 +17,6 @@ type workflowPayload struct { 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 { @@ -67,38 +49,7 @@ 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 - }} + tasks []verifiedTask } func workflowSpec(queue, handler string, timeout time.Duration) orbit.WorkflowSpecification { @@ -128,16 +79,12 @@ func measureClient(parent context.Context, cfg config, api *orbit.Client, runID return r.finish() } r.Queue = &queue - if queue.ID != cfg.QueueID || queue.Paused || queue.MaxConcurrency == 0 || queue.RatePerSecond == 0 { + if queue.ID != cfg.QueueID || !validID(queue.NamespaceID) || 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) + spec := workflowSpec(cfg.QueueID, handler, cfg.RunTimeout) 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() @@ -146,7 +93,7 @@ func measureClient(parent context.Context, cfg config, api *orbit.Client, runID return r.finish() } r.DefinitionID = definition.Definition.ID - if !validID(r.DefinitionID) { + if !validID(r.DefinitionID) || definition.Deduplicated || definition.Definition.NamespaceID != queue.NamespaceID || definition.Definition.Name != definitionKey(runID) { r.Errors["INVALID_DEFINITION_RECEIPT"]++ return r.finish() } @@ -158,7 +105,7 @@ func measureClient(parent context.Context, cfg config, api *orbit.Client, runID return r.finish() } r.VersionID = version.Version.ID - if !validID(r.VersionID) || version.Version.Version != "v1" || version.Version.DefinitionID != r.DefinitionID { + if !validID(r.VersionID) || version.Version.Version != "v1" || version.Version.DefinitionID != r.DefinitionID || version.Deduplicated || version.Version.NamespaceID != queue.NamespaceID || !sameSpecification(version.Version.Specification, spec) { r.Errors["INVALID_VERSION_RECEIPT"]++ return r.finish() } @@ -185,13 +132,21 @@ func measureClient(parent context.Context, cfg config, api *orbit.Client, runID }(worker) } - admissions := submitWorkflows(ctx, cfg, api, runID, r.DefinitionID) + admissions := submitWorkflows(ctx, cfg, api, runID, r.DefinitionID, r.VersionID, queue.NamespaceID) r.Attempted = len(admissions) r.NotSubmitted = cfg.Requests - r.Attempted var accepted []workflowAdmission var admissionDurations []time.Duration + seenInstances := map[string]bool{} for _, a := range admissions { admissionDurations = append(admissionDurations, a.duration) + if a.code == "" && seenInstances[a.instance.ID] { + a.code = "DUPLICATE_INSTANCE" + } + if a.code == "" { + seenInstances[a.instance.ID] = true + } + r.Instances = append(r.Instances, instanceReport{Sequence: a.sequence, Key: requestKey(runID, a.sequence), ID: a.instance.ID, Code: a.code}) if a.code != "" { r.AdmissionErrors++ r.Errors["ADMISSION_"+a.code]++ @@ -204,12 +159,20 @@ func measureClient(parent context.Context, cfg config, api *orbit.Client, runID r.AdmissionFinishedAt = time.Now().UTC() verified := verifyWorkflows(ctx, cfg, api, runID, r.DefinitionID, accepted) var completions []time.Duration + var proven []verifiedTask for _, v := range verified { if v.code != "" { r.Errors[v.code]++ continue } r.VerifiedSuccess++ + proven = append(proven, v.tasks...) + for i := range r.Instances { + if r.Instances[i].Sequence == v.sequence { + r.Instances[i].Verified = true + break + } + } completions = append(completions, v.duration) } if ctx.Err() != nil { @@ -227,23 +190,12 @@ func measureClient(parent context.Context, cfg config, api *orbit.Client, runID 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() + log.summarize(&r, proven) r.Completion = distributionOf(completions) return r.finish() } -func submitWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string) []workflowAdmission { +func submitWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, definitionID, versionID, namespace string) []workflowAdmission { jobs := make(chan int) out := make(chan workflowAdmission, cfg.Requests) var wg sync.WaitGroup @@ -263,7 +215,7 @@ func submitWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, de 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) { + } else if result.Deduplicated || !matchesInstance(result.Instance, cfg, runID, sequence, definitionID, versionID, namespace) { a.code = "INVALID_RECEIPT" } out <- a @@ -301,8 +253,8 @@ func verifyWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, de 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} + duration, tasks, code := verifyWorkflow(ctx, cfg, api, runID, definitionID, admission) + out <- workflowVerification{sequence: admission.sequence, duration: duration, tasks: tasks, code: code} } }() } @@ -315,110 +267,61 @@ func verifyWorkflows(ctx context.Context, cfg config, api workflowAPI, runID, de return results } -func verifyWorkflow(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string, admission workflowAdmission) (time.Duration, string) { +func verifyWorkflow(ctx context.Context, cfg config, api workflowAPI, runID, definitionID string, admission workflowAdmission) (time.Duration, []verifiedTask, string) { for { if ctx.Err() != nil { - return 0, "VERIFY_" + classify(ctx.Err()) + return 0, nil, "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) + return 0, nil, "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" + if instance.ID != admission.instance.ID || !matchesInstance(instance, cfg, runID, admission.sequence, definitionID, admission.instance.DefinitionVersionID, admission.instance.NamespaceID) || !instance.CreatedAt.Equal(admission.instance.CreatedAt) || !instance.DeadlineAt.Equal(admission.instance.DeadlineAt) { + return 0, nil, "WORKFLOW_MISMATCH" } switch instance.State { case "SUCCEEDED": if instance.CompletedAt == nil || instance.CreatedAt.IsZero() || instance.CompletedAt.Before(instance.CreatedAt) { - return 0, "COMPLETION_MISMATCH" + return 0, nil, "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) + return 0, nil, "STEPS_" + classify(e) } - if code := verifySteps(ctx, cfg, api, instance, runID, admission.sequence, steps); code != "" { - return 0, code + tasks, code := verifySteps(ctx, cfg, api, instance, runID, admission.sequence, steps) + if code != "" { + return 0, nil, code } if !matchesOutput(instance.Output, runID, admission.sequence) { - return 0, "OUTPUT_MISMATCH" + return 0, nil, "OUTPUT_MISMATCH" } - return instance.CompletedAt.Sub(instance.CreatedAt), "" + return instance.CompletedAt.Sub(instance.CreatedAt), tasks, "" case "NEEDS_REVIEW", "FAILED", "CANCELLED": - return 0, "WORKFLOW_" + instance.State + return 0, nil, "WORKFLOW_" + instance.State case "RUNNING", "WAITING", "PAUSED", "PENDING": default: - return 0, "UNKNOWN_WORKFLOW_STATE" + return 0, nil, "UNKNOWN_WORKFLOW_STATE" } timer := time.NewTimer(cfg.PollInterval) select { case <-ctx.Done(): timer.Stop() - return 0, "VERIFY_" + classify(ctx.Err()) + return 0, nil, "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 sameSpecification(a, b orbit.WorkflowSpecification) bool { + left, e := json.Marshal(a) + right, e2 := json.Marshal(b) + return e == nil && e2 == nil && bytes.Equal(left, right) } - -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 +func matchesInstance(i orbit.WorkflowInstance, c config, runID string, sequence int, definitionID, versionID, namespace string) bool { + p, ok := decodePayload(i.Input) + return validID(i.ID) && i.NamespaceID == namespace && i.DefinitionID == definitionID && i.DefinitionVersionID == versionID && i.Version == "v1" && i.BusinessKey == businessKey(runID, sequence) && ok && p == (workflowPayload{RunID: runID, Sequence: sequence}) && !i.CreatedAt.IsZero() && i.DeadlineAt.Equal(i.CreatedAt.Add(time.Duration(c.RunTimeout.Milliseconds())*time.Millisecond)) && !i.Paused && i.CancelRequestedAt == nil && i.StopKind == nil && i.StopRequestedAt == nil } diff --git a/sdk/go/cmd/orbit-workflow-load/payload.go b/sdk/go/cmd/orbit-workflow-load/payload.go new file mode 100644 index 0000000..4629a36 --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/payload.go @@ -0,0 +1,64 @@ +package main + +import ( + "bytes" + "encoding/json" + "io" +) + +// exactFields 验证精确字段集合；原文重复键、大小写别名和尾随JSON均拒绝。 +func exactFields(raw json.RawMessage, names ...string) (map[string]json.RawMessage, bool) { + decoder := json.NewDecoder(bytes.NewReader(raw)) + token, err := decoder.Token() + if err != nil || token != json.Delim('{') { + return nil, false + } + fields := make(map[string]json.RawMessage, len(names)) + for decoder.More() { + token, err = decoder.Token() + if err != nil { + return nil, false + } + key, ok := token.(string) + if !ok { + return nil, false + } + allowed := false + for _, name := range names { + allowed = allowed || name == key + } + if !allowed { + return nil, false + } + if _, exists := fields[key]; exists { + return nil, false + } + var value json.RawMessage + if decoder.Decode(&value) != nil || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return nil, false + } + fields[key] = value + } + if _, err = decoder.Token(); err != nil { + return nil, false + } + var tail any + return fields, len(fields) == len(names) && decoder.Decode(&tail) == io.EOF +} +func decodePayload(raw json.RawMessage) (workflowPayload, bool) { + fields, ok := exactFields(raw, "run_id", "sequence") + var out workflowPayload + if !ok || json.Unmarshal(fields["run_id"], &out.RunID) != nil || json.Unmarshal(fields["sequence"], &out.Sequence) != nil { + return out, false + } + return out, true +} +func matchesOutput(raw json.RawMessage, runID string, sequence int) bool { + fields, ok := exactFields(raw, "left", "right") + if !ok { + return false + } + left, ok := decodePayload(fields["left"]) + right, ok2 := decodePayload(fields["right"]) + return ok && ok2 && left == (workflowPayload{RunID: runID, Sequence: sequence}) && right == left +} diff --git a/sdk/go/cmd/orbit-workflow-load/report.go b/sdk/go/cmd/orbit-workflow-load/report.go index b964378..284bff8 100644 --- a/sdk/go/cmd/orbit-workflow-load/report.go +++ b/sdk/go/cmd/orbit-workflow-load/report.go @@ -18,44 +18,55 @@ type distribution struct { Max *float64 `json:"max_ms"` // Max 纳入样本的最大毫秒数。 } +// instanceReport 保存每个启动请求的原身份，便于未知结果按原键核对。 +type instanceReport struct { + Sequence int `json:"sequence"` // Sequence 从零开始的请求序号。 + Key string `json:"key"` // Key 启动幂等键，同时作为业务键。 + ID string `json:"id,omitempty"` // ID 已取得的实例身份；为空不证明未提交。 + Code string `json:"admission_error,omitempty"` // Code 受理错误分类，不包含原异常正文。 + Verified bool `json:"verified"` // Verified 是否通过持久结果核验。 +} + // 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 仅全部实例和两倍处理器调用逐项通过核验时为真。 + Instances []instanceReport `json:"instances"` // Instances 已尝试启动的请求身份，包含失败或未知结果。 + UnconfirmedInvocations int `json:"unconfirmed_invocations"` // UnconfirmedInvocations 尚未关联到已核验分支任务/尝试的调用数。 + 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{}} + return workflowReport{Instances: []instanceReport{}, 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 { @@ -66,7 +77,7 @@ func (r workflowReport) finish() workflowReport { } 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 + 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 && r.UnconfirmedInvocations == 0 && len(r.Errors) == 0 return r } diff --git a/sdk/go/cmd/orbit-workflow-load/verify.go b/sdk/go/cmd/orbit-workflow-load/verify.go new file mode 100644 index 0000000..57dd9e5 --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/verify.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +// verifiedTask 保存已核验分支的任务和尝试，最终与本地调用逐个关联。 +type verifiedTask struct { + sequence int // sequence 实例请求序号。 + taskID string // taskID 分支唯一任务身份。 + attemptID string // attemptID 唯一成功尝试身份。 +} + +func verifySteps(ctx context.Context, cfg config, api workflowAPI, instance orbit.WorkflowInstance, runID string, sequence int, steps orbit.WorkflowSteps) ([]verifiedTask, string) { + if steps.InstanceID != instance.ID || steps.Revision != instance.Revision || len(steps.Steps) != 3 { + return nil, "STEP_COUNT_OR_PARENT_MISMATCH" + } + keys, ids, tasks, activations := map[string]bool{}, map[string]bool{}, map[string]bool{}, map[string]bool{} + for _, s := range steps.Steps { + kind := map[string]string{"pair": "PARALLEL", "left": "TASK", "right": "TASK"}[s.Key] + if kind == "" || s.Kind != kind || s.State != "SUCCEEDED" || !validID(s.ID) || s.Revision == 0 || ids[s.ID] || keys[s.Key] || s.ActivationSequence != 1 || s.ActivationID == nil || !validID(*s.ActivationID) || activations[*s.ActivationID] { + return nil, "STEP_MISMATCH" + } + keys[s.Key] = true + ids[s.ID] = true + activations[*s.ActivationID] = true + if kind == "PARALLEL" { + if s.TaskID != nil { + return nil, "JOIN_TASK_MISMATCH" + } + continue + } + if s.TaskID == nil || !validID(*s.TaskID) || tasks[*s.TaskID] { + return nil, "DUPLICATE_OR_INVALID_TASK" + } + tasks[*s.TaskID] = true + } + proofs := make([]verifiedTask, 0, 2) + for _, step := range steps.Steps { + if step.Kind != "TASK" { + continue + } + call, stop := context.WithTimeout(ctx, cfg.Timeout) + task, err := api.GetTask(call, *step.TaskID) + stop() + if err != nil { + return nil, "TASK_" + classify(err) + } + result, ok := decodePayload(task.Result) + if task.ID != *step.TaskID || !matchesTask(task, cfg, runID, sequence) || task.NamespaceID != instance.NamespaceID || task.State != orbit.TaskSucceeded || task.EffectUncertain || task.AttemptCount != 1 || task.CompletedAt == nil || task.CancelRequestedAt != nil || !ok || result != (workflowPayload{RunID: runID, Sequence: sequence}) { + return nil, "TASK_MISMATCH" + } + if task.CreatedAt.Before(instance.CreatedAt) || !task.AvailableAt.Equal(task.CreatedAt) || !task.DeadlineAt.Equal(instance.DeadlineAt) || task.CompletedAt.Before(task.CreatedAt) || instance.CompletedAt == nil || task.CompletedAt.After(*instance.CompletedAt) { + return nil, "TASK_TIMESTAMP_MISMATCH" + } + call, stop = context.WithTimeout(ctx, cfg.Timeout) + history, err := api.ListTaskAttempts(call, task.ID, orbit.HistoryPageRequest{Limit: 2}) + stop() + if err != nil { + return nil, "HISTORY_" + classify(err) + } + if len(history.Items) != 1 || history.NextCursor != "" { + return nil, "ATTEMPT_COUNT_MISMATCH" + } + a := history.Items[0] + if !validID(a.ID) || a.TaskID != task.ID || a.State != orbit.AttemptSucceeded || a.Epoch != 1 || a.AuthorizedAt == nil || a.EndedAt == nil || a.ErrorCode != "" || a.Retryable || a.ConcurrencyToken != 0 { + return nil, "ATTEMPT_MISMATCH" + } + if a.CreatedAt.Before(task.AvailableAt) || a.AuthorizedAt.Before(a.CreatedAt) || a.EndedAt.Before(*a.AuthorizedAt) || !a.EndedAt.Equal(*task.CompletedAt) { + return nil, "ATTEMPT_TIMESTAMP_MISMATCH" + } + call, stop = context.WithTimeout(ctx, cfg.Timeout) + source, err := api.GetTaskSource(call, task.ID) + stop() + if err != nil { + return nil, "SOURCE_" + classify(err) + } + s := source.Workflow + if source.TaskID != task.ID || source.Kind != "WORKFLOW" || source.RetryOf != nil || source.Schedule != nil || s == nil || s.WorkflowID != instance.ID || s.DefinitionID != instance.DefinitionID || s.DefinitionVersionID != instance.DefinitionVersionID || s.Version != instance.Version || s.StepID != step.ID || s.StepKey != step.Key || s.StepKind != "TASK" || s.ActivationID != *step.ActivationID || s.ActivationSequence != 1 || s.PollRound != 0 || s.PollRoundID != nil || s.RetryOf != nil { + return nil, "SOURCE_MISMATCH" + } + proofs = append(proofs, verifiedTask{sequence: sequence, taskID: task.ID, attemptID: a.ID}) + } + return proofs, "" +} +func matchesTask(t orbit.Task, c config, runID string, sequence int) bool { + input, ok := decodePayload(t.Payload) + return validID(t.ID) && t.QueueID == c.QueueID && t.Handler == handlerName(runID) && t.HandlerVersion == "v1" && t.PayloadVersion == "v1" && t.BusinessKey == "" && t.ConcurrencyKey == "" && t.Priority == 0 && t.RetryOf == nil && t.StartDeadlineAt == nil && t.RecoverySafety == orbit.RecoveryReconcileOnly && t.Retry == (orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(c.RunTimeout.Milliseconds())}) && ok && input == (workflowPayload{RunID: runID, Sequence: sequence}) +} diff --git a/sdk/go/cmd/orbit-workflow-load/verify_test.go b/sdk/go/cmd/orbit-workflow-load/verify_test.go new file mode 100644 index 0000000..0993345 --- /dev/null +++ b/sdk/go/cmd/orbit-workflow-load/verify_test.go @@ -0,0 +1,210 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "sync" + "testing" + "time" +) + +func testID(n int) string { return fmt.Sprintf("%032x", n) } +func testConfig() config { + return config{QueueID: testID(1), Requests: 1, Concurrency: 1, Workers: 1, Capacity: 2, Timeout: time.Second, RunTimeout: 90 * time.Second, PollInterval: 100 * time.Millisecond} +} + +type proofAPI struct { + workflowAPI + instance orbit.WorkflowInstance + steps orbit.WorkflowSteps + tasks map[string]orbit.Task + attempts map[string]orbit.AttemptPage + sources map[string]orbit.TaskSource +} + +func (a *proofAPI) GetWorkflow(context.Context, string) (orbit.WorkflowInstance, error) { + return a.instance, nil +} +func (a *proofAPI) GetWorkflowSteps(context.Context, string) (orbit.WorkflowSteps, error) { + return a.steps, nil +} +func (a *proofAPI) GetTask(_ context.Context, id string) (orbit.Task, error) { return a.tasks[id], nil } +func (a *proofAPI) ListTaskAttempts(_ context.Context, id string, _ orbit.HistoryPageRequest) (orbit.AttemptPage, error) { + return a.attempts[id], nil +} +func (a *proofAPI) GetTaskSource(_ context.Context, id string) (orbit.TaskSource, error) { + return a.sources[id], nil +} +func (a *proofAPI) StartWorkflow(context.Context, string, orbit.StartWorkflowRequest) (orbit.WorkflowStartResult, error) { + return orbit.WorkflowStartResult{Instance: a.instance}, nil +} +func proofFixture() (*proofAPI, workflowAdmission) { + c := testConfig() + start := time.Date(2026, 9, 20, 0, 0, 0, 0, time.UTC) + end := start.Add(3 * time.Second) + raw := json.RawMessage(`{"run_id":"run","sequence":0}`) + instance := orbit.WorkflowInstance{ID: testID(2), NamespaceID: testID(3), DefinitionID: testID(4), DefinitionVersionID: testID(5), Version: "v1", State: "SUCCEEDED", Revision: 4, Input: raw, Output: json.RawMessage(`{"left":{"run_id":"run","sequence":0},"right":{"run_id":"run","sequence":0}}`), BusinessKey: businessKey("run", 0), DeadlineAt: start.Add(c.RunTimeout), CreatedAt: start, CompletedAt: &end} + api := &proofAPI{instance: instance, steps: orbit.WorkflowSteps{InstanceID: instance.ID, Revision: instance.Revision}, tasks: map[string]orbit.Task{}, attempts: map[string]orbit.AttemptPage{}, sources: map[string]orbit.TaskSource{}} + for i, key := range []string{"pair", "left", "right"} { + activation := testID(10 + i) + step := orbit.WorkflowStepStatus{ID: testID(20 + i), Key: key, Kind: "TASK", State: "SUCCEEDED", Revision: 3, ActivationID: &activation, ActivationSequence: 1} + if i == 0 { + step.Kind = "PARALLEL" + api.steps.Steps = append(api.steps.Steps, step) + continue + } + taskID := testID(30 + i) + step.TaskID = &taskID + api.steps.Steps = append(api.steps.Steps, step) + authorized, completed := start.Add(time.Second), start.Add(2*time.Second) + task := orbit.Task{ID: taskID, NamespaceID: instance.NamespaceID, QueueID: c.QueueID, Handler: handlerName("run"), HandlerVersion: "v1", PayloadVersion: "v1", State: orbit.TaskSucceeded, Payload: raw, Result: raw, RecoverySafety: orbit.RecoveryReconcileOnly, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(c.RunTimeout.Milliseconds())}, AttemptCount: 1, CreatedAt: start, AvailableAt: start, DeadlineAt: instance.DeadlineAt, CompletedAt: &completed} + api.tasks[taskID] = task + api.attempts[taskID] = orbit.AttemptPage{Items: []orbit.AttemptSummary{{ID: testID(40 + i), TaskID: taskID, State: orbit.AttemptSucceeded, Epoch: 1, CreatedAt: start, AuthorizedAt: &authorized, EndedAt: &completed}}} + api.sources[taskID] = orbit.TaskSource{TaskID: taskID, Kind: "WORKFLOW", Workflow: &orbit.TaskWorkflowSource{WorkflowID: instance.ID, DefinitionID: instance.DefinitionID, DefinitionVersionID: instance.DefinitionVersionID, Version: "v1", StepID: step.ID, StepKey: key, StepKind: "TASK", ActivationID: activation, ActivationSequence: 1}} + } + return api, workflowAdmission{sequence: 0, instance: instance} +} +func TestWorkflowProofComplete(t *testing.T) { + api, admission := proofFixture() + cfg := testConfig() + duration, proofs, code := verifyWorkflow(context.Background(), cfg, api, "run", admission.instance.DefinitionID, admission) + if code != "" || duration != 3*time.Second || len(proofs) != 2 || proofs[0].taskID == proofs[1].taskID { + t.Fatalf("valid proof rejected %s %#v", code, proofs) + } + log := newInvocationLog(1) + handler := log.handler(context.Background(), cfg, "run") + var wg sync.WaitGroup + for _, p := range proofs { + wg.Add(1) + go func(p verifiedTask) { + defer wg.Done() + _, err := handler.Handle(context.Background(), orbit.WorkerExecution{Task: api.tasks[p.taskID], AttemptID: p.attemptID, Epoch: 1}) + if err != nil { + t.Error(err) + } + }(p) + } + wg.Wait() + r := newReport(cfg, "run") + log.summarize(&r, proofs) + if r.HandlerCalls != 2 || r.DuplicateInvocations != 0 || r.UnconfirmedInvocations != 0 || len(r.finish().Errors) != 0 { + t.Fatalf("call proof mismatch %#v", r) + } +} +func TestWorkflowProofRejectsWrongIdentitiesAndHistory(t *testing.T) { + for name, change := range map[string]func(*proofAPI){ + "version substitution": func(a *proofAPI) { a.instance.DefinitionVersionID = testID(80) }, + "changed input": func(a *proofAPI) { a.instance.Input = json.RawMessage(`{"run_id":"run"}`) }, + "wrong steps owner": func(a *proofAPI) { a.steps.InstanceID = testID(80) }, + "missing join": func(a *proofAPI) { a.steps.Steps = a.steps.Steps[1:] }, + "duplicate branch": func(a *proofAPI) { a.steps.Steps[2] = a.steps.Steps[1] }, + "wrong branch name": func(a *proofAPI) { a.steps.Steps[2].Key = "third" }, + "reused task": func(a *proofAPI) { a.steps.Steps[2].TaskID = a.steps.Steps[1].TaskID }, + "reused activation": func(a *proofAPI) { a.steps.Steps[2].ActivationID = a.steps.Steps[1].ActivationID }, + "wrong task policy": func(a *proofAPI) { v := a.tasks[testID(31)]; v.Retry.MaxAttempts = 2; a.tasks[v.ID] = v }, + "wrong task queue": func(a *proofAPI) { v := a.tasks[testID(31)]; v.QueueID = testID(80); a.tasks[v.ID] = v }, + "missing result sequence": func(a *proofAPI) { + v := a.tasks[testID(31)] + v.Result = json.RawMessage(`{"run_id":"run"}`) + a.tasks[v.ID] = v + }, + "uncertain task": func(a *proofAPI) { v := a.tasks[testID(31)]; v.EffectUncertain = true; a.tasks[v.ID] = v }, + "wrong task deadline": func(a *proofAPI) { + v := a.tasks[testID(31)] + v.DeadlineAt = v.DeadlineAt.Add(time.Second) + a.tasks[v.ID] = v + }, + "extra attempt": func(a *proofAPI) { + v := a.attempts[testID(31)] + v.Items = append(v.Items, v.Items[0]) + a.attempts[testID(31)] = v + }, + "missing authorization": func(a *proofAPI) { a.attempts[testID(31)].Items[0].AuthorizedAt = nil }, + "wrong epoch": func(a *proofAPI) { a.attempts[testID(31)].Items[0].Epoch = 2 }, + "wrong attempt time": func(a *proofAPI) { + v := a.instance.CreatedAt.Add(-time.Second) + a.attempts[testID(31)].Items[0].EndedAt = &v + }, + "wrong source task": func(a *proofAPI) { v := a.sources[testID(31)]; v.TaskID = testID(80); a.sources[testID(31)] = v }, + "wrong source branch": func(a *proofAPI) { a.sources[testID(31)].Workflow.StepKey = "right" }, + "wrong source kind": func(a *proofAPI) { a.sources[testID(31)].Workflow.StepKind = "POLL" }, + "wrong output zero": func(a *proofAPI) { + a.instance.Output = json.RawMessage(`{"left":{"run_id":"run"},"right":{"run_id":"run"}}`) + }, + } { + t.Run(name, func(t *testing.T) { + a, admission := proofFixture() + change(a) + _, proofs, code := verifyWorkflow(context.Background(), testConfig(), a, "run", admission.instance.DefinitionID, admission) + if code == "" || len(proofs) != 0 { + t.Fatalf("bad evidence counted %s %#v", code, proofs) + } + }) + } +} +func TestStartBindsPublishedVersion(t *testing.T) { + a, _ := proofFixture() + c := testConfig() + results := submitWorkflows(context.Background(), c, a, "run", a.instance.DefinitionID, a.instance.DefinitionVersionID, a.instance.NamespaceID) + if len(results) != 1 || results[0].code != "" { + t.Fatalf("valid start rejected %#v", results) + } + results = submitWorkflows(context.Background(), c, a, "run", a.instance.DefinitionID, testID(80), a.instance.NamespaceID) + if len(results) != 1 || results[0].code != "INVALID_RECEIPT" { + t.Fatal("unpublished version accepted") + } +} +func TestDuplicateInvocationCannotCountAsSecondBranch(t *testing.T) { + a, _ := proofFixture() + c := testConfig() + log := newInvocationLog(1) + h := log.handler(context.Background(), c, "run") + proof := verifiedTask{sequence: 0, taskID: testID(31), attemptID: testID(41)} + for i := 0; i < 2; i++ { + _, _ = h.Handle(context.Background(), orbit.WorkerExecution{Task: a.tasks[proof.taskID], AttemptID: proof.attemptID, Epoch: 1}) + } + r := newReport(c, "run") + log.summarize(&r, []verifiedTask{proof, {sequence: 0, taskID: testID(32), attemptID: testID(42)}}) + if r.HandlerCalls != 2 || r.DuplicateInvocations != 1 || r.Errors["LOCAL_PROOF_MISMATCH"] != 1 { + t.Fatalf("duplicate filled missing branch %#v", r) + } + r = newReport(c, "run") + log.summarize(&r, nil) + if r.UnconfirmedInvocations != 2 { + t.Fatal("unverified calls disappeared") + } +} +func TestStrictOutputKeepsSequenceZeroRequired(t *testing.T) { + for _, raw := range []string{ + `{"left":{"run_id":"run"},"right":{"run_id":"run","sequence":0}}`, + `{"left":{"run_id":"run","sequence":null},"right":{"run_id":"run","sequence":0}}`, + `{"left":{"run_id":"run","sequence":0,"sequence":0},"right":{"run_id":"run","sequence":0}}`, + `{"left":{"run_id":"run","Sequence":0},"right":{"run_id":"run","sequence":0}}`, + `{"left":{"run_id":"run","sequence":0},"right":{"run_id":"run","sequence":0},"right":{"run_id":"run","sequence":0}}`, + } { + if matchesOutput(json.RawMessage(raw), "run", 0) { + t.Fatalf("invalid zero payload accepted %s", raw) + } + } +} +func TestSpecificationRetainsConfiguredTimeout(t *testing.T) { + c := testConfig() + spec := workflowSpec(c.QueueID, handlerName("run"), c.RunTimeout) + if spec.TimeoutMS != 90000 || spec.Body.Steps[0].Parallel.Branches[0].Body.Steps[0].Task.TimeoutMS != 90000 { + t.Fatal("run timeout silently capped") + } + var stored orbit.WorkflowSpecification + raw, _ := json.Marshal(spec) + if err := json.Unmarshal(raw, &stored); err != nil { + t.Fatal(err) + } + if !sameSpecification(stored, spec) { + t.Fatal("serialized specification rejected") + } + stored.TimeoutMS = 30000 + if sameSpecification(stored, spec) { + t.Fatal("changed published spec accepted") + } +}