From 29734596a34a83add56087827d664934e439d37e Mon Sep 17 00:00:00 2001 From: k2safe Date: Sun, 20 Sep 2026 13:56:44 +0800 Subject: [PATCH] feat: verify bounded simultaneous schedule execution --- AGENTS.md | 2 + README.md | 2 +- docs/08-development-plan.md | 2 +- docs/api/README.md | 1 + docs/api/schedule-load.md | 74 +++ .../data/phase-6-schedule-load/database.json | 10 + .../phase-6-schedule-load/deadline-24.json | 110 ++++ .../phase-6-schedule-load/environment.json | 113 ++++ .../phase-6-schedule-load/sample-128.json | 618 ++++++++++++++++++ .../data/phase-6-schedule-load/sample-24.json | 142 ++++ .../data/phase-6-schedule-load/schema.json | 7 + .../data/phase-6-schedule-load/sdk-check.log | 30 + docs/reports/phase-6-schedule-load.md | 42 ++ docs/reports/phase-7-regression-status.md | 2 + scripts/test-mysql.sh | 2 + sdk/go/cmd/orbit-schedule-load/handler.go | 82 +++ sdk/go/cmd/orbit-schedule-load/main.go | 153 +++++ sdk/go/cmd/orbit-schedule-load/measure.go | 351 ++++++++++ .../cmd/orbit-schedule-load/measure_test.go | 313 +++++++++ sdk/go/cmd/orbit-schedule-load/observe.go | 206 ++++++ sdk/go/cmd/orbit-schedule-load/report.go | 111 ++++ 21 files changed, 2371 insertions(+), 2 deletions(-) create mode 100644 docs/api/schedule-load.md create mode 100644 docs/reports/data/phase-6-schedule-load/database.json create mode 100644 docs/reports/data/phase-6-schedule-load/deadline-24.json create mode 100644 docs/reports/data/phase-6-schedule-load/environment.json create mode 100644 docs/reports/data/phase-6-schedule-load/sample-128.json create mode 100644 docs/reports/data/phase-6-schedule-load/sample-24.json create mode 100644 docs/reports/data/phase-6-schedule-load/schema.json create mode 100644 docs/reports/data/phase-6-schedule-load/sdk-check.log create mode 100644 docs/reports/phase-6-schedule-load.md create mode 100644 sdk/go/cmd/orbit-schedule-load/handler.go create mode 100644 sdk/go/cmd/orbit-schedule-load/main.go create mode 100644 sdk/go/cmd/orbit-schedule-load/measure.go create mode 100644 sdk/go/cmd/orbit-schedule-load/measure_test.go create mode 100644 sdk/go/cmd/orbit-schedule-load/observe.go create mode 100644 sdk/go/cmd/orbit-schedule-load/report.go diff --git a/AGENTS.md b/AGENTS.md index acc056e..c91bfd6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -88,3 +88,5 @@ - `sdk/go/cmd/orbit-load` 只用于隔离环境受理测量：每项使用稳定独立幂等键，固定请求形状和并发上限，按请求超时，不自动重试，不启动Worker或执行外部副作用；输出延迟百分位与有限错误分类，失败必须以非零状态返回。其结果不能替代真实Worker、流程推进、数据库拓扑或生产容量验收。 - `sdk/go/cmd/orbit-exec-load` 独立测量真实Worker执行：每次运行固定专属处理器/业务键/序号/模拟时长，最多一次尝试且无外部副作用；任务成功须核对持久化结果、唯一尝试、时间顺序及单次本地调用。队列额度只读，提交不重试；整体截止和有界排空后输出实际分母，未完成/未知结果必须非零退出。数据库阶段耗时不得混用客户端轮询观察时间，不将隔离样本解释为生产容量。 - `sdk/go/cmd/orbit-workflow-load` 独立测量真实Weave流程：每次运行创建固定版本的双分支并行定义，启动实例不自动重试，必须核对流程/步骤/任务/尝试/来源/输出及两倍分支调用；队列只读，处理器无外部副作用，截止后保留实际分母并以非零状态退出。隔离样本不代表生产容量、高可用或跨故障域验收。 + +- `sdk/go/cmd/orbit-schedule-load` 使用同时起止的有限FIXED_RATE计划测量集中触发；固定版本/时区快照、触发时间槽、任务/尝试/来源、本地调用逐项核对。配置总数、已观察、已触发和已核验分开计数；截止保留原操作身份并有界暂停，不取消已生成任务；暂停只在明确版本冲突后换键，传输不明不得新键重试。样本不代表长期积压容量或生产HA。 diff --git a/README.md b/README.md index 83d5692..81c9418 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)：短/混合/中长模拟任务、持久结果和单次尝试核验、阶段延迟分布；不修改队列额度、不调用外部业务系统。Weave 并行流程链路使用[`orbit-workflow-load`](docs/api/workflow-load.md)，核对固定版本、步骤、来源和最终输出。 +真实Worker与任务完成路径测量使用[`orbit-exec-load`](docs/api/execution-load.md)：短/混合/中长模拟任务、持久结果和单次尝试核验、阶段延迟分布；不修改队列额度、不调用外部业务系统。Weave 并行流程链路使用[`orbit-workflow-load`](docs/api/workflow-load.md)，核对固定版本、步骤、来源和最终输出。定时集中触发使用[`orbit-schedule-load`](docs/api/schedule-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 20fe137..8cecfa1 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)。标准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)，用固定版本的双分支并行流程核对步骤、来源、尝试和输出，见[流程测量报告](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/README.md b/docs/api/README.md index 9e75425..8c19849 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -26,6 +26,7 @@ | `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) | +| 定时集中触发测量 | 已实现 | [有界计划测量工具](schedule-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/schedule-load.md b/docs/api/schedule-load.md new file mode 100644 index 0000000..8f13294 --- /dev/null +++ b/docs/api/schedule-load.md @@ -0,0 +1,74 @@ +# Pulse 定时触发测量工具 + +`orbit-schedule-load` 在隔离环境创建多个有限的固定间隔计划，让它们共享同一组触发时间，再通过真实 Worker 和持久化记录逐项核验。处理器只回显固定 JSON，没有外部业务副作用；工具不修改队列额度，不需要额外消息队列或缓存。 + +## 运行场景 + +在 SDK 模块目录执行，凭据需要队列读取、Worker 执行、任务读取及计划创建/读取/控制权限。隔离测试凭据可由 `orbit-admin bootstrap --worker-access --schedule-access` 签发，完整步骤见[管理手册](operations.md)；不要使用生产队列做样本。 + +```sh +# 当前目录：sdk/go +ORBIT_URL=https://orbit.example.test \ +ORBIT_CREDENTIALS_FILE=/run/secrets/orbit-sample.json \ +go run ./cmd/orbit-schedule-load \ + -schedules 4 -requests 6 -interval 1s \ + -workers 2 -capacity 4 -start-delay 5s \ + -run-timeout 90s -poll-interval 100ms +``` + +这个例子预期产生 **4 个计划 × 每计划 6 次 = 24 个任务**。每计划先以暂停状态创建，全部创建完后启动 Worker 并恢复计划。恢复必须在共同起点之前完成，迟到会判定本次样本失败。排他的 `end_at = start_at + requests × interval` 限制触发总量；即使工具被强制终止，计划也有固定终点。 + +可以把 `-schedules` 调到 32、`-requests` 设为 4，观察同一秒集中触发及现有队列速率限制下的排队。队列默认并发和每秒授权量仍由部署配置决定，工具只记录预检快照，不提高配额。另可使用 `-schedules 2 -requests 12 -start-delay 1s -run-timeout 5s` 验证截止不足的失败报告；它必须非零退出，未观察到的触发点不能算成功。 + +## 参数边界 + +| 参数 | 默认值 | 限制/语义 | +| --- | --- | --- | +| `-requests` | 6 | 每计划 1–100 次，不是总任务数 | +| `-schedules` | 1 | 1–32 个，共享起止和间隔 | +| `-interval` | 1s | 1s–1m，整毫秒 | +| `-start-delay` | 5s | 1s–1m，整毫秒；预留创建、注册和恢复时间，不保证 Worker 已连接 | +| `-workers` | 2 | 1–16 个 WSS 连接，均在一个负载进程中 | +| `-capacity` | 2 | 每 Worker 1–128，总容量不超过 256 | +| `-timeout` | 5s | 单个 HTTP 请求 1ms–1m | +| `-run-timeout` | 2m | 整次测量 5s–1h，另外最多 5s Worker 收尾及 5s 计划暂停收尾 | +| `-poll-interval` | 250ms | 100ms–10s，每轮顺序查询各计划及未核验任务 | +| `-credentials` | 环境变量路径 | 最大 16 KiB 的私密 JSON，包含 `token`、`queue_id` | + +URL、凭据路径和队列也可从 `ORBIT_URL`、`ORBIT_CREDENTIALS_FILE`、`ORBIT_QUEUE_ID` 读取。HTTP 明文仅用于 SDK 允许的本机隔离测试；远端使用 HTTPS/WSS。 + +计划使用 `FIXED_RATE`、`UTC`、`CATCH_UP`、`PARALLEL`，批量补跑上限 100，补跑窗口覆盖整次样本及计划跨度。发布版本固定为 1，服务端时区快照标识与生效时间从首次回执保存，后续不得变化。任务策略固定一次尝试、`RECONCILE_ONLY`，总时限和尝试时限为 `run-timeout`；未知结果不会被工具自动重跑。 + +## 核验和报告 + +输出一行 JSON，退出码 0 只表示本次有界样本完整通过。以下分母分别报告： + +| 字段 | 解释 | +| --- | --- | +| `expected_occurrences` | 配置的总触发数，包含尚未观察到的部分 | +| `occurrences` | 已看到的唯一预期触发点 | +| `fired` | 单点触发且关联唯一任务的数量，不等于执行成功 | +| `verified_success` | 任务结果、唯一尝试、来源与时间顺序全部通过的数量 | +| `unobserved` | 预期数减已观察触发点数 | +| `unverified` | 已触发数减已核验成功数 | +| `handler_calls` / `duplicate_invocations` | 有效输入的本地调用数 / 同任务额外调用数 | +| `unconfirmed_invocations` | 尚未对应到已核验任务和同一尝试的本地调用数 | +| `plans` | 原计划身份、创建/恢复/暂停键、版本、生效时间和逐计划计数 | + +核验拒绝错配计划、固定版本变化、非预期时间槽、重复触发/任务、历史消失、分页溢出、跳过/合并、错误输入/结果、额外尝试、来源错配和时间倒置。只有 `SUCCEEDED` 任务与唯一成功尝试一致才计为成功。已通过核验的槽位只计一次，重复轮询不会放大分母。 + +`trigger_lag` 是计划时间到任务受理，`queue_wait` 是受理到开始授权，`authorized_execution` 是授权到结果落库，`completion` 是计划时间到结果落库。这些时间均来自数据库，不混入客户端轮询等待；授权执行耗时包含通信和事务，不是纯函数执行时间。分布只纳入已核验成功样本，空分布百分位为 `null`。 + +`complete=true` 还要求全部计划恢复成功、固定触发范围耗尽、暂停已确认，且每个已核验任务能对应到本地唯一调用/尝试，Worker 启动数一致、无错误或重复。超时、调用过多或收尾不明均不得显示完整通过。 + +## 超时与原身份核查 + +创建和恢复请求不自动重试。创建请求结果不明时，`plans[].id` 可能为空；原创建请求的 `paused=true` 保证它不会自行开始触发。用报告里的 `create_key`、`run_id`、计划序号和时间/策略字段重建原请求，按原键读取回执，不能生成新键替代。收到计划 ID 后的所有退出路径都会尝试暂停。 + +暂停先读当前 revision，再带固定键提交。仅明确收到 `409 STATE_CONFLICT` 时，才重新读版本并使用下一个已记录键，最多三次；超时、断线或其他冲突停止重试，保留最后的 `pause_key` 和 `pause_revision`。恢复原因固定为 `start bounded schedule verification`，暂停原因为 `stop bounded schedule verification`。收尾使用独立上下文，最多 5 秒，多个已知计划并发暂停。 + +`pause_confirmed=true` 只表示停止后续触发，已经生成的任务仍可执行。截止后报告是观察快照，任务可能在工具停止后完成；应沿原计划的触发历史、任务与尝试查询核对。工具不删除计划、任务、操作回执或审计。若创建回执不明、暂停不明或进程被强杀，报告不构成清理完成的证明。 + +## 验证范围 + +复现样本、原始 JSON、构建哈希和实际工具链见[定时测量报告](../reports/phase-6-schedule-load.md)。该工具覆盖固定间隔的同时触发；Cron/DST、其他漏跑和重叠策略、计划版本竞争由原调度功能集成测试覆盖。此次样本不证明这些场景的容量，也不证明长期积压、资源曲线、跨故障域高可用或生产 SLA。 diff --git a/docs/reports/data/phase-6-schedule-load/database.json b/docs/reports/data/phase-6-schedule-load/database.json new file mode 100644 index 0000000..3202421 --- /dev/null +++ b/docs/reports/data/phase-6-schedule-load/database.json @@ -0,0 +1,10 @@ +{ + "mysql": "8.4.10", + "tasks": 158, + "schedules": 38, + "occurrences": 158, + "succeeded_tasks": 154, + "foreign_key_count": 0, + "unpaused_schedules": 0, + "duplicate_tasks_in_occurrences": 0 +} diff --git a/docs/reports/data/phase-6-schedule-load/deadline-24.json b/docs/reports/data/phase-6-schedule-load/deadline-24.json new file mode 100644 index 0000000..dc49a88 --- /dev/null +++ b/docs/reports/data/phase-6-schedule-load/deadline-24.json @@ -0,0 +1,110 @@ +{ + "run_id": "abd51b013ef637e214f3b868d73d1e3c", + "go_version": "go1.25.14", + "goos": "linux", + "goarch": "arm64", + "started_at": "2026-09-20T05:52:03.516120763Z", + "finished_at": "2026-09-20T05:52:08.536435043Z", + "requests_per_schedule": 12, + "schedules": 2, + "expected_occurrences": 24, + "workers": 2, + "capacity_per_worker": 2, + "interval_ms": 1000, + "start_delay_ms": 1000, + "run_timeout_ms": 5000, + "poll_interval_ms": 100, + "http_timeout_ms": 5000, + "queue": { + "max_unfinished_tasks": 100000, + "unfinished_tasks": "0", + "id": "17f6a89f5d68895600d3bb2185fbadb4", + "namespace_id": "65aefc4c4d7c0441cda02794632c7408", + "name": "schedule", + "paused": false, + "max_concurrency": 4, + "rate_per_second": 10, + "revision": "1" + }, + "schedule_start": "2026-09-20T05:52:04.517253Z", + "schedule_end": "2026-09-20T05:52:16.517253Z", + "plans": [ + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:52:03.517994Z", + "index": 0, + "id": "c7d2bc3caea7b7e5ea7403783e98306a", + "create_key": "schedule-abd51b013ef637e214f3b868d73d1e3c-00-create", + "resume_key": "schedule-abd51b013ef637e214f3b868d73d1e3c-00-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": false, + "pause_key": "schedule-abd51b013ef637e214f3b868d73d1e3c-00-pause-0", + "pause_revision": "4", + "pause_confirmed": true, + "occurrences": 3, + "fired": 3, + "verified_success": 1 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:52:03.521008Z", + "index": 1, + "id": "f86e04e63a42f805d5c146b2594d4d0f", + "create_key": "schedule-abd51b013ef637e214f3b868d73d1e3c-01-create", + "resume_key": "schedule-abd51b013ef637e214f3b868d73d1e3c-01-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": false, + "pause_key": "schedule-abd51b013ef637e214f3b868d73d1e3c-01-pause-0", + "pause_revision": "4", + "pause_confirmed": true, + "occurrences": 3, + "fired": 3, + "verified_success": 1 + } + ], + "occurrences": 6, + "fired": 6, + "verified_success": 2, + "unobserved": 18, + "unverified": 4, + "handler_calls": 2, + "duplicate_invocations": 0, + "unconfirmed_invocations": 0, + "worker_started": 2, + "worker_reconnects": 0, + "worker_renewals_confirmed": 0, + "trigger_lag": { + "count": 2, + "p50_ms": 504.563, + "p95_ms": 515.272, + "p99_ms": 515.272, + "max_ms": 515.272 + }, + "queue_wait": { + "count": 2, + "p50_ms": 1530.163, + "p95_ms": 1535.71, + "p99_ms": 1535.71, + "max_ms": 1535.71 + }, + "authorized_execution": { + "count": 2, + "p50_ms": 9.803, + "p95_ms": 20.264, + "p99_ms": 20.264, + "max_ms": 20.264 + }, + "completion": { + "count": 2, + "p50_ms": 2055.238, + "p95_ms": 2060.537, + "p99_ms": 2060.537, + "max_ms": 2060.537 + }, + "errors": { + "RUN_TIMEOUT": 1 + }, + "complete": false +} diff --git a/docs/reports/data/phase-6-schedule-load/environment.json b/docs/reports/data/phase-6-schedule-load/environment.json new file mode 100644 index 0000000..2389783 --- /dev/null +++ b/docs/reports/data/phase-6-schedule-load/environment.json @@ -0,0 +1,113 @@ +{ + "date_local": "2026-09-20", + "timezone": "Asia/Shanghai", + "source_base_commit": "71c33ea", + "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, + "orbit_processes": 1, + "mysql_nodes": 1, + "worker_processes": 1, + "worker_connections": 2, + "queue_max_concurrency": 4, + "queue_rate_per_second": 10, + "metrics_enabled": false, + "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, + "cleanup": "Test-created containers stopped and removed; no existing databases used", + "sdk_checks": [ + "go test -race ./...", + "go vet ./...", + "go build ./...", + "go list -deps ./...; no database/server dependencies" + ], + "modfile_note": "All Go commands use GOWORK=off and temporary sdk.mod/root.mod changing only declared Go version for available toolchain; repository go.mod unchanged", + "sample_commands": [ + [ + "-schedules", + "4", + "-requests", + "6", + "-workers", + "2", + "-capacity", + "4", + "-start-delay", + "5s", + "-run-timeout", + "90s", + "-poll-interval", + "100ms" + ], + [ + "-schedules", + "32", + "-requests", + "4", + "-workers", + "2", + "-capacity", + "4", + "-start-delay", + "5s", + "-run-timeout", + "90s", + "-poll-interval", + "100ms" + ], + [ + "-schedules", + "2", + "-requests", + "12", + "-workers", + "2", + "-capacity", + "2", + "-start-delay", + "1s", + "-run-timeout", + "5s", + "-poll-interval", + "100ms" + ] + ], + "exit_codes": [ + 0, + 0, + 1 + ], + "schema_verification": { + "schema_version": 28, + "migration_count": 28, + "foreign_key_count": 0, + "required_index_count": 6, + "complete": true + }, + "source_sha256": { + "sdk/go/cmd/orbit-schedule-load/handler.go": "560426c5eaaad08cce386b304d0b4673757c1d82978d39557b6a72c8b00f12f8", + "sdk/go/cmd/orbit-schedule-load/main.go": "2378bc59bde8e219618848206f5e4aa3e5158fdb0f419b1b8c286754e5e74136", + "sdk/go/cmd/orbit-schedule-load/measure.go": "93bf30957af66631acc202fb65bab81dab2ea6c40259ad1d619eb47d23935832", + "sdk/go/cmd/orbit-schedule-load/measure_test.go": "bd5cf497c427678edf29d6b48af9e95c410d656470f2c7340de19dc5705e9772", + "sdk/go/cmd/orbit-schedule-load/observe.go": "50d43aa8b4b4690ceac976f3d65135a79fcf17530f5263ff5af5453074ad9e90", + "sdk/go/cmd/orbit-schedule-load/report.go": "5a949c2904c21116d19f219d18a2e81a1153f578d4b2d7907b43bbd522f09b32" + }, + "binary_sha256": { + "orbit": "6290b47a74459cdc9f45c4288df7e0cf16ed67c1bfff7acb4e6e9fd4b636c9d7", + "orbit-admin": "9325990fdedbc5faeb351b1333fd33b27ea35d2d0ee6b59b425deb30e973b396", + "orbit-migrate": "56be44ae4e5d387aa13a6d0cd2e22a533a2ca6d4845b45548e7feac8f0c56635", + "orbit-schedule-load": "67cf4d0ee141e2669744dcce33b032ab8ffd8c463d41662c647840e8ae1510db" + }, + "artifacts_sha256": { + "database.json": "462a9ea2921c1960bf0a4d97d846129af7eb7ba37784acad55b2bfb07cd1c35d", + "deadline-24.json": "035696ab376dae1b5f8bcfa94033a94690d99d38c2bbc4d191299c7d1786457e", + "sample-128.json": "bdfe818a94cfcb6184ca3fc0e057802f7f3cb8c223af43bca4a44488ba444fe5", + "sample-24.json": "af78f339c497f5ddca02560005ece49354a98a1343bd08b3dac25c3fcbbbf3ae", + "schema.json": "9ee118dd4306cddde44582b8d3983f2cd25331148d993672fd70cca86dc06d0a" + } +} diff --git a/docs/reports/data/phase-6-schedule-load/sample-128.json b/docs/reports/data/phase-6-schedule-load/sample-128.json new file mode 100644 index 0000000..a4959c8 --- /dev/null +++ b/docs/reports/data/phase-6-schedule-load/sample-128.json @@ -0,0 +1,618 @@ +{ + "run_id": "7056d28993e47c01647892a53952c7b1", + "go_version": "go1.25.14", + "goos": "linux", + "goarch": "arm64", + "started_at": "2026-09-20T05:51:38.180345251Z", + "finished_at": "2026-09-20T05:52:03.51048843Z", + "requests_per_schedule": 4, + "schedules": 32, + "expected_occurrences": 128, + "workers": 2, + "capacity_per_worker": 4, + "interval_ms": 1000, + "start_delay_ms": 5000, + "run_timeout_ms": 90000, + "poll_interval_ms": 100, + "http_timeout_ms": 5000, + "queue": { + "max_unfinished_tasks": 100000, + "unfinished_tasks": "0", + "id": "17f6a89f5d68895600d3bb2185fbadb4", + "namespace_id": "65aefc4c4d7c0441cda02794632c7408", + "name": "schedule", + "paused": false, + "max_concurrency": 4, + "rate_per_second": 10, + "revision": "1" + }, + "schedule_start": "2026-09-20T05:51:43.181786Z", + "schedule_end": "2026-09-20T05:51:47.181786Z", + "plans": [ + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.182744Z", + "index": 0, + "id": "fcdd4ba396d4eaa0ee0fa972c699b051", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-00-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-00-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-00-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.185432Z", + "index": 1, + "id": "983feafac5dbdb69bb39c7e8b89517b7", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-01-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-01-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-01-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.187835Z", + "index": 2, + "id": "51f0d54d467cef5325122ecda038ce78", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-02-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-02-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-02-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.190392Z", + "index": 3, + "id": "f01d813f12e16b4b47929fe927f1bc4a", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-03-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-03-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-03-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.193883Z", + "index": 4, + "id": "0b498972511743f1c735a26da75a29d7", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-04-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-04-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-04-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.196309Z", + "index": 5, + "id": "97000fdccd6ae61fbebeaec477151714", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-05-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-05-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-05-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.198861Z", + "index": 6, + "id": "4d369519d8fe46cf787f6d1018cb4253", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-06-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-06-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-06-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.203173Z", + "index": 7, + "id": "c834d23a6ec817df44cea6794c6f6639", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-07-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-07-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-07-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.205847Z", + "index": 8, + "id": "8a58343f784b63bc915d11041712b4f2", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-08-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-08-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-08-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.20941Z", + "index": 9, + "id": "c7212dab959cffb639fe39798bf20385", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-09-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-09-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-09-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.211796Z", + "index": 10, + "id": "1c2ae5b9d1583d0607b34fe0ade46621", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-10-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-10-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-10-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.213893Z", + "index": 11, + "id": "cb44442045923c5fe2687f8840c9fb37", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-11-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-11-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-11-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.219422Z", + "index": 12, + "id": "2f2c444436c2ee51d0eefc2e78a2e799", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-12-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-12-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-12-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.222664Z", + "index": 13, + "id": "5a49395323073eff76470d378cc2bc01", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-13-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-13-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-13-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.225018Z", + "index": 14, + "id": "6fe18947edcbd95aa4288b1ef2611124", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-14-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-14-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-14-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.22847Z", + "index": 15, + "id": "582591eb99f06745e3fa07ea54b54006", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-15-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-15-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-15-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.231576Z", + "index": 16, + "id": "829e2428e144ab924715b255091c3757", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-16-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-16-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-16-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.234664Z", + "index": 17, + "id": "3204baf132a3cb7954d59c5a84896504", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-17-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-17-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-17-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.239704Z", + "index": 18, + "id": "85a944a2dbf10940e30e306eeb6207c1", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-18-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-18-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-18-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.242679Z", + "index": 19, + "id": "754ea55797e8574625385c4762fdb346", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-19-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-19-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-19-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.245641Z", + "index": 20, + "id": "b54c4a596e4c900fc915e8845d351e67", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-20-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-20-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-20-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.248371Z", + "index": 21, + "id": "9b6c85c9a8b6a290ce7ff6dd458638eb", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-21-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-21-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-21-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.251722Z", + "index": 22, + "id": "25171ecf0b2b5aa5ac96abd06de41ba0", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-22-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-22-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-22-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.25387Z", + "index": 23, + "id": "902b9ad707391d67f76ba6e70cffcb3f", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-23-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-23-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-23-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.256482Z", + "index": 24, + "id": "4214b53489a8cdb75202b639ee574427", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-24-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-24-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-24-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.258774Z", + "index": 25, + "id": "4d4c51d538861102eac9fa76f4d90df4", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-25-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-25-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-25-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.261484Z", + "index": 26, + "id": "351abd4c1b1cccb07ec1f988c199c2f0", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-26-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-26-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-26-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.263584Z", + "index": 27, + "id": "c3e19c402d45239b04b20a7adc526141", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-27-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-27-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-27-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.26587Z", + "index": 28, + "id": "5652ed089f0834abaad62311c898007f", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-28-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-28-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-28-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.27058Z", + "index": 29, + "id": "2cbab1577b43ae69bf308687f3d67785", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-29-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-29-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-29-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.27287Z", + "index": 30, + "id": "8d3c0d2151c69e4c50dcd476c6737eee", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-30-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-30-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-30-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:38.275023Z", + "index": 31, + "id": "0b90d1c46cd178c49d63e7dfb5125057", + "create_key": "schedule-7056d28993e47c01647892a53952c7b1-31-create", + "resume_key": "schedule-7056d28993e47c01647892a53952c7b1-31-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-7056d28993e47c01647892a53952c7b1-31-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 4, + "fired": 4, + "verified_success": 4 + } + ], + "occurrences": 128, + "fired": 128, + "verified_success": 128, + "unobserved": 0, + "unverified": 0, + "handler_calls": 128, + "duplicate_invocations": 0, + "unconfirmed_invocations": 0, + "worker_started": 128, + "worker_reconnects": 0, + "worker_renewals_confirmed": 0, + "trigger_lag": { + "count": 128, + "p50_ms": 992.747, + "p95_ms": 1964.089, + "p99_ms": 1985.97, + "max_ms": 1992.747 + }, + "queue_wait": { + "count": 128, + "p50_ms": 8192.259, + "p95_ms": 14277.183, + "p99_ms": 15230.422, + "max_ms": 15232.355 + }, + "authorized_execution": { + "count": 128, + "p50_ms": 8.856, + "p95_ms": 20.414, + "p99_ms": 25.195, + "max_ms": 34.262 + }, + "completion": { + "count": 128, + "p50_ms": 9191.23, + "p95_ms": 16200.649, + "p99_ms": 17172.61, + "max_ms": 17175.759 + }, + "errors": {}, + "complete": true +} diff --git a/docs/reports/data/phase-6-schedule-load/sample-24.json b/docs/reports/data/phase-6-schedule-load/sample-24.json new file mode 100644 index 0000000..502d097 --- /dev/null +++ b/docs/reports/data/phase-6-schedule-load/sample-24.json @@ -0,0 +1,142 @@ +{ + "run_id": "278555409a9d68e7de5b635087c73d04", + "go_version": "go1.25.14", + "goos": "linux", + "goarch": "arm64", + "started_at": "2026-09-20T05:51:25.035193676Z", + "finished_at": "2026-09-20T05:51:38.173794626Z", + "requests_per_schedule": 6, + "schedules": 4, + "expected_occurrences": 24, + "workers": 2, + "capacity_per_worker": 4, + "interval_ms": 1000, + "start_delay_ms": 5000, + "run_timeout_ms": 90000, + "poll_interval_ms": 100, + "http_timeout_ms": 5000, + "queue": { + "max_unfinished_tasks": 100000, + "unfinished_tasks": "0", + "id": "17f6a89f5d68895600d3bb2185fbadb4", + "namespace_id": "65aefc4c4d7c0441cda02794632c7408", + "name": "schedule", + "paused": false, + "max_concurrency": 4, + "rate_per_second": 10, + "revision": "1" + }, + "schedule_start": "2026-09-20T05:51:30.039771Z", + "schedule_end": "2026-09-20T05:51:36.039771Z", + "plans": [ + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:25.043209Z", + "index": 0, + "id": "713696f120e328bcedd8b307432745ec", + "create_key": "schedule-278555409a9d68e7de5b635087c73d04-00-create", + "resume_key": "schedule-278555409a9d68e7de5b635087c73d04-00-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-278555409a9d68e7de5b635087c73d04-00-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 6, + "fired": 6, + "verified_success": 6 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:25.05288Z", + "index": 1, + "id": "2c7cc5f7278677e48c1fcb594f762a99", + "create_key": "schedule-278555409a9d68e7de5b635087c73d04-01-create", + "resume_key": "schedule-278555409a9d68e7de5b635087c73d04-01-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-278555409a9d68e7de5b635087c73d04-01-pause-0", + "pause_revision": "6", + "pause_confirmed": true, + "occurrences": 6, + "fired": 6, + "verified_success": 6 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:25.057968Z", + "index": 2, + "id": "147de621f4e54d96d1a0c90fae9bb3ca", + "create_key": "schedule-278555409a9d68e7de5b635087c73d04-02-create", + "resume_key": "schedule-278555409a9d68e7de5b635087c73d04-02-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-278555409a9d68e7de5b635087c73d04-02-pause-0", + "pause_revision": "6", + "pause_confirmed": true, + "occurrences": 6, + "fired": 6, + "verified_success": 6 + }, + { + "calendar_version": "go1.27.1-b2d18a7c8fa8", + "effective_at": "2026-09-20T05:51:25.063146Z", + "index": 3, + "id": "93272d0b600756362052ec09dac56f7f", + "create_key": "schedule-278555409a9d68e7de5b635087c73d04-03-create", + "resume_key": "schedule-278555409a9d68e7de5b635087c73d04-03-resume", + "resume_revision": "1", + "resumed": true, + "exhausted": true, + "pause_key": "schedule-278555409a9d68e7de5b635087c73d04-03-pause-0", + "pause_revision": "5", + "pause_confirmed": true, + "occurrences": 6, + "fired": 6, + "verified_success": 6 + } + ], + "occurrences": 24, + "fired": 24, + "verified_success": 24, + "unobserved": 0, + "unverified": 0, + "handler_calls": 24, + "duplicate_invocations": 0, + "unconfirmed_invocations": 0, + "worker_started": 24, + "worker_reconnects": 0, + "worker_renewals_confirmed": 0, + "trigger_lag": { + "count": 24, + "p50_ms": 999.646, + "p95_ms": 2005.396, + "p99_ms": 2015.96, + "max_ms": 2015.96 + }, + "queue_wait": { + "count": 24, + "p50_ms": 1111.19, + "p95_ms": 2126.701, + "p99_ms": 2130.041, + "max_ms": 2130.041 + }, + "authorized_execution": { + "count": 24, + "p50_ms": 10.082, + "p95_ms": 21.111, + "p99_ms": 27.578, + "max_ms": 27.578 + }, + "completion": { + "count": 24, + "p50_ms": 3059.006, + "p95_ms": 4067.032, + "p99_ms": 4094.876, + "max_ms": 4094.876 + }, + "errors": {}, + "complete": true +} diff --git a/docs/reports/data/phase-6-schedule-load/schema.json b/docs/reports/data/phase-6-schedule-load/schema.json new file mode 100644 index 0000000..718bdc7 --- /dev/null +++ b/docs/reports/data/phase-6-schedule-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-schedule-load/sdk-check.log b/docs/reports/data/phase-6-schedule-load/sdk-check.log new file mode 100644 index 0000000..286e946 --- /dev/null +++ b/docs/reports/data/phase-6-schedule-load/sdk-check.log @@ -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-schedule-load.md b/docs/reports/phase-6-schedule-load.md new file mode 100644 index 0000000..dfebbbb --- /dev/null +++ b/docs/reports/phase-6-schedule-load.md @@ -0,0 +1,42 @@ +# P6 定时集中触发测量 + +日期：2026-09-20。本批新增独立 SDK 命令 `orbit-schedule-load`，完成多个固定间隔计划同一时刻触发、真实 Worker 执行和持久结果核验。P6/P7 仍进行中；样本不作为生产容量或高可用结论。 + +## 已实现与实际检查 + +计划先暂停创建再恢复，固定起止/间隔、版本 1、UTC 时区快照、CATCH_UP/PARALLEL；每任务只有一次尝试。工具核对预期时间槽、触发唯一性、任务身份/模板/结果、成功尝试和时间顺序、固定来源及本地单次调用，并等待计划耗尽。所有退出路径执行有限 Worker 收尾和计划暂停；传输不明不换键重试，保留原操作身份。 + +初次真实连接发现工具误把 API `v1` 当成时区快照标识，修正为保存服务端返回的具体标识和版本生效时间，后续读取核对不变。下表和原始文件全部来自修正后**重新构建并运行**的二进制；首次失败的创建发生在另一已清理的一次性数据库，不计入最终样本。 + +SDK 全模块 `go test -race ./...`、`go vet ./...`、`go build ./...` 通过，完整依赖图没有服务端或 MySQL 驱动依赖。新增单测覆盖等待不算成功、重复轮询不重复计数、错配输入/来源/历史/时间、跨计划重复任务、本地并发重复调用、暂停明确冲突与传输不明的处理差异、原时区快照/生效时间不变及部分分母。标准 `scripts/test-mysql.sh` 已接入工具编译和 2×3 触发样本，脚本语法检查通过；本批没有重新跑整份标准集成脚本。 + +## 环境与结果 + +一次性 MySQL 8.4.10，1 个 Orbit 进程，1 个测量进程中 2 个 Worker 连接。共享本机 Docker VM，12 个 CPU、12,529,274,880 字节内存，无独占资源额度；其他工作负载可能影响结果。队列并发上限 4、每秒开始授权上限 10，工具未修改配置；指标监听关闭。Go/MySQL 容器共享网络命名空间，仅本机回环通信，无宿主机发布端口。 + +实际工具链是 **Go 1.25.14 Linux arm64**，通过临时 modfile 使用，仓库声明的 Go 1.27.1 未改。本批不能替代正式工具链全量验收。精确配置、源码/二进制哈希、退出码和原始文件哈希见[环境清单](data/phase-6-schedule-load/environment.json)，SDK 输出见[检查日志](data/phase-6-schedule-load/sdk-check.log)。 + +| 样本 | 预期触发 | 已观察/生成任务 | 已核验成功 | 本地调用/重复 | 结果 | +| --- | ---: | ---: | ---: | ---: | --- | +| 4计划×6次，1秒间隔 | 24 | 24/24 | 24 | 24/0 | `complete=true`，退出0 | +| 32计划×4次，1秒间隔 | 128 | 128/128 | 128 | 128/0 | `complete=true`，退出0 | +| 2计划×12次，1秒间隔，整次仅5秒 | 24 | 6/6 | 2 | 2/0 | `RUN_TIMEOUT`，`complete=false`，退出1 | + +原始结果：[24次](data/phase-6-schedule-load/sample-24.json)、[128次](data/phase-6-schedule-load/sample-128.json)、[截止负例](data/phase-6-schedule-load/deadline-24.json)。正常组各有两个 Worker，每个容量4；负例每个容量2。正常组预留5秒设置，整体截止90秒；负例预留1秒设置，整体截止5秒。每次HTTP截止5秒，轮询100毫秒。 + +| 正常样本 | 触发延迟P95 | 排队至授权P95 | 授权执行P95 | 计划时间至完成P95 | +| --- | ---: | ---: | ---: | ---: | +| 24次 | 2005.396ms | 2126.701ms | 21.111ms | 4067.032ms | +| 128次 | 1964.089ms | 14277.183ms | 20.414ms | 16200.649ms | + +32 个同时触发的计划在现有并发/速率限制下形成排队，上表记录这次现象；未做控制变量或持续资源采样，不能据此判定瓶颈或推导生产吞吐。耗时分布均由持久时间计算，分母分别为24和128，只包含已核验成功任务。 + +负例保留18个未观察到的计划触发点、4个已经触发但未核验成功的任务，没有把超时隐藏为成功。`pause_confirmed` 在全部计划中为真；暂停不取消已生成任务。测量后独立数据库回读共有38个计划且全部暂停、158个触发记录、158个任务，其中154个成功，触发记录间重复任务数0；其余4个任务没有被该样本核验为成功，不能推测为已撤销或最终失败。见[数据库快照](data/phase-6-schedule-load/database.json)。 + +`orbit-admin verify` 独立确认 schema version28、28条迁移记录、6个必需索引和**实际外键0**，见[结构核验](data/phase-6-schedule-load/schema.json)。本批未修改数据库模型或迁移；不可变旧迁移仍保留历史外键及其前向清理记录。 + +## 复现与下一步 + +按[工具手册](../api/schedule-load.md)创建隔离命名空间/队列，签发 Worker 与计划权限，依次运行环境清单中的三组参数；前两组应退出0，第三组应退出1并报告剩余分母。使用新创建的一次性MySQL、当前源码构建的服务端和测量工具，最终执行 `orbit-admin verify`。校验时不得把源码修改后的哈希附到之前构建的二进制结果。 + +当前只验证同节点、固定间隔、有限计划数的集中触发和超时收尾。Cron/DST及其他策略的功能测试已有独立记录，本批未做这些场景的容量测量。下一步继续长期积压、资源曲线、生产采集/通知、目标数据库拓扑与正式工具链/版本兼容门禁。 diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index 3cf79b6..cbb4be2 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -20,6 +20,8 @@ | 真实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；生产容量未验收 | +| 真实定时集中触发测量 | `orbit-schedule-load`；4×6及32×4触发，2×12截止负例，见[定时测量报告](phase-6-schedule-load.md) | 正常152项逐项核验；负例18个未观察点、4个未核验任务，非零退出；38个计划均暂停，实际库外键0。SDK全模块竞态/vet/build通过；生产容量未验收 | + ## 当前环境仍未覆盖 此前使用容器网络别名运行存储集成时，两组 COMMIT 故障测试曾因自身 `127.0.0.1` 明文连接前置条件被拒绝；本轮已改用共享网络命名空间单独复跑并通过，因此不再把它们列为未验证场景。共享网络命名空间只解决测试进程的 loopback 观察条件，不改变生产部署约束。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 0fe1399..55d3cd6 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -48,6 +48,7 @@ ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" "$TASK_GO" test -race -count=1 - (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) +(cd sdk/go && GOWORK=off "$TASK_GO" build -o "$TASK_TEST_DIR/orbit-schedule-load" ./cmd/orbit-schedule-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) @@ -97,6 +98,7 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/retry) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/delayed-task) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/schedules) +(cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_TEST_DIR/orbit-schedule-load" -schedules 2 -requests 3 -workers 2 -capacity 2 -run-timeout 30s -poll-interval 100ms) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/cancellation) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/resources) (cd sdk/go && ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./examples/batch) diff --git a/sdk/go/cmd/orbit-schedule-load/handler.go b/sdk/go/cmd/orbit-schedule-load/handler.go new file mode 100644 index 0000000..c6aeb99 --- /dev/null +++ b/sdk/go/cmd/orbit-schedule-load/handler.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "encoding/json" + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "sync" +) + +// invocation 固定首次调用的任务/尝试身份，重复调用不得覆盖首条证据。 +type invocation struct { + index int // index 输入中的计划序号。 + 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: 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) { + payload, ok := decodePayload(e.Task.Payload) + if !ok || payload.Schedule < 0 || payload.Schedule >= c.Schedules || !matchesTask(e.Task, c, runID, payload.Schedule) || !validID(e.AttemptID) || e.Epoch != 1 || e.ConcurrencyToken != 0 { + l.mu.Lock() + l.invalid++ + l.mu.Unlock() + return nil, &orbit.WorkerFailure{Code: "SCHEDULE_INPUT_MISMATCH"} + } + l.mu.Lock() + existing := l.calls[e.Task.ID] + if existing != nil { + existing.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{index: payload.Schedule, 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 *scheduleReport, o *observations) { + l.mu.Lock() + defer l.mu.Unlock() + r.Errors["INVALID_HANDLER_INPUT"] += l.invalid + r.Errors["HANDLER_IDENTITY_LIMIT"] += l.overflow + for taskID, call := range l.calls { + r.HandlerCalls += call.calls + if call.calls > 1 { + r.DuplicateInvocations += call.calls - 1 + } + s := o.tasks[taskID] + if s == nil || !s.verified || s.attemptID != call.attemptID || o.plans[s.occurrence.ScheduleID].index != call.index { + r.UnconfirmedInvocations += call.calls + } + } + for taskID, s := range o.tasks { + if s.verified && l.calls[taskID] == nil { + r.Errors["MISSING_LOCAL_INVOCATION"]++ + } + } +} diff --git a/sdk/go/cmd/orbit-schedule-load/main.go b/sdk/go/cmd/orbit-schedule-load/main.go new file mode 100644 index 0000000..7f149c5 --- /dev/null +++ b/sdk/go/cmd/orbit-schedule-load/main.go @@ -0,0 +1,153 @@ +// orbit-schedule-load 在隔离环境测量固定间隔计划的持久化触发、真实Worker执行和来源核对。 +// 计划模板只包含固定JSON输入，处理器没有外部业务副作用。 +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, Schedules, Workers, Capacity int + Interval, Timeout, RunTimeout time.Duration + PollInterval, StartDelay 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("schedule 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: 6, Schedules: 1, StartDelay: 5 * time.Second, Workers: 2, Capacity: 2, Interval: time.Second, Timeout: 5 * time.Second, RunTimeout: 2 * time.Minute, PollInterval: 250 * time.Millisecond} + flags := flag.NewFlagSet("orbit-schedule-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, "expected occurrences per schedule, 1..100") + flags.IntVar(&cfg.Schedules, "schedules", cfg.Schedules, "schedules sharing the same trigger times, 1..32") + flags.DurationVar(&cfg.StartDelay, "start-delay", cfg.StartDelay, "lead time for setup, 1s..1m; late setup fails the sample") + flags.IntVar(&cfg.Workers, "workers", cfg.Workers, "independent WSS Workers in this process, 1..16") + flags.IntVar(&cfg.Capacity, "capacity", cfg.Capacity, "concurrent handlers per Worker, 1..128; total at most 256") + flags.DurationVar(&cfg.Interval, "interval", cfg.Interval, "fixed-rate interval in whole milliseconds, 1s..1m") + flags.DurationVar(&cfg.Timeout, "timeout", cfg.Timeout, "per HTTP request timeout, 1ms..1m") + flags.DurationVar(&cfg.RunTimeout, "run-timeout", cfg.RunTimeout, "whole sample deadline, 5s..1h, plus at most 5s Worker shutdown and 5s schedule cleanup") + flags.DurationVar(&cfg.PollInterval, "poll-interval", cfg.PollInterval, "occurrence 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 > 100 || cfg.Schedules < 1 || cfg.Schedules > 32 || cfg.Workers < 1 || cfg.Workers > 16 || cfg.Capacity < 1 || cfg.Capacity > 128 || cfg.Workers*cfg.Capacity > 256 { + return config{}, errors.New("request or Worker count is out of bounds") + } + if cfg.Interval < time.Second || cfg.Interval > time.Minute || cfg.Interval%time.Millisecond != 0 || cfg.StartDelay < time.Second || cfg.StartDelay > time.Minute || cfg.StartDelay%time.Millisecond != 0 || cfg.Timeout < time.Millisecond || cfg.Timeout > time.Minute || cfg.RunTimeout < 5*time.Second || cfg.RunTimeout > time.Hour || cfg.PollInterval < 100*time.Millisecond || cfg.PollInterval > 10*time.Second { + return config{}, errors.New("interval, timeout or poll interval is out of bounds") + } + 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-schedule-load/measure.go b/sdk/go/cmd/orbit-schedule-load/measure.go new file mode 100644 index 0000000..7027f0c --- /dev/null +++ b/sdk/go/cmd/orbit-schedule-load/measure.go @@ -0,0 +1,351 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "reflect" + "sync" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +// schedulePayload 冻结运行和计划序号；精确字段集合用于检测错配输入。 +type schedulePayload struct { + RunID string `json:"run_id"` // RunID 本次测量身份，不是业务幂等保证。 + Schedule int `json:"schedule"` // Schedule 从零开始的计划序号。 +} + +func decodePayload(raw json.RawMessage) (schedulePayload, bool) { + // map中的RawMessage保留整数文本，不接受缺字段、null和非精确字段名。 + var fields map[string]json.RawMessage + var out schedulePayload + if json.Unmarshal(raw, &fields) != nil || len(fields) != 2 { + return out, false + } + run, hasRun := fields["run_id"] + index, hasIndex := fields["schedule"] + if !hasRun || !hasIndex || string(run) == "null" || string(index) == "null" || json.Unmarshal(run, &out.RunID) != nil || json.Unmarshal(index, &out.Schedule) != nil { + return schedulePayload{}, false + } + // 按原文计数，拒绝重复键；字段次序可以不同。 + var keys []string + decoder := json.NewDecoder(bytes.NewReader(raw)) + if _, err := decoder.Token(); err != nil { + return schedulePayload{}, false + } + for decoder.More() { + k, err := decoder.Token() + if err != nil { + return schedulePayload{}, false + } + keys = append(keys, k.(string)) + var value json.RawMessage + if decoder.Decode(&value) != nil { + return schedulePayload{}, false + } + } + return out, len(keys) == 2 +} + +func handlerName(runID string) string { return "load.schedule." + runID } +func operationKey(runID string, index int, op string) string { + return fmt.Sprintf("schedule-%s-%02d-%s", runID, index, op) +} +func businessKey(runID string, index int) string { return operationKey(runID, index, "task") } + +type scheduleAPI interface { + GetQueue(context.Context, string) (orbit.Queue, error) + CreateSchedule(context.Context, string, orbit.CreateScheduleRequest) (orbit.ScheduleResult, error) + GetSchedule(context.Context, string) (orbit.Schedule, error) + ResumeSchedule(context.Context, string, string, orbit.ControlScheduleRequest) (orbit.ScheduleResult, error) + PauseSchedule(context.Context, string, string, orbit.ControlScheduleRequest) (orbit.ScheduleResult, error) + ListScheduleOccurrences(context.Context, string, orbit.SchedulePageRequest) (orbit.ScheduleOccurrencePage, error) + GetTask(context.Context, string) (orbit.Task, error) + ListTaskAttempts(context.Context, string, orbit.HistoryPageRequest) (orbit.AttemptPage, error) + GetTaskSource(context.Context, string) (orbit.TaskSource, error) +} + +func scheduleDefinition(cfg config, runID string, index int, start time.Time) orbit.ScheduleDefinition { + end := start.Add(time.Duration(cfg.Requests) * cfg.Interval) + payload, _ := json.Marshal(schedulePayload{RunID: runID, Schedule: index}) + return orbit.ScheduleDefinition{ + Rule: orbit.ScheduleRule{Kind: "FIXED_RATE", Timezone: "UTC", StartAt: start, EndAt: &end, IntervalMS: uint64(cfg.Interval.Milliseconds())}, + Policy: orbit.SchedulePolicy{Misfire: "CATCH_UP", Overlap: "PARALLEL", CatchupLimit: 100, CatchupWindowMS: uint64((cfg.RunTimeout + time.Duration(cfg.Requests)*cfg.Interval + cfg.StartDelay).Milliseconds()), GraceMS: 1000}, + Task: orbit.SubmitTaskRequest{QueueID: cfg.QueueID, Handler: handlerName(runID), HandlerVersion: "v1", PayloadVersion: "v1", BusinessKey: businessKey(runID, index), Payload: payload, TaskTimeoutMS: uint64(cfg.RunTimeout.Milliseconds()), RecoverySafety: orbit.RecoveryReconcileOnly, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(cfg.RunTimeout.Milliseconds())}}, + } +} + +func measure(parent context.Context, cfg config, api *orbit.Client, runID string) (r scheduleReport) { + r = newReport(cfg, runID) + ctx, cancel := context.WithTimeout(parent, cfg.RunTimeout) + defer cancel() + var workers []*orbit.Worker + var stopWorkers context.CancelFunc + var done chan error + log := newInvocationLog(cfg.Requests * cfg.Schedules) + observations := newObservations(cfg, runID) + // 所有返回路径都执行有界收尾；超时后的暂停使用独立上下文，结果不明不换键重试。 + defer func() { + if stopWorkers != nil { + stopWorkers() + for range workers { + if err := <-done; err != nil { + r.Errors["WORKER_STOP"]++ + } + } + } + for _, w := range workers { + s := w.Stats() + r.WorkerStarted += s.Started + r.Reconnects += s.Reconnects + r.Renewals += s.RenewalsConfirmed + } + cleanup, stop := context.WithTimeout(context.Background(), 5*time.Second) + var wg sync.WaitGroup + codes := make(chan string, len(r.Plans)) + for i := range r.Plans { + wg.Add(1) + go func(p *planReport) { + defer wg.Done() + if validID(p.ID) { + codes <- pauseSchedule(cleanup, cfg, api, runID, p) + } + }(&r.Plans[i]) + } + wg.Wait() + stop() + close(codes) + for code := range codes { + if code != "" { + r.Errors[code]++ + } + } + observations.summarize(&r) + log.summarize(&r, observations) + if ctx.Err() != nil { + r.Errors["RUN_"+classify(ctx.Err())]++ + } + r = r.finish() + }() + call, stop := context.WithTimeout(ctx, cfg.Timeout) + queue, err := api.GetQueue(call, cfg.QueueID) + stop() + if err != nil { + r.Errors["PREFLIGHT_"+classify(err)]++ + return + } + r.Queue = &queue + if queue.ID != cfg.QueueID || !validID(queue.NamespaceID) || queue.Paused || queue.MaxConcurrency == 0 || queue.RatePerSecond == 0 { + r.Errors["QUEUE_UNAVAILABLE"]++ + return + } + r.ScheduleStart = time.Now().UTC().Truncate(time.Microsecond).Add(cfg.StartDelay) + r.ScheduleEnd = r.ScheduleStart.Add(time.Duration(cfg.Requests) * cfg.Interval) + // 在恢复任何计划之前完成全部创建；计划有限终止，即使进程被强杀也不会无限触发。 + seen := map[string]bool{} + for i := 0; i < cfg.Schedules; i++ { + if ctx.Err() != nil { + return + } + definition := scheduleDefinition(cfg, runID, i, r.ScheduleStart) + key := operationKey(runID, i, "create") + p := planReport{Index: i, CreateKey: key} + r.Plans = append(r.Plans, p) + call, stop = context.WithTimeout(ctx, cfg.Timeout) + res, e := api.CreateSchedule(call, key, orbit.CreateScheduleRequest{Name: key, Paused: true, Definition: definition}) + stop() + if e != nil { + r.Errors["CREATE_"+classify(e)]++ + return + } + p.ID = res.Schedule.ID + p.CalendarVersion = res.Schedule.Current.Definition.Rule.CalendarVersion + p.EffectiveAt = res.Schedule.Current.EffectiveAt + r.Plans[i] = p + if res.Deduplicated || !validID(p.ID) || seen[p.ID] || !matchesSchedule(res.Schedule, cfg, runID, i, queue.NamespaceID, r.ScheduleStart) || !res.Schedule.Paused || res.Schedule.Current.EffectiveAt.After(r.ScheduleStart) { + r.Errors["CREATE_MISMATCH"]++ + return + } + seen[p.ID] = true + observations.addPlan(p.ID, i, r.ScheduleStart, queue.NamespaceID) + } + for i := 0; i < cfg.Workers; i++ { + w, 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 + } + workers = append(workers, w) + } + workerCtx, workerCancel := context.WithCancel(ctx) + stopWorkers = workerCancel + done = make(chan error, len(workers)) + for _, w := range workers { + go func(w *orbit.Worker) { + err := w.Run(workerCtx) + if workerCtx.Err() == nil { + cancel() + } + done <- err + }(w) + } + for i := range r.Plans { + p := &r.Plans[i] + call, stop = context.WithTimeout(ctx, cfg.Timeout) + current, e := api.GetSchedule(call, p.ID) + stop() + if e != nil { + r.Errors["RESUME_READ_"+classify(e)]++ + return + } + if !time.Now().Before(r.ScheduleStart) || current.ID != p.ID || !matchesSchedule(current, cfg, runID, i, queue.NamespaceID, r.ScheduleStart) || !current.Paused || !sameVersion(current, *p) { + r.Errors["SETUP_LATE_OR_CHANGED"]++ + return + } + p.ResumeKey = operationKey(runID, i, "resume") + p.ResumeRevision = current.Revision + call, stop = context.WithTimeout(ctx, cfg.Timeout) + res, e := api.ResumeSchedule(call, p.ID, p.ResumeKey, orbit.ControlScheduleRequest{ExpectedRevision: current.Revision, Reason: "start bounded schedule verification"}) + stop() + if e != nil { + r.Errors["RESUME_"+classify(e)]++ + return + } + if res.Deduplicated || res.Schedule.ID != p.ID || res.Schedule.Paused || !matchesSchedule(res.Schedule, cfg, runID, i, queue.NamespaceID, r.ScheduleStart) || !res.Schedule.UpdatedAt.Before(r.ScheduleStart) || !sameVersion(res.Schedule, *p) { + r.Errors["RESUME_MISMATCH_OR_LATE"]++ + return + } + p.Resumed = true + } + for { + if ctx.Err() != nil { + return + } + all := true + for i := range r.Plans { + p := &r.Plans[i] + call, stop = context.WithTimeout(ctx, cfg.Timeout) + page, e := api.ListScheduleOccurrences(call, p.ID, orbit.SchedulePageRequest{Limit: 100}) + stop() + if e != nil { + r.Errors["OCCURRENCES_"+classify(e)]++ + return + } + if code := observations.ingest(p.ID, page); code != "" { + r.Errors[code]++ + return + } + complete, code := observations.verify(ctx, api, p.ID) + if code != "" { + r.Errors[code]++ + return + } + if complete { + call, stop = context.WithTimeout(ctx, cfg.Timeout) + current, e := api.GetSchedule(call, p.ID) + stop() + if e != nil { + r.Errors["FINAL_READ_"+classify(e)]++ + return + } + if current.ID != p.ID || !matchesSchedule(current, cfg, runID, i, queue.NamespaceID, r.ScheduleStart) || current.Paused || !sameVersion(current, *p) { + r.Errors["FINAL_SCHEDULE_MISMATCH"]++ + return + } + p.Exhausted = current.CheckAt == nil && current.NextFireAt == nil && current.ProcessingVersion == 1 + } + all = all && complete && p.Exhausted + } + if all { + return + } + timer := time.NewTimer(cfg.PollInterval) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } +} + +func matchesSchedule(s orbit.Schedule, cfg config, runID string, index int, namespace string, start time.Time) bool { + want := scheduleDefinition(cfg, runID, index, start) + got := s.Current.Definition + // 服务端冻结具体时区快照标识；不把API版本v1误当作时区数据版本。 + if !validCalendarVersion(got.Rule.CalendarVersion) { + return false + } + got.Rule.CalendarVersion = "" + return s.NamespaceID == namespace && s.QueueID == cfg.QueueID && s.Name == operationKey(runID, index, "create") && s.CurrentVersion == 1 && s.Current.Version == 1 && s.Revision > 0 && !s.Current.EffectiveAt.IsZero() && sameDefinition(got, want) +} + +func sameDefinition(a, b orbit.ScheduleDefinition) bool { + // 原始JSON可有空白差异；模板和结果保持精确整数，不经float64。 + pa, ok := decodePayload(a.Task.Payload) + pb, ok2 := decodePayload(b.Task.Payload) + a.Task.Payload = nil + b.Task.Payload = nil + return ok && ok2 && pa == pb && reflect.DeepEqual(a, b) +} + +func pauseSchedule(ctx context.Context, cfg config, api scheduleAPI, runID string, p *planReport) string { + for attempt := 0; attempt < 3; attempt++ { + call, stop := context.WithTimeout(ctx, cfg.Timeout) + current, err := api.GetSchedule(call, p.ID) + stop() + if err != nil { + return "PAUSE_READ_" + classify(err) + } + if current.ID != p.ID || current.Name != operationKey(runID, p.Index, "create") || current.QueueID != cfg.QueueID { + return "PAUSE_IDENTITY_MISMATCH" + } + if current.Paused { + p.PauseConfirmed = true + return "" + } + p.PauseKey = operationKey(runID, p.Index, fmt.Sprintf("pause-%d", attempt)) + p.PauseRevision = current.Revision + call, stop = context.WithTimeout(ctx, cfg.Timeout) + res, err := api.PauseSchedule(call, p.ID, p.PauseKey, orbit.ControlScheduleRequest{ExpectedRevision: p.PauseRevision, Reason: "stop bounded schedule verification"}) + stop() + if err == nil { + if res.Schedule.ID != p.ID || !res.Schedule.Paused { + return "PAUSE_RECEIPT_MISMATCH" + } + p.PauseConfirmed = true + return "" + } + var apiErr *orbit.APIError + if !errors.As(err, &apiErr) || apiErr.Code != "STATE_CONFLICT" || apiErr.StatusCode != 409 { + return "PAUSE_" + classify(err) + } + timer := time.NewTimer(25 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + return "PAUSE_" + classify(ctx.Err()) + case <-timer.C: + } + } + return "PAUSE_CONFLICT_LIMIT" +} + +func validCalendarVersion(v string) bool { + if len(v) == 0 || len(v) > 64 { + return false + } + for _, c := range v { + if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '.' || c == '-' || c == '_') { + return false + } + } + return true +} +func sameVersion(s orbit.Schedule, p planReport) bool { + return s.Current.Definition.Rule.CalendarVersion == p.CalendarVersion && s.Current.EffectiveAt.Equal(p.EffectiveAt) +} diff --git a/sdk/go/cmd/orbit-schedule-load/measure_test.go b/sdk/go/cmd/orbit-schedule-load/measure_test.go new file mode 100644 index 0000000..9b1c31d --- /dev/null +++ b/sdk/go/cmd/orbit-schedule-load/measure_test.go @@ -0,0 +1,313 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "reflect" + "sync" + "testing" + "time" +) + +func testID(n int) string { return fmt.Sprintf("%032x", n) } +func testConfig() config { + return config{QueueID: testID(1), Requests: 2, Schedules: 1, Workers: 1, Capacity: 2, StartDelay: 5 * time.Second, Interval: time.Second, RunTimeout: time.Minute, Timeout: time.Second, PollInterval: 100 * time.Millisecond} +} +func TestDecodePayloadStrict(t *testing.T) { + want := schedulePayload{RunID: "run", Schedule: 0} + for _, raw := range []string{`{"run_id":"run","schedule":0}`, ` {"schedule":0,"run_id":"run"} `} { + if got, ok := decodePayload(json.RawMessage(raw)); !ok || got != want { + t.Fatalf("valid payload rejected %s", raw) + } + } + for _, raw := range []string{`{}`, `null`, `[]`, `{"run_id":"run"}`, `{"schedule":0}`, `{"run_id":null,"schedule":0}`, `{"run_id":"run","schedule": null }`, `{"run_id":"run","schedule":0.0}`, `{"run_id":"run","schedule":0,"extra":1}`, `{"run_id":"run","Schedule":0}`, `{"run_id":"run","schedule":0,"schedule":0}`, `{"run_id":"run","schedule":0} {}`} { + if _, ok := decodePayload(json.RawMessage(raw)); ok { + t.Fatalf("invalid payload accepted %s", raw) + } + } +} + +// memoryAPI 仅为观察/收尾提供持久快照替身；未使用方法通过嵌入接口留空。 +type memoryAPI struct { + scheduleAPI + task orbit.Task + history orbit.AttemptPage + source orbit.TaskSource + current orbit.Schedule + reads, pauses int + keys []string + pauseError error + conflictOnce bool +} + +func (m *memoryAPI) GetTask(context.Context, string) (orbit.Task, error) { + m.reads++ + return m.task, nil +} +func (m *memoryAPI) ListTaskAttempts(context.Context, string, orbit.HistoryPageRequest) (orbit.AttemptPage, error) { + return m.history, nil +} +func (m *memoryAPI) GetTaskSource(context.Context, string) (orbit.TaskSource, error) { + return m.source, nil +} +func (m *memoryAPI) GetSchedule(context.Context, string) (orbit.Schedule, error) { + return m.current, nil +} +func (m *memoryAPI) PauseSchedule(_ context.Context, _ string, key string, in orbit.ControlScheduleRequest) (orbit.ScheduleResult, error) { + m.pauses++ + m.keys = append(m.keys, key) + if m.pauseError != nil { + return orbit.ScheduleResult{}, m.pauseError + } + if m.conflictOnce && m.pauses == 1 { + m.current.Revision++ + return orbit.ScheduleResult{}, &orbit.APIError{StatusCode: 409, Code: "STATE_CONFLICT"} + } + if in.ExpectedRevision != m.current.Revision { + return orbit.ScheduleResult{}, errors.New("stale revision") + } + m.current.Paused = true + return orbit.ScheduleResult{Schedule: m.current}, nil +} +func fixture() (*observations, *memoryAPI, orbit.ScheduleOccurrencePage) { + c := testConfig() + c.Requests = 1 + start := time.Date(2026, 9, 20, 0, 0, 0, 0, time.UTC) + id, taskID := testID(2), testID(3) + o := newObservations(c, "run") + o.addPlan(id, 0, start, testID(4)) + item := orbit.ScheduleOccurrence{ID: testID(5), ScheduleID: id, ScheduleVersion: 1, ScheduledAt: start, Decision: "FIRED", TaskID: &taskID, Reason: "SCHEDULED", CreatedAt: start.Add(time.Second)} + raw, _ := json.Marshal(schedulePayload{RunID: "run", Schedule: 0}) + authorized, end := start.Add(2*time.Second), start.Add(3*time.Second) + task := orbit.Task{ID: taskID, NamespaceID: testID(4), QueueID: c.QueueID, State: orbit.TaskSucceeded, Handler: handlerName("run"), HandlerVersion: "v1", PayloadVersion: "v1", BusinessKey: businessKey("run", 0), Payload: raw, Result: raw, RecoverySafety: orbit.RecoveryReconcileOnly, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(c.RunTimeout.Milliseconds())}, AttemptCount: 1, CreatedAt: item.CreatedAt, AvailableAt: item.CreatedAt, DeadlineAt: item.CreatedAt.Add(c.RunTimeout), CompletedAt: &end} + attempt := orbit.AttemptSummary{ID: testID(6), TaskID: taskID, Epoch: 1, State: orbit.AttemptSucceeded, AuthorizedAt: &authorized, EndedAt: &end, CreatedAt: item.CreatedAt} + source := orbit.TaskSource{TaskID: taskID, Kind: "SCHEDULE", Schedule: &orbit.TaskScheduleSource{ScheduleID: id, ScheduleVersion: 1, OccurrenceID: item.ID, ScheduledAt: start, Decision: "FIRED"}} + return o, &memoryAPI{task: task, history: orbit.AttemptPage{Items: []orbit.AttemptSummary{attempt}}, source: source}, orbit.ScheduleOccurrencePage{Items: []orbit.ScheduleOccurrence{item}} +} +func TestObserverWaitsAndCountsOnce(t *testing.T) { + o, api, page := fixture() + id := page.Items[0].ScheduleID + api.task.State = orbit.TaskRunning + if code := o.ingest(id, page); code != "" { + t.Fatal(code) + } + if complete, code := o.verify(context.Background(), api, id); complete || code != "" { + t.Fatalf("running counted as success %t %s", complete, code) + } + api.task.State = orbit.TaskSucceeded + for i := 0; i < 3; i++ { + if code := o.ingest(id, page); code != "" { + t.Fatal(code) + } + if complete, code := o.verify(context.Background(), api, id); !complete || code != "" { + t.Fatalf("verify %t %s", complete, code) + } + } + if api.reads != 2 { + t.Fatalf("verified task reread %d", api.reads) + } + r := newReport(o.cfg, "run") + r.Plans = []planReport{{ID: id, Resumed: true, Exhausted: true, PauseConfirmed: true}} + o.summarize(&r) + r = r.finish() + if r.Occurrences != 1 || r.Fired != 1 || r.VerifiedSuccess != 1 || r.Completion.Count != 1 || *r.Completion.Max != 3000 || *r.TriggerLag.Max != 1000 || *r.QueueWait.Max != 1000 || *r.Execution.Max != 1000 || r.Complete { + t.Fatalf("incorrect report %#v", r) + } + log := newInvocationLog(1) + _, err := log.handler(context.Background(), o.cfg, "run").Handle(context.Background(), orbit.WorkerExecution{Task: api.task, AttemptID: api.history.Items[0].ID, Epoch: 1}) + if err != nil { + t.Fatal(err) + } + log.summarize(&r, o) + r.WorkerStarted = 1 + r = r.finish() + if !r.Complete { + t.Fatalf("full evidence not recognized: %#v", r) + } + r.Plans[0].PauseConfirmed = false + if r.finish().Complete { + t.Fatal("unconfirmed cleanup accepted") + } +} +func TestObserverRejectsMalformedSuccess(t *testing.T) { + tests := map[string]func(*memoryAPI){ + "wrong queue": func(m *memoryAPI) { m.task.QueueID = testID(90) }, + "wrong namespace": func(m *memoryAPI) { m.task.NamespaceID = testID(90) }, + "missing result field": func(m *memoryAPI) { m.task.Result = json.RawMessage(`{"run_id":"run"}`) }, + "wrong input": func(m *memoryAPI) { m.task.Payload = json.RawMessage(`{"run_id":"else","schedule":0}`) }, + "uncertain": func(m *memoryAPI) { m.task.EffectUncertain = true }, + "retry policy": func(m *memoryAPI) { m.task.Retry.MaxAttempts = 2 }, + "extra attempt": func(m *memoryAPI) { m.history.Items = append(m.history.Items, m.history.Items[0]) }, + "attempt cursor": func(m *memoryAPI) { m.history.NextCursor = "next" }, + "wrong attempt owner": func(m *memoryAPI) { m.history.Items[0].TaskID = testID(90) }, + "missing authorization": func(m *memoryAPI) { m.history.Items[0].AuthorizedAt = nil }, + "completion reversed": func(m *memoryAPI) { v := m.task.CreatedAt.Add(-time.Second); m.history.Items[0].EndedAt = &v }, + "source task": func(m *memoryAPI) { m.source.TaskID = testID(90) }, + "source occurrence": func(m *memoryAPI) { m.source.Schedule.OccurrenceID = testID(90) }, + "source version": func(m *memoryAPI) { m.source.Schedule.ScheduleVersion = 2 }, + "source time": func(m *memoryAPI) { m.source.Schedule.ScheduledAt = m.source.Schedule.ScheduledAt.Add(time.Second) }, + "source decision": func(m *memoryAPI) { m.source.Schedule.Decision = "COALESCED" }, + } + for name, change := range tests { + t.Run(name, func(t *testing.T) { + o, api, page := fixture() + id := page.Items[0].ScheduleID + if code := o.ingest(id, page); code != "" { + t.Fatal(code) + } + change(api) + if ok, code := o.verify(context.Background(), api, id); ok || code == "" { + t.Fatal("malformed success accepted") + } + if o.plans[id].slots[0].verified { + t.Fatal("invalid proof counted") + } + }) + } +} +func TestOccurrenceUniquenessAndHistory(t *testing.T) { + for name, change := range map[string]func(*orbit.ScheduleOccurrencePage){ + "wrong parent": func(p *orbit.ScheduleOccurrencePage) { p.Items[0].ScheduleID = testID(90) }, + "wrong version": func(p *orbit.ScheduleOccurrencePage) { p.Items[0].ScheduleVersion = 2 }, + "off slot": func(p *orbit.ScheduleOccurrencePage) { + p.Items[0].ScheduledAt = p.Items[0].ScheduledAt.Add(time.Millisecond) + }, + "skipped": func(p *orbit.ScheduleOccurrencePage) { p.Items[0].Decision = "SKIPPED" }, + "cursor": func(p *orbit.ScheduleOccurrencePage) { p.NextCursor = "more" }, + "duplicate": func(p *orbit.ScheduleOccurrencePage) { p.Items = append(p.Items, p.Items[0]) }, + } { + t.Run(name, func(t *testing.T) { + o, _, page := fixture() + id := page.Items[0].ScheduleID + change(&page) + if o.ingest(id, page) == "" { + t.Fatal("invalid page accepted") + } + }) + } + o, _, page := fixture() + id := page.Items[0].ScheduleID + if code := o.ingest(id, page); code != "" { + t.Fatal(code) + } + if o.ingest(id, orbit.ScheduleOccurrencePage{Items: []orbit.ScheduleOccurrence{}}) != "OCCURRENCE_DISAPPEARED" { + t.Fatal("lost history accepted") + } + changed := page + changed.Items = append([]orbit.ScheduleOccurrence(nil), page.Items...) + changed.Items[0].ID = testID(91) + if o.ingest(id, changed) != "OCCURRENCE_CHANGED" { + t.Fatal("replaced slot accepted") + } + second := testID(92) + o.addPlan(second, 1, o.plans[id].start, testID(4)) + changed.Items[0].ScheduleID = second + if o.ingest(second, changed) != "DUPLICATE_TASK" { + t.Fatal("task shared by two slots accepted") + } +} +func TestHandlerDetectsDuplicateAndUnconfirmedCalls(t *testing.T) { + o, api, page := fixture() + id := page.Items[0].ScheduleID + if code := o.ingest(id, page); code != "" { + t.Fatal(code) + } + log := newInvocationLog(1) + handler := log.handler(context.Background(), o.cfg, "run") + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, _ = handler.Handle(context.Background(), orbit.WorkerExecution{Task: api.task, AttemptID: testID(6), Epoch: 1}) + }() + } + wg.Wait() + r := newReport(o.cfg, "run") + log.summarize(&r, o) + if r.HandlerCalls != 4 || r.DuplicateInvocations != 3 || r.UnconfirmedInvocations != 4 { + t.Fatalf("calls lost: %#v", r) + } + if ok, code := o.verify(context.Background(), api, id); !ok || code != "" { + t.Fatal(code) + } + r = newReport(o.cfg, "run") + log.summarize(&r, o) + if r.UnconfirmedInvocations != 0 || r.DuplicateInvocations != 3 { + t.Fatal("duplicate hidden after verification") + } +} +func TestPauseOnlyRekeysExplicitRevisionConflict(t *testing.T) { + cfg := testConfig() + current := orbit.Schedule{ScheduleSummary: orbit.ScheduleSummary{ID: testID(2), QueueID: cfg.QueueID, Name: operationKey("run", 0, "create"), Revision: 7}} + m := &memoryAPI{current: current, conflictOnce: true} + p := planReport{ID: current.ID} + if code := pauseSchedule(context.Background(), cfg, m, "run", &p); code != "" { + t.Fatal(code) + } + if m.pauses != 2 || !p.PauseConfirmed || p.PauseRevision != 8 || !reflect.DeepEqual(m.keys, []string{operationKey("run", 0, "pause-0"), operationKey("run", 0, "pause-1")}) { + t.Fatalf("conflict recovery lost identity %#v %#v", p, m) + } + for _, err := range []error{context.DeadlineExceeded, errors.New("transport lost"), &orbit.APIError{StatusCode: 409, Code: "IDEMPOTENCY_CONFLICT"}} { + m = &memoryAPI{current: current, pauseError: err} + p = planReport{ID: current.ID} + if code := pauseSchedule(context.Background(), cfg, m, "run", &p); code == "" || m.pauses != 1 || p.PauseConfirmed || p.PauseKey == "" || p.PauseRevision != 7 { + t.Fatalf("unknown outcome reissued: %#v %#v", p, m) + } + } +} +func TestScheduleDefinitionFrozenBoundedAndCompared(t *testing.T) { + cfg := testConfig() + start := time.Date(2026, 9, 20, 0, 0, 0, 0, time.UTC) + def := scheduleDefinition(cfg, "run", 0, start) + if def.Rule.EndAt == nil || !def.Rule.EndAt.Equal(start.Add(2*time.Second)) || def.Task.Retry.MaxAttempts != 1 || def.Policy.Overlap != "PARALLEL" { + t.Fatal("schedule is not bounded as configured") + } + s := orbit.Schedule{ScheduleSummary: orbit.ScheduleSummary{NamespaceID: testID(4), QueueID: cfg.QueueID, Name: operationKey("run", 0, "create"), Revision: 1, CurrentVersion: 1}, Current: orbit.ScheduleVersion{Version: 1, EffectiveAt: start.Add(-time.Second), Definition: def}} + s.Current.Definition.Rule.CalendarVersion = "go1.27.1-b2d18a7c8fa8" + if !matchesSchedule(s, cfg, "run", 0, testID(4), start) { + t.Fatal("normalized frozen schedule rejected") + } + s.Current.Definition.Task.Payload = json.RawMessage(`{"run_id":"run","schedule":1}`) + if matchesSchedule(s, cfg, "run", 0, testID(4), start) { + t.Fatal("changed template accepted") + } +} +func TestPartialReportPreservesPlannedDenominator(t *testing.T) { + c := testConfig() + c.Schedules = 2 + r := newReport(c, "run") + r.Occurrences = 3 + r.Fired = 2 + r.VerifiedSuccess = 1 + r = r.finish() + if r.Expected != 4 || r.Unobserved != 1 || r.Unverified != 1 || r.Complete { + t.Fatalf("bad denominators %#v", r) + } + if d := distributionOf(nil); d.P50 != nil || d.P99 != nil { + t.Fatal("empty percentile is not null") + } +} + +func TestCalendarSnapshotRemainsFixed(t *testing.T) { + c := testConfig() + start := time.Date(2026, 9, 20, 0, 0, 0, 0, time.UTC) + p := planReport{CalendarVersion: "go1.27.1-b2d18a7c8fa8", EffectiveAt: start} + s := orbit.Schedule{Current: orbit.ScheduleVersion{Definition: scheduleDefinition(c, "run", 0, start), EffectiveAt: start}} + s.Current.Definition.Rule.CalendarVersion = p.CalendarVersion + if !sameVersion(s, p) { + t.Fatal("unchanged snapshot rejected") + } + s.Current.Definition.Rule.CalendarVersion = "replacement" + if sameVersion(s, p) { + t.Fatal("changed timezone snapshot accepted") + } + s.Current.Definition.Rule.CalendarVersion = p.CalendarVersion + s.Current.EffectiveAt = start.Add(time.Microsecond) + if sameVersion(s, p) { + t.Fatal("changed effective time accepted") + } +} diff --git a/sdk/go/cmd/orbit-schedule-load/observe.go b/sdk/go/cmd/orbit-schedule-load/observe.go new file mode 100644 index 0000000..84f82e8 --- /dev/null +++ b/sdk/go/cmd/orbit-schedule-load/observe.go @@ -0,0 +1,206 @@ +package main + +import ( + "context" + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "reflect" + "time" +) + +// slotObservation 只缓存一个固定时间槽的不可变触发和核验证据。 +type slotObservation struct { + occurrence *orbit.ScheduleOccurrence // occurrence 首次读到的快照；不得改写。 + verified bool // verified 是否已完成全部持久化核验。 + attemptID string // attemptID 唯一成功尝试，用于与本地调用关联。 + trigger, queue, execution, completion time.Duration // 各阶段只使用数据库时间。 +} + +// planObservation 预分配固定槽位，内存上限由配置限制。 +type planObservation struct { + index int // index 计划在本次配置中的序号。 + start time.Time // start 固定触发基准。 + namespace string // namespace 预检认证队列命名空间。 + slots []slotObservation // slots 预期触发槽位，不随服务端输入增长。 +} + +// observations 合并轮询快照，跨计划检查唯一触发和任务身份。 +type observations struct { + cfg config // cfg 固定样本配置。 + runID string // runID 固定测量身份。 + plans map[string]*planObservation // plans 创建时登记的有限计划集合。 + occurrences map[string]string // occurrences 触发ID到计划ID，避免跨计划重复。 + tasks map[string]*slotObservation // tasks 已观察任务到唯一槽位。 +} + +func newObservations(cfg config, runID string) *observations { + return &observations{cfg: cfg, runID: runID, plans: map[string]*planObservation{}, occurrences: map[string]string{}, tasks: map[string]*slotObservation{}} +} +func (o *observations) addPlan(id string, index int, start time.Time, namespace string) { + o.plans[id] = &planObservation{index: index, start: start, namespace: namespace, slots: make([]slotObservation, o.cfg.Requests)} +} +func (o *observations) ingest(id string, page orbit.ScheduleOccurrencePage) string { + p := o.plans[id] + if p == nil || page.Items == nil || page.NextCursor != "" || len(page.Items) > len(p.slots) { + return "OCCURRENCE_PAGE_MISMATCH" + } + seen := map[string]bool{} + for _, item := range page.Items { + delta := item.ScheduledAt.Sub(p.start) + if !validID(item.ID) || item.ScheduleID != id || item.ScheduleVersion != 1 || delta < 0 || delta%o.cfg.Interval != 0 || delta/o.cfg.Interval >= time.Duration(len(p.slots)) || item.CreatedAt.Before(item.ScheduledAt) { + return "OCCURRENCE_MISMATCH" + } + if seen[item.ID] { + return "DUPLICATE_OCCURRENCE" + } + seen[item.ID] = true + slot := &p.slots[int(delta/o.cfg.Interval)] + if slot.occurrence != nil { + if !reflect.DeepEqual(*slot.occurrence, item) { + return "OCCURRENCE_CHANGED" + } + continue + } + if _, exists := o.occurrences[item.ID]; exists { + return "DUPLICATE_OCCURRENCE" + } + // 保存已观察证据再判定策略异常；未验证的触发不会充当成功分母。 + snapshot := item + slot.occurrence = &snapshot + o.occurrences[item.ID] = id + if item.Decision != "FIRED" || item.CoveredUntil != nil || item.Reason != "SCHEDULED" { + return "OCCURRENCE_DECISION_MISMATCH" + } + if item.TaskID == nil || !validID(*item.TaskID) { + return "OCCURRENCE_TASK_MISMATCH" + } + if _, exists := o.tasks[*item.TaskID]; exists { + return "DUPLICATE_TASK" + } + o.tasks[*item.TaskID] = slot + } + // 此样本总量不超过单页上限，历史不可消失；分页异常不降级继续。 + known := 0 + for _, s := range p.slots { + if s.occurrence != nil { + known++ + } + } + if len(page.Items) != known { + return "OCCURRENCE_DISAPPEARED" + } + return "" +} +func (o *observations) verify(ctx context.Context, api scheduleAPI, id string) (bool, string) { + p := o.plans[id] + complete := true + for i := range p.slots { + slot := &p.slots[i] + if slot.occurrence == nil { + complete = false + continue + } + if slot.verified { + continue + } + if ctx.Err() != nil { + return false, "VERIFY_" + classify(ctx.Err()) + } + item := slot.occurrence + if item.TaskID == nil || o.tasks[*item.TaskID] != slot { + return false, "OCCURRENCE_TASK_MISMATCH" + } + call, stop := context.WithTimeout(ctx, o.cfg.Timeout) + task, err := api.GetTask(call, *item.TaskID) + stop() + if err != nil { + return false, "TASK_" + classify(err) + } + if task.ID != *item.TaskID || task.NamespaceID != p.namespace || !matchesTask(task, o.cfg, o.runID, p.index) || !task.CreatedAt.Equal(item.CreatedAt) || !task.AvailableAt.Equal(task.CreatedAt) || !task.DeadlineAt.Equal(task.CreatedAt.Add(time.Duration(o.cfg.RunTimeout.Milliseconds())*time.Millisecond)) { + return false, "TASK_MISMATCH" + } + switch task.State { + case orbit.TaskWaiting, orbit.TaskLeased, orbit.TaskRunning: + complete = false + continue + case orbit.TaskFailed, orbit.TaskCancelled, orbit.TaskNeedsReview: + return false, "TASK_" + string(task.State) + case orbit.TaskSucceeded: + default: + return false, "UNKNOWN_TASK_STATE" + } + output, valid := decodePayload(task.Result) + if !valid || output != (schedulePayload{RunID: o.runID, Schedule: p.index}) || task.CompletedAt == nil || task.EffectUncertain || task.AttemptCount != 1 || task.CancelRequestedAt != nil { + return false, "COMPLETION_MISMATCH" + } + call, stop = context.WithTimeout(ctx, o.cfg.Timeout) + history, err := api.ListTaskAttempts(call, task.ID, orbit.HistoryPageRequest{Limit: 2}) + stop() + if err != nil { + return false, "HISTORY_" + classify(err) + } + if len(history.Items) != 1 || history.NextCursor != "" { + return false, "ATTEMPT_COUNT_MISMATCH" + } + a := history.Items[0] + if !validID(a.ID) || a.TaskID != task.ID || a.Epoch != 1 || a.State != orbit.AttemptSucceeded || a.AuthorizedAt == nil || a.EndedAt == nil || a.ErrorCode != "" || a.Retryable || a.ConcurrencyToken != 0 { + return false, "ATTEMPT_MISMATCH" + } + if a.CreatedAt.Before(task.AvailableAt) || a.AuthorizedAt.Before(a.CreatedAt) || a.EndedAt.Before(*a.AuthorizedAt) || !a.EndedAt.Equal(*task.CompletedAt) { + return false, "TIMESTAMP_MISMATCH" + } + call, stop = context.WithTimeout(ctx, o.cfg.Timeout) + source, err := api.GetTaskSource(call, task.ID) + stop() + if err != nil { + return false, "SOURCE_" + classify(err) + } + s := source.Schedule + if source.TaskID != task.ID || source.Kind != "SCHEDULE" || source.Workflow != nil || source.RetryOf != nil || s == nil || s.ScheduleID != id || s.ScheduleVersion != 1 || s.OccurrenceID != item.ID || !s.ScheduledAt.Equal(item.ScheduledAt) || s.Decision != "FIRED" || s.CoveredUntil != nil { + return false, "SOURCE_MISMATCH" + } + slot.verified = true + slot.attemptID = a.ID + slot.trigger = task.CreatedAt.Sub(item.ScheduledAt) + slot.queue = a.AuthorizedAt.Sub(task.CreatedAt) + slot.execution = a.EndedAt.Sub(*a.AuthorizedAt) + slot.completion = task.CompletedAt.Sub(item.ScheduledAt) + } + return complete, "" +} +func matchesTask(t orbit.Task, c config, runID string, index int) bool { + payload, ok := decodePayload(t.Payload) + return ok && payload == (schedulePayload{RunID: runID, Schedule: index}) && validID(t.ID) && t.QueueID == c.QueueID && t.Handler == handlerName(runID) && t.HandlerVersion == "v1" && t.PayloadVersion == "v1" && t.BusinessKey == businessKey(runID, index) && t.ConcurrencyKey == "" && t.Priority == 0 && t.RetryOf == nil && t.StartDeadlineAt == nil && t.Retry == (orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(c.RunTimeout.Milliseconds())}) && t.RecoverySafety == orbit.RecoveryReconcileOnly +} +func (o *observations) summarize(r *scheduleReport) { + var trigger, queue, execution, completion []time.Duration + for i := range r.Plans { + p := &r.Plans[i] + data := o.plans[p.ID] + if data == nil { + continue + } + for j := range data.slots { + s := &data.slots[j] + if s.occurrence != nil { + p.Occurrences++ + } + if s.occurrence != nil && s.occurrence.TaskID != nil && o.tasks[*s.occurrence.TaskID] == s { + p.Fired++ + } + if s.verified { + p.VerifiedSuccess++ + trigger = append(trigger, s.trigger) + queue = append(queue, s.queue) + execution = append(execution, s.execution) + completion = append(completion, s.completion) + } + } + r.Occurrences += p.Occurrences + r.Fired += p.Fired + r.VerifiedSuccess += p.VerifiedSuccess + } + r.TriggerLag = distributionOf(trigger) + r.QueueWait = distributionOf(queue) + r.Execution = distributionOf(execution) + r.Completion = distributionOf(completion) +} diff --git a/sdk/go/cmd/orbit-schedule-load/report.go b/sdk/go/cmd/orbit-schedule-load/report.go new file mode 100644 index 0000000..cba8315 --- /dev/null +++ b/sdk/go/cmd/orbit-schedule-load/report.go @@ -0,0 +1,111 @@ +package main + +import ( + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "math" + "runtime" + "sort" + "time" +) + +// 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 最大耗时。 +} + +// planReport 保存原操作身份；请求结果不明时只能按原键和原请求核查。 +type planReport struct { + CalendarVersion string `json:"calendar_version,omitempty"` // CalendarVersion 创建回执中的时区快照标识，后续读取不得变化。 + EffectiveAt time.Time `json:"effective_at"` // EffectiveAt 固定计划版本生效时间。 + Index int `json:"index"` // Index 从零开始的计划序号。 + ID string `json:"id,omitempty"` // ID 已取得的计划身份；空不证明创建未提交。 + CreateKey string `json:"create_key"` // CreateKey 创建键，同时作为计划名称。 + ResumeKey string `json:"resume_key,omitempty"` // ResumeKey 原恢复键。 + ResumeRevision uint64 `json:"resume_revision,string"` // ResumeRevision 原恢复请求版本，零表示尚未发送。 + Resumed bool `json:"resumed"` // Resumed 已收到且核对恢复回执。 + Exhausted bool `json:"exhausted"` // Exhausted 已观察到固定版本无下个触发点和扫描工作。 + PauseKey string `json:"pause_key,omitempty"` // PauseKey 最后发送的暂停键，仅明确版本冲突才换键。 + PauseRevision uint64 `json:"pause_revision,string"` // PauseRevision 最后暂停请求的预期版本。 + PauseConfirmed bool `json:"pause_confirmed"` // PauseConfirmed 已确认计划暂停；不取消已生成任务。 + Occurrences int `json:"occurrences"` // Occurrences 已观察唯一触发点数。 + Fired int `json:"fired"` // Fired 已关联唯一任务的单点触发数。 + VerifiedSuccess int `json:"verified_success"` // VerifiedSuccess 完成任务/尝试/来源核对数。 +} + +// scheduleReport 是隔离环境有界样本，不作为生产容量或高可用承诺。 +type scheduleReport 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 预检开始本机时间。 + FinishedAt time.Time `json:"finished_at"` // FinishedAt 全部有界收尾完成本机时间。 + Requests int `json:"requests_per_schedule"` // Requests 每计划预期触发数。 + Schedules int `json:"schedules"` // Schedules 预期计划数。 + Expected int `json:"expected_occurrences"` // Expected 总计划触发数，包含未观察的触发点。 + Workers int `json:"workers"` // Workers 本进程独立WSS Worker数。 + Capacity int `json:"capacity_per_worker"` // Capacity 每Worker并行槽数。 + IntervalMS int64 `json:"interval_ms"` // IntervalMS 固定间隔毫秒。 + StartDelayMS int64 `json:"start_delay_ms"` // StartDelayMS 创建/恢复的预留时间毫秒。 + 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 预检队列快照，只读配置。 + ScheduleStart time.Time `json:"schedule_start"` // ScheduleStart 所有计划共同起点，包含边界。 + ScheduleEnd time.Time `json:"schedule_end"` // ScheduleEnd 所有计划共同截止，排除边界。 + Plans []planReport `json:"plans"` // Plans 已尝试创建的固定身份和收尾证据。 + Occurrences int `json:"occurrences"` // Occurrences 已观察唯一触发数。 + Fired int `json:"fired"` // Fired 单点生成且关联唯一任务数，不代表成功。 + VerifiedSuccess int `json:"verified_success"` // VerifiedSuccess 持久化结果/唯一尝试/来源全部通过数。 + Unobserved int `json:"unobserved"` // Unobserved 预期但未观察到的触发点数。 + Unverified int `json:"unverified"` // Unverified 已触发但未核验成功的任务数。 + HandlerCalls int `json:"handler_calls"` // HandlerCalls 通过输入核对的处理器调用总数。 + DuplicateInvocations int `json:"duplicate_invocations"` // DuplicateInvocations 同一任务的额外调用数。 + UnconfirmedInvocations int `json:"unconfirmed_invocations"` // UnconfirmedInvocations 未对应到已核验任务及尝试的调用数。 + WorkerStarted uint64 `json:"worker_started"` // WorkerStarted SDK实际调用处理器数。 + Reconnects uint64 `json:"worker_reconnects"` // Reconnects SDK连接重建次数。 + Renewals uint64 `json:"worker_renewals_confirmed"` // Renewals 已确认租约续期次数。 + TriggerLag distribution `json:"trigger_lag"` // TriggerLag 计划时间至任务受理数据库耗时。 + QueueWait distribution `json:"queue_wait"` // QueueWait 任务受理至开始授权数据库耗时。 + Execution distribution `json:"authorized_execution"` // Execution 授权至结果落库耗时，包含网络和事务。 + Completion distribution `json:"completion"` // Completion 计划时间至结果落库耗时。 + Errors map[string]int `json:"errors"` // Errors 有限分类，不含凭据或原异常正文。 + Complete bool `json:"complete"` // Complete 全部计数、证据和收尾满足时为真。 +} + +func newReport(c config, runID string) scheduleReport { + return scheduleReport{RunID: runID, GoVersion: runtime.Version(), GOOS: runtime.GOOS, GOARCH: runtime.GOARCH, StartedAt: time.Now().UTC(), Requests: c.Requests, Schedules: c.Schedules, Expected: c.Requests * c.Schedules, Workers: c.Workers, Capacity: c.Capacity, IntervalMS: c.Interval.Milliseconds(), StartDelayMS: c.StartDelay.Milliseconds(), RunTimeoutMS: c.RunTimeout.Milliseconds(), PollIntervalMS: c.PollInterval.Milliseconds(), HTTPTimeoutMS: c.Timeout.Milliseconds(), Plans: []planReport{}, Errors: map[string]int{}} +} +func (r scheduleReport) finish() scheduleReport { + for code, n := range r.Errors { + if n == 0 { + delete(r.Errors, code) + } + } + r.FinishedAt = time.Now().UTC() + r.Unobserved = r.Expected - r.Occurrences + r.Unverified = r.Fired - r.VerifiedSuccess + plansReady := len(r.Plans) == r.Schedules + for _, p := range r.Plans { + plansReady = plansReady && p.Resumed && p.Exhausted && p.PauseConfirmed && p.VerifiedSuccess == r.Requests + } + r.Complete = r.Expected > 0 && plansReady && r.Occurrences == r.Expected && r.Fired == r.Expected && r.VerifiedSuccess == r.Expected && r.HandlerCalls == r.Expected && r.WorkerStarted == uint64(r.Expected) && r.DuplicateInvocations == 0 && r.UnconfirmedInvocations == 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 +}