From 4d4b2e9f7dfb3ba18773e26882dd3be257853962 Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 14:55:23 +0800 Subject: [PATCH] feat: add execution history queries and preserve bounded JSON payloads --- README.md | 4 +- docs/07-data-model.md | 6 +- docs/08-development-plan.md | 4 +- docs/README.md | 2 +- docs/api/README.md | 2 + docs/api/execution-history.md | 166 +++++++++++++ docs/api/tasks.md | 8 +- docs/api/worker.md | 2 +- docs/examples/README.md | 11 +- docs/reports/phase-2-execution-history.md | 55 +++++ docs/sdk/README.md | 49 +++- docs/sdk/worker-runtime.md | 2 +- internal/jsonwire/marshal.go | 18 ++ .../000004_execution_history.up.sql | 2 + internal/storage/mysql/task_history.go | 224 ++++++++++++++++++ .../mysql/task_history_integration_test.go | 207 ++++++++++++++++ internal/storage/mysql/task_store.go | 3 +- internal/transport/httpserver/history.go | 65 +++++ internal/transport/httpserver/server.go | 1 + internal/transport/httpserver/tasks.go | 3 +- internal/transport/httpserver/worker.go | 11 +- pulse/service/history.go | 122 ++++++++++ pulse/service/types.go | 4 + scripts/test-mysql.sh | 1 + sdk/go/examples/history/main.go | 151 ++++++++++++ sdk/go/history.go | 95 ++++++++ sdk/go/history_integration_test.go | 81 +++++++ sdk/go/history_test.go | 83 +++++++ sdk/go/history_types.go | 79 ++++++ sdk/go/json_wire.go | 18 ++ sdk/go/tasks.go | 2 +- sdk/go/worker_connection.go | 8 +- 32 files changed, 1461 insertions(+), 28 deletions(-) create mode 100644 docs/api/execution-history.md create mode 100644 docs/reports/phase-2-execution-history.md create mode 100644 internal/jsonwire/marshal.go create mode 100644 internal/storage/mysql/migrations/000004_execution_history.up.sql create mode 100644 internal/storage/mysql/task_history.go create mode 100644 internal/storage/mysql/task_history_integration_test.go create mode 100644 internal/transport/httpserver/history.go create mode 100644 pulse/service/history.go create mode 100644 sdk/go/examples/history/main.go create mode 100644 sdk/go/history.go create mode 100644 sdk/go/history_integration_test.go create mode 100644 sdk/go/history_test.go create mode 100644 sdk/go/history_types.go create mode 100644 sdk/go/json_wire.go diff --git a/README.md b/README.md index a1c3940..8a99fec 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,9 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已实现 32 个数据模型、版本化迁移、凭据及范围校验、队列创建查询、任务幂等提交/查询/分页/待执行取消，以及对应 Go SDK 和示例。P2 服务端、WSS、底层 SDK 与自动 Worker 运行器已通过阶段内回归，支持自动心跳/续租/重连及有界排空；查询补全、调度器和流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已实现 32 个数据模型、版本化迁移、凭据及范围校验、队列创建查询、任务幂等提交/查询/分页/待执行取消，以及对应 Go SDK 和示例。P2 服务端、WSS、底层 SDK 与自动 Worker 运行器已通过阶段内回归，支持自动心跳/续租/重连、有界排空及 [执行尝试/证据查询](docs/api/execution-history.md)；协议验收、调度器和流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 -readiness 只表示数据库与迁移记录可用，不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 服务端记录](docs/reports/phase-2-server.md)、[P2 SDK 协议记录](docs/reports/phase-2-sdk-protocol.md) 与 [自动运行器记录](docs/reports/phase-2-worker-runtime.md)。 +readiness 只表示数据库与迁移记录可用，不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 服务端记录](docs/reports/phase-2-server.md)、[P2 SDK 协议记录](docs/reports/phase-2-sdk-protocol.md)、[自动运行器记录](docs/reports/phase-2-worker-runtime.md) 与 [执行历史记录](docs/reports/phase-2-execution-history.md)。 ## 目录 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 28f3abb..9cf0811 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -88,7 +88,7 @@ 外键与 `CHECK` 无法代替执行引擎。P1 已完成 HTTP 动作/队列授权、幂等摘要与原回执、受理事件，以及待执行取消的 CAS/审计事务。执行相关的以下条件继续分阶段实现： -P2 服务端已新增会话隔离、两步派发、租约、并发许可、结果回执/证据和恢复扫描，见 [执行协议](api/worker.md) 与 [阶段记录](reports/phase-2-server.md)。Worker SDK 和完整 P2 验收仍在推进。 +P2 服务端已新增会话隔离、两步派发、租约、并发许可、结果回执/证据和恢复扫描，见 [执行协议](api/worker.md) 与 [阶段记录](reports/phase-2-server.md)。底层及自动 Worker SDK、尝试/证据查询已实现，完整 P2 验收仍在推进。 1. 权限验证、凭据校验、命名空间过滤和队列消费授权。 2. 同一次事务更新任务、尝试、租约、并发许可、事件与流程交付记录。 @@ -108,4 +108,6 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 000002_task_api 增加任务分页索引，将业务参数、结果、回执、流程定义/输入/输出等字段改为校验过的 JSON 文本，避免原生 JSON 数字重编码。公共 Go JSON 扫描类型保持兼容，原模型职责不变。响应 DTO 与持久化模型分开；SQL 空结果先扫描到可空字节切片再转换，避免驱动与 json.RawMessage 的可空类型不兼容。 -命令与验证记录见 [开发计划](08-development-plan.md) 和 [基础阶段验证报告](reports/phase-0.md)。 +000004_execution_history 为已有证据表增加 `(namespace_id, attempt_id, created_at, id)` 分页索引，不新增实体或字段，不回改已有迁移。尝试列表复用执行代次唯一索引。查询 DTO 与持久化模型分离，只在详情读取业务正文，不返回执行秘密或内部摘要；接口见 [执行历史](api/execution-history.md)。 + +命令与验证记录见 [开发计划](08-development-plan.md)、[基础阶段验证报告](reports/phase-0.md) 与 [执行历史回归](reports/phase-2-execution-history.md)。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index b9da778..2b30021 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -26,7 +26,7 @@ | --- | --- | --- | --- | | P0 基础与数据模型 | Go workspace、独立 SDK module、公共/Pulse/Weave 模型、MySQL 迁移、配置、健康检查 | 编译/竞态/静态检查；真实 MySQL 建表、隔离、唯一性、预算、不可变版本、迁移失败防护；SDK 健康调用 | 已完成并通过阶段回归（见 P0 报告） | | P1 任务受理闭环 | 应用凭据与作用域权限、队列、幂等提交、查询、待执行取消、受理事件；SDK 客户端 | API/SDK 创建查询闭环；重复键与内容冲突、响应未确认重报、跨命名空间拒绝、参数边界 | 已完成并通过阶段回归（见 P1 报告；真实断路/提交 ACK 丢失注入留后续故障阶段） | -| P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复；Worker SDK | 真 Worker 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 开发中：服务端、SDK 协议和自动运行器通过阶段内回归；跨初始租约执行、自动重连/证据重报、有界退出及并发验证通过；尝试/证据查询、完整 Schema 和阶段验收待完成 | +| P2 Worker 执行闭环 | WSS 接入、能力/容量、领取与开始、租约、续租、结果确认、排空与恢复；Worker SDK | 真 Worker 执行；失联/重连/重复确认/旧尝试拒绝；业务效果未知转核查；节点故障接管 | 开发中：服务端、SDK 协议、自动运行器及尝试/证据 HTTP+SDK 查询通过阶段内回归；自动续租/重连、有界退出、权限/稳定分页和大载荷链路验证通过；完整 Schema 与进程级故障阶段验收待完成 | | P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 待开发 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | @@ -47,6 +47,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 外部作业轮询 | 提交一次 → 有界查询处理中状态 → 回执或最终结果 → 超限核查 | P5 | | 暂停与人工恢复 | 暂停 → 保存结果 → 核查/恢复 → 查看审计 | P5 | | 故障恢复 | 节点/连接中断 → 相同身份核对 → 恢复或人工核查 | P2–P6 | +| 执行历史核对 | 读取尝试 → 查看冲突/迟到证据 → 对照权威结果 | P2，已实现完整示例 | 每个正式示例需要：前置配置、完整可运行源码、启动命令、输入、预期输出、失败路径、幂等/清理说明。SDK API 尚未实现时，文档只能标为规划，不能给出伪装可运行的代码。 @@ -76,3 +77,4 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P2 服务端阶段记录](reports/phase-2-server.md) - [P2 SDK 协议阶段记录](reports/phase-2-sdk-protocol.md) - [P2 自动 Worker 运行器记录](reports/phase-2-worker-runtime.md) +- [P2 执行历史与大载荷记录](reports/phase-2-execution-history.md) diff --git a/docs/README.md b/docs/README.md index 627bb0b..99bf2cb 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本：0.2 - 整理日期：2026-09-17 -- 阶段：P0/P1 已完成，P2 服务端、底层 SDK 与自动 Worker 运行器已通过阶段内回归，查询/协议验收继续推进；尚未完成生产验收 +- 阶段：P0/P1 已完成，P2 服务端、底层 SDK、自动 Worker 运行器及执行历史查询已通过阶段内回归，协议与进程故障验收继续推进；尚未完成生产验收 - 适用范围：Orbit 统一平台，包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 diff --git a/docs/api/README.md b/docs/api/README.md index 65e4c7d..fbc4566 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -9,6 +9,8 @@ | `GET /health/live` | 已实现 | [健康检查](health.md) | | `GET /health/ready` | 已实现，仅验证数据库与迁移记录 | [健康检查](health.md) | | `/v1/tasks` 提交、查询、分页与待执行取消 | P1 已实现 | [任务接口](tasks.md) | +| `/v1/tasks/{task_id}/attempts` 与尝试详情 | P2 已实现 | [执行历史](execution-history.md) | +| `/v1/tasks/{task_id}/attempts/{attempt_id}/evidence` 与证据详情 | P2 已实现 | [执行历史](execution-history.md) | | `/v1/queues` 创建与查询 | P1 已实现 | [队列接口](queues.md) | | 凭据校验与初始化 CLI | P1 已实现 | [鉴权](authentication.md) | | `/v1/worker/connect` | P2 服务端、底层 SDK 与自动 Worker 已实现；完整阶段验收继续推进 | [Worker 契约](worker.md) | diff --git a/docs/api/execution-history.md b/docs/api/execution-history.md new file mode 100644 index 0000000..74aa436 --- /dev/null +++ b/docs/api/execution-history.md @@ -0,0 +1,166 @@ +# 执行尝试与证据查询 v1 + +更新：2026-09-17。四个只读接口及独立 Go SDK 已实现；对应 [可运行示例](../../sdk/go/examples/history/main.go) 和 [回归记录](../reports/phase-2-execution-history.md)。 + +## 1. 用途与权限 + +任务代表一次受理的工作，尝试代表其中一次执行代次。领取后未获开始授权就过期的尝试也会保留，并消耗尝试预算。重连恢复同一次尝试，不创建新的尝试记录。 + +证据保存 Worker 的迟到、未开始或冲突报告。`outcome=SUCCEEDED` 只是原报告的声明；`disposition=EVIDENCE_ONLY` 表示采集时仅作为证据保存，不能解释为任务成功。权威任务状态与结果使用 [GetTask](tasks.md) 查询。当前接口不执行人工裁定、恢复或重试。 + +四个接口均要求 `Authorization: Bearer `、`tasks:read` 动作及任务所属队列范围。凭据确定命名空间，每次调用都重新校验有效性。ID 是 32 位小写十六进制字符串，全部父子关系都校验；知道尝试 ID 或证据 ID 不代表有权读取。 + +响应为 `application/json`，含 `Cache-Control: no-store`、`X-Content-Type-Options: nosniff`；追踪 ID 同时出现在 `X-Request-ID` 和正文 `request_id`。这些 JSON 供解析使用，不能直接嵌入 HTML。SDK 保留结果中的数字文本，不先转换为 float64。 + +## 2. 请求与分页 + +| 方法和路径 | data 对象 | SDK 方法 | +| --- | --- | --- | +| `GET /v1/tasks/{task_id}/attempts` | AttemptPage | `ListTaskAttempts(ctx, taskID, page)` | +| `GET /v1/tasks/{task_id}/attempts/{attempt_id}` | Attempt | `GetTaskAttempt(ctx, taskID, attemptID)` | +| `GET /v1/tasks/{task_id}/attempts/{attempt_id}/evidence` | AttemptEvidencePage | `ListAttemptEvidence(ctx, taskID, attemptID, page)` | +| `GET /v1/tasks/{task_id}/attempts/{attempt_id}/evidence/{evidence_id}` | AttemptEvidence | `GetAttemptEvidence(ctx, taskID, attemptID, evidenceID)` | + +没有请求体，不需要 Idempotency-Key。仅两个列表接受查询参数： + +| 参数 | 类型 / 默认 | 限制 | +| --- | --- | --- | +| limit | 整数，默认 20 | 1–100；SDK 的 0 表示不传、采用默认值，HTTP 显式 0 被拒绝 | +| cursor | 字符串，默认空 | 最长 1024 字节；原样使用前一页的 next_cursor，并进行 URL 编码 | + +参数只能出现一次；未知字段、重复参数、错误 URL 编码、非法游标均返回 400。详情接口不接受查询参数。游标绑定命名空间、任务、尝试和列表类型，不能跨资源复用，也不能作为授权证明。 + +尝试按 `epoch DESC`，证据按 `(created_at DESC, id DESC)` 分页，相同微秒时间以 ID 区分。使用索引和位置条件，不使用 offset 或全量 count。`items` 始终为数组，空时为 `[]`；没有下一页时省略 next_cursor。列表不带业务结果正文。 + +分页不承诺跨请求快照。读取期间尝试状态可能更新、可能出现新记录；刷新最新记录应从空游标重新读取。limit 可以调整，父资源不可改变。 + +```sh +curl --get "$ORBIT_URL/v1/tasks/$TASK_ID/attempts" \ + -H "Authorization: Bearer $ORBIT_TOKEN" --data-urlencode 'limit=20' +curl "$ORBIT_URL/v1/tasks/$TASK_ID/attempts/$ATTEMPT_ID" \ + -H "Authorization: Bearer $ORBIT_TOKEN" +curl --get "$ORBIT_URL/v1/tasks/$TASK_ID/attempts/$ATTEMPT_ID/evidence" \ + -H "Authorization: Bearer $ORBIT_TOKEN" \ + --data-urlencode 'limit=20' --data-urlencode "cursor=$CURSOR" +curl "$ORBIT_URL/v1/tasks/$TASK_ID/attempts/$ATTEMPT_ID/evidence/$EVIDENCE_ID" \ + -H "Authorization: Bearer $ORBIT_TOKEN" +``` + +变量使用实际资源 ID，首次证据查询的 CURSOR 设为空。成功均为 HTTP 200；响应信封见下方完整示例，ID 和时间为示意值。 + +## 3. 尝试字段与响应 + +| 字段 | 类型 / 是否始终返回 | 含义 | +| --- | --- | --- | +| id、task_id | string / 是 | 尝试与父任务标识 | +| epoch | 十进制字符串 / 是 | 单任务执行代次；SDK uint64 | +| state | string / 是 | LEASED、RUNNING、SUCCEEDED、FAILED、CANCELLED、EXPIRED、UNKNOWN；枚举不表示全部状态转换已实现 | +| revision | 十进制字符串 / 是 | 记录版本，续租及归属变化可递增；SDK uint64 | +| lease_until | UTC RFC 3339 / 是 | 数据库记录的租约期限，**查询不提供执行授权** | +| authorized_at | UTC RFC 3339 / 否 | 获批开始时间；省略表示尚未获批，不证明外部效果是否发生 | +| ended_at | UTC RFC 3339 / 否 | 尝试结束时间；尚未结束时省略 | +| error_code | string / 否 | 稳定错误分类，无错误时省略 | +| created_at、updated_at | UTC RFC 3339 / 是 | 创建和最近更新时间，最多微秒精度 | +| session_id、worker_instance_id | string / 仅详情 | 当前或最后归属会话及进程身份；会话可能因恢复改变，不是完整归属变更日志 | +| result | JSON / 仅详情，可省略 | 此尝试已保存的结果；无结果时省略，与追加证据分开 | + +列表响应： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{"items":[{ + "id":"22222222222222222222222222222222", + "task_id":"11111111111111111111111111111111", + "epoch":"1","state":"SUCCEEDED","revision":"3", + "lease_until":"2026-09-17T08:00:30Z", + "authorized_at":"2026-09-17T08:00:01Z","ended_at":"2026-09-17T08:00:02Z", + "created_at":"2026-09-17T08:00:00Z","updated_at":"2026-09-17T08:00:02Z" + }]} +} +``` + +详情响应为展开对象： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{ + "id":"22222222222222222222222222222222", + "task_id":"11111111111111111111111111111111", + "epoch":"1","state":"SUCCEEDED","revision":"3", + "lease_until":"2026-09-17T08:00:30Z", + "authorized_at":"2026-09-17T08:00:01Z","ended_at":"2026-09-17T08:00:02Z", + "created_at":"2026-09-17T08:00:00Z","updated_at":"2026-09-17T08:00:02Z", + "session_id":"33333333333333333333333333333333", + "worker_instance_id":"55555555555555555555555555555555", + "result":{"version":"v1","sequence":9007199254740993} + } +} +``` + +不返回 lease_secret、instance_secret、凭据或内部摘要。执行权限必须经过 [Worker 开始/恢复与当前租约确认](worker.md)，不能通过历史详情推导。 + +## 4. 证据字段与响应 + +| 字段 | 类型 / 是否始终返回 | 含义 | +| --- | --- | --- | +| id、task_id、attempt_id | string / 是 | 证据及已校验的父资源标识 | +| outcome | string / 是 | 原报告的 SUCCEEDED、FAILED 或 UNKNOWN | +| error_code | string / 否 | 原报告的错误分类，空时省略 | +| reason | string / 是 | LEASE_LOST（授权失效）、NOT_STARTED（未获批开始）、RESULT_CONFLICT（与已保存结果冲突） | +| disposition | string / 是 | 固定 EVIDENCE_ONLY，描述采集处理方式 | +| created_at | UTC RFC 3339 / 是 | 首次保存时间，相同身份的重复报告不刷新 | +| message_id | string / 仅详情 | 原报告身份，便于追踪同一次重报 | +| payload | JSON / 仅详情 | 原报告结果或证据引用，最大 256 KiB，保留精确数字文本 | + +列表响应： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{"items":[{ + "id":"66666666666666666666666666666666", + "task_id":"11111111111111111111111111111111", + "attempt_id":"22222222222222222222222222222222", + "outcome":"SUCCEEDED","reason":"RESULT_CONFLICT","disposition":"EVIDENCE_ONLY", + "created_at":"2026-09-17T08:00:03Z" + }]} +} +``` + +详情响应： + +```json +{ + "request_id":"44444444444444444444444444444444", + "data":{ + "id":"66666666666666666666666666666666", + "task_id":"11111111111111111111111111111111", + "attempt_id":"22222222222222222222222222222222", + "outcome":"SUCCEEDED","reason":"RESULT_CONFLICT","disposition":"EVIDENCE_ONLY", + "created_at":"2026-09-17T08:00:03Z", + "message_id":"77777777777777777777777777777777","payload":{"version":"v2"} + } +} +``` + +保存 v2 证据不会覆盖尝试中的 v1 结果。查询没有状态变更或去重写入；人工裁定、证据采纳与审计管理属于后续能力。 + +## 5. 错误、超时与兼容 + +| HTTP / code | 处理 | +| --- | --- | +| 400 INVALID_ARGUMENT | 修正 ID、limit、游标及查询参数；不可盲目重试 | +| 401 UNAUTHENTICATED | 核对凭据撤销/到期、应用及命名空间状态 | +| 403 PERMISSION_DENIED | 凭据缺少 tasks:read | +| 404 NOT_FOUND | 资源不存在、父子不匹配、跨命名空间或不在队列范围内；不区分具体原因 | +| 503 UNAVAILABLE | 只读查询失败，可在总体截止内有限退避重试 | + +```json +{"request_id":"44444444444444444444444444444444","error":{"code":"NOT_FOUND","message":"resource not found"}} +``` + +服务端操作期限 8 秒；SDK 默认 HTTP 超时 10 秒，可用 context 缩短。SDK 不自动重试查询；取消查询不取消任务。SDK 业务响应上限 1 MiB，列表通过省略大正文和最多 100 条保持有界。每次重读获得当次数据，不保证与前次完全相同。 + +当前开发契约为 v1，SDK 尚未发布版本，接入方式见 [SDK 手册](../sdk/README.md)。SDK 忽略新增响应字段；已有字段含义改变需升级契约并回归。跨租户/队列、父子错配、重复证据、同时间戳分页、精确数字和大载荷均有实际回归，见 [验证记录](../reports/phase-2-execution-history.md)。 diff --git a/docs/api/tasks.md b/docs/api/tasks.md index 2ceccc2..65d8869 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -1,8 +1,8 @@ # 任务接口 v1 -状态：P1 已实现并通过真实 MySQL 与 SDK 回归。当前提供受理、查询、分页和待执行取消；Worker 执行与自动重试在 P2–P3 实现。 +状态：P1 已实现并通过真实 MySQL 与 SDK 回归。当前提供受理、查询、分页和待执行取消；P2 已提供 Worker 执行与历史查询，策略重试在 P3 实现。 -P2 进展：服务端执行内核和 WSS 已可产生 LEASED、RUNNING、SUCCEEDED、FAILED、NEEDS_REVIEW 等状态，见 [Worker 契约](worker.md)；独立 Worker SDK 仍在开发，自动策略重试尚未完成。 +P2 进展：服务端执行内核、WSS 与独立自动 Worker SDK 已可产生 LEASED、RUNNING、SUCCEEDED、FAILED、NEEDS_REVIEW 等状态，见 [Worker 契约](worker.md)；[尝试与证据查询](execution-history.md) 可核对每次执行，自动策略重试尚未完成。 ## 公共约定 @@ -28,7 +28,7 @@ P2 进展：服务端执行内核和 WSS 已可产生 LEASED、RUNNING、SUCCEED | start_deadline_at | time / null | 无独立限制 | 必须晚于当前时间，不早于 available_at，不晚于总截止时间 | | task_timeout_ms | uint64 | 0 或省略为 86400000 | 从受理时计算的总时限，最大 30 天（2592000000 毫秒） | | recovery_safety | string | RECONCILE_ONLY | 也可为 IDEMPOTENT；必须基于处理器真实的幂等能力声明 | -| retry | object | 见下表 | 受理时冻结的执行策略；当前阶段仅保存 | +| retry | object | 见下表 | 受理时冻结策略；P2 已执行尝试预算及时限，策略退避重试在 P3 实现 | 处理器及版本名称以 ASCII 字母或数字开头，后续只能包含字母、数字、`_ . : -`。时间年份为 2000–9998。 @@ -186,4 +186,4 @@ SDK：`Client.CancelTask(ctx, taskID, key, request)`。此操作不承诺回滚 {"request_id":"44444444444444444444444444444444","error":{"code":"IDEMPOTENCY_CONFLICT","message":"idempotency key was used with different content"}} ``` -批量受理、执行历史、人工重试和运行中取消尚未实现。完整可运行程序见 [SDK 任务示例](../../sdk/go/examples/tasks/main.go)，已验证异常场景见 [P1 回归报告](../reports/phase-1.md)。SDK 忽略未知响应字段；更改已有字段语义需升级契约和回归。 +批量受理、人工重试和运行中取消尚未实现。[执行历史](execution-history.md) 已提供尝试与证据的列表和详情。完整可运行程序见 [SDK 任务示例](../../sdk/go/examples/tasks/main.go)，已验证异常场景见 [P1 回归报告](../reports/phase-1.md)。SDK 忽略未知响应字段；更改已有字段语义需升级契约和回归。 diff --git a/docs/api/worker.md b/docs/api/worker.md index 1e0cd82..ce55c65 100644 --- a/docs/api/worker.md +++ b/docs/api/worker.md @@ -1,6 +1,6 @@ # Worker 通信与执行契约 v1（开发中） -已实现服务端执行事务、WebSocket 接入、[独立 Go 底层协议 SDK](../sdk/worker.md) 和 [自动 Worker 运行器](../sdk/worker-runtime.md)，通过真实 WSS、自动续租、ACK 丢失/持续断网恢复及有界退出回归。**P2 尚未完成**：尝试/证据查询、完整协议 Schema 与完整阶段验收继续推进。底层 WorkerConnection 不自动管理处理器；NewWorker / Run 提供自动管理。 +已实现服务端执行事务、WebSocket 接入、[独立 Go 底层协议 SDK](../sdk/worker.md)、[自动 Worker 运行器](../sdk/worker-runtime.md) 与 [尝试/证据查询](execution-history.md)，通过真实 WSS、自动续租、ACK 丢失/持续断网恢复、有界退出及历史权限/分页回归。**P2 尚未完成**：完整协议 Schema 与完整阶段验收继续推进。底层 WorkerConnection 不自动管理处理器；NewWorker / Run 提供自动管理。 ## 1. 接入与身份 diff --git a/docs/examples/README.md b/docs/examples/README.md index 3ef7477..fe11d82 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -36,11 +36,20 @@ - 异常：自动心跳/续租与原身份重连；普通错误/panic 按未知效果报告，超时退出不声称杀死 goroutine；完整故障证据见 [回归记录](../reports/phase-2-worker-runtime.md)。 - 清理：正常排空关闭连接，保留历史；进程被强杀后的在途状态由租约恢复与核查衔接。 +## 已可运行：执行历史与冲突证据核对 + +- 源码：[执行历史示例](../../sdk/go/examples/history/main.go)，[启动命令和输入输出](../sdk/README.md#6-查询执行历史与证据)。 +- 前置：迁移到当前版本、启动服务，专用凭据具备提交/读取与 Worker 权限。 +- 步骤：创建模拟任务 → 开始授权 → 保存 v1 结果 → 用另一消息身份报告 v2 → 查询尝试/证据列表与详情 → 核对任务仍保留 v1。 +- 预期：SUCCEEDED、attempts=1、evidence=1、reason=RESULT_CONFLICT、disposition=EVIDENCE_ONLY；实际结果中的大整数保持精确。 +- 异常：迟到或矛盾内容不能覆盖权威结果；缺少动作权限返回 403，错配父资源、跨命名空间或队列越权返回 404，详见 [接口](../api/execution-history.md)。示例不调用外部业务系统。 +- 清理：正常结束确认排空并关闭连接；任务及证据保留用于追踪，不硬删除。重复运行创建独立任务。 + ## 计划场景与验收标准 | 场景 | 输入与步骤 | 预期结果 | 必测异常 | 状态 | | --- | --- | --- | --- | --- | -| 文件转换 | 对象引用 → 提交任务 → Worker 转换 → 结果引用 | 同一操作获得一次业务结果，历史可查 | 重复提交、Worker 在结果上报前中断 | P1 受理完成，P2 执行待开发 | +| 文件转换 | 对象引用 → 提交任务 → Worker 转换 → 结果引用 | 同一操作获得一次业务结果，历史可查 | 重复提交、Worker 在结果上报前中断 | 受理与执行框架已实现，实际转换示例待补齐 | | 定期同步 | 发布计划 → 周期触发 → 暂停 → 恢复 | 触发身份唯一，漏跑策略可追溯 | 多节点竞争、停机后补跑上限 | P3 待开发 | | 并行生成 | 输入对象 → 并行计算两份结果 → 汇合 | 两支满足条件后只创建一份后继工作 | 一支失败、重复结果、汇合竞争 | P4 待开发 | | 等待审核 | 校验 → 等待信号 → 发布结果 | 只接受授权、匹配当前操作的信号 | 早到、重复、过期与越权信号 | P4 待开发 | diff --git a/docs/reports/phase-2-execution-history.md b/docs/reports/phase-2-execution-history.md new file mode 100644 index 0000000..41a34ef --- /dev/null +++ b/docs/reports/phase-2-execution-history.md @@ -0,0 +1,55 @@ +# P2 执行历史与大载荷回归 + +日期：2026-09-17。本轮完成尝试及证据查询，P2 的完整 WS Schema 与进程级故障阶段验收仍待完成。接口与 SDK 模型均提供中文职责和字段注释。 + +## 实现范围 + +- 四个 HTTP GET 与对应独立 Go SDK 方法：尝试列表/详情、证据列表/详情。 +- 每次请求重新校验凭据、命名空间、tasks:read、队列范围及父子关系；详情不暴露执行秘密或内部摘要。 +- 列表省略业务正文，以执行代次或时间/ID 稳定分页；详情提供原结果和证据内容，两者分开。 +- 新增 000004 前向迁移，为证据查询增加四列索引；既有迁移不变，仍为 32 个持久化模型。 +- 新增可运行 `examples/history`，演示矛盾回报被保留为证据而不覆盖权威结果。 + +完整接口、字段、错误码和 SDK 对应方法见 [执行历史](../api/execution-history.md)，运行步骤见 [SDK 手册](../sdk/README.md#6-查询执行历史与证据)。 + +## 修复与验证过程 + +大载荷回归发现默认 JSON HTML 转义会将合法的约 240 KiB `<>&` 文本放大至约 1.4 MiB，超出 HTTP/WS 限制。服务端 JSON 响应、WS 帧、提交幂等回执与 SDK 传输现在使用关闭 HTML 转义的编码器；仍使用 JSON 通道，不允许把结果直接嵌入 HTML。没有放宽消息上限，没有改变幂等语义摘要算法或重写历史数据。 + +首轮新测试还发现测试用例重复使用解码结构而残留被省略的字段，已改为每页新建结果；随后真实 MySQL 捕获尝试详情 JOIN 错用了 Worker 实例列名，已修复为实际表字段。修复后完整集成脚本重跑通过，没有跳过失败项。 + +## 环境与入口 + +macOS arm64、Go 1.27.1、Docker MySQL 8.4.10，WS 依赖锁定 coder/websocket v1.8.15。数据库使用随机凭据和回环随机端口，结束删除临时容器。真实 Orbit 进程提供 HTTP/WS，已有 WSS 故障回归使用受信任测试证书。 + +```sh +make test +make vet +make build +make integration +cd sdk/go +GOWORK=off go test -race ./... +GOWORK=off go vet ./... +``` + +## 本轮通过的回归 + +| 场景 | 实际验证 | +| --- | --- | +| 执行尝试历史 | 同一任务第一次未开始即过期，第二次获批并成功；逆序分页、次数、时间、归属和精确结果正确 | +| 追加证据及去重 | 未开始报告保留 NOT_STARTED；成功后矛盾内容保留 RESULT_CONFLICT，同消息重报只出现一次 | +| 同时间戳分页 | 一条 SQL 插入同微秒的两条受控证据，逐条翻页无重复或遗漏 | +| 访问控制 | 四个路由覆盖跨命名空间、队列越权和错配父资源；缺失动作权限 403、撤销凭据 401；不返回秘密和摘要 | +| 输入边界 | 非法/过长游标、跨任务/尝试/类型复用、重复/未知查询参数、错误编码、非法 ID 和分页上限被拒绝 | +| 列表与详情 | 空 items 为 []；摘要不含 result/payload；详情不丢失超过 2^53 的整数或高精度小数 | +| 真实大载荷链路 | 约 240 KiB `<>&` JSON 经提交、同键重复回执、WS 派发、开始、结果上报、详情与冲突证据读取保持内容；任务完整输入+结果仍在 SDK 响应限制内 | +| SDK 契约 | 四个 GET 路径、授权头、URL 编码、uint64 字符串、精确 RawMessage 和本地参数拒绝；独立 module 不引用服务端 | +| 迁移 | 全部迁移应用成功，证据索引四列实际存在；既有约束/不可变证据/迁移校验回归继续通过 | +| 既有链路 | 模型、任务事务、Worker、WSS、自动续租、ACK 丢失与断网恢复、有界退出和并发回归全部通过 | +| 命令行示例 | 健康、任务受理、底层 Worker、自动 Worker、执行历史五个例子均在真实服务上运行通过 | + +## 当前边界 + +查询是当前快照，不是跨页一致快照或完整事件/审计导出。尝试详情显示当前/最后归属会话，不重建全部归属历史。只读结果不授予执行权，EVIDENCE_ONLY 不构成人工裁定。 + +本轮没有完成旧数据转义重编码迁移、策略重试、人工恢复、运行中取消、可靠事件订阅、数据库故障转移或容量测试。大载荷测试使用真实 WS，已有 WSS 回归另外覆盖 TLS 和故障恢复；不将这些结果描述为多主机生产高可用验收。下一步继续 P2 协议 Schema 与进程级故障验收。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index b167f87..366021d 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -1,6 +1,6 @@ # Go SDK 接入说明 -已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询、[自动 Worker](worker-runtime.md) 和 [Worker 底层协议 SDK](worker.md)。流程构建等后续接口按 [开发计划](../08-development-plan.md) 继续实现。 +已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询、[自动 Worker](worker-runtime.md)、[Worker 底层协议 SDK](worker.md) 和 [执行历史查询](../api/execution-history.md)。流程构建等后续接口按 [开发计划](../08-development-plan.md) 继续实现。 module：`git.yuntuo.hk/k2safe/orbit/sdk/go`。SDK 使用独立 go.mod，依赖 Go 标准库与 `github.com/coder/websocket v1.8.15`，不引用服务端模型或数据库实现。 @@ -112,6 +112,10 @@ fmt.Println(cancelled.Task.State) | CancelTask | taskID、key、CancelTaskRequest / CancelTaskResult | [任务](../api/tasks.md) | | CreateQueue | key、CreateQueueRequest / CreateQueueResult | [队列](../api/queues.md) | | GetQueue | queueID / Queue | [队列](../api/queues.md) | +| ListTaskAttempts | taskID、HistoryPageRequest / AttemptPage | [执行历史](../api/execution-history.md) | +| GetTaskAttempt | taskID、attemptID / Attempt | [执行历史](../api/execution-history.md) | +| ListAttemptEvidence | taskID、attemptID、HistoryPageRequest / AttemptEvidencePage | [执行历史](../api/execution-history.md) | +| GetAttemptEvidence | taskID、attemptID、evidenceID / AttemptEvidence | [执行历史](../api/execution-history.md) | 所有方法第一个参数为 context.Context，返回值之外都返回 error。命令方法要求调用方提供稳定 key，不自动生成或重试。ListTasksRequest 接受 QueueID、State、Limit、Cursor。 @@ -131,6 +135,45 @@ go mod tidy 将示例路径换成真实路径，不要把本机绝对路径作为发布依赖提交。后续会验证独立 module 标签和真实远端取包，再提供 go get 发布命令。当前源码同时验证 GOWORK=off 的独立构建。 -## 6. 后续交付 +## 6. 查询执行历史与证据 -P2 已提供底层协议与 [自动 Worker 运行器及示例](worker-runtime.md)，继续补齐查询与阶段验收；P3 提供计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 +只读凭据需要 tasks:read 和任务所属队列范围。下方为接入函数片段，遍历尝试摘要并查询每条尝试详情；每次页面请求均需检查错误： + +```go +pageRequest := orbit.HistoryPageRequest{Limit: 20} +for { + page, err := client.ListTaskAttempts(ctx, taskID, pageRequest) + if err != nil { return err } + for _, summary := range page.Items { + detail, err := client.GetTaskAttempt(ctx, taskID, summary.ID) + if err != nil { return err } + fmt.Println(detail.ID, detail.Epoch, detail.State) + // Result 是 json.RawMessage；读取业务字段时保留精确类型。 + } + if page.NextCursor == "" { break } + pageRequest.Cursor = page.NextCursor +} +``` + +查询证据对应 `ListAttemptEvidence(ctx, taskID, attemptID, pageRequest)` 与 `GetAttemptEvidence(ctx, taskID, attemptID, evidenceID)`，列表相同方式使用 NextCursor。结果正文只在详情返回，列表适合按需展示。查询快照不授予执行权；证据的 outcome 是 Worker 原始声明，EVIDENCE_ONLY 不能当作任务成功。当前权威任务结果用 GetTask 核对。 + +完整 [执行历史示例](../../sdk/go/examples/history/main.go) 会创建一条模拟任务，提交固定成功结果，再模拟矛盾报告，并通过上述四个查询核对历史。先按 [Worker 初始化步骤](worker.md#1-可运行的文本规范化示例) 准备具有提交、读取及 Worker 权限的专用凭据，再从仓库根目录执行： + +```sh +cd sdk/go +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/worker.credentials.json \ + go run ./examples/history +``` + +预期输出： + +```text +task=<任务ID> state=SUCCEEDED attempts=1 evidence=1 reason=RESULT_CONFLICT disposition=EVIDENCE_ONLY +``` + +示例是受控故障演示，不调用外部业务系统。原结果为 `{"version":"v1","sequence":9007199254740993}`，矛盾内容 `{"version":"v2"}` 只进入证据；任务仍保留原成功结果。每次运行创建独立任务/处理器身份，正常结束会排空并关闭连接，历史保留。示例不自动处理全部网络故障，失败时非零退出；应按输出的稳定错误分类核查，持续消费请使用自动 Worker。 + +## 7. 后续交付 + +P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) 与历史查询，继续补齐协议 Schema 及阶段验收；P3 提供计划与结果订阅；P4–P5 提供流程构建发布、信号、等待、轮询与人工恢复。每项随实现同步交付可运行示例及异常回归。 diff --git a/docs/sdk/worker-runtime.md b/docs/sdk/worker-runtime.md index 62d66fa..cf06b31 100644 --- a/docs/sdk/worker-runtime.md +++ b/docs/sdk/worker-runtime.md @@ -1,6 +1,6 @@ # 自动 Worker 接入 -更新：2026-09-17。自动 Worker 已实现，并通过真实 MySQL、实际服务命令、WSS 断线/ACK 丢失/持续断网回归。P2 的尝试与证据查询、完整协议 Schema 和进程级故障验收仍需继续完成。 +更新：2026-09-17。自动 Worker 已实现，并通过真实 MySQL、实际服务命令、WSS 断线/ACK 丢失/持续断网回归。[尝试与证据查询](../api/execution-history.md) 已可用于核对结果；P2 的完整协议 Schema 和进程级故障验收仍需继续完成。 源码：[运行器](../../sdk/go/worker_runtime.go)、[配置与中文注释](../../sdk/go/worker_runtime_types.go)、[完整示例](../../sdk/go/examples/worker/main.go)。特殊传输集成可以使用 [底层协议 SDK](worker.md)，普通消费直接使用本页的 Worker。 diff --git a/internal/jsonwire/marshal.go b/internal/jsonwire/marshal.go new file mode 100644 index 0000000..59408b9 --- /dev/null +++ b/internal/jsonwire/marshal.go @@ -0,0 +1,18 @@ +// Package jsonwire 统一 JSON 传输与回执编码；响应以 application/json 提供，不嵌入 HTML。 +package jsonwire + +import ( + "bytes" + "encoding/json" +) + +// Marshal 保留 RawMessage 数字文本，避免 HTML 转义把合法载荷放大到传输上限之外。 +func Marshal(value any) ([]byte, error) { + var buffer bytes.Buffer + encoder := json.NewEncoder(&buffer) + encoder.SetEscapeHTML(false) + if err := encoder.Encode(value); err != nil { + return nil, err + } + return bytes.TrimSuffix(buffer.Bytes(), []byte{'\n'}), nil +} diff --git a/internal/storage/mysql/migrations/000004_execution_history.up.sql b/internal/storage/mysql/migrations/000004_execution_history.up.sql new file mode 100644 index 0000000..f4894e5 --- /dev/null +++ b/internal/storage/mysql/migrations/000004_execution_history.up.sql @@ -0,0 +1,2 @@ +-- 证据分页按父尝试、创建时间和唯一标识查找，不扫描整个命名空间的业务正文。 +ALTER TABLE pulse_attempt_evidence ADD KEY ix_evidence_history(namespace_id,attempt_id,created_at,id) COMMENT '执行证据按尝试和时间稳定分页'; diff --git a/internal/storage/mysql/task_history.go b/internal/storage/mysql/task_history.go new file mode 100644 index 0000000..5def588 --- /dev/null +++ b/internal/storage/mysql/task_history.go @@ -0,0 +1,224 @@ +package mysql + +import ( + "bytes" + "context" + "database/sql" + "encoding/base64" + "encoding/json" + "errors" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +// historyCursor 只定位已授权父资源中的后续位置，不作为身份或授权证明。 +type historyCursor struct { + Version int `json:"v"` + Kind string `json:"k"` + Namespace core.ID `json:"n"` + Task core.ID `json:"t"` + Attempt core.ID `json:"a,omitempty"` + Epoch uint64 `json:"e,omitempty,string"` + Created *time.Time `json:"c,omitempty"` + ID core.ID `json:"i,omitempty"` +} + +func decodeHistoryCursor(value string, ns, task, attempt core.ID) (historyCursor, error) { + var c historyCursor + raw, err := base64.RawURLEncoding.Strict().DecodeString(value) + if err != nil || len(raw) > 768 { + return c, fault.Invalid("invalid history cursor") + } + if _, err := jsonvalue.Canonical(raw); err != nil { + return c, fault.Invalid("invalid history cursor") + } + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil || fields == nil { + return c, fault.Invalid("invalid history cursor") + } + for k := range fields { + switch k { + case "v", "k", "n", "t", "a", "e", "c", "i": + default: + return c, fault.Invalid("invalid history cursor") + } + } + d := json.NewDecoder(bytes.NewReader(raw)) + d.DisallowUnknownFields() + if d.Decode(&c) != nil || c.Version != 1 || c.Namespace != ns || c.Task != task || c.Attempt != attempt { + return c, fault.Invalid("cursor does not match this history query") + } + if attempt == "" { + if c.Kind != "attempts" || c.Epoch == 0 || c.ID != "" || c.Created != nil { + return c, fault.Invalid("invalid attempts cursor") + } + } else { + if c.Kind != "evidence" || c.Epoch != 0 || c.ID.Validate() != nil || c.Created == nil || c.Created.Year() < 2000 || c.Created.Year() > 9998 || c.Created.Nanosecond()%1000 != 0 { + return c, fault.Invalid("invalid evidence cursor") + } + } + return c, nil +} +func encodeHistoryCursor(c historyCursor) (string, error) { + raw, err := json.Marshal(c) + return base64.RawURLEncoding.EncodeToString(raw), err +} +func authorizeHistoryTask(ctx context.Context, tx *sql.Tx, ns, task core.ID, p auth.Permissions) error { + var queue core.ID + err := tx.QueryRowContext(ctx, `SELECT queue_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, task).Scan(&queue) + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + if err != nil { + return err + } + if !p.AllowsQueue(queue) { + return fault.NotFound + } + return nil +} +func historyAttemptExists(ctx context.Context, tx *sql.Tx, ns, task, attempt core.ID) error { + var found core.ID + err := tx.QueryRowContext(ctx, `SELECT id FROM pulse_attempts WHERE namespace_id=? AND task_id=? AND id=?`, ns, task, attempt).Scan(&found) + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + return err +} + +const attemptHistoryColumns = `a.id,a.task_id,a.epoch,a.state,a.revision,a.lease_until,a.authorized_at,a.ended_at,COALESCE(a.error_code,''),a.created_at,a.updated_at` + +func scanAttemptSummary(row interface{ Scan(...any) error }) (service.AttemptSummary, error) { + var a service.AttemptSummary + err := row.Scan(&a.ID, &a.TaskID, &a.Epoch, &a.State, &a.Revision, &a.LeaseUntil, &a.AuthorizedAt, &a.EndedAt, &a.ErrorCode, &a.CreatedAt, &a.UpdatedAt) + return a, err +} +func (s *TaskStore) ListTaskAttempts(ctx context.Context, t auth.Token, task core.ID, in service.HistoryPageInput) (service.AttemptPage, error) { + out := service.AttemptPage{Items: []service.AttemptSummary{}} + err := s.transaction(ctx, t, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error { + if err := authorizeHistoryTask(ctx, tx, t.NamespaceID, task, p); err != nil { + return err + } + q := `SELECT ` + attemptHistoryColumns + ` FROM pulse_attempts a WHERE a.namespace_id=? AND a.task_id=?` + args := []any{t.NamespaceID, task} + if in.Cursor != "" { + c, err := decodeHistoryCursor(in.Cursor, t.NamespaceID, task, "") + if err != nil { + return err + } + q += ` AND a.epoch in.Limit { + out.Items = out.Items[:in.Limit] + last := out.Items[len(out.Items)-1] + out.NextCursor, err = encodeHistoryCursor(historyCursor{Version: 1, Kind: "attempts", Namespace: t.NamespaceID, Task: task, Epoch: last.Epoch}) + return err + } + return nil + }) + return out, err +} +func (s *TaskStore) GetTaskAttempt(ctx context.Context, t auth.Token, task, attempt core.ID) (service.AttemptView, error) { + var out service.AttemptView + err := s.transaction(ctx, t, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error { + if err := authorizeHistoryTask(ctx, tx, t.NamespaceID, task, p); err != nil { + return err + } + var result []byte + err := tx.QueryRowContext(ctx, `SELECT `+attemptHistoryColumns+`,a.session_id,w.worker_instance_id,a.result FROM pulse_attempts a JOIN pulse_worker_sessions w ON w.namespace_id=a.namespace_id AND w.id=a.session_id WHERE a.namespace_id=? AND a.task_id=? AND a.id=?`, t.NamespaceID, task, attempt).Scan(&out.ID, &out.TaskID, &out.Epoch, &out.State, &out.Revision, &out.LeaseUntil, &out.AuthorizedAt, &out.EndedAt, &out.ErrorCode, &out.CreatedAt, &out.UpdatedAt, &out.SessionID, &out.WorkerInstanceID, &result) + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + out.Result = result + return err + }) + return out, err +} +func (s *TaskStore) ListAttemptEvidence(ctx context.Context, t auth.Token, task, attempt core.ID, in service.HistoryPageInput) (service.EvidencePage, error) { + out := service.EvidencePage{Items: []service.EvidenceSummary{}} + err := s.transaction(ctx, t, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error { + if err := authorizeHistoryTask(ctx, tx, t.NamespaceID, task, p); err != nil { + return err + } + if err := historyAttemptExists(ctx, tx, t.NamespaceID, task, attempt); err != nil { + return err + } + q := `SELECT id,attempt_id,outcome,COALESCE(error_code,''),reason,created_at FROM pulse_attempt_evidence WHERE namespace_id=? AND attempt_id=?` + args := []any{t.NamespaceID, attempt} + if in.Cursor != "" { + c, err := decodeHistoryCursor(in.Cursor, t.NamespaceID, task, attempt) + if err != nil { + return err + } + q += ` AND (created_at in.Limit { + out.Items = out.Items[:in.Limit] + last := out.Items[len(out.Items)-1] + out.NextCursor, err = encodeHistoryCursor(historyCursor{Version: 1, Kind: "evidence", Namespace: t.NamespaceID, Task: task, Attempt: attempt, Created: &last.CreatedAt, ID: last.ID}) + return err + } + return nil + }) + return out, err +} +func (s *TaskStore) GetAttemptEvidence(ctx context.Context, t auth.Token, task, attempt, evidence core.ID) (service.EvidenceView, error) { + out := service.EvidenceView{EvidenceSummary: service.EvidenceSummary{TaskID: task, Disposition: "EVIDENCE_ONLY"}} + err := s.transaction(ctx, t, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error { + if err := authorizeHistoryTask(ctx, tx, t.NamespaceID, task, p); err != nil { + return err + } + if err := historyAttemptExists(ctx, tx, t.NamespaceID, task, attempt); err != nil { + return err + } + var payload []byte + err := tx.QueryRowContext(ctx, `SELECT id,attempt_id,outcome,COALESCE(error_code,''),reason,created_at,message_id,payload FROM pulse_attempt_evidence WHERE namespace_id=? AND attempt_id=? AND id=?`, t.NamespaceID, attempt, evidence).Scan(&out.ID, &out.AttemptID, &out.Outcome, &out.ErrorCode, &out.Reason, &out.CreatedAt, &out.MessageID, &payload) + if errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } + out.Payload = payload + return err + }) + return out, err +} diff --git a/internal/storage/mysql/task_history_integration_test.go b/internal/storage/mysql/task_history_integration_test.go new file mode 100644 index 0000000..b20e8a7 --- /dev/null +++ b/internal/storage/mysql/task_history_integration_test.go @@ -0,0 +1,207 @@ +package mysql + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" +) + +func TestIntegrationTaskHistory(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 16, 4) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + f := newExecutionFixture(t, ctx, db) + other := newExecutionFixture(t, ctx, db) + task := f.submit(3) + _, first := f.claim() + _, err = f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: first, Outcome: "UNKNOWN", ErrorCode: "EARLY_REPORT", Result: json.RawMessage(`{"source":"early"}`)}) + if err != nil { + t.Fatal(err) + } + f.expire(first) + if _, err := f.storeA.Recover(ctx, 10); err != nil { + t.Fatal(err) + } + f.exec(`UPDATE pulse_tasks SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID) + _, second := f.claim() + if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), second); err != nil { + t.Fatal(err) + } + result := json.RawMessage(`{"value":9007199254740993,"ratio":0.1234567890123456789012345}`) + finish := worker.FinishInput{Ref: second, Outcome: "SUCCEEDED", Result: result} + if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), finish); err != nil { + t.Fatal(err) + } + message := executionID(t) + finish.Result = json.RawMessage(`{"claimed_value":9007199254740995}`) + if _, err := f.a.Finish(ctx, f.token, f.binding, message, finish); err != nil { + t.Fatal(err) + } + if _, err := f.a.Finish(ctx, f.token, f.binding, message, finish); err != nil { + t.Fatal(err) + } + // 同一条 INSERT 中两条证据具有相同微秒时间，验证 ID 是稳定的分页次序补充。 + e1, e2 := executionID(t), executionID(t) + f.exec(`INSERT INTO pulse_attempt_evidence(namespace_id,id,attempt_id,message_id,outcome,payload,request_digest,reason,created_at) VALUES (?,?,?,?,'SUCCEEDED','{"tie":1}',?,'RESULT_CONFLICT',UTC_TIMESTAMP(6)),(?,?,?,?,'UNKNOWN','{"tie":2}',?,'LEASE_LOST',UTC_TIMESTAMP(6))`, f.resources.NamespaceID, e1, second.AttemptID, executionID(t), core.Digest{}, f.resources.NamespaceID, e2, second.AttemptID, executionID(t), core.Digest{}) + server := httptest.NewServer(httpserver.New("", func(context.Context) error { return nil }, f.tasks).Handler) + defer server.Close() + call := func(path, token string, status int) json.RawMessage { + t.Helper() + req, err := http.NewRequestWithContext(ctx, "GET", server.URL+path, nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer "+token) + resp, err := server.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + raw, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + if resp.StatusCode != status { + t.Fatalf("history HTTP %d wanted %d: %s", resp.StatusCode, status, raw) + } + if resp.Header.Get("Cache-Control") != "no-store" || resp.Header.Get("X-Content-Type-Options") != "nosniff" { + t.Fatal("missing history privacy headers") + } + var envelope struct { + RequestID core.ID `json:"request_id"` + Data json.RawMessage `json:"data"` + } + if json.Unmarshal(raw, &envelope) != nil || envelope.RequestID.Validate() != nil || resp.Header.Get("X-Request-ID") != string(envelope.RequestID) { + t.Fatal("invalid response envelope") + } + for _, secret := range []string{first.LeaseSecret, second.LeaseSecret, f.hello.InstanceSecret, "token_digest", "secret_digest", "request_digest", "result_digest"} { + if strings.Contains(string(raw), secret) { + t.Fatal("execution history exposed an internal proof") + } + } + return envelope.Data + } + base := "/v1/tasks/" + string(task.ID) + "/attempts" + var page service.AttemptPage + if err := json.Unmarshal(call(base+"?limit=1", f.resources.Token, 200), &page); err != nil { + t.Fatal(err) + } + if len(page.Items) != 1 || page.Items[0].ID != second.AttemptID || page.Items[0].Epoch != 2 || page.NextCursor == "" { + t.Fatal("wrong first attempt page") + } + if strings.Contains(string(call(base, f.resources.Token, 200)), `"result"`) { + t.Fatal("attempt list included large result") + } + attemptCursor := page.NextCursor + page = service.AttemptPage{} + if err := json.Unmarshal(call(base+"?cursor="+url.QueryEscape(attemptCursor), f.resources.Token, 200), &page); err != nil { + t.Fatal(err) + } + if len(page.Items) != 1 || page.Items[0].ID != first.AttemptID || page.Items[0].AuthorizedAt != nil || page.Items[0].EndedAt == nil || page.NextCursor != "" { + t.Fatal("wrong expired attempt history") + } + var detail service.AttemptView + if err := json.Unmarshal(call(base+"/"+string(second.AttemptID), f.resources.Token, 200), &detail); err != nil { + t.Fatal(err) + } + if detail.WorkerInstanceID != f.hello.InstanceID || detail.SessionID != f.binding.SessionID || detail.AuthorizedAt == nil || detail.EndedAt == nil || string(detail.Result) != string(result) { + t.Fatal("attempt detail lost owner, times or precise result") + } + evidenceBase := base + "/" + string(second.AttemptID) + "/evidence" + seen := map[core.ID]bool{} + cursor := "" + var evidenceCursor string + for { + var ep service.EvidencePage + raw := call(evidenceBase+"?limit=1&cursor="+url.QueryEscape(cursor), f.resources.Token, 200) + if json.Unmarshal(raw, &ep) != nil || len(ep.Items) != 1 { + t.Fatal("invalid evidence page") + } + if strings.Contains(string(raw), `"payload"`) { + t.Fatal("evidence page included business payload") + } + e := ep.Items[0] + if seen[e.ID] || e.Disposition != "EVIDENCE_ONLY" || e.TaskID != task.ID || e.AttemptID != second.AttemptID { + t.Fatal("evidence pagination duplicated or crossed identity") + } + seen[e.ID] = true + var ev service.EvidenceView + if json.Unmarshal(call(evidenceBase+"/"+string(e.ID), f.resources.Token, 200), &ev) != nil || !json.Valid(ev.Payload) { + t.Fatal("invalid evidence detail") + } + if ev.MessageID == message && string(ev.Payload) != string(finish.Result) { + t.Fatal("evidence numeric precision lost") + } + if ep.NextCursor == "" { + break + } + cursor = ep.NextCursor + evidenceCursor = cursor + } + if len(seen) != 3 || !seen[e1] || !seen[e2] { + t.Fatal("same-time pagination or duplicate evidence identity failed") + } + earlyBase := base + "/" + string(first.AttemptID) + "/evidence" + var early service.EvidencePage + if json.Unmarshal(call(earlyBase, f.resources.Token, 200), &early) != nil || len(early.Items) != 1 || early.Items[0].Reason != "NOT_STARTED" { + t.Fatal("early report is missing") + } + call(earlyBase+"/"+string(e1), f.resources.Token, 404) + call(earlyBase+"?cursor="+url.QueryEscape(evidenceCursor), f.resources.Token, 400) + call(base+"?cursor="+url.QueryEscape(evidenceCursor), f.resources.Token, 400) + call(evidenceBase+"?cursor="+url.QueryEscape(attemptCursor), f.resources.Token, 400) + newTask := f.submit(1) + emptyBase := "/v1/tasks/" + string(newTask.ID) + "/attempts" + var empty service.AttemptPage + if json.Unmarshal(call(emptyBase, f.resources.Token, 200), &empty) != nil || empty.Items == nil || len(empty.Items) != 0 { + t.Fatal("empty history must be []") + } + call(emptyBase+"/"+string(second.AttemptID), f.resources.Token, 404) + call(emptyBase+"?cursor="+url.QueryEscape(attemptCursor), f.resources.Token, 400) + for _, path := range []string{base, base + "/" + string(second.AttemptID), evidenceBase, evidenceBase + "/" + string(e1)} { + call(path, other.resources.Token, 404) + } + for _, query := range []string{"limit=0", "limit=101", "limit=1&limit=2", "cursor=%zz", "cursor=invalid", "unknown=1", "limit=1;limit=2", "cursor=" + strings.Repeat("a", 1025)} { + call(base+"?"+query, f.resources.Token, 400) + } + call(base+"/"+string(second.AttemptID)+"?cursor=x", f.resources.Token, 400) + call("/v1/tasks/invalid/attempts", f.resources.Token, 400) + permissions, _ := json.Marshal(auth.Permissions{Actions: []string{"tasks:read"}, Queues: []string{string(executionID(t))}}) + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(permissions), f.resources.NamespaceID, f.resources.CredentialID) + for _, path := range []string{base, base + "/" + string(second.AttemptID), evidenceBase, evidenceBase + "/" + string(e1)} { + call(path, f.resources.Token, 404) + } + permissions, _ = json.Marshal(auth.Permissions{Actions: []string{"queues:read"}, Queues: []string{"*"}}) + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(permissions), f.resources.NamespaceID, f.resources.CredentialID) + call(base, f.resources.Token, 403) + f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID) + call(base, f.resources.Token, 401) + var count int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='pulse_attempt_evidence' AND index_name='ix_evidence_history'`).Scan(&count); err != nil || count != 4 { + t.Fatal("history index migration missing") + } +} diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index 1686773..6873a77 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -13,6 +13,7 @@ import ( "git.yuntuo.hk/k2safe/orbit/internal/fault" "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" core "git.yuntuo.hk/k2safe/orbit/model" "git.yuntuo.hk/k2safe/orbit/platform/auth" pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" @@ -108,7 +109,7 @@ func reserveIdentity(ctx context.Context, tx *sql.Tx, ns core.ID, operation, sco return storedID, response, nil } func saveIdentity(ctx context.Context, tx *sql.Tx, ns, id core.ID, value any) error { - raw, err := json.Marshal(value) + raw, err := jsonwire.Marshal(value) if err != nil { return err } diff --git a/internal/transport/httpserver/history.go b/internal/transport/httpserver/history.go new file mode 100644 index 0000000..fe2a247 --- /dev/null +++ b/internal/transport/httpserver/history.go @@ -0,0 +1,65 @@ +package httpserver + +import ( + "net/http" + "net/url" + "strconv" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/pulse/service" +) + +func historyQuery(r *http.Request, paged bool) (service.HistoryPageInput, error) { + var in service.HistoryPageInput + q, err := url.ParseQuery(r.URL.RawQuery) + if err != nil { + return in, fault.Invalid("invalid query encoding") + } + for key, values := range q { + if !paged || len(values) != 1 || (key != "limit" && key != "cursor") { + return in, fault.Invalid("invalid history query parameter") + } + } + if q.Has("limit") { + in.Limit, err = strconv.Atoi(q.Get("limit")) + if err != nil || in.Limit < 1 { + return in, fault.Invalid("invalid history limit") + } + } + in.Cursor = q.Get("cursor") + return in, nil +} +func registerHistory(mux *http.ServeMux, s *service.Service) { + mux.HandleFunc("GET /v1/tasks/{task}/attempts", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + in, err := historyQuery(r, true) + if err != nil { + return nil, 0, err + } + out, err := s.ListTaskAttempts(r.Context(), t, core.ID(r.PathValue("task")), in) + return out, 200, err + })) + mux.HandleFunc("GET /v1/tasks/{task}/attempts/{attempt}", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + if _, err := historyQuery(r, false); err != nil { + return nil, 0, err + } + out, err := s.GetTaskAttempt(r.Context(), t, core.ID(r.PathValue("task")), core.ID(r.PathValue("attempt"))) + return out, 200, err + })) + mux.HandleFunc("GET /v1/tasks/{task}/attempts/{attempt}/evidence", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + in, err := historyQuery(r, true) + if err != nil { + return nil, 0, err + } + out, err := s.ListAttemptEvidence(r.Context(), t, core.ID(r.PathValue("task")), core.ID(r.PathValue("attempt")), in) + return out, 200, err + })) + mux.HandleFunc("GET /v1/tasks/{task}/attempts/{attempt}/evidence/{evidence}", wrap(func(r *http.Request, t auth.Token) (any, int, error) { + if _, err := historyQuery(r, false); err != nil { + return nil, 0, err + } + out, err := s.GetAttemptEvidence(r.Context(), t, core.ID(r.PathValue("task")), core.ID(r.PathValue("attempt")), core.ID(r.PathValue("evidence"))) + return out, 200, err + })) +} diff --git a/internal/transport/httpserver/server.go b/internal/transport/httpserver/server.go index ac1d0d1..08b18ab 100644 --- a/internal/transport/httpserver/server.go +++ b/internal/transport/httpserver/server.go @@ -32,6 +32,7 @@ func New(addr string, ready Checker, tasks ...*service.Service) *http.Server { }) if len(tasks) > 0 && tasks[0] != nil { registerTasks(mux, tasks[0]) + registerHistory(mux, tasks[0]) } return &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 15 * time.Second, IdleTimeout: 60 * time.Second, MaxHeaderBytes: 16 << 10} } diff --git a/internal/transport/httpserver/tasks.go b/internal/transport/httpserver/tasks.go index 4cf408e..8e061f7 100644 --- a/internal/transport/httpserver/tasks.go +++ b/internal/transport/httpserver/tasks.go @@ -17,6 +17,7 @@ import ( "git.yuntuo.hk/k2safe/orbit/internal/fault" "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" core "git.yuntuo.hk/k2safe/orbit/model" "git.yuntuo.hk/k2safe/orbit/platform/auth" pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" @@ -149,7 +150,7 @@ func wrap(next endpoint) http.HandlerFunc { } } func writeJSON(w http.ResponseWriter, status int, body any) { - raw, err := json.Marshal(body) + raw, err := jsonwire.Marshal(body) if err != nil { status = 503 raw = []byte(`{"error":{"code":"UNAVAILABLE","message":"operation unavailable"}}`) diff --git a/internal/transport/httpserver/worker.go b/internal/transport/httpserver/worker.go index c5178f3..0c60795 100644 --- a/internal/transport/httpserver/worker.go +++ b/internal/transport/httpserver/worker.go @@ -13,6 +13,7 @@ import ( "git.yuntuo.hk/k2safe/orbit/internal/fault" "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" core "git.yuntuo.hk/k2safe/orbit/model" "git.yuntuo.hk/k2safe/orbit/platform/auth" "git.yuntuo.hk/k2safe/orbit/pulse/worker" @@ -154,7 +155,7 @@ func (h *WorkerHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { _ = writeWorkerFrame(ctx, conn, workerFrame{Version: "1", Type: "error", MessageID: requestID, ReplyTo: hello.MessageID, Error: publicWorkerError(err)}) return } - welcomeRaw, err := json.Marshal(welcome) + welcomeRaw, err := jsonwire.Marshal(welcome) if err != nil { return } @@ -219,7 +220,7 @@ func (h *WorkerHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { return false } } - raw, err := json.Marshal(frame) + raw, err := jsonwire.Marshal(frame) if err != nil { return false } @@ -320,7 +321,7 @@ func (h *WorkerHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { reply.Type = "error" reply.Error = publicWorkerError(commandErr) } else { - reply.Body, commandErr = json.Marshal(body) + reply.Body, commandErr = jsonwire.Marshal(body) if commandErr != nil { return } @@ -354,7 +355,7 @@ func (h *WorkerHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { if assignment == nil { break } - raw, err := json.Marshal(assignment) + raw, err := jsonwire.Marshal(assignment) if err != nil { return } @@ -374,7 +375,7 @@ func publicWorkerError(err error) *fault.Error { return fault.Unavailable } func writeWorkerFrame(ctx context.Context, conn *websocket.Conn, frame workerFrame) error { - raw, err := json.Marshal(frame) + raw, err := jsonwire.Marshal(frame) if err != nil { return err } diff --git a/pulse/service/history.go b/pulse/service/history.go new file mode 100644 index 0000000..90410cb --- /dev/null +++ b/pulse/service/history.go @@ -0,0 +1,122 @@ +package service + +import ( + "context" + "encoding/json" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" +) + +// HistoryPageInput 用于任务执行历史的有界分页；游标必须沿原父资源继续使用。 +type HistoryPageInput struct { + Limit int // Limit 默认 20，最大 100；列表不返回业务结果正文。 + Cursor string // Cursor 服务端不透明游标，最长 1024 字节；为空从最新记录开始。 +} + +// AttemptSummary 是执行尝试的查询快照，不提供当前执行授权。 +type AttemptSummary struct { + ID core.ID `json:"id"` // ID 尝试标识，不是执行秘密。 + TaskID core.ID `json:"task_id"` // TaskID 所属任务；查询始终限定在该任务下。 + Epoch uint64 `json:"epoch,string"` // Epoch 单任务执行代次，JSON 为十进制字符串。 + State pulse.AttemptState `json:"state"` // State 尝试状态，与整个任务状态分开。 + Revision uint64 `json:"revision,string"` // Revision 当前记录版本，可因续租/恢复变化。 + LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 数据库记录的租约期限；查询不能授权业务执行。 + AuthorizedAt *time.Time `json:"authorized_at,omitempty"` // AuthorizedAt 开始批准时间；空表示未批准，不证明副作用已经发生。 + EndedAt *time.Time `json:"ended_at,omitempty"` // EndedAt 尝试结束时间；仍在执行时为空。 + ErrorCode string `json:"error_code,omitempty"` // ErrorCode 稳定错误分类；无错误时为空。 + CreatedAt time.Time `json:"created_at"` // CreatedAt 创建时间，UTC 微秒精度。 + UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最新变更时间，不能代替版本校验。 +} + +// AttemptView 增加当前归属与已保存结果，不返回令牌、摘要或原始请求秘密。 +type AttemptView struct { + AttemptSummary // AttemptSummary 尝试状态、执行代次与时间快照。 + SessionID core.ID `json:"session_id"` // SessionID 当前或最后归属会话；重连迁移后可变化。 + WorkerInstanceID core.ID `json:"worker_instance_id"` // WorkerInstanceID 对应进程身份，不是认证凭据。 + Result json.RawMessage `json:"result,omitempty"` // Result 已保存的尝试结果，未保存时省略；不包含迟到证据。 +} + +// AttemptPage 返回一页尝试摘要及可选的后续位置。 +type AttemptPage struct { + Items []AttemptSummary `json:"items"` // Items 按 Epoch 降序排列的摘要，空列表返回 []。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 后续页游标，空表示本次已读到末尾。 +} + +// EvidenceSummary 描述追加证据，原报告并未改变权威任务结果。 +type EvidenceSummary struct { + ID core.ID `json:"id"` // ID 不可变证据标识。 + TaskID core.ID `json:"task_id"` // TaskID 所属任务，从已验证的父关联确定。 + AttemptID core.ID `json:"attempt_id"` // AttemptID 所属执行尝试。 + Outcome string `json:"outcome"` // Outcome Worker 声明的 SUCCEEDED、FAILED 或 UNKNOWN，不代表平台认可。 + ErrorCode string `json:"error_code,omitempty"` // ErrorCode 原报告的稳定分类，可为空。 + Reason string `json:"reason"` // Reason LEASE_LOST、NOT_STARTED 或 RESULT_CONFLICT。 + Disposition string `json:"disposition"` // Disposition 固定为 EVIDENCE_ONLY，表示采集时仅保存证据。 + CreatedAt time.Time `json:"created_at"` // CreatedAt 首次收到时间，重复报告不改变它。 +} + +// EvidenceView 提供单条追加证据的原报告身份和正文。 +type EvidenceView struct { + EvidenceSummary // EvidenceSummary 原报告分类及采集时间。 + MessageID core.ID `json:"message_id"` // MessageID 原结果报告身份，用于追踪同一次重报。 + Payload json.RawMessage `json:"payload"` // Payload 原始结果/证据引用 JSON，保留数字文本；不提升为任务结果。 +} + +// EvidencePage 返回一页证据摘要，正文通过详情单独读取。 +type EvidencePage struct { + Items []EvidenceSummary `json:"items"` // Items 按 created_at、ID 降序的证据摘要，不含大正文。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 必须沿同一任务及尝试继续使用。 +} + +func normalizeHistoryPage(in HistoryPageInput) (HistoryPageInput, error) { + if in.Limit == 0 { + in.Limit = 20 + } + if in.Limit < 1 || in.Limit > 100 || len(in.Cursor) > 1024 { + return in, fault.Invalid("invalid history page limit or cursor") + } + return in, nil +} +func validateHistoryIDs(ids ...core.ID) error { + for _, id := range ids { + if id.Validate() != nil { + return fault.Invalid("invalid history resource id") + } + } + return nil +} +func (s *Service) ListTaskAttempts(ctx context.Context, t auth.Token, task core.ID, in HistoryPageInput) (AttemptPage, error) { + if err := validateHistoryIDs(task); err != nil { + return AttemptPage{}, err + } + in, err := normalizeHistoryPage(in) + if err != nil { + return AttemptPage{}, err + } + return s.store.ListTaskAttempts(ctx, t, task, in) +} +func (s *Service) GetTaskAttempt(ctx context.Context, t auth.Token, task, attempt core.ID) (AttemptView, error) { + if err := validateHistoryIDs(task, attempt); err != nil { + return AttemptView{}, err + } + return s.store.GetTaskAttempt(ctx, t, task, attempt) +} +func (s *Service) ListAttemptEvidence(ctx context.Context, t auth.Token, task, attempt core.ID, in HistoryPageInput) (EvidencePage, error) { + if err := validateHistoryIDs(task, attempt); err != nil { + return EvidencePage{}, err + } + in, err := normalizeHistoryPage(in) + if err != nil { + return EvidencePage{}, err + } + return s.store.ListAttemptEvidence(ctx, t, task, attempt, in) +} +func (s *Service) GetAttemptEvidence(ctx context.Context, t auth.Token, task, attempt, evidence core.ID) (EvidenceView, error) { + if err := validateHistoryIDs(task, attempt, evidence); err != nil { + return EvidenceView{}, err + } + return s.store.GetAttemptEvidence(ctx, t, task, attempt, evidence) +} diff --git a/pulse/service/types.go b/pulse/service/types.go index 35ffeb3..4476a20 100644 --- a/pulse/service/types.go +++ b/pulse/service/types.go @@ -183,4 +183,8 @@ type Store interface { CancelTask(context.Context, auth.Token, core.ID, string, CancelInput) (CancelResult, error) CreateQueue(context.Context, auth.Token, string, QueueInput) (QueueResult, error) GetQueue(context.Context, auth.Token, core.ID) (QueueView, error) + ListTaskAttempts(context.Context, auth.Token, core.ID, HistoryPageInput) (AttemptPage, error) + GetTaskAttempt(context.Context, auth.Token, core.ID, core.ID) (AttemptView, error) + ListAttemptEvidence(context.Context, auth.Token, core.ID, core.ID, HistoryPageInput) (EvidencePage, error) + GetAttemptEvidence(context.Context, auth.Token, core.ID, core.ID, core.ID) (EvidenceView, error) } diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 1dc9fa5..6811b41 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -52,3 +52,4 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - (cd sdk/go && ORBIT_TEST_WORKER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_WORKER_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" test -race -count=1 -run 'IntegrationWorker(Connection|Runtime)' -v .) (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/worker-protocol) (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/worker) +(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/history) diff --git a/sdk/go/examples/history/main.go b/sdk/go/examples/history/main.go new file mode 100644 index 0000000..f42d584 --- /dev/null +++ b/sdk/go/examples/history/main.go @@ -0,0 +1,151 @@ +// history 创建一次受控的冲突报告，演示通过公开查询区分权威结果和追加证据。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} +func run() error { + token, queue := os.Getenv("ORBIT_TOKEN"), os.Getenv("ORBIT_QUEUE_ID") + if path := os.Getenv("ORBIT_CREDENTIALS_FILE"); path != "" { + raw, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read private credentials: %w", err) + } + var c struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if json.Unmarshal(raw, &c) != nil { + return errors.New("invalid credentials file") + } + if token == "" { + token = c.Token + } + if queue == "" { + queue = c.QueueID + } + } + client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + if err != nil { + return err + } + op, err := orbit.NewWorkerMessageID() + if err != nil { + return err + } + handler := "history.inspect." + op + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + accepted, err := client.SubmitTask(ctx, "history-"+op, orbit.SubmitTaskRequest{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"object_ref":"examples/input"}`), TaskTimeoutMS: 60000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 10000}}) + if err != nil { + return err + } + registration, err := orbit.NewWorkerRegistration(queue, 1, []orbit.WorkerCapability{{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + return err + } + w, err := client.DialWorker(ctx, registration) + if err != nil { + return err + } + defer w.Close() + nextID := func() (string, error) { return orbit.NewWorkerMessageID() } + id, err := nextID() + if err != nil { + return err + } + if _, err := w.Capacity(ctx, id, orbit.WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + return err + } + a, err := w.NextAssignment(ctx) + if err != nil { + return err + } + if a.Task.ID != accepted.Task.ID { + return errors.New("unexpected task") + } + id, err = nextID() + if err != nil { + return err + } + started, err := w.Start(ctx, id, a.Attempt()) + if err != nil { + return err + } + if _, active := started.Lease.ExecutionDeadline(); !active { + return errors.New("missing current execution authority") + } + // 这是故障演示，不对外部系统执行操作；先提交固定的模拟处理结果。 + id, err = nextID() + if err != nil { + return err + } + original := json.RawMessage(`{"version":"v1","sequence":9007199254740993}`) + if _, err := w.Finish(ctx, id, orbit.WorkerFinish{Ref: a.Attempt(), Outcome: orbit.WorkerSucceeded, Result: original}); err != nil { + return err + } + // 使用另一消息身份模拟矛盾回报；只能追加证据，不能覆盖先前权威结果。 + id, err = nextID() + if err != nil { + return err + } + conflict, err := w.Finish(ctx, id, orbit.WorkerFinish{Ref: a.Attempt(), Outcome: orbit.WorkerSucceeded, Result: json.RawMessage(`{"version":"v2"}`)}) + if err != nil { + return err + } + if conflict.Receipt.Disposition != "EVIDENCE_ONLY" { + return errors.New("conflicting report was not isolated as evidence") + } + attempts, err := client.ListTaskAttempts(ctx, a.Task.ID, orbit.HistoryPageRequest{Limit: 20}) + if err != nil { + return err + } + if len(attempts.Items) != 1 { + return errors.New("expected one attempt") + } + detail, err := client.GetTaskAttempt(ctx, a.Task.ID, attempts.Items[0].ID) + if err != nil { + return err + } + evidence, err := client.ListAttemptEvidence(ctx, a.Task.ID, detail.ID, orbit.HistoryPageRequest{Limit: 20}) + if err != nil { + return err + } + if len(evidence.Items) != 1 { + return errors.New("expected one evidence record") + } + item, err := client.GetAttemptEvidence(ctx, a.Task.ID, detail.ID, evidence.Items[0].ID) + if err != nil { + return err + } + current, err := client.GetTask(ctx, a.Task.ID) + if err != nil { + return err + } + if current.State != orbit.TaskSucceeded || string(current.Result) != string(original) || item.Reason != "RESULT_CONFLICT" { + return errors.New("history did not preserve authoritative outcome") + } + id, err = nextID() + if err != nil { + return err + } + if _, err := w.Capacity(ctx, id, orbit.WorkerCapacity{Sequence: 2, Capacity: 0, Drain: true}); err != nil { + return err + } + fmt.Printf("task=%s state=%s attempts=%d evidence=%d reason=%s disposition=%s\n", current.ID, current.State, len(attempts.Items), len(evidence.Items), item.Reason, item.Disposition) + return nil +} diff --git a/sdk/go/history.go b/sdk/go/history.go new file mode 100644 index 0000000..4cad566 --- /dev/null +++ b/sdk/go/history.go @@ -0,0 +1,95 @@ +package orbit + +import ( + "context" + "errors" + "net/url" + "strconv" +) + +func historyResourcePath(ids ...string) (string, error) { + for _, id := range ids { + if !validID(id) { + return "", errors.New("invalid history resource id") + } + } + path := "/v1/tasks/" + ids[0] + "/attempts" + if len(ids) > 1 { + path += "/" + ids[1] + "/evidence" + } + if len(ids) > 2 { + path += "/" + ids[2] + } + return path, nil +} +func historyPageQuery(in HistoryPageRequest) (string, error) { + if in.Limit < 0 || in.Limit > 100 || len(in.Cursor) > 1024 { + return "", errors.New("invalid history limit or cursor") + } + q := url.Values{} + if in.Limit > 0 { + q.Set("limit", strconv.Itoa(in.Limit)) + } + if in.Cursor != "" { + q.Set("cursor", in.Cursor) + } + if len(q) == 0 { + return "", nil + } + return "?" + q.Encode(), nil +} + +// ListTaskAttempts 按执行代次降序读取摘要；当前查询不会授予执行权。 +func (c *Client) ListTaskAttempts(ctx context.Context, taskID string, in HistoryPageRequest) (AttemptPage, error) { + var out AttemptPage + path, err := historyResourcePath(taskID) + if err != nil { + return out, err + } + query, err := historyPageQuery(in) + if err != nil { + return out, err + } + err = c.call(ctx, "GET", path+query, "", nil, &out) + return out, err +} + +// GetTaskAttempt 获取当前归属及已保存结果，不提供执行秘密或执行授权。 +func (c *Client) GetTaskAttempt(ctx context.Context, taskID, attemptID string) (Attempt, error) { + var out Attempt + if !validID(attemptID) { + return out, errors.New("invalid attempt id") + } + path, err := historyResourcePath(taskID) + if err != nil { + return out, err + } + err = c.call(ctx, "GET", path+"/"+attemptID, "", nil, &out) + return out, err +} + +// ListAttemptEvidence 列出原先仅保存为证据的报告，不把声明结果提升为权威结果。 +func (c *Client) ListAttemptEvidence(ctx context.Context, taskID, attemptID string, in HistoryPageRequest) (AttemptEvidencePage, error) { + var out AttemptEvidencePage + path, err := historyResourcePath(taskID, attemptID) + if err != nil { + return out, err + } + query, err := historyPageQuery(in) + if err != nil { + return out, err + } + err = c.call(ctx, "GET", path+query, "", nil, &out) + return out, err +} + +// GetAttemptEvidence 读取单条原报告正文，不改变任务或尝试的权威状态。 +func (c *Client) GetAttemptEvidence(ctx context.Context, taskID, attemptID, evidenceID string) (AttemptEvidence, error) { + var out AttemptEvidence + path, err := historyResourcePath(taskID, attemptID, evidenceID) + if err != nil { + return out, err + } + err = c.call(ctx, "GET", path, "", nil, &out) + return out, err +} diff --git a/sdk/go/history_integration_test.go b/sdk/go/history_integration_test.go new file mode 100644 index 0000000..9e17f66 --- /dev/null +++ b/sdk/go/history_integration_test.go @@ -0,0 +1,81 @@ +package orbit + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" +) + +func TestIntegrationWorkerConnectionHistoryAndPayloadBounds(t *testing.T) { + client, queue, _ := integrationWorkerClient(t) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + handler := "history.bounds." + workerTestID(t) + // 原先默认 HTML 转义会把这份约 240 KiB 文档扩大到约 1.4 MiB。 + payload := json.RawMessage(`{"text":"` + strings.Repeat("<>&", 80000) + `","n":9007199254740993}`) + in := SubmitTaskRequest{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, TaskTimeoutMS: 60000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 30000}} + key := "history-bounds-" + workerTestID(t) + first, err := client.SubmitTask(ctx, key, in) + if err != nil { + t.Fatal(err) + } + again, err := client.SubmitTask(ctx, key, in) + if err != nil { + t.Fatal(err) + } + if !again.Deduplicated || again.Task.ID != first.Task.ID || string(again.Task.Payload) != string(payload) { + t.Fatal("durable receipt inflated or altered payload") + } + r, err := NewWorkerRegistration(queue, 1, []WorkerCapability{{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}}) + if err != nil { + t.Fatal(err) + } + w, err := client.DialWorker(ctx, r) + if err != nil { + t.Fatal(err) + } + defer w.Close() + if _, err := w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil { + t.Fatal(err) + } + assigned, err := w.NextAssignment(ctx) + if err != nil { + t.Fatal(err) + } + if assigned.Task.ID != first.Task.ID || string(assigned.Task.Payload) != string(payload) { + t.Fatal("assignment inflated or changed payload") + } + if _, err := w.Start(ctx, workerTestID(t), assigned.Attempt()); err != nil { + t.Fatal(err) + } + finish := WorkerFinish{Ref: assigned.Attempt(), Outcome: WorkerSucceeded, Result: payload} + if _, err := w.Finish(ctx, workerTestID(t), finish); err != nil { + t.Fatal(err) + } + page, err := client.ListTaskAttempts(ctx, first.Task.ID, HistoryPageRequest{Limit: 1}) + if err != nil || len(page.Items) != 1 || page.Items[0].ID != assigned.AttemptID { + t.Fatalf("attempt list: %v", err) + } + detail, err := client.GetTaskAttempt(ctx, first.Task.ID, assigned.AttemptID) + if err != nil || detail.State != AttemptSucceeded || detail.WorkerInstanceID != r.InstanceID || string(detail.Result) != string(payload) { + t.Fatalf("attempt detail: %v", err) + } + finish.Result = json.RawMessage(strings.Replace(string(payload), "9007199254740993", "9007199254740995", 1)) + if out, err := w.Finish(ctx, workerTestID(t), finish); err != nil || out.Receipt.Disposition != "EVIDENCE_ONLY" { + t.Fatalf("conflicting report: %v", err) + } + evidence, err := client.ListAttemptEvidence(ctx, first.Task.ID, assigned.AttemptID, HistoryPageRequest{Limit: 1}) + if err != nil || len(evidence.Items) != 1 || evidence.Items[0].Reason != "RESULT_CONFLICT" { + t.Fatalf("evidence summary: %v", err) + } + e, err := client.GetAttemptEvidence(ctx, first.Task.ID, assigned.AttemptID, evidence.Items[0].ID) + if err != nil || string(e.Payload) != string(finish.Result) || e.Disposition != "EVIDENCE_ONLY" { + t.Fatalf("evidence detail: %v", err) + } + task, err := client.GetTask(ctx, first.Task.ID) + if err != nil || string(task.Result) != string(payload) || task.State != TaskSucceeded { + t.Fatalf("authoritative result changed: %v", err) + } +} diff --git a/sdk/go/history_test.go b/sdk/go/history_test.go new file mode 100644 index 0000000..43395fe --- /dev/null +++ b/sdk/go/history_test.go @@ -0,0 +1,83 @@ +package orbit + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" +) + +func TestHistoryClientPathsPrecisionAndBounds(t *testing.T) { + const task = "11111111111111111111111111111111" + const attempt = "22222222222222222222222222222222" + const evidence = "33333333333333333333333333333333" + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if r.Method != "GET" || r.Header.Get("Authorization") != "Bearer example-token" || r.Header.Get("Idempotency-Key") != "" { + t.Error("invalid history request headers") + } + var data any + switch r.URL.Path { + case "/v1/tasks/" + task + "/attempts": + if r.URL.Query().Get("cursor") != "opaque+cursor/=" || r.URL.Query().Get("limit") != "2" { + t.Error("cursor encoding changed") + } + data = AttemptPage{Items: []AttemptSummary{{ID: attempt, TaskID: task, Epoch: 9007199254740993, Revision: 9007199254740995}}} + case "/v1/tasks/" + task + "/attempts/" + attempt: + data = Attempt{AttemptSummary: AttemptSummary{ID: attempt, TaskID: task}, Result: json.RawMessage(`{"n":9007199254740993}`)} + case "/v1/tasks/" + task + "/attempts/" + attempt + "/evidence": + data = AttemptEvidencePage{Items: []AttemptEvidenceSummary{{ID: evidence, AttemptID: attempt, TaskID: task, Disposition: "EVIDENCE_ONLY"}}} + case "/v1/tasks/" + task + "/attempts/" + attempt + "/evidence/" + evidence: + data = AttemptEvidence{AttemptEvidenceSummary: AttemptEvidenceSummary{ID: evidence}, Payload: json.RawMessage(`{"ratio":0.1234567890123456789012345}`)} + default: + t.Error("unexpected history path") + } + raw, _ := json.Marshal(map[string]any{"request_id": task, "data": data}) + w.Write(raw) + })) + defer server.Close() + c, _ := NewClient(server.URL, Options{Token: "example-token"}) + ctx := context.Background() + page, err := c.ListTaskAttempts(ctx, task, HistoryPageRequest{Limit: 2, Cursor: "opaque+cursor/="}) + if err != nil || page.Items[0].Epoch != 9007199254740993 || page.Items[0].Revision != 9007199254740995 { + t.Fatal("attempt integers lost precision") + } + a, err := c.GetTaskAttempt(ctx, task, attempt) + if err != nil || string(a.Result) != `{"n":9007199254740993}` { + t.Fatal("attempt result changed") + } + ep, err := c.ListAttemptEvidence(ctx, task, attempt, HistoryPageRequest{}) + if err != nil || len(ep.Items) != 1 || ep.Items[0].Disposition != "EVIDENCE_ONLY" { + t.Fatal("evidence classification changed") + } + e, err := c.GetAttemptEvidence(ctx, task, attempt, evidence) + if err != nil || string(e.Payload) != `{"ratio":0.1234567890123456789012345}` { + t.Fatal("evidence precision changed") + } + for _, id := range []string{"../queues", strings.Repeat("A", 32), "id?secret=x"} { + if _, err := c.ListTaskAttempts(ctx, id, HistoryPageRequest{}); err == nil { + t.Fatal("invalid parent id accepted") + } + if _, err := c.GetTaskAttempt(ctx, task, id); err == nil { + t.Fatal("invalid attempt id accepted") + } + if _, err := c.ListAttemptEvidence(ctx, task, id, HistoryPageRequest{}); err == nil { + t.Fatal("invalid evidence parent accepted") + } + if _, err := c.GetAttemptEvidence(ctx, task, attempt, id); err == nil { + t.Fatal("invalid evidence id accepted") + } + } + for _, page := range []HistoryPageRequest{{Limit: -1}, {Limit: 101}, {Cursor: strings.Repeat("x", 1025)}} { + if _, err := c.ListTaskAttempts(ctx, task, page); err == nil { + t.Fatal("invalid page accepted") + } + } + if calls.Load() != 4 { + t.Fatal("invalid history input reached server") + } +} diff --git a/sdk/go/history_types.go b/sdk/go/history_types.go new file mode 100644 index 0000000..d5810b1 --- /dev/null +++ b/sdk/go/history_types.go @@ -0,0 +1,79 @@ +package orbit + +import ( + "encoding/json" + "time" +) + +// HistoryPageRequest 用于任务执行历史的有界分页；游标必须沿原父资源继续使用。 +type HistoryPageRequest struct { + Limit int // Limit 默认 20，最大 100；列表不返回业务结果正文。 + Cursor string // Cursor 服务端不透明游标，最长 1024 字节；为空从最新记录开始。 +} + +// AttemptSummary 是执行尝试的查询快照，不提供当前执行授权。 +type AttemptSummary struct { + ID string `json:"id"` // ID 尝试标识，不是执行秘密。 + TaskID string `json:"task_id"` // TaskID 所属任务；查询始终限定在该任务下。 + Epoch uint64 `json:"epoch,string"` // Epoch 单任务执行代次，JSON 为十进制字符串。 + State AttemptState `json:"state"` // State 尝试状态，与整个任务状态分开。 + Revision uint64 `json:"revision,string"` // Revision 当前记录版本，可因续租/恢复变化。 + LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 数据库记录的租约期限；查询不能授权业务执行。 + AuthorizedAt *time.Time `json:"authorized_at,omitempty"` // AuthorizedAt 开始批准时间；空表示未批准，不证明副作用已经发生。 + EndedAt *time.Time `json:"ended_at,omitempty"` // EndedAt 尝试结束时间；仍在执行时为空。 + ErrorCode string `json:"error_code,omitempty"` // ErrorCode 稳定错误分类；无错误时为空。 + CreatedAt time.Time `json:"created_at"` // CreatedAt 创建时间，UTC 微秒精度。 + UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最新变更时间，不能代替版本校验。 +} + +// Attempt 增加当前归属与已保存结果，不返回令牌、摘要或原始请求秘密。 +type Attempt struct { + AttemptSummary // AttemptSummary 尝试状态、执行代次与时间快照。 + SessionID string `json:"session_id"` // SessionID 当前或最后归属会话；重连迁移后可变化。 + WorkerInstanceID string `json:"worker_instance_id"` // WorkerInstanceID 对应进程身份，不是认证凭据。 + Result json.RawMessage `json:"result,omitempty"` // Result 已保存的尝试结果，未保存时省略；不包含迟到证据。 +} + +// AttemptPage 返回一页尝试摘要及可选的后续位置。 +type AttemptPage struct { + Items []AttemptSummary `json:"items"` // Items 按 Epoch 降序排列的摘要，空列表返回 []。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 后续页游标，空表示本次已读到末尾。 +} + +// AttemptEvidenceSummary 描述追加证据，原报告并未改变权威任务结果。 +type AttemptEvidenceSummary struct { + ID string `json:"id"` // ID 不可变证据标识。 + TaskID string `json:"task_id"` // TaskID 所属任务，从已验证的父关联确定。 + AttemptID string `json:"attempt_id"` // AttemptID 所属执行尝试。 + Outcome string `json:"outcome"` // Outcome Worker 声明的 SUCCEEDED、FAILED 或 UNKNOWN，不代表平台认可。 + ErrorCode string `json:"error_code,omitempty"` // ErrorCode 原报告的稳定分类，可为空。 + Reason string `json:"reason"` // Reason LEASE_LOST、NOT_STARTED 或 RESULT_CONFLICT。 + Disposition string `json:"disposition"` // Disposition 固定为 EVIDENCE_ONLY，表示采集时仅保存证据。 + CreatedAt time.Time `json:"created_at"` // CreatedAt 首次收到时间，重复报告不改变它。 +} + +// AttemptEvidence 提供单条追加证据的原报告身份和正文。 +type AttemptEvidence struct { + AttemptEvidenceSummary // AttemptEvidenceSummary 原报告分类及采集时间。 + MessageID string `json:"message_id"` // MessageID 原结果报告身份，用于追踪同一次重报。 + Payload json.RawMessage `json:"payload"` // Payload 原始结果/证据引用 JSON，保留数字文本；不提升为任务结果。 +} + +// AttemptEvidencePage 返回一页证据摘要，正文通过详情单独读取。 +type AttemptEvidencePage struct { + Items []AttemptEvidenceSummary `json:"items"` // Items 按 created_at、ID 降序的证据摘要，不含大正文。 + NextCursor string `json:"next_cursor,omitempty"` // NextCursor 必须沿同一任务及尝试继续使用。 +} + +// AttemptState 表示一次尝试的状态，与整个任务的状态分开。 +type AttemptState string + +const ( + AttemptLeased AttemptState = "LEASED" + AttemptRunning AttemptState = "RUNNING" + AttemptSucceeded AttemptState = "SUCCEEDED" + AttemptFailed AttemptState = "FAILED" + AttemptCancelled AttemptState = "CANCELLED" + AttemptExpired AttemptState = "EXPIRED" + AttemptUnknown AttemptState = "UNKNOWN" +) diff --git a/sdk/go/json_wire.go b/sdk/go/json_wire.go new file mode 100644 index 0000000..5d8f67d --- /dev/null +++ b/sdk/go/json_wire.go @@ -0,0 +1,18 @@ +package orbit + +import ( + "bytes" + "encoding/json" +) + +// marshalWire 不做 HTML 转义，保持 RawMessage 精确数字及受控载荷大小。 +// 这些文档只用于 JSON/WS 通道，不能直接嵌入 HTML。 +func marshalWire(value any) ([]byte, error) { + var buffer bytes.Buffer + encoder := json.NewEncoder(&buffer) + encoder.SetEscapeHTML(false) + if err := encoder.Encode(value); err != nil { + return nil, err + } + return bytes.TrimSuffix(buffer.Bytes(), []byte{'\n'}), nil +} diff --git a/sdk/go/tasks.go b/sdk/go/tasks.go index a4a0379..7e62b9a 100644 --- a/sdk/go/tasks.go +++ b/sdk/go/tasks.go @@ -106,7 +106,7 @@ func (c *Client) call(ctx context.Context, method, path, key string, in, out any var body []byte var err error if in != nil { - body, err = json.Marshal(in) + body, err = marshalWire(in) if err != nil { return err } diff --git a/sdk/go/worker_connection.go b/sdk/go/worker_connection.go index fa9cf53..2189d71 100644 --- a/sdk/go/worker_connection.go +++ b/sdk/go/worker_connection.go @@ -95,8 +95,8 @@ func (c *Client) DialWorker(ctx context.Context, registration WorkerRegistration if err != nil { return nil, err } - body, _ := json.Marshal(r) - frame, _ := json.Marshal(workerFrame{Version: "1", Type: "worker.hello", MessageID: id, Body: body}) + body, _ := marshalWire(r) + frame, _ := marshalWire(workerFrame{Version: "1", Type: "worker.hello", MessageID: id, Body: body}) if err = conn.Write(handshake, websocket.MessageText, frame); err != nil { return nil, workerIOError(err) } @@ -249,11 +249,11 @@ func (w *WorkerConnection) call(ctx context.Context, kind, id string, in, out an w.mu.Unlock() defer func() { w.mu.Lock(); delete(w.pending, id); w.mu.Unlock() }() // 先占用有界槽再分配编码缓冲，超额调用不先复制大结果。 - body, err := json.Marshal(in) + body, err := marshalWire(in) if err != nil { return started, errors.New("invalid Worker request JSON") } - raw, err := json.Marshal(workerFrame{Version: "1", Type: kind, MessageID: id, SessionID: w.welcome.SessionID, Body: body}) + raw, err := marshalWire(workerFrame{Version: "1", Type: kind, MessageID: id, SessionID: w.welcome.SessionID, Body: body}) if err != nil || len(raw) > 512<<10 { return started, errors.New("Worker request exceeds frame limit") }