From b17d9b19771c578b9d46fb927d0b9e9aee7d54c8 Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 20:22:41 +0800 Subject: [PATCH] feat: add reliable event consumer SDK and process recovery tests --- AGENTS.md | 2 +- README.md | 2 +- docs/07-data-model.md | 6 +- docs/08-development-plan.md | 5 +- docs/09-result-subscription-core.md | 4 +- docs/api/README.md | 2 +- docs/api/authentication.md | 2 +- docs/api/subscriptions.md | 2 +- docs/examples/README.md | 9 +- docs/examples/event-protocol.md | 30 ++ docs/reports/phase-3-event-consumer-sdk.md | 75 +++++ docs/sdk/README.md | 6 +- docs/sdk/event-consumer.md | 94 ++++++ scripts/test-mysql.sh | 4 +- sdk/go/doc.go | 3 +- sdk/go/event_codec.go | 152 +++++++++ sdk/go/event_connection.go | 285 ++++++++++++++++ sdk/go/event_connection_test.go | 283 ++++++++++++++++ sdk/go/event_consumer.go | 368 +++++++++++++++++++++ sdk/go/event_consumer_test.go | 342 +++++++++++++++++++ sdk/go/event_integration_test.go | 190 +++++++++++ sdk/go/event_process_integration_test.go | 103 ++++++ sdk/go/event_types.go | 87 +++++ sdk/go/examples/event-protocol/main.go | 112 +++++++ sdk/go/worker_integration_test.go | 30 +- 25 files changed, 2170 insertions(+), 28 deletions(-) create mode 100644 docs/examples/event-protocol.md create mode 100644 docs/reports/phase-3-event-consumer-sdk.md create mode 100644 docs/sdk/event-consumer.md create mode 100644 sdk/go/event_codec.go create mode 100644 sdk/go/event_connection.go create mode 100644 sdk/go/event_connection_test.go create mode 100644 sdk/go/event_consumer.go create mode 100644 sdk/go/event_consumer_test.go create mode 100644 sdk/go/event_integration_test.go create mode 100644 sdk/go/event_process_integration_test.go create mode 100644 sdk/go/event_types.go create mode 100644 sdk/go/examples/event-protocol/main.go diff --git a/AGENTS.md b/AGENTS.md index 27d486f..8ac6bd3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,4 +14,4 @@ - 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 - 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 -- 可靠订阅已完成核心、HTTP/WSS及管理SDK；事件消费SDK、可运行Inbox示例和真实订阅进程故障回归完成前，不标记P3订阅能力完成。 +- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK和节点进程故障回归；业务侧持久化Inbox示例及消费者进程退出后的事务去重回归完成前，不标记P3订阅能力完成。 diff --git a/README.md b/README.md index 81e705f..52de16a 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已完成 P0–P2 阶段验收：35 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 、[队列资源控制](docs/api/queues.md) 、[应用共享配额](docs/api/application-resources.md) 、[资源键互斥](docs/api/keyed-concurrency.md) 和 [优先级/节点公平派发](docs/api/priority-and-fairness.md) 已接入服务端和 SDK，[可靠结果订阅](docs/api/subscriptions.md) 已实现持久化核心、HTTP/WSS、管理 SDK 与示例，消费 SDK 与 Inbox 场景待完成；剩余 Pulse 能力与流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 +已完成 P0–P2 阶段验收：35 个数据模型、版本化迁移、凭据及范围校验、队列与任务 API、WSS 执行、独立 SDK、自动 Worker 和 [执行尝试/证据查询](docs/api/execution-history.md)。[Worker Schema](docs/api/worker-schema.md) 与真实进程强杀恢复回归已通过；[安全重试](docs/api/retries.md)、[定时计划](docs/api/schedules.md)、[运行中取消](docs/api/cancellation.md) 、[队列资源控制](docs/api/queues.md) 、[应用共享配额](docs/api/application-resources.md) 、[资源键互斥](docs/api/keyed-concurrency.md) 和 [优先级/节点公平派发](docs/api/priority-and-fairness.md) 已接入服务端和 SDK，[可靠结果订阅](docs/api/subscriptions.md) 已实现持久化核心、HTTP/WSS、管理 SDK 与示例，[消费 SDK](docs/sdk/event-consumer.md) 已实现，Inbox 场景待完成；剩余 Pulse 能力与流程引擎按 [持续开发计划](docs/08-development-plan.md) 继续实现。 数据库禁止外键，关联一致性通过服务端事务校验，索引、唯一约束和 CHECK 保留；readiness 同时检查迁移及零外键要求。这不表示完整平台已具备生产服务能力。验证证据见 [P0](docs/reports/phase-0.md)、[P1](docs/reports/phase-1.md)、[P2 阶段验收](docs/reports/phase-2.md)、[安全重试](docs/reports/phase-3-safe-retry.md)、[计划调度](docs/reports/phase-3-schedules.md)、[取消与无外键回归](docs/reports/phase-3-cancellation.md) 及 [队列资源回归](docs/reports/phase-3-queue-resources.md)。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index f8b5603..74f7249 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -1,6 +1,6 @@ # 数据模型与实现基线 -更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS 与管理 SDK 已实现，消费 SDK 及流程编排继续开发。 +更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现，Inbox 场景及流程编排继续开发。 ## 1. 模块与依赖 @@ -95,7 +95,7 @@ 禁止“先查后另开事务写入”。需要防止并发状态改变的关联行在事务内持锁，统一锁顺序并保留必要索引。当前不提供硬删除；将来增加清理时，必须核验在途任务、引用、幂等恢复窗口与审计保留期，按有限批次处理，不能依赖数据库级联。直接绕过服务写 SQL 不受应用校验保护。 -P1 已完成动作/队列授权、幂等摘要与原回执、受理事件及取消 CAS/审计事务；P3 扩展到运行中取消。订阅已实现服务层及持久化核心，包含事务内队列、所有者和交付归属校验；HTTP/WSS 和管理 SDK 已接入，消费 SDK 待完成。流程目前仍只具备模型，后续引擎也必须补齐事务内父资源归属验证。 +P1 已完成动作/队列授权、幂等摘要与原回执、受理事件及取消 CAS/审计事务；P3 扩展到运行中取消。订阅已实现服务层及持久化核心，包含事务内队列、所有者和交付归属校验；HTTP/WSS、管理与消费 SDK 已接入，业务 Inbox 示例待完成。流程目前仍只具备模型，后续引擎也必须补齐事务内父资源归属验证。 P2 服务端已新增会话隔离、两步派发、租约、并发许可、结果回执/证据和恢复扫描，见 [执行协议](api/worker.md)。底层及自动 Worker SDK、尝试/证据查询、Schema 与进程故障恢复已通过 [P2 阶段验收](reports/phase-2.md)。 @@ -151,7 +151,7 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 ## 13. 可靠结果订阅核心 -000013_result_subscriptions 新增 EventStream、ResultEvent，扩展 Subscription 和 Delivery。289 个直接声明字段均有中文注释。生产事务在独立流行锁下分配序号，与任务、原事件和结果快照一起提交；消费事务不锁流行、任务、队列或应用额度。交付记录和扫描位置一起落库，每条交付单独持有租约和证明。重复确认不重复扣减积压，耗尽保留记录，人工追加预算不重置历史。详见 [订阅核心](09-result-subscription-core.md)。对外 HTTP/WSS 与管理 SDK 已实现，消费 SDK 与完整场景继续推进。 +000013_result_subscriptions 新增 EventStream、ResultEvent，扩展 Subscription 和 Delivery。289 个直接声明字段均有中文注释。生产事务在独立流行锁下分配序号，与任务、原事件和结果快照一起提交；消费事务不锁流行、任务、队列或应用额度。交付记录和扫描位置一起落库，每条交付单独持有租约和证明。重复确认不重复扣减积压，耗尽保留记录，人工追加预算不重置历史。详见 [订阅核心](09-result-subscription-core.md)。对外 HTTP/WSS、管理与消费 SDK 已实现，持久化 Inbox 完整场景继续推进。 ## 14. 交付历史查询索引 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 114a518..5d7f0b1 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -27,7 +27,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 验收报告 | -| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、Inbox 场景及订阅进程故障回归继续开发 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景及消费者进程故障回归继续开发 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -48,7 +48,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 跨命名空间应用配额 | first 受理 → second 全局满额 → 释放 → 原键成功 → 恢复配置 | P3，已实现完整示例 | | 跨队列共享资源 | 两队列同键 → 一个有效许可 → 递增序号 → 资源端拒绝旧写入 | P3，已实现完整示例 | | 优先级与应用隔离 | 近同时到期按优先级 → 旧任务等待补偿 → 应用/队列轮转 | P3，已实现完整示例 | -| 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，核心、HTTP/WSS、管理 SDK/示例已实现；消费 SDK 与 Inbox 场景待完成 | +| 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，核心、HTTP/WSS、管理 SDK/示例已实现；消费 SDK 已实现，持久化 Inbox 场景待完成 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | | 审核与外部回执 | 提交流程 → 等待信号 → 授权信号注入 → 继续 | P4 | | 外部作业轮询 | 提交一次 → 有界查询处理中状态 → 回执或最终结果 → 超限核查 | P5 | @@ -78,6 +78,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [可靠结果订阅核心](09-result-subscription-core.md) - [P3 订阅核心回归](reports/phase-3-result-subscription-core.md) - [P3 订阅通信与管理 SDK 回归](reports/phase-3-result-subscription-transport.md) +- [P3 消费 SDK 与节点故障回归](reports/phase-3-event-consumer-sdk.md) - [接口手册](api/README.md) - [SDK 接入](sdk/README.md) - [通用场景示例](examples/README.md) diff --git a/docs/09-result-subscription-core.md b/docs/09-result-subscription-core.md index 99bf22f..6c0b6da 100644 --- a/docs/09-result-subscription-core.md +++ b/docs/09-result-subscription-core.md @@ -1,6 +1,6 @@ # 可靠结果订阅：持久化核心 -状态：P3 的存储与服务层已实现并通过本地全量回归，见 [验证记录](reports/phase-3-result-subscription-core.md)。[HTTP/WSS、管理 SDK 与管理示例](api/subscriptions.md) 已接入。事件连接 SDK、自动消费和持久化 Inbox 示例继续开发；本页的存储方法属于服务端内部模块。 +状态：P3 的存储与服务层已实现并通过本地全量回归，见 [验证记录](reports/phase-3-result-subscription-core.md)。[HTTP/WSS、管理 SDK 与管理示例](api/subscriptions.md) 已接入。[事件连接 SDK 与自动消费器](sdk/event-consumer.md) 已实现，持久化 Inbox 示例继续开发；本页的存储方法属于服务端内部模块。 ## 1. 已实现范围 @@ -91,4 +91,4 @@ retained_from_sequence 初始且目前固定为 1。未来实现清理后，如 ## 8. 下一步 -HTTP 创建/查询/分页、暂停/继续、交付与耗尽记录查询、人工重试，以及 orbit.events.v1 WSS、消息 Schema、管理 SDK 和管理示例已实现，见 [接口手册](api/subscriptions.md)。接下来完成事件连接 SDK、自动消费及持久化 Inbox 去重示例，并补 SDK 断线/ACK 丢失和节点进程强杀回归，再验收 P3 的可靠订阅能力。 +HTTP 创建/查询/分页、暂停/继续、交付与耗尽记录查询、人工重试，以及 orbit.events.v1 WSS、消息 Schema、管理 SDK 和管理示例已实现，见 [接口手册](api/subscriptions.md)。[事件连接 SDK 和自动消费器](sdk/event-consumer.md) 已实现；继续完成持久化 Inbox 去重示例与消费者进程故障回归，再验收 P3 的可靠订阅能力。 diff --git a/docs/api/README.md b/docs/api/README.md index 2652377..579e378 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -21,7 +21,7 @@ | `/v1/tasks/{task_id}:cancel` 与 `task.cancel` 通知 | P3 运行中取消已实现 | [取消及 SDK 场景](cancellation.md) | | `/v1/schedules`、不可变版本、暂停/恢复及触发查询 | P3 已实现 | [计划接口与 SDK 场景](schedules.md) | | `/v1/subscriptions` 及管理、交付查询与重试 | 已实现 HTTP、管理 SDK 和完整管理示例 | [订阅手册](subscriptions.md) | -| `/v1/events/connect` | 已实现服务端 WSS 与消息 Schema；消费 SDK 待交付 | [消息协议](subscriptions.md#5-websocket-建连与边界) | +| `/v1/events/connect` | 已实现 WSS、消息 Schema、连接 SDK 和自动消费器 | [消息协议](subscriptions.md#5-websocket-建连与边界) / [消费 SDK](../sdk/event-consumer.md) | | `/v1/workflows` 与流程管理 | 计划 P4–P5，尚未实现 | [流程接口规划](workflows.md) | 未实现路由返回 404。健康检查就绪不表示任务执行、定时调度或流程引擎已经可用。 diff --git a/docs/api/authentication.md b/docs/api/authentication.md index 3894338..c545c80 100644 --- a/docs/api/authentication.md +++ b/docs/api/authentication.md @@ -58,4 +58,4 @@ P2 增加 workers:connect 和 workers:consume。bootstrap 只有显式提供 --w P3 应用额度管理新增 `application:resources:read` 和 `application:resources:manage`。只有显式 `--resource-access` 才授予，作用域是所属应用的全部命名空间，普通队列管理不能替代。查询返回聚合计数，不开放其他命名空间的任务数据。`bootstrap --application-id <现有应用ID>` 可由数据库管理员增加命名空间，与 `--application` 二选一；验证应用启用及额度记录存在后，同事务创建新资源，所有秘密交付约定保持不变。应用默认未终结上限 1000000、并发 10000、开始速率 100000/秒；详见 [应用共享配额](application-resources.md)。 -P3 结果订阅核心新增 `events:read`、`events:manage` 和 `events:consume`，bootstrap 仅显式 `--event-access` 时授予。订阅绑定创建凭据与单个队列，不能用同命名空间的另一张凭据消费或管理。当前 HTTP/WSS 和管理 SDK 已接入；消费 SDK、凭据重绑继续开发，详见 [订阅接口](subscriptions.md)。 +P3 结果订阅核心新增 `events:read`、`events:manage` 和 `events:consume`，bootstrap 仅显式 `--event-access` 时授予。订阅绑定创建凭据与单个队列，不能用同命名空间的另一张凭据消费或管理。当前 HTTP/WSS、管理及消费 SDK 已接入；凭据重绑继续开发，详见 [订阅接口](subscriptions.md)。 diff --git a/docs/api/subscriptions.md b/docs/api/subscriptions.md index eccdea5..8c18446 100644 --- a/docs/api/subscriptions.md +++ b/docs/api/subscriptions.md @@ -322,6 +322,6 @@ _, err = client.ConfigureSubscription(ctx, current.ID, pauseKey, orbit.Configure return err ``` -[完整管理示例](../../sdk/go/examples/subscriptions/main.go) 的启动、输入输出、失败和保留规则见 [场景文档](../examples/subscriptions.md)。它演示管理与幂等，不启动结果消费者。WSS 连接 SDK、自动重连和持久化 Inbox 示例仍在下一阶段开发，不能把测试中的原始 WebSocket 辅助代码当成正式 SDK。 +[完整管理示例](../../sdk/go/examples/subscriptions/main.go) 的启动、输入输出、失败和保留规则见 [场景文档](../examples/subscriptions.md)。它演示管理与幂等，不启动结果消费者。正式 [事件连接 SDK 与自动消费器](../sdk/event-consumer.md) 已提供接收、ACK、心跳、重连与有界收尾；[完整协议示例](../examples/event-protocol.md) 验证跨连接重报原确认。持久化 Inbox 示例继续开发。 机器可读消息契约见 [events.schema.json](../../protocol/eventsv1/events.schema.json)，使用 Draft 2020-12，clientFrame/serverFrame 分方向约束。运行时额外拒绝重复 JSON 键、超范围 uint64 及非规范 base64url。Schema 不代替授权、状态机、租约及事务校验。 diff --git a/docs/examples/README.md b/docs/examples/README.md index 62707cd..4d42d2c 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -114,4 +114,11 @@ - 步骤：创建 → 同键核对 → 暂停 → 恢复 → 查询与分页 → 最终暂停。 - 预期：duplicate=true、resumed_revision=3、final_enabled=false、revision=4、pending=0。 - 保留：每次运行留下一个已暂停订阅及审计；没有硬删除，受64个订阅上限约束。 -- 范围：本例只演示管理SDK；自动消费、断线重连与持久化Inbox例子继续开发。 +- 范围：本例只演示管理SDK；自动消费 SDK 已实现，持久化Inbox例子继续开发。 + +## 已可运行：结果消费协议与跨连接确认 + +- 源码：[事件协议示例](../../sdk/go/examples/event-protocol/main.go)，[启动与输出说明](event-protocol.md)。 +- 操作：创建订阅 → 提交并取消任务 → 接收冻结结果 → ACK → 重连重报原证明 → 核对原确认时间 → 暂停订阅。 +- 输出：ack_replayed=true、same_receipt=true、pending=0、subscription_enabled=false。 +- 范围：本例只做读取验证，不写业务数据；[自动消费器](../sdk/event-consumer.md) 的生产接入须提供事务Inbox，不能把示例当作业务恰好一次保证。 diff --git a/docs/examples/event-protocol.md b/docs/examples/event-protocol.md new file mode 100644 index 0000000..1b604c2 --- /dev/null +++ b/docs/examples/event-protocol.md @@ -0,0 +1,30 @@ +# 结果消费协议示例 + +[完整代码](../../sdk/go/examples/event-protocol/main.go) 使用正式 Go SDK 创建取消结果订阅，提交并取消一个待执行任务，接收并验证冻结结果，然后 ACK；关闭连接后在新连接中重报原证明，验证原确认时间和去重回执，最后暂停本次订阅。 + +本例只读取并验证结果，没有业务数据库写入。生产业务应使用 [自动消费器及事务去重约定](../sdk/event-consumer.md)。持久化 Inbox 完整示例尚待交付。 + +## 启动 + +先按 [订阅管理示例](subscriptions.md) 建立隔离示例队列和带 --event-access 的凭据，启动 Orbit。凭据需要 tasks:submit、tasks:cancel、events:manage/read/consume 及该队列范围；订阅绑定创建凭据，不能临时换凭据消费。 + +从 sdk/go 目录运行： + +```sh +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/event.credentials.json \ + go run ./examples/event-protocol +``` + +也可以传 ORBIT_TOKEN 与 ORBIT_QUEUE_ID，环境变量覆盖凭据文件对应值。每次运行新建随机订阅及任务身份，全部流程最多 15 秒。本例是短流程，两个连接均短于一次心跳周期；长期运行必须用自动消费器或主动 Ping。 + +## 输出与核对 + +```text +event=<本次事件ID> task=<本次任务ID> state=CANCELLED +ack_replayed=true same_receipt=true pending=0 subscription_enabled=false +``` + +该输出验证的是事件快照和平台 ACK 去重，不表示业务副作用获得恰好一次保证。任何请求失败、收到非本次任务的事件或确认时间不一致，都非零退出；保留已生成的状态供核查。 + +需要独立示例队列，避免其他程序同时产生取消结果。正常结束暂停本次订阅，保留任务、交付和审计；不删除历史。每个命名空间最多 64 个订阅，重复演示应使用独立测试环境。`make integration` 在一次性 MySQL 和真实进程上实际执行本例。 diff --git a/docs/reports/phase-3-event-consumer-sdk.md b/docs/reports/phase-3-event-consumer-sdk.md new file mode 100644 index 0000000..3fd01e2 --- /dev/null +++ b/docs/reports/phase-3-event-consumer-sdk.md @@ -0,0 +1,75 @@ +# P3 事件消费 SDK 与节点故障回归 + +验证日期：2026-09-17。已交付 [消费 SDK](../sdk/event-consumer.md) 和 [协议示例](../examples/event-protocol.md)。业务端持久化 Inbox 示例及消费者进程强杀恢复尚未完成，P3 和持续目标继续进行。 + +## 1. 实现范围 + +- [EventConnection](../../sdk/go/event_connection.go)：严格事件协议、接收、请求关联、ACK、心跳和关闭；控制最多32个在途，接收缓冲为声明容量，超时关闭连接后保留原证明核对。 +- [EventConsumer](../../sdk/go/event_consumer.go)：自动心跳、指数退避重连、本地1–32个在途事件、成功后确认、ACK不明时原证明核对、处理器失败不确认、取消与有界收尾。 +- [类型及秘密隐藏](../../sdk/go/event_types.go)：ResultEvent不给业务处理器租约秘密；Grant/Proof仍可显式序列化用于受保护的存储，普通fmt隐藏敏感数据。 +- [严格解码](../../sdk/go/event_codec.go)：大小写、重复键、方向、会话、规范uint64、证明编码和事件/交付/快照关联。大整数不经float64转换。协议类WebSocket关闭按永久错误处理。 +- [协议例子](../../sdk/go/examples/event-protocol/main.go)：产生取消结果、读取、ACK、重连重报、验证原确认时间、暂停订阅。它只读取结果，不执行接收方业务写入。 +- SDK保持独立模块，无新增数据库驱动、服务端包或持久化模型。本阶段不增加迁移、外键或级联删除。 + +## 2. 验证环境与入口 + +macOS、Go1.27.1、Docker MySQL8.4.10、一次性数据库；普通SDK集成连接真实Orbit进程，故障网关使用本地TLS，进程回归使用真实子进程和SIGKILL。注入后用Wait确认进程已经退出，不能以发送了Kill调用代替终止证据。 + +```sh +make fmt test vet build +make integration +cd sdk/go +GOWORK=off go test -race ./... +``` + +| 分组 | 结果 | +| --- | --- | +| 全量MySQL集成 | PASS，92.995秒 | +| SDK Worker连接/运行器 | PASS，88.494秒 | +| SDK计划、队列、订阅管理、应用配额 | PASS，5.704 / 4.466 / 33.386 / 1.363秒 | +| 新消费SDK三项集成 | PASS，2.18 / 15.28 / 33.14秒，包总计51.843秒 | +| 新节点交付租约恢复 | PASS，32.15秒 | +| 新ACK提交/丢响应/节点强杀 | PASS，1.31秒 | +| 九项真实进程故障（原七项加新两项） | PASS，包总计61.810秒 | +| 完整SDK示例 | 14个全部通过，包括新的event-protocol | +| fmt、race、vet、build | PASS | +| GOWORK=off独立SDK模块 | PASS，编译及竞态测试不依赖服务端workspace | + +完整SDK示例输出包含： + +```text +event=<本次事件ID> task=<本次任务ID> state=CANCELLED +ack_replayed=true same_receipt=true pending=0 subscription_enabled=false +``` + +## 3. 协议与运行器边界 + +[连接单元回归](../../sdk/go/event_connection_test.go) 在TLS服务器上混合投递与乱序控制响应，验证独立关联、精确数字、秘密隐藏；取消Next不关闭连接，已发ACK取消关闭连接并报告未知确认状态。32个控制槽满后拒绝新控制，交付缓冲满后关闭。拒绝重复键、大小写别名、非规范计数、null布尔/错误、缺失创建时间、错误事件关联、状态不匹配、额外字段及不匹配ACK；1008关闭不无限重连。 + +[运行器回归](../../sdk/go/event_consumer_test.go) 验证ACK回应丢失后原证明重连，不重复运行仍保留的已提交处理器；新代次替换后重新进入业务去重；error/panic/Goexit均不确认、不立即本地重试。取消后收集已提交结果，忽略取消的处理器在收尾期限到达后报告Running与Unconfirmed。永久401只连接一次，实例不能重复Run，容量2实际并行处理并限制并发。 + +处理器测试中的原子计数只用于确认调用次数，不能作为已实现持久化Inbox的证据。 + +## 4. 真实服务与MySQL + +[三项SDK集成](../../sdk/go/event_integration_test.go)： + +1. 接收已取消任务的冻结快照，暂停后Ping返回enabled=false，已有Proof仍能ACK；关闭连接后重报返回同一确认时间。暂停时新增结果不投递，恢复后继续消费，最终pending_count=0。 +2. 处理器持续等待真实15秒自动心跳，故障网关收到上游心跳ACK才放行业务完成。随后上游提交事件ACK，网关丢弃回应并断开。消费器重连核对原证明，处理器调用数和交付次数均保持1。 +3. 首次处理器返回error，SDK不ACK。测试等待服务端真实30秒租约及退避，不修改数据库时间；同一交付以第二代次发送，处理成功后确认，最终attempts=2、HandlerFailures=1。 + +## 5. 真实节点强杀 + +[进程故障测试](../../sdk/go/event_process_integration_test.go)： + +- **领取提交后节点退出**：节点A成功投递后被SIGKILL并确认退出。节点B读取同一MySQL，在租约自然到期后恢复相同event_id/delivery_id，代次和次数增加到2。旧Proof返回DELIVERY_LEASE_LOST，新Proof成功ACK，订阅积压清零。 +- **ACK提交后响应丢失**：网关截获成功ACK回应并阻断后续重连；通过查询确认ACKED和已保存的确认时间，然后强杀节点A。启动节点B，切换网关并恢复连接；Consumer仍持有原Proof，返回原确认时间，不新建交付尝试，不重跑处理器。 + +这两项强杀的是Orbit服务端进程。第二项消费器保持运行，其内存证明未丢失；不能将此证据当作消费者进程被强杀后业务去重也已验证。 + +## 6. 保留边界 + +- 后续需要交付独立业务数据库的Inbox事务例子，验证提交后/ACK前消费者进程退出、并发重复投递以及接收方事务回滚/提交不明等窗口，再验收P3可靠订阅。 +- Consumer取消或传输断线不能撤销已经发出的业务效果；交付租约不是业务互斥锁。生产业务需要持久化唯一键和事务，外部副作用需要业务幂等/Outbox/核查。 +- 当前仍按启用订阅的领取活动恢复过期交付；独立清扫、历史保留与清理、凭据安全重绑和生产容量/RPO/RTO仍属后续阶段。 +- 结果证明只覆盖本报告的单数据库、进程/TLS网关拓扑；未将本机回归等同于百万租户容量或生产高可用验收。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index fe028b5..6d3bf3f 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) 和 [执行历史查询](../api/execution-history.md)。[安全重试](../api/retries.md) 已提供有限策略、WorkerFailure.Retryable、历史查询及完整例子。[计划调度](../api/schedules.md) 已支持创建、查询、版本发布、暂停/恢复、触发历史与完整示例。[订阅管理](../api/subscriptions.md) 已支持创建、查询、暂停恢复、交付历史与人工重试；事件消费 SDK 继续开发。流程构建等后续接口按 [开发计划](../08-development-plan.md) 继续实现。 +已实现健康查询、任务提交/查询/分页/待执行取消、队列创建/查询、[自动 Worker](worker-runtime.md)、[Worker 底层协议 SDK](worker.md) 和 [执行历史查询](../api/execution-history.md)。[安全重试](../api/retries.md) 已提供有限策略、WorkerFailure.Retryable、历史查询及完整例子。[计划调度](../api/schedules.md) 已支持创建、查询、版本发布、暂停/恢复、触发历史与完整示例。[订阅管理](../api/subscriptions.md) 已支持创建、查询、暂停恢复、交付历史与人工重试；[事件消费 SDK](event-consumer.md) 已支持连接、自动心跳、重连与确认。流程构建等后续接口按 [开发计划](../08-development-plan.md) 继续实现。 module：`git.yuntuo.hk/k2safe/orbit/sdk/go`。SDK 使用独立 go.mod，依赖 Go 标准库与 `github.com/coder/websocket v1.8.15`，不引用服务端模型或数据库实现。 @@ -108,6 +108,8 @@ fmt.Println(cancelled.Task.State) | CreateSubscription / GetSubscription / ListSubscriptions | 订阅受理、当前配置与队列分页 | [订阅](../api/subscriptions.md) | | ConfigureSubscription | 显式暂停/继续与版本、原因 | [订阅](../api/subscriptions.md) | | ListEventDeliveries / GetEventDelivery / RetryEventDelivery | 交付历史、冻结结果和耗尽后追加预算 | [订阅](../api/subscriptions.md) | +| DialEvents / Next / Ack / Ping | 单连接交付、确认与心跳，可用原证明跨连接重报 | [消费 SDK](event-consumer.md) | +| NewEventConsumer / Run / Events / Stats | 自动心跳、重连、受控处理器并发及有界收尾 | [消费 SDK](event-consumer.md) | | CreateSchedule / GetSchedule / ListSchedules | 计划受理、详情与队列分页 | [计划](../api/schedules.md) | | PublishSchedule / GetScheduleVersion | 不可变发布与历史版本 | [计划](../api/schedules.md) | | PauseSchedule / ResumeSchedule / ListScheduleOccurrences | 控制与触发记录 | [计划](../api/schedules.md) | @@ -186,4 +188,4 @@ P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) ## 订阅管理示例 -[完整运行方法](../examples/subscriptions.md) 使用显式 --event-access 凭据，执行创建/同键核对/暂停/恢复/查询，再暂停本次订阅并保留审计。源码见 [examples/subscriptions](../../sdk/go/examples/subscriptions/main.go)。WSS 连接 SDK、自动消费与持久化 Inbox 示例尚未交付，不能将此管理示例当作完整可靠消费链路。 +[完整运行方法](../examples/subscriptions.md) 使用显式 --event-access 凭据，执行创建/同键核对/暂停/恢复/查询，再暂停本次订阅并保留审计。源码见 [examples/subscriptions](../../sdk/go/examples/subscriptions/main.go)。WSS 连接 SDK 与自动消费器见 [消费手册](event-consumer.md)，可运行 [协议示例](../examples/event-protocol.md) 验证跨连接 ACK；持久化 Inbox 示例继续开发。 diff --git a/docs/sdk/event-consumer.md b/docs/sdk/event-consumer.md new file mode 100644 index 0000000..37151eb --- /dev/null +++ b/docs/sdk/event-consumer.md @@ -0,0 +1,94 @@ +# Go 结果消费 SDK + +对应 [HTTP/WSS 契约](../api/subscriptions.md)。SDK 是独立模块，只依赖 HTTP/WebSocket，不导入服务端模型、存储或数据库驱动。订阅先用管理 API 创建，并固定在创建凭据与队列范围内。 + +已提供 [EventConnection](../../sdk/go/event_connection.go)、[EventConsumer](../../sdk/go/event_consumer.go)、[公开类型](../../sdk/go/event_types.go) 与 [可运行协议示例](../examples/event-protocol.md)。业务 MySQL Inbox 示例和消费者进程强杀验证尚在后续阶段，不能把 SDK 内存状态作为持久化去重。 + +## 1. 自动消费 + +下面是可复用的应用函数。`applyOnce` 由接收方提供，必须在同一数据库事务内保存事件去重记录和业务变更；事务明确提交或读取到已经提交的同一事件后才返回 nil。它不能仅向内存队列投递后返回。 + +```go +func consumeResults( + ctx context.Context, + client *orbit.Client, + subscriptionID string, + applyOnce func(context.Context, orbit.ResultEvent) error, +) error { + consumer, err := orbit.NewEventConsumer(client, orbit.EventConsumerOptions{ + SubscriptionID: subscriptionID, + Capacity: 4, + Handle: applyOnce, + ShutdownTimeout: 30 * time.Second, + ReconnectMinDelay: 200 * time.Millisecond, + ReconnectMaxDelay: 5 * time.Second, + }) + if err != nil { return err } + return consumer.Run(ctx) +} +``` + +所在文件需导入 `context`、`time` 和 `orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"`。同一 Consumer 只能 Run 一次；多实例可共同绑定同一个订阅，按服务端交付租约竞争，不保证事件处理/完成顺序。 + +| 配置 | 默认 / 范围 | +| --- | --- | +| SubscriptionID | 必填；有效的 32 位小写十六进制 ID | +| Handle | 必填；处理冻结的 ResultEvent，不接收交付秘密 | +| Capacity | 默认 4；1–32 个本地在途事件及处理器 | +| ShutdownTimeout | 默认 30 秒；1 毫秒至 5 分钟 | +| ReconnectMinDelay | 默认 200 毫秒；至少 10 毫秒 | +| ReconnectMaxDelay | 默认 5 秒；不小于起点且不超过 30 秒 | + +每个连接另有 Capacity 条未读交付缓冲，满后关闭连接并按退避重连。处理器结果缓冲不超过 Capacity，本地观察缓冲 128；不会为积压无限创建 goroutine。自动消费器每条连接至多一个 ACK 在途，心跳独立并发，慢 ACK 不阻塞 WebSocket 读取。 + +## 2. 提交、补发和去重 + +推荐的接收方事务步骤： + +1. 使用固定消费者用途与事件 ID 建立唯一键，例如 `(consumer_name, event_id)`。不要使用连接 ID、交付代次或消息 ID 作为业务去重键。 +2. 在一个业务事务中插入 Inbox 记录并更新业务数据。唯一键竞争必须等待对方事务提交或回滚；遇到重复键时核对已有记录，而不是忽略所有 SQL 错误。 +3. 业务事务提交成功后返回 nil，SDK 再发送 ACK。提交结果未知时返回 error，保留核查信息，下次投递先按唯一键核对。 +4. ACK 已提交但回应丢失，SDK 保留原证明，在新连接上重报；在该本地事件仍保留期间不会重新调用处理器。 +5. 原证明已被新代次替代时，SDK 记录 proof_lost，等待新投递。新投递会再次进入处理器，必须命中持久化 Inbox 后才能确认新证明。 + +处理器 error、panic 或 Goexit 都不会 ACK，也不会在 SDK 内存中立即重新执行。平台在原 30 秒租约自然到期后，按退避和累计预算重新投递；耗尽后保留 EXPIRED，占用订阅窗口，需要管理 API 追加预算。仅在启用订阅有领取活动时触发恢复，当前尚无独立后台交付清扫器。 + +事件交付租约不提供业务独占执行授权，Ping 不延长它。长于租约的处理、网络分区或多个接收进程可能造成同一事件并行进入业务侧；本地 Capacity 和事件合并不能代替跨进程的唯一约束及事务。外部 HTTP 副作用也不能和 Inbox SQL 自动组成原子事务，需要接收方 Outbox、业务幂等及结果核查设计。 + +`ResultEvent.Snapshot` 是原事件时刻的冻结快照，不能通过后来查询到的当前任务状态替换。SDK 保留 RawMessage 中大整数及业务 JSON；例如 `9007199254740993` 不经 float64 重编码。`task.needs_review` 表示需要核查，不能按业务成功或最终失败处理。 + +## 3. 重连与关闭 + +自动心跳间隔为当前 v1 固定的 15 秒。心跳也重新验证凭据和订阅权限，暂停的订阅仍可心跳及确认已有交付。 + +网络中断、超时、接收缓冲耗尽、UNAVAILABLE、HTTP 408/429/5xx 使用指数退避及半区间抖动重连；延迟上限为 ReconnectMaxDelay。确认成功后重置退避。认证、权限、资源不存在、历史缺口、协议错误及其他永久错误停止消费，返回支持 `errors.Is/As` 的 EventConsumerRunError。 + +取消 Run 的 context 后，SDK 停止启动新处理器并取消已有处理器 context。在 ShutdownTimeout 内收集处理器返回，已提交的结果仍尝试 ACK 或重连核对；正常收尾返回 nil。未成功处理的交付保留给平台按预算恢复。 + +超时返回 `ErrEventShutdownTimeout` 包装错误：`Unconfirmed` 是本进程已知且未确认的事件数，`Running` 是仍未退出的处理器数。Go 不能强杀忽略 context 的函数，不能在收到超时后声称这些函数的业务效果已经撤销。永久连接错误也会取消并等待在途函数，等待受相同收尾时限约束。 + +调用方主动停止后，暂停订阅是独立管理操作；关闭一个 Consumer 不会暂停共享订阅。进程被强杀时，SDK 内存中的结果和证明不保证保留，接收方 Inbox 必须独立持久化。 + +## 4. 底层连接 + +| 方法 | 语义 | +| --- | --- | +| Client.DialEvents(ctx, EventRegistration) | 使用 Authorization 与 orbit.events.v1 握手；ctx 只控制握手，容量必须显式传 1–32 | +| Welcome() | 初始握手快照；不是实时订阅状态 | +| Next(ctx) | 等待交付；等待取消不关闭连接 | +| Ack(ctx, messageID, EventProof) | 确认原交付，可在同订阅的新连接中重报 | +| Ping(ctx, messageID) | 查询 Enabled 并维持连接，不续交付租约 | +| Done() / Err() | 等待读取循环退出及查询关闭原因 | +| Close() | 关闭传输并等待读取循环退出，不取消服务端交付 | + +EventConnection 不自动心跳、重连或执行处理器。应用必须每 15 秒 Ping，并在退出时 Close。最多 32 个在途控制请求；超出返回 ErrEventBusy。控制请求总时限最多 10 秒，包含写入排队；握手最多 8 秒；每条文本帧最多 512 KiB。控制超时关闭连接，防止无法关联的迟到回应被当作另一次操作成功。 + +`grant.Proof()` 包含订阅、事件、交付、代次和秘密；SDK 本地核对订阅与回执事件，线上 ACK 只发送交付 ID、代次和秘密。重连后使用原 Proof，可通过 NewEventMessageID 生成新的关联 ID。ACK 持久化去重不依赖消息 ID。 + +SDK 严格验证消息方向、会话、回复关联、规范十进制 uint64、秘密编码、必填字段、重复键、字段大小写和交付/事件/快照关联。无效响应返回 ErrEventProtocol 并关闭连接；不会将无效回执当作已确认。 + +## 5. 观察与敏感信息 + +`Stats()` 返回进程内 Running、Started、HandlerFailures、Acknowledged、Reconnects 和 DroppedEvents；重启清零。`Events()` 返回有界本地通知，不读取也不会阻塞执行，Run 返回时关闭。它们不能替代数据库状态和持久化审计；Acknowledged 包括对已提交 ACK 的重报确认。 + +EventGrant、EventProof、EventConnection、EventConsumer 和 EventError 的常见 fmt 输出隐藏秘密、业务 JSON 或远端错误正文。显式 JSON 序列化 Grant/Proof 仍包含租约秘密，不可放到普通日志；若业务确需保存 Proof，应加密并限制访问。调用方记录自己的 handler error 时也应脱敏，SDK 观察事件只给出 HANDLER_FAILED 分类。 diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index b326459..e00b396 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -53,6 +53,7 @@ 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 '^TestIntegrationSchedule' -v .) (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 '^TestIntegrationQueueResourcesSDK$' -v .) (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 '^TestIntegrationSubscriptionManagementSDK$' -v .) +(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 '^TestIntegrationEvent(Connection|Consumer)' -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) @@ -63,6 +64,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/keyed-concurrency) (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/priority) (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/subscriptions) +(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/event-protocol) # 应用配额场景使用独立应用与两个命名空间，普通 Worker 凭据不扩权。 @@ -76,4 +78,4 @@ ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap - kill "$TASK_SERVER_PID" wait "$TASK_SERVER_PID" TASK_SERVER_PID= -(cd sdk/go && ORBIT_TEST_APP_CREDENTIALS_FILE="$TASK_TEST_DIR/app.credentials.json" ORBIT_TEST_APP_SECOND_CREDENTIALS_FILE="$TASK_TEST_DIR/app-second.credentials.json" ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" 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 '^TestIntegrationWorkerProcess' -v .) +(cd sdk/go && ORBIT_TEST_APP_CREDENTIALS_FILE="$TASK_TEST_DIR/app.credentials.json" ORBIT_TEST_APP_SECOND_CREDENTIALS_FILE="$TASK_TEST_DIR/app-second.credentials.json" ORBIT_TEST_SERVER_BINARY="$TASK_TEST_DIR/orbit" 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 '^TestIntegration(Worker|Event)Process' -v .) diff --git a/sdk/go/doc.go b/sdk/go/doc.go index e81d17c..f10d435 100644 --- a/sdk/go/doc.go +++ b/sdk/go/doc.go @@ -1,5 +1,6 @@ -// Package orbit 是 Orbit 的独立 Go SDK，提供健康、任务、队列与 Worker 底层协议。 +// Package orbit 是 Orbit 的独立 Go SDK，提供健康、任务、队列、计划及结果订阅。 // WorkerConnection 是底层接口；NewWorker/Run 自动管理处理器、续租、恢复与排空。 +// EventConnection 是结果交付协议；EventConsumer 自动管理心跳、重连与确认，业务端负责事务去重。 // 流程 SDK 等后续能力按阶段继续开发。 // SDK 不导入服务端模型、迁移或数据库依赖。 package orbit diff --git a/sdk/go/event_codec.go b/sdk/go/event_codec.go new file mode 100644 index 0000000..383a027 --- /dev/null +++ b/sdk/go/event_codec.go @@ -0,0 +1,152 @@ +package orbit + +import ( + "bytes" + "encoding/base64" + "encoding/json" + "reflect" + "strconv" + "strings" + "time" + + "github.com/coder/websocket" +) + +type eventFrame struct { + Version string `json:"protocol_version"` + Type string `json:"type"` + MessageID string `json:"message_id"` + ReplyTo string `json:"reply_to,omitempty"` + SessionID string `json:"session_id,omitempty"` + Body json.RawMessage `json:"body,omitempty"` + Error *EventError `json:"error,omitempty"` +} + +// 仅协议结构严格匹配字段；业务RawMessage保持原数字、键和值。 +func eventFields(raw []byte, typ reflect.Type) error { + if typ == reflect.TypeOf(json.RawMessage{}) { + return nil + } + if bytes.Equal(bytes.TrimSpace(raw), []byte("null")) { + return ErrEventProtocol + } + if typ.Kind() == reflect.Pointer { + return eventFields(raw, typ.Elem()) + } + if typ == reflect.TypeOf(time.Time{}) { + return nil + } + if typ.Kind() != reflect.Struct { + return nil + } + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil || fields == nil { + return ErrEventProtocol + } + for i := 0; i < typ.NumField(); i++ { + f := typ.Field(i) + tags := strings.Split(f.Tag.Get("json"), ",") + if tags[0] == "" || tags[0] == "-" { + continue + } + v, found := fields[tags[0]] + optional, quoted := false, false + for _, tag := range tags[1:] { + optional = optional || tag == "omitempty" || tag == "omitzero" + quoted = quoted || tag == "string" + } + if !found { + if optional { + continue + } + return ErrEventProtocol + } + delete(fields, tags[0]) + if f.Type.Kind() == reflect.Uint64 && quoted { + var s string + if json.Unmarshal(v, &s) != nil { + return ErrEventProtocol + } + n, err := strconv.ParseUint(s, 10, 64) + if err != nil || n == 0 || strconv.FormatUint(n, 10) != s { + return ErrEventProtocol + } + } + if err := eventFields(v, f.Type); err != nil { + return err + } + } + if len(fields) > 0 { + return ErrEventProtocol + } + return nil +} +func decodeEventValue(raw []byte, out any) error { + if validateWorkerJSON(raw) != nil || eventFields(raw, reflect.TypeOf(out).Elem()) != nil || json.Unmarshal(raw, out) != nil { + return ErrEventProtocol + } + return nil +} +func parseEventFrame(kind websocket.MessageType, raw []byte) (eventFrame, error) { + var f eventFrame + if kind != websocket.MessageText || len(raw) > 512<<10 || decodeEventValue(raw, &f) != nil || f.Version != "1" || !validID(f.MessageID) { + return f, ErrEventProtocol + } + var fields map[string]json.RawMessage + _ = json.Unmarshal(raw, &fields) + if fields["reply_to"] != nil && !validID(f.ReplyTo) || fields["session_id"] != nil && !validID(f.SessionID) { + return f, ErrEventProtocol + } + switch f.Type { + case "error": + if f.Error == nil || fields["body"] != nil || len(f.Error.Message) == 0 || len(f.Error.Message) > 1024 { + return f, ErrEventProtocol + } + switch f.Error.Code { + case "INVALID_ARGUMENT", "UNAUTHENTICATED", "PERMISSION_DENIED", "NOT_FOUND", "STATE_CONFLICT", "DELIVERY_LEASE_LOST", "EVENT_HISTORY_EXPIRED", "UNAVAILABLE": + default: + return f, ErrEventProtocol + } + case "events.welcome", "ack", "event.deliver": + if fields["error"] != nil || len(f.Body) == 0 || bytes.Equal(bytes.TrimSpace(f.Body), []byte("null")) || !validID(f.SessionID) { + return f, ErrEventProtocol + } + if f.Type == "event.deliver" { + if fields["reply_to"] != nil { + return f, ErrEventProtocol + } + } else if !validID(f.ReplyTo) { + return f, ErrEventProtocol + } + default: + return f, ErrEventProtocol + } + return f, nil +} +func validEventProof(p EventProof) bool { + b, err := base64.RawURLEncoding.Strict().DecodeString(p.LeaseSecret) + return validID(p.SubscriptionID) && validID(p.EventID) && validID(p.DeliveryID) && p.Epoch > 0 && len(p.LeaseSecret) == 43 && err == nil && len(b) == 32 +} +func validEventGrant(g EventGrant, sub string) bool { + d, e, s := g.Delivery, g.Event, g.Event.Snapshot + if !validEventProof(g.Proof()) || d.SubscriptionID != sub || d.EventID != e.ID || d.Sequence != e.Sequence || d.Sequence == 0 || d.Revision == 0 || d.State != DeliveryLeased || d.CreatedAt.IsZero() || d.AvailableAt.IsZero() || d.LeaseUntil == nil || d.LeaseUntil.IsZero() || d.AcknowledgedAt != nil || d.Attempts == 0 || d.Attempts > d.AttemptBudget || d.AttemptBudget > 1000000 { + return false + } + if e.AggregateKind != "task" || !validID(e.AggregateID) || e.AggregateID != s.TaskID || e.AggregateRevision != s.Revision || s.Revision == 0 || e.OccurredAt.IsZero() || s.SchemaVersion != 1 || !validID(s.QueueID) || s.CurrentAttemptID != nil && !validID(*s.CurrentAttemptID) { + return false + } + if len(bytes.TrimSpace(e.Details)) == 0 || bytes.TrimSpace(e.Details)[0] != '{' { + return false + } + switch e.Type { + case "task.succeeded": + return s.State == TaskSucceeded + case "task.failed": + return s.State == TaskFailed + case "task.cancelled": + return s.State == TaskCancelled + case "task.needs_review": + return s.State == TaskNeedsReview + } + return false +} diff --git a/sdk/go/event_connection.go b/sdk/go/event_connection.go new file mode 100644 index 0000000..a7a2c98 --- /dev/null +++ b/sdk/go/event_connection.go @@ -0,0 +1,285 @@ +package orbit + +import ( + "context" + "errors" + "fmt" + "net/http" + "sync" + "time" + + "github.com/coder/websocket" +) + +const eventSubprotocol = "orbit.events.v1" +const eventMaxPending = 32 + +// EventConnection 管理单次连接、请求关联和有界收发。可并发调用Ack/Ping。 +// 本类型不自动心跳/重连/调用处理器；高层消费请使用EventConsumer。 +type EventConnection struct { + conn *websocket.Conn + ctx context.Context + cancel context.CancelFunc + welcome EventWelcome + deliveries chan EventGrant + done chan struct{} + mu sync.Mutex + err error + pending map[string]chan eventFrame +} + +// DialEvents 的ctx仅控制握手；成功后必须Close，取消握手ctx不会关闭已建立的连接。 +func (c *Client) DialEvents(ctx context.Context, in EventRegistration) (*EventConnection, error) { + if c.token == "" || !validID(in.SubscriptionID) || in.Capacity < 1 || in.Capacity > 32 { + return nil, errors.New("authenticated client, valid subscription and capacity 1..32 are required") + } + bounded, stop := context.WithTimeout(ctx, 8*time.Second) + defer stop() + headers := http.Header{} + headers.Set("Authorization", "Bearer "+c.token) + conn, response, err := websocket.Dial(bounded, c.baseURL+"/v1/events/connect", &websocket.DialOptions{HTTPClient: c.http, HTTPHeader: headers, Subprotocols: []string{eventSubprotocol}, CompressionMode: websocket.CompressionDisabled}) + if err != nil { + if response != nil && response.StatusCode != http.StatusSwitchingProtocols { + return nil, &HTTPError{StatusCode: response.StatusCode} + } + return nil, eventIOError(err) + } + ok := false + defer func() { + if !ok { + _ = conn.CloseNow() + } + }() + if conn.Subprotocol() != eventSubprotocol { + return nil, ErrEventProtocol + } + conn.SetReadLimit(512 << 10) + id, err := NewEventMessageID() + if err != nil { + return nil, err + } + body, _ := marshalWire(in) + raw, _ := marshalWire(eventFrame{Version: "1", Type: "events.hello", MessageID: id, Body: body}) + if err = conn.Write(bounded, websocket.MessageText, raw); err != nil { + return nil, eventIOError(err) + } + kind, raw, err := conn.Read(bounded) + if err != nil { + return nil, eventIOError(err) + } + f, err := parseEventFrame(kind, raw) + if err != nil || f.ReplyTo != id { + return nil, ErrEventProtocol + } + if f.Type == "error" { + return nil, f.Error + } + var welcome EventWelcome + if f.Type != "events.welcome" || decodeEventValue(f.Body, &welcome) != nil || welcome.SessionID != f.SessionID || welcome.SubscriptionID != in.SubscriptionID || welcome.Capacity != in.Capacity || welcome.ServerTime.IsZero() || welcome.LeaseDurationMS != 30000 || welcome.HeartbeatIntervalMS != 15000 { + return nil, ErrEventProtocol + } + lifetime, cancel := context.WithCancel(context.Background()) + e := &EventConnection{conn: conn, ctx: lifetime, cancel: cancel, welcome: welcome, deliveries: make(chan EventGrant, in.Capacity), done: make(chan struct{}), pending: make(map[string]chan eventFrame)} + ok = true + go e.readLoop() + return e, nil +} +func (e *EventConnection) Welcome() EventWelcome { return e.welcome } +func (e *EventConnection) Done() <-chan struct{} { return e.done } +func (e *EventConnection) Err() error { e.mu.Lock(); defer e.mu.Unlock(); return e.err } +func (e *EventConnection) Close() error { e.fail(ErrEventClosed); <-e.done; return nil } +func (e *EventConnection) fail(err error) { + e.mu.Lock() + if e.err != nil { + e.mu.Unlock() + return + } + e.err = err + e.cancel() + e.mu.Unlock() + _ = e.conn.CloseNow() +} +func (e *EventConnection) readLoop() { + defer close(e.done) + for { + ctx, stop := context.WithTimeout(e.ctx, 60*time.Second) + kind, raw, err := e.conn.Read(ctx) + stop() + if err != nil { + e.fail(eventIOError(err)) + return + } + f, err := parseEventFrame(kind, raw) + if err != nil || f.SessionID != e.welcome.SessionID { + e.fail(ErrEventProtocol) + return + } + if f.Type == "event.deliver" { + var grant EventGrant + if decodeEventValue(f.Body, &grant) != nil || !validEventGrant(grant, e.welcome.SubscriptionID) { + e.fail(ErrEventProtocol) + return + } + select { + case e.deliveries <- grant: + default: + e.fail(ErrEventBackpressure) + return + } + continue + } + if f.Type != "ack" && f.Type != "error" { + e.fail(ErrEventProtocol) + return + } + if f.Type == "error" && f.ReplyTo == "" { + e.fail(f.Error) + return + } + e.mu.Lock() + waiter := e.pending[f.ReplyTo] + e.mu.Unlock() + if waiter == nil { + e.fail(ErrEventProtocol) + return + } + select { + case waiter <- f: + default: + e.fail(ErrEventProtocol) + return + } + if f.Error != nil && (f.Error.Code == "UNAUTHENTICATED" || f.Error.Code == "PERMISSION_DENIED" || f.Error.Code == "NOT_FOUND" || f.Error.Code == "UNAVAILABLE") { + e.fail(f.Error) + return + } + } +} + +// Next 等待下一条交付。ctx到期只中断本次等待；未确认交付按服务端预算补发。 +func (e *EventConnection) Next(ctx context.Context) (EventGrant, error) { + if err := ctx.Err(); err != nil { + return EventGrant{}, err + } + select { + case <-ctx.Done(): + return EventGrant{}, ctx.Err() + case <-e.ctx.Done(): + return EventGrant{}, e.Err() + case g := <-e.deliveries: + if err := e.Err(); err != nil { + return EventGrant{}, err + } + return g, nil + } +} +func (e *EventConnection) call(ctx context.Context, kind, id string, in, out any) error { + if !validID(id) { + return errors.New("invalid event message id") + } + if err := ctx.Err(); err != nil { + return err + } + bounded, stop := context.WithTimeout(ctx, 10*time.Second) + defer stop() + waiter := make(chan eventFrame, 1) + e.mu.Lock() + if e.err != nil { + err := e.err + e.mu.Unlock() + return err + } + if len(e.pending) >= eventMaxPending { + e.mu.Unlock() + return ErrEventBusy + } + if e.pending[id] != nil { + e.mu.Unlock() + return errors.New("event message id is already in flight") + } + e.pending[id] = waiter + e.mu.Unlock() + defer func() { e.mu.Lock(); delete(e.pending, id); e.mu.Unlock() }() + body, err := marshalWire(in) + if err != nil { + return errors.New("invalid event request") + } + raw, err := marshalWire(eventFrame{Version: "1", Type: kind, MessageID: id, SessionID: e.welcome.SessionID, Body: body}) + if err != nil || len(raw) > 512<<10 { + return errors.New("event request exceeds frame limit") + } + if err = e.conn.Write(bounded, websocket.MessageText, raw); err != nil { + err = eventIOError(err) + e.fail(err) + return err + } + var f eventFrame + select { + case f = <-waiter: + case <-e.ctx.Done(): + select { + case f = <-waiter: + default: + return e.Err() + } + case <-bounded.Done(): + e.fail(bounded.Err()) + return bounded.Err() + } + if f.Error != nil { + return f.Error + } + if decodeEventValue(f.Body, out) != nil { + e.fail(ErrEventProtocol) + return ErrEventProtocol + } + return nil +} + +// Ack 仅在业务事务已提交或持久化去重命中后调用。重连可用原Proof重报，MessageID可新建。 +// 超时/断线不证明ACK未提交；DELIVERY_LEASE_LOST表示证明已失效，不应撤销已经提交的业务。 +func (e *EventConnection) Ack(ctx context.Context, id string, p EventProof) (EventAckResult, error) { + var out EventAckResult + if !validEventProof(p) || p.SubscriptionID != e.welcome.SubscriptionID { + return out, errors.New("invalid event proof or subscription") + } + in := struct { + DeliveryID string `json:"delivery_id"` + Epoch uint64 `json:"epoch,string"` + LeaseSecret string `json:"lease_secret"` + }{p.DeliveryID, p.Epoch, p.LeaseSecret} + err := e.call(ctx, "event.ack", id, in, &out) + if err == nil && (out.DeliveryID != p.DeliveryID || out.EventID != p.EventID || out.Epoch != p.Epoch || out.AcknowledgedAt.IsZero()) { + err = ErrEventProtocol + e.fail(err) + } + return out, err +} + +// Ping 重新核验凭据及订阅，并维持连接；不会续期交付租约。 +func (e *EventConnection) Ping(ctx context.Context, id string) (EventPong, error) { + var out EventPong + err := e.call(ctx, "events.ping", id, struct{}{}, &out) + if err == nil && out.ServerTime.IsZero() { + err = ErrEventProtocol + e.fail(err) + } + return out, err +} +func eventIOError(err error) error { + if errors.Is(err, context.Canceled) { + return context.Canceled + } + if errors.Is(err, context.DeadlineExceeded) { + return context.DeadlineExceeded + } + switch websocket.CloseStatus(err) { + case websocket.StatusProtocolError, websocket.StatusUnsupportedData, websocket.StatusInvalidFramePayloadData, websocket.StatusPolicyViolation, websocket.StatusMessageTooBig: + return ErrEventProtocol + } + return ErrEventDisconnected +} +func (e *EventConnection) String() string { + return fmt.Sprintf("EventConnection{SubscriptionID:%s SessionID:%s secrets:redacted}", e.welcome.SubscriptionID, e.welcome.SessionID) +} +func (e *EventConnection) GoString() string { return e.String() } diff --git a/sdk/go/event_connection_test.go b/sdk/go/event_connection_test.go new file mode 100644 index 0000000..16980aa --- /dev/null +++ b/sdk/go/event_connection_test.go @@ -0,0 +1,283 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +const eventTestSub = "55555555555555555555555555555555" +const eventTestID = "66666666666666666666666666666666" +const eventTestDelivery = "77777777777777777777777777777777" + +func eventTestGrant() EventGrant { + now := time.Now().UTC() + until := now.Add(30 * time.Second) + return EventGrant{Delivery: EventDelivery{ID: eventTestDelivery, SubscriptionID: eventTestSub, EventID: eventTestID, Sequence: 9007199254740993, Revision: 2, State: DeliveryLeased, CreatedAt: now, AvailableAt: now, LeaseEpoch: 1, LeaseUntil: &until, Attempts: 1, AttemptBudget: 20}, Event: ResultEvent{ID: eventTestID, Sequence: 9007199254740993, Type: "task.succeeded", AggregateKind: "task", AggregateID: workerTestTask, AggregateRevision: 9007199254740993, OccurredAt: now, Details: json.RawMessage(`{}`), Snapshot: TaskResultSnapshot{SchemaVersion: 1, TaskID: workerTestTask, QueueID: workerTestQueue, State: TaskSucceeded, Revision: 9007199254740993, Result: json.RawMessage(`{"n":9007199254740993,"text":"<&>"}`)}}, LeaseSecret: strings.Repeat("A", 43)} +} +func eventTestRead(ctx context.Context, c *websocket.Conn) (eventFrame, error) { + kind, raw, err := c.Read(ctx) + var f eventFrame + if err != nil { + return f, err + } + if kind != websocket.MessageText || json.Unmarshal(raw, &f) != nil { + return f, ErrEventProtocol + } + return f, nil +} +func eventTestWrite(ctx context.Context, c *websocket.Conn, f eventFrame) error { + f.Version = "1" + f.SessionID = workerTestSession + f.MessageID, _ = NewEventMessageID() + raw, err := marshalWire(f) + if err != nil { + return err + } + return c.Write(ctx, websocket.MessageText, raw) +} +func eventBody(in any) json.RawMessage { b, _ := marshalWire(in); return b } +func eventMock(t *testing.T, serve func(context.Context, *websocket.Conn, EventRegistration, int)) *Client { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + var count atomic.Int64 + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/events/connect" || r.URL.RawQuery != "" || r.Header.Get("Authorization") != "Bearer event-test-token" { + t.Error("unexpected event handshake") + } + c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{eventSubprotocol}}) + if err != nil { + return + } + defer c.CloseNow() + f, err := eventTestRead(ctx, c) + var in EventRegistration + if err != nil || f.Type != "events.hello" || f.SessionID != "" || !validID(f.MessageID) || json.Unmarshal(f.Body, &in) != nil { + t.Error("invalid event hello") + return + } + welcome := EventWelcome{SessionID: workerTestSession, SubscriptionID: in.SubscriptionID, Capacity: in.Capacity, LeaseDurationMS: 30000, HeartbeatIntervalMS: 15000, ServerTime: time.Now().UTC(), Enabled: true} + if eventTestWrite(ctx, c, eventFrame{Type: "events.welcome", ReplyTo: f.MessageID, Body: eventBody(welcome)}) != nil { + return + } + serve(ctx, c, in, int(count.Add(1))) + })) + t.Cleanup(func() { cancel(); server.Close() }) + client, err := NewClient(server.URL, Options{Token: "event-test-token", HTTPClient: server.Client()}) + if err != nil { + t.Fatal(err) + } + return client +} +func eventTestDial(t *testing.T, client *Client, capacity uint32) *EventConnection { + t.Helper() + ctx, stop := context.WithTimeout(context.Background(), 3*time.Second) + defer stop() + c, err := client.DialEvents(ctx, EventRegistration{SubscriptionID: eventTestSub, Capacity: capacity}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Close() }) + return c +} +func eventAckFor(g EventGrant) EventAckResult { + return EventAckResult{DeliveryID: g.Delivery.ID, EventID: g.Event.ID, Epoch: g.Delivery.LeaseEpoch, AcknowledgedAt: time.Now().UTC()} +} + +func TestEventConnectionMultiplexAndPrecision(t *testing.T) { + g := eventTestGrant() + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + if eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(g)}) != nil { + return + } + first, err := eventTestRead(ctx, c) + if err != nil { + return + } + second, err := eventTestRead(ctx, c) + if err != nil { + return + } + for _, f := range []eventFrame{second, first} { + var body any = EventPong{ServerTime: time.Now().UTC(), Enabled: true} + if f.Type == "event.ack" { + body = eventAckFor(g) + } + if eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(body)}) != nil { + return + } + } + _, _ = eventTestRead(ctx, c) + }) + c := eventTestDial(t, client, 1) + ctx, stop := context.WithTimeout(context.Background(), 3*time.Second) + defer stop() + got, err := c.Next(ctx) + if err != nil || got.Event.Sequence != 9007199254740993 || string(got.Event.Snapshot.Result) != string(g.Event.Snapshot.Result) { + t.Fatal("event snapshot changed", err) + } + errs := make(chan error, 2) + go func() { _, err := c.Ping(ctx, workerTestID(t)); errs <- err }() + go func() { _, err := c.Ack(ctx, workerTestID(t), got.Proof()); errs <- err }() + for range 2 { + if err := <-errs; err != nil { + t.Fatal(err) + } + } + if c.Err() != nil { + t.Fatal(c.Err()) + } + for _, value := range []any{g, g.Proof(), c, &EventError{Code: "UNAVAILABLE", Message: "secret-text"}} { + for _, format := range []string{"%v", "%+v", "%#v"} { + printed := fmt.Sprintf(format, value) + if strings.Contains(printed, g.LeaseSecret) || strings.Contains(printed, "secret-text") || strings.Contains(printed, "9007199254740993") { + t.Fatal("event formatter exposed secrets or business data") + } + } + } +} +func TestEventConnectionWaitCancellationAndAckUncertainty(t *testing.T) { + received := make(chan struct{}, 1) + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + f, err := eventTestRead(ctx, c) + if err != nil { + return + } + if f.Type != "event.ack" { + t.Error("unexpected control") + } + received <- struct{}{} + _, _ = eventTestRead(ctx, c) + }) + c := eventTestDial(t, client, 1) + short, stop := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer stop() + if _, err := c.Next(short); !errors.Is(err, context.DeadlineExceeded) || c.Err() != nil { + t.Fatal("Next timeout closed connection", err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { _, err := c.Ack(ctx, workerTestID(t), eventTestGrant().Proof()); done <- err }() + <-received + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + select { + case <-c.Done(): + case <-time.After(time.Second): + t.Fatal("uncertain ACK did not close connection") + } +} +func TestEventConnectionBoundedControlsAndDeliveries(t *testing.T) { + t.Run("controls", func(t *testing.T) { + received := make(chan struct{}, 40) + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + for { + if _, err := eventTestRead(ctx, c); err != nil { + return + } + received <- struct{}{} + } + }) + c := eventTestDial(t, client, 1) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 32) + for range 32 { + go func() { _, err := c.Ping(ctx, workerTestID(t)); done <- err }() + <-received + } + if _, err := c.Ping(ctx, workerTestID(t)); !errors.Is(err, ErrEventBusy) { + t.Fatal("unbounded pending controls", err) + } + cancel() + for range 32 { + <-done + } + }) + t.Run("deliveries", func(t *testing.T) { + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + for range 2 { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(eventTestGrant())}) + } + _, _ = eventTestRead(ctx, c) + }) + c := eventTestDial(t, client, 1) + select { + case <-c.Done(): + case <-time.After(time.Second): + t.Fatal("unbounded deliveries") + } + if !errors.Is(c.Err(), ErrEventBackpressure) { + t.Fatal(c.Err()) + } + }) +} +func TestEventConnectionRejectsUntrustedProtocol(t *testing.T) { + g := eventTestGrant() + valid := eventFrame{Version: "1", Type: "event.deliver", MessageID: workerTestQueue, SessionID: workerTestSession, Body: eventBody(g)} + raw, _ := marshalWire(valid) + cases := map[string][]byte{ + "duplicate": []byte(strings.Replace(string(raw), `"protocol_version":"1"`, `"protocol_version":"1","protocol_version":"1"`, 1)), + "alias": []byte(strings.Replace(string(raw), `"lease_secret"`, `"Lease_Secret"`, 1)), + "noncanonical": []byte(strings.Replace(string(raw), `"sequence":"9007199254740993"`, `"sequence":"09007199254740993"`, 1)), + "null_bool": []byte(strings.Replace(string(raw), `"effect_uncertain":false`, `"effect_uncertain":null`, 1)), + "null_error": []byte(strings.Replace(string(raw), `"body":`, `"error":null,"body":`, 1)), + "mismatched_event": []byte(strings.Replace(string(raw), `"event_id":"`+eventTestID+`"`, `"event_id":"`+workerTestTask+`"`, 1)), + "missing_created": []byte(strings.Replace(string(raw), `"created_at":`+string(eventBody(g.Delivery.CreatedAt))+`,`, "", 1)), + "wrong_state": []byte(strings.Replace(string(raw), `"state":"SUCCEEDED"`, `"state":"FAILED"`, 1)), + "unknown_body_field": []byte(strings.Replace(string(raw), `"schema_version":1`, `"extra":true,"schema_version":1`, 1)), + } + for name, bad := range cases { + t.Run(name, func(t *testing.T) { + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + _ = c.Write(ctx, websocket.MessageText, bad) + _, _ = eventTestRead(ctx, c) + }) + c := eventTestDial(t, client, 1) + ctx, stop := context.WithTimeout(context.Background(), time.Second) + defer stop() + if _, err := c.Next(ctx); !errors.Is(err, ErrEventProtocol) { + t.Fatal("invalid response accepted", err) + } + }) + } +} +func TestEventConnectionRejectsMismatchedAck(t *testing.T) { + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + f, err := eventTestRead(ctx, c) + if err != nil { + return + } + out := eventAckFor(eventTestGrant()) + out.Epoch++ + _ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(out)}) + _, _ = eventTestRead(ctx, c) + }) + c := eventTestDial(t, client, 1) + if _, err := c.Ack(context.Background(), workerTestID(t), eventTestGrant().Proof()); !errors.Is(err, ErrEventProtocol) { + t.Fatal("wrong receipt accepted", err) + } +} + +func TestEventConnectionPolicyClosureIsPermanent(t *testing.T) { + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + _ = c.Close(websocket.StatusPolicyViolation, "private peer reason") + }) + c := eventTestDial(t, client, 1) + ctx, stop := context.WithTimeout(context.Background(), time.Second) + defer stop() + if _, err := c.Next(ctx); !errors.Is(err, ErrEventProtocol) || eventRetryable(err) { + t.Fatal("policy violation classified as retryable", err) + } +} diff --git a/sdk/go/event_consumer.go b/sdk/go/event_consumer.go new file mode 100644 index 0000000..cce0cbf --- /dev/null +++ b/sdk/go/event_consumer.go @@ -0,0 +1,368 @@ +package orbit + +import ( + "context" + "errors" + "fmt" + "math/rand/v2" + "sync/atomic" + "time" +) + +var ErrEventConsumerAlreadyRun = errors.New("event consumer may only Run once") +var ErrEventShutdownTimeout = errors.New("event consumer shutdown grace period expired") + +// EventConsumerOptions 定义自动消费器。Handle必须以事件ID做持久化去重， +// 在去重记录和业务效果同一事务提交后才返回nil；提交结果未知时返回error。 +// error或panic不ACK，平台在租约到期后按原预算补发；SDK不在内存中立即重跑。 +type EventConsumerOptions struct { + SubscriptionID string + Capacity uint32 // Capacity 最大本地在途事件/处理器，默认4，范围1至32。 + Handle func(context.Context, ResultEvent) error // Handle不接收租约秘密；必须响应context取消。 + ShutdownTimeout time.Duration // ShutdownTimeout停止时取消处理器，等待返回及ACK，默认30秒，最多5分钟。 + ReconnectMinDelay time.Duration // ReconnectMinDelay默认200毫秒，至少10毫秒。 + ReconnectMaxDelay time.Duration // ReconnectMaxDelay默认5秒，最多30秒；指数退避并加入抖动。 +} + +// EventConsumerEvent 只用于本地观察，缓冲满可丢弃；不是可靠结果或审计。 +type EventConsumerEvent struct { + Type string // Type为connected、reconnecting、handler_started、handler_failed、acknowledged、proof_lost、draining。 + EventID string + Code string // Code不含业务异常、数据或秘密。 +} +type EventConsumerStats struct { + Running int64 + Started uint64 + HandlerFailures uint64 + Acknowledged uint64 + Reconnects uint64 + DroppedEvents uint64 +} +type EventConsumerRunError struct { + Cause error + Unconfirmed int // Unconfirmed本进程已知且未确认的事件数，不是订阅全局积压。 + Running int64 // Running仍未退出的处理器；Go不能强杀忽略context的函数。 +} + +func (e *EventConsumerRunError) Error() string { + return fmt.Sprintf("event consumer stopped with %d unconfirmed events and %d running handlers: %v", e.Unconfirmed, e.Running, e.Cause) +} +func (e *EventConsumerRunError) Unwrap() error { return e.Cause } + +// EventConsumer 自动维护心跳、重连、处理器并发和ACK核对。只能Run一次。 +// 服务端交付租约不是业务锁；跨进程重复及并行消费由业务Inbox唯一键和事务处理。 +type EventConsumer struct { + client *Client + options EventConsumerOptions + used atomic.Bool + events chan EventConsumerEvent + running atomic.Int64 + started, failures, acked, reconnects, dropped atomic.Uint64 +} + +func NewEventConsumer(client *Client, in EventConsumerOptions) (*EventConsumer, error) { + if client == nil || client.token == "" || !validID(in.SubscriptionID) || in.Handle == nil { + return nil, errors.New("authenticated client, subscription and durable event handler are required") + } + if in.Capacity == 0 { + in.Capacity = 4 + } + if in.ShutdownTimeout == 0 { + in.ShutdownTimeout = 30 * time.Second + } + if in.ReconnectMinDelay == 0 { + in.ReconnectMinDelay = 200 * time.Millisecond + } + if in.ReconnectMaxDelay == 0 { + in.ReconnectMaxDelay = 5 * time.Second + } + if in.Capacity > 32 || in.ShutdownTimeout < time.Millisecond || in.ShutdownTimeout > 5*time.Minute || in.ReconnectMinDelay < 10*time.Millisecond || in.ReconnectMaxDelay < in.ReconnectMinDelay || in.ReconnectMaxDelay > 30*time.Second { + return nil, errors.New("invalid event consumer capacity or durations") + } + return &EventConsumer{client: client, options: in, events: make(chan EventConsumerEvent, 128)}, nil +} +func (c *EventConsumer) Events() <-chan EventConsumerEvent { return c.events } +func (c *EventConsumer) Stats() EventConsumerStats { + return EventConsumerStats{Running: c.running.Load(), Started: c.started.Load(), HandlerFailures: c.failures.Load(), Acknowledged: c.acked.Load(), Reconnects: c.reconnects.Load(), DroppedEvents: c.dropped.Load()} +} +func (c *EventConsumer) emit(typ, id, code string) { + select { + case c.events <- EventConsumerEvent{Type: typ, EventID: id, Code: code}: + default: + c.dropped.Add(1) + } +} +func (c *EventConsumer) String() string { + return fmt.Sprintf("EventConsumer{SubscriptionID:%s Capacity:%d secrets:redacted}", c.options.SubscriptionID, c.options.Capacity) +} +func (c *EventConsumer) GoString() string { return c.String() } + +type consumerJob struct { + proof EventProof + complete bool +} +type consumerResult struct { + id string + failed bool +} + +// Run在ctx取消后停止启动新处理器，取消在途处理器，在收尾时限内确认已提交结果。 +// 收到认证/协议等永久错误时也取消并等待在途处理器；未确认结果仍保留在服务端。 +// 重连先核对已完成处理器的原证明；证明被替代时等待新投递，由业务去重命中后确认。 +func (c *EventConsumer) Run(parent context.Context) error { + if !c.used.CompareAndSwap(false, true) { + return ErrEventConsumerAlreadyRun + } + defer close(c.events) + if parent.Err() != nil { + return nil + } + ctx, cancel := context.WithCancel(context.WithoutCancel(parent)) + defer cancel() + handlerCtx, stopHandlers := context.WithCancel(ctx) + defer stopHandlers() + abort := make(chan struct{}) + watchDone := make(chan struct{}) + go func() { + defer close(watchDone) + select { + case <-ctx.Done(): + return + case <-parent.Done(): + case <-abort: + } + stopHandlers() + timer := time.NewTimer(c.options.ShutdownTimeout) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + cancel() + } + }() + defer func() { cancel(); <-watchDone }() + var conn *EventConnection + var heartbeatDone <-chan struct{} + closeConnection := func() { + if conn != nil { + _ = conn.Close() + <-heartbeatDone + conn = nil + } + } + defer closeConnection() + jobs := make(map[string]*consumerJob) + results := make(chan consumerResult, c.options.Capacity) + draining := false + var cause error + beginDrain := func(err error) { + if !draining { + draining = true + stopHandlers() + close(abort) + c.emit("draining", "", "") + } + if cause == nil { + cause = err + } + } + finish := func(err error) error { + if err == nil { + return nil + } + return &EventConsumerRunError{Cause: err, Unconfirmed: len(jobs), Running: c.running.Load()} + } + backoff := c.options.ReconnectMinDelay + var dialAt time.Time + everConnected := false + disconnect := func(err error) { + closeConnection() + if !eventRetryable(err) { + beginDrain(err) + return + } + floor := backoff / 2 + dialAt = time.Now().Add(floor + time.Duration(rand.Int64N(int64(backoff-floor)+1))) + backoff = min(backoff*2, c.options.ReconnectMaxDelay) + c.emit("reconnecting", "", eventErrorCode(err)) + } + for { + if parent.Err() != nil { + beginDrain(nil) + } + if ctx.Err() != nil { + return finish(ErrEventShutdownTimeout) + } + if draining && (len(jobs) == 0 || cause != nil && c.running.Load() == 0) { + return finish(cause) + } + if conn == nil && cause == nil && !time.Now().Before(dialAt) { + if everConnected { + c.reconnects.Add(1) + } + var err error + conn, err = c.client.DialEvents(ctx, EventRegistration{SubscriptionID: c.options.SubscriptionID, Capacity: c.options.Capacity}) + if err != nil { + disconnect(err) + continue + } + everConnected = true + heartbeatDone = startEventHeartbeat(ctx, conn) + c.emit("connected", "", "") + continue + } + if conn != nil { + if err := conn.Err(); err != nil { + disconnect(err) + continue + } + // 一个ACK在途；读取与自动心跳独立，不需要接收方不断调用Next维持连接。 + acknowledged := false + for id, j := range jobs { + if !j.complete { + continue + } + messageID, err := NewEventMessageID() + if err == nil { + _, err = conn.Ack(ctx, messageID, j.proof) + } + if err == nil { + delete(jobs, id) + c.acked.Add(1) + c.emit("acknowledged", id, "") + backoff = c.options.ReconnectMinDelay + } else { + var remote *EventError + if errors.As(err, &remote) && remote.Code == "DELIVERY_LEASE_LOST" { + delete(jobs, id) + c.emit("proof_lost", id, remote.Code) + } else { + disconnect(err) + } + } + acknowledged = true + break + } + if acknowledged { + continue + } + } + var deliveries <-chan EventGrant + var disconnected <-chan struct{} + if conn != nil { + disconnected = conn.Done() + if !draining && len(jobs) < int(c.options.Capacity) { + deliveries = conn.deliveries + } + } + var parentDone <-chan struct{} + if !draining { + parentDone = parent.Done() + } + // 定时器只负责重连；活跃连接没有每轮轮询。 + var wake <-chan time.Time + var timer *time.Timer + if conn == nil && cause == nil { + timer = time.NewTimer(max(time.Millisecond, time.Until(dialAt))) + wake = timer.C + } + select { + case <-ctx.Done(): + case <-parentDone: + beginDrain(nil) + case <-wake: + case <-disconnected: + disconnect(conn.Err()) + case result := <-results: + j := jobs[result.id] + if result.failed { + delete(jobs, result.id) + c.failures.Add(1) + c.emit("handler_failed", result.id, "HANDLER_FAILED") + } else { + j.complete = true + } + case grant := <-deliveries: + if parent.Err() != nil || conn.Err() != nil { + break + } + id := grant.Event.ID + if j := jobs[id]; j != nil { + if j.proof.DeliveryID != grant.Delivery.ID { + disconnect(ErrEventProtocol) + break + } + if grant.Delivery.LeaseEpoch > j.proof.Epoch { + j.proof = grant.Proof() + } + break + } + jobs[id] = &consumerJob{proof: grant.Proof()} + c.running.Add(1) + c.started.Add(1) + c.emit("handler_started", id, "") + go func(event ResultEvent) { + failed := true + defer func() { + _ = recover() + c.running.Add(-1) + select { + case results <- consumerResult{id: event.ID, failed: failed}: + case <-ctx.Done(): + } + }() + failed = c.options.Handle(handlerCtx, event) != nil + }(grant.Event) + } + if timer != nil { + timer.Stop() + } + } +} +func startEventHeartbeat(ctx context.Context, conn *EventConnection) <-chan struct{} { + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(time.Duration(conn.Welcome().HeartbeatIntervalMS) * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-conn.Done(): + return + case <-ticker.C: + id, err := NewEventMessageID() + if err == nil { + _, err = conn.Ping(ctx, id) + } + if err != nil { + conn.fail(err) + return + } + } + } + }() + return done +} +func eventRetryable(err error) bool { + if errors.Is(err, ErrEventDisconnected) || errors.Is(err, ErrEventBackpressure) || errors.Is(err, context.DeadlineExceeded) { + return true + } + var remote *EventError + if errors.As(err, &remote) { + return remote.Code == "UNAVAILABLE" + } + var httpErr *HTTPError + if errors.As(err, &httpErr) { + return httpErr.StatusCode == 408 || httpErr.StatusCode == 429 || httpErr.StatusCode >= 500 && httpErr.StatusCode <= 599 + } + return false +} +func eventErrorCode(err error) string { + var remote *EventError + if errors.As(err, &remote) { + return remote.Code + } + return "TRANSPORT_INTERRUPTED" +} diff --git a/sdk/go/event_consumer_test.go b/sdk/go/event_consumer_test.go new file mode 100644 index 0000000..fe6e8a1 --- /dev/null +++ b/sdk/go/event_consumer_test.go @@ -0,0 +1,342 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "runtime" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" +) + +func runEventConsumer(t *testing.T, client *Client, handle func(context.Context, ResultEvent) error) (*EventConsumer, context.CancelFunc, <-chan error) { + t.Helper() + consumer, err := NewEventConsumer(client, EventConsumerOptions{SubscriptionID: eventTestSub, Capacity: 1, Handle: handle, ReconnectMinDelay: 10 * time.Millisecond, ReconnectMaxDelay: 50 * time.Millisecond, ShutdownTimeout: time.Second}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- consumer.Run(ctx) }() + t.Cleanup(cancel) + return consumer, cancel, done +} +func awaitConsumerEvent(t *testing.T, c *EventConsumer, typ string) EventConsumerEvent { + t.Helper() + timer := time.NewTimer(3 * time.Second) + defer timer.Stop() + for { + select { + case event, ok := <-c.Events(): + if !ok { + t.Fatal("consumer stopped before " + typ) + } + if event.Type == typ { + return event + } + case <-timer.C: + t.Fatal("consumer did not emit " + typ) + } + } +} +func awaitConsumerStop(t *testing.T, done <-chan error) error { + t.Helper() + select { + case err := <-done: + return err + case <-time.After(3 * time.Second): + t.Fatal("consumer did not stop") + return nil + } +} +func TestEventConsumerReconcilesLostAckWithoutRerunningHandler(t *testing.T) { + g := eventTestGrant() + var calls atomic.Int64 + var proofs [2]EventProof + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, n int) { + if n > 2 { + t.Error("unexpected third connection") + return + } + if n == 1 { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(g)}) + } + f, err := eventTestRead(ctx, c) + if err != nil { + return + } + if f.Type != "event.ack" || json.Unmarshal(f.Body, &proofs[n-1]) != nil { + t.Error("missing ACK proof") + return + } + if n == 1 { + return + } // ACK已在服务端提交，回应在网络中丢失。 + if proofs[0].DeliveryID != proofs[1].DeliveryID || proofs[0].Epoch != proofs[1].Epoch || proofs[0].LeaseSecret != proofs[1].LeaseSecret { + t.Error("reconnect changed acknowledgement identity") + } + out := eventAckFor(g) + out.Deduplicated = true + _ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(out)}) + _, _ = eventTestRead(ctx, c) + }) + consumer, cancel, done := runEventConsumer(t, client, func(context.Context, ResultEvent) error { calls.Add(1); return nil }) + awaitConsumerEvent(t, consumer, "acknowledged") + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + if calls.Load() != 1 || consumer.Stats().Acknowledged != 1 || consumer.Stats().Reconnects != 1 { + t.Fatal("lost receipt re-executed committed handler", consumer.Stats()) + } + if !errors.Is(consumer.Run(context.Background()), ErrEventConsumerAlreadyRun) { + t.Fatal("consumer ran twice") + } +} +func TestEventConsumerSupersededProofRequiresDurableDeduplication(t *testing.T) { + g := eventTestGrant() + var calls, effects atomic.Int64 + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(g)}) + f, err := eventTestRead(ctx, c) + if err != nil { + return + } + _ = eventTestWrite(ctx, c, eventFrame{Type: "error", ReplyTo: f.MessageID, Error: &EventError{Code: "DELIVERY_LEASE_LOST", Message: "proof superseded"}}) + next := g + next.Delivery.LeaseEpoch = 2 + next.Delivery.Attempts = 2 + next.Delivery.Revision = 4 + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(next)}) + f, err = eventTestRead(ctx, c) + if err != nil { + return + } + _ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(eventAckFor(next))}) + _, _ = eventTestRead(ctx, c) + }) + consumer, cancel, done := runEventConsumer(t, client, func(context.Context, ResultEvent) error { calls.Add(1); effects.CompareAndSwap(0, 1); return nil }) + awaitConsumerEvent(t, consumer, "acknowledged") + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + if calls.Load() != 2 || effects.Load() != 1 { + t.Fatal("replacement proof bypassed handler deduplication") + } +} +func TestEventConsumerHandlerFailureDoesNotAcknowledge(t *testing.T) { + for _, panicHandler := range []bool{false, true} { + name := "error" + if panicHandler { + name = "panic" + } + t.Run(name, func(t *testing.T) { + g := eventTestGrant() + retry := make(chan struct{}) + var calls atomic.Int64 + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(g)}) + select { + case <-retry: + case <-ctx.Done(): + return + } + next := g + next.Delivery.LeaseEpoch = 2 + next.Delivery.Attempts = 2 + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(next)}) + f, err := eventTestRead(ctx, c) + if err != nil { + return + } + var proof EventProof + if f.Type != "event.ack" || json.Unmarshal(f.Body, &proof) != nil || proof.Epoch != 2 { + t.Error("failed handler acknowledged original delivery") + } + _ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(eventAckFor(next))}) + _, _ = eventTestRead(ctx, c) + }) + consumer, cancel, done := runEventConsumer(t, client, func(context.Context, ResultEvent) error { + if calls.Add(1) == 1 { + if panicHandler { + panic("private handler data") + } + return errors.New("private handler data") + } + return nil + }) + awaitConsumerEvent(t, consumer, "handler_failed") + if calls.Load() != 1 { + t.Fatal("handler retried locally") + } + close(retry) + awaitConsumerEvent(t, consumer, "acknowledged") + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + if consumer.Stats().HandlerFailures != 1 || consumer.Stats().Acknowledged != 1 { + t.Fatal(consumer.Stats()) + } + }) + } +} +func TestEventConsumerShutdownCancelsAndConfirmsCommittedHandler(t *testing.T) { + g := eventTestGrant() + started := make(chan struct{}) + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(g)}) + f, err := eventTestRead(ctx, c) + if err != nil { + return + } + _ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(eventAckFor(g))}) + _, _ = eventTestRead(ctx, c) + }) + consumer, cancel, done := runEventConsumer(t, client, func(ctx context.Context, _ ResultEvent) error { close(started); <-ctx.Done(); return nil }) + <-started + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + if consumer.Stats().Acknowledged != 1 || consumer.Stats().Running != 0 { + t.Fatal("committed result not drained", consumer.Stats()) + } +} +func TestEventConsumerShutdownBoundsIgnoringHandler(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + defer close(release) + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(eventTestGrant())}) + _, _ = eventTestRead(ctx, c) + }) + consumer, err := NewEventConsumer(client, EventConsumerOptions{SubscriptionID: eventTestSub, Capacity: 1, ShutdownTimeout: 30 * time.Millisecond, Handle: func(context.Context, ResultEvent) error { close(started); <-release; return nil }}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(ctx) }() + <-started + cancel() + err = awaitConsumerStop(t, done) + var run *EventConsumerRunError + if !errors.Is(err, ErrEventShutdownTimeout) || !errors.As(err, &run) || run.Unconfirmed != 1 || run.Running != 1 { + t.Fatal("uncooperative handler not reported", err) + } +} +func TestEventConsumerPermanentFailureDoesNotReconnect(t *testing.T) { + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls.Add(1); w.WriteHeader(http.StatusUnauthorized) })) + defer server.Close() + client, _ := NewClient(server.URL, Options{Token: "token"}) + _, _, done := runEventConsumer(t, client, func(context.Context, ResultEvent) error { t.Error("unauthorized consumer invoked handler"); return nil }) + err := awaitConsumerStop(t, done) + var status *HTTPError + if !errors.As(err, &status) || status.StatusCode != 401 || calls.Load() != 1 { + t.Fatal("permanent handshake error retried", err) + } +} + +func TestEventConsumerHandlerGoexitReleasesSlotWithoutAck(t *testing.T) { + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, _ EventRegistration, _ int) { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(eventTestGrant())}) + if f, err := eventTestRead(ctx, c); err == nil && f.Type == "event.ack" { + t.Error("Goexit handler was acknowledged") + } + }) + consumer, cancel, done := runEventConsumer(t, client, func(context.Context, ResultEvent) error { runtime.Goexit(); return nil }) + awaitConsumerEvent(t, consumer, "handler_failed") + cancel() + if err := awaitConsumerStop(t, done); err != nil || consumer.Stats().Running != 0 { + t.Fatal("Goexit leaked handler capacity", err) + } +} + +func TestEventConsumerConcurrentCapacity(t *testing.T) { + start := make(chan string, 3) + release := make(chan struct{}) + var active, peak atomic.Int64 + grants := []EventGrant{eventTestGrant(), eventTestGrant(), eventTestGrant()} + for i := range grants { + grants[i].Event.ID = workerTestID(t) + grants[i].Delivery.ID = workerTestID(t) + grants[i].Delivery.EventID = grants[i].Event.ID + } + client := eventMock(t, func(ctx context.Context, c *websocket.Conn, in EventRegistration, _ int) { + if in.Capacity != 2 { + t.Error("capacity changed") + } + for _, g := range grants[:2] { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(g)}) + } + for i := 0; i < 3; i++ { + f, err := eventTestRead(ctx, c) + if err != nil { + return + } + var p EventProof + if f.Type != "event.ack" || json.Unmarshal(f.Body, &p) != nil { + t.Error("missing ACK") + return + } + var grant EventGrant + for _, g := range grants { + if g.Delivery.ID == p.DeliveryID { + grant = g + } + } + _ = eventTestWrite(ctx, c, eventFrame{Type: "ack", ReplyTo: f.MessageID, Body: eventBody(eventAckFor(grant))}) + if i == 0 { + _ = eventTestWrite(ctx, c, eventFrame{Type: "event.deliver", Body: eventBody(grants[2])}) + } + } + _, _ = eventTestRead(ctx, c) + }) + consumer, err := NewEventConsumer(client, EventConsumerOptions{SubscriptionID: eventTestSub, Capacity: 2, Handle: func(ctx context.Context, event ResultEvent) error { + n := active.Add(1) + defer active.Add(-1) + for { + old := peak.Load() + if n <= old || peak.CompareAndSwap(old, n) { + break + } + } + start <- event.ID + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(ctx) }() + for range 2 { + select { + case <-start: + case <-ctx.Done(): + t.Fatal("capacity not utilized") + } + } + close(release) + for range 3 { + awaitConsumerEvent(t, consumer, "acknowledged") + } + cancel() + if err := awaitConsumerStop(t, done); err != nil || peak.Load() != 2 || consumer.Stats().Started != 3 { + t.Fatal("capacity or drain invariant", err, consumer.Stats()) + } +} diff --git a/sdk/go/event_integration_test.go b/sdk/go/event_integration_test.go new file mode 100644 index 0000000..b75661c --- /dev/null +++ b/sdk/go/event_integration_test.go @@ -0,0 +1,190 @@ +package orbit + +import ( + "context" + "encoding/json" + "errors" + "sync/atomic" + "testing" + "time" +) + +func integrationEventSubscription(t *testing.T, ctx context.Context, api *Client, maxAttempts uint32) Subscription { + t.Helper() + queue, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "event-sdk-" + workerTestID(t), MaxConcurrency: 1, RatePerSecond: 10}) + if err != nil { + t.Fatal(err) + } + sub, err := api.CreateSubscription(ctx, workerTestID(t), CreateSubscriptionRequest{Name: "sdk-" + workerTestID(t), QueueID: queue.Queue.ID, Types: []string{"task.cancelled"}, MaxPending: 10, MaxAttempts: maxAttempts}) + if err != nil { + t.Fatal(err) + } + return sub.Subscription +} +func integrationCancelledEvent(t *testing.T, ctx context.Context, api *Client, sub Subscription) Task { + t.Helper() + task, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: sub.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}) + if err != nil { + t.Fatal(err) + } + cancelled, err := api.CancelTask(ctx, task.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "subscription SDK regression"}) + if err != nil { + t.Fatal(err) + } + return cancelled.Task +} +func eventDialIntegration(t *testing.T, ctx context.Context, api *Client, sub Subscription, capacity uint32) *EventConnection { + t.Helper() + c, err := api.DialEvents(ctx, EventRegistration{SubscriptionID: sub.ID, Capacity: capacity}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Close() }) + return c +} +func TestIntegrationEventConnectionSDK(t *testing.T) { + api, _, _ := integrationWorkerClient(t) + ctx, stop := context.WithTimeout(context.Background(), 15*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, api, 3) + task := integrationCancelledEvent(t, ctx, api, sub) + a := eventDialIntegration(t, ctx, api, sub, 1) + grant, err := a.Next(ctx) + if err != nil || grant.Event.Snapshot.TaskID != task.ID || grant.Event.Snapshot.Revision != task.Revision || grant.Event.Type != "task.cancelled" { + t.Fatal("SDK snapshot mismatch", err) + } + paused, err := api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: sub.Revision, Reason: "pause with active delivery", Enabled: false}) + if err != nil { + t.Fatal(err) + } + pong, err := a.Ping(ctx, workerTestID(t)) + if err != nil || pong.Enabled { + t.Fatal("paused subscription heartbeat", err) + } + first, err := a.Ack(ctx, workerTestID(t), grant.Proof()) + if err != nil { + t.Fatal(err) + } + a.Close() + b := eventDialIntegration(t, ctx, api, sub, 1) + duplicate, err := b.Ack(ctx, workerTestID(t), grant.Proof()) + if err != nil || !duplicate.Deduplicated || !duplicate.AcknowledgedAt.Equal(first.AcknowledgedAt) { + t.Fatal("ACK proof could not cross connections", err) + } + _ = integrationCancelledEvent(t, ctx, api, sub) + short, cancel := context.WithTimeout(ctx, 100*time.Millisecond) + _, err = b.Next(short) + cancel() + if !errors.Is(err, context.DeadlineExceeded) || b.Err() != nil { + t.Fatal("paused subscription delivered or Next closed connection", err) + } + if _, err = api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: paused.Subscription.Revision, Reason: "resume", Enabled: true}); err != nil { + t.Fatal(err) + } + second, err := b.Next(ctx) + if err != nil || second.Event.ID == grant.Event.ID { + t.Fatal("resume failed", err) + } + if _, err = b.Ack(ctx, workerTestID(t), second.Proof()); err != nil { + t.Fatal(err) + } + current, err := api.GetSubscription(ctx, sub.ID) + if err != nil || current.PendingCount != 0 { + t.Fatal("SDK acknowledgement accounting", err) + } +} +func TestIntegrationEventConsumerHeartbeatAndLostAck(t *testing.T) { + api, _, token := integrationWorkerClient(t) + ctx, stop := context.WithTimeout(context.Background(), 25*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, api, 3) + _ = integrationCancelledEvent(t, ctx, api, sub) + gateway := newProtocolFaultGateway(t, api.baseURL, "/v1/events/connect", eventSubprotocol) + gateway.drop.Store("event.ack") + client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()}) + if err != nil { + t.Fatal(err) + } + var calls atomic.Int64 + consumer, err := NewEventConsumer(client, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, event ResultEvent) error { + calls.Add(1) + // 慢业务期间真实等待15秒自动心跳；确认计数来自上游服务端ACK，不是客户端写入。 + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + for gateway.confirmedEventPings.Load() == 0 { + select { + case <-work.Done(): + return work.Err() + case <-ticker.C: + } + } + return nil + }}) + if err != nil { + t.Fatal(err) + } + run, cancel := context.WithCancel(ctx) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(run) }() + for consumer.Stats().Acknowledged == 0 { + select { + case <-ctx.Done(): + t.Fatal("consumer did not reconcile committed ACK") + case err := <-done: + t.Fatal("consumer stopped", err) + case <-time.After(20 * time.Millisecond): + } + } + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + if calls.Load() != 1 || consumer.Stats().Reconnects < 1 || gateway.confirmedEventPings.Load() < 1 { + t.Fatal("heartbeat or reconciliation missing", consumer.Stats()) + } + page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked}) + if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 1 { + t.Fatal("lost ACK created another delivery attempt", err) + } + t.Log("real 15-second heartbeat and committed ACK loss recovered; handler and delivery attempt each remained one") +} +func TestIntegrationEventConsumerFailureRedeliversNaturally(t *testing.T) { + api, _, _ := integrationWorkerClient(t) + ctx, stop := context.WithTimeout(context.Background(), 45*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, api, 2) + _ = integrationCancelledEvent(t, ctx, api, sub) + var calls atomic.Int64 + consumer, err := NewEventConsumer(api, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(context.Context, ResultEvent) error { + if calls.Add(1) == 1 { + return errors.New("receiver transaction rolled back") + } + return nil + }}) + if err != nil { + t.Fatal(err) + } + run, cancel := context.WithCancel(ctx) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(run) }() + for consumer.Stats().Acknowledged == 0 { + select { + case <-ctx.Done(): + t.Fatal("delivery did not retry after natural lease expiry") + case err := <-done: + t.Fatal("consumer stopped", err) + case <-time.After(50 * time.Millisecond): + } + } + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked}) + if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 2 || calls.Load() != 2 || consumer.Stats().HandlerFailures != 1 { + t.Fatal("handler error or retry budget mishandled", err) + } + t.Log("handler error caused no ACK; same delivery reappeared after its real 30-second lease and backoff, then acknowledged at attempt two") +} diff --git a/sdk/go/event_process_integration_test.go b/sdk/go/event_process_integration_test.go new file mode 100644 index 0000000..de68874 --- /dev/null +++ b/sdk/go/event_process_integration_test.go @@ -0,0 +1,103 @@ +package orbit + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" +) + +func TestIntegrationEventProcessLeaseRecovery(t *testing.T) { + a := startOrbitProcess(t) + _, _, token := integrationWorkerClient(t) + api := processClient(t, a, token) + ctx, stop := context.WithTimeout(context.Background(), 45*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, api, 3) + _ = integrationCancelledEvent(t, ctx, api, sub) + firstConn := eventDialIntegration(t, ctx, api, sub, 1) + first, err := firstConn.Next(ctx) + if err != nil { + t.Fatal(err) + } + a.kill(t) + select { + case <-firstConn.Done(): + case <-time.After(3 * time.Second): + t.Fatal("dead server connection remained live") + } + b := startOrbitProcess(t) + other := processClient(t, b, token) + secondConn := eventDialIntegration(t, ctx, other, sub, 1) + second, err := secondConn.Next(ctx) + if err != nil || first.Event.ID != second.Event.ID || first.Delivery.ID != second.Delivery.ID || second.Delivery.LeaseEpoch != 2 || second.Delivery.Attempts != 2 { + t.Fatal("node restart did not recover original delivery", err) + } + _, err = secondConn.Ack(ctx, workerTestID(t), first.Proof()) + var remote *EventError + if !errors.As(err, &remote) || remote.Code != "DELIVERY_LEASE_LOST" { + t.Fatal("dead node proof acknowledged newer lease", err) + } + if _, err = secondConn.Ack(ctx, workerTestID(t), second.Proof()); err != nil { + t.Fatal(err) + } + current, err := other.GetSubscription(ctx, sub.ID) + if err != nil || current.PendingCount != 0 { + t.Fatal("recovered acknowledgement not persisted", err) + } + t.Log("SIGKILL confirmed by Wait; new process recovered original delivery after natural lease expiry, rejected old proof and committed new ACK") +} +func TestIntegrationEventProcessAckCommitResponseLoss(t *testing.T) { + a := startOrbitProcess(t) + _, _, token := integrationWorkerClient(t) + api := processClient(t, a, token) + ctx, stop := context.WithTimeout(context.Background(), 20*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, api, 3) + _ = integrationCancelledEvent(t, ctx, api, sub) + gateway := newProtocolFaultGateway(t, a.url, "/v1/events/connect", eventSubprotocol) + gateway.drop.Store("event.ack") + gateway.dropOffline.Store(true) + client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()}) + if err != nil { + t.Fatal(err) + } + var calls atomic.Int64 + consumer, err := NewEventConsumer(client, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(context.Context, ResultEvent) error { calls.Add(1); return nil }}) + if err != nil { + t.Fatal(err) + } + run, cancel := context.WithCancel(ctx) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(run) }() + awaitProcessDrop(t, ctx, gateway, "event.ack") + before, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked}) + if err != nil || len(before.Items) != 1 || before.Items[0].AcknowledgedAt == nil { + t.Fatal("fault occurred before durable ACK", err) + } + a.kill(t) + b := startOrbitProcess(t) + other := processClient(t, b, token) + gateway.target.Store(b.url) + gateway.offline.Store(false) + for consumer.Stats().Acknowledged == 0 { + select { + case <-ctx.Done(): + t.Fatal("consumer failed to reconcile ACK on new process") + case err := <-done: + t.Fatal("consumer stopped", err) + case <-time.After(20 * time.Millisecond): + } + } + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + after, err := other.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked}) + if err != nil || len(after.Items) != 1 || after.Items[0].Attempts != 1 || after.Items[0].AcknowledgedAt == nil || !after.Items[0].AcknowledgedAt.Equal(*before.Items[0].AcknowledgedAt) || calls.Load() != 1 { + t.Fatal("ACK receipt changed or committed handler reran", err) + } + t.Log("ACK commit verified before response loss and SIGKILL; replacement process returned original receipt without another handler or delivery attempt") +} diff --git a/sdk/go/event_types.go b/sdk/go/event_types.go new file mode 100644 index 0000000..cd22468 --- /dev/null +++ b/sdk/go/event_types.go @@ -0,0 +1,87 @@ +package orbit + +import ( + "errors" + "fmt" + "time" +) + +var ( + ErrEventClosed = errors.New("event connection closed") + ErrEventDisconnected = errors.New("event connection interrupted; acknowledgement may be unconfirmed") + ErrEventProtocol = errors.New("invalid event protocol response") + ErrEventBusy = errors.New("event control request limit reached") + ErrEventBackpressure = errors.New("event delivery buffer exhausted") +) + +// EventError 是服务端公开错误。Error/String/GoString只展示分类，不输出远端正文。 +type EventError struct { + Code string `json:"code"` + Message string `json:"message"` +} + +func (e *EventError) Error() string { return "orbit event " + e.Code } +func (e *EventError) String() string { return e.Error() } +func (e *EventError) GoString() string { return e.Error() } + +// EventRegistration 绑定一个已有订阅。容量固定为1至32，重连可以重新选择。 +type EventRegistration struct { + SubscriptionID string `json:"subscription_id"` // SubscriptionID 已创建且属于当前凭据的订阅。 + Capacity uint32 `json:"capacity"` +} + +// EventWelcome 是当前连接的握手快照；Enabled变化通过Ping查询。 +type EventWelcome struct { + SessionID string `json:"session_id"` + SubscriptionID string `json:"subscription_id"` + Capacity uint32 `json:"capacity"` + LeaseDurationMS int64 `json:"lease_duration_ms"` + HeartbeatIntervalMS int64 `json:"heartbeat_interval_ms"` + ServerTime time.Time `json:"server_time"` + Enabled bool `json:"enabled"` +} +type EventPong struct { + ServerTime time.Time `json:"server_time"` + Enabled bool `json:"enabled"` +} + +// EventGrant 包含冻结结果及当前交付证明。不得将JSON写到普通日志。 +// 租约只控制平台补发，不授权接收方跳过去重；不同消费者仍可能并行收到同一事件。 +type EventGrant struct { + Delivery EventDelivery `json:"delivery"` + Event ResultEvent `json:"event"` + LeaseSecret string `json:"lease_secret"` +} + +// EventProof 保留原身份及秘密，在同订阅的新连接上仍可核对ACK。 +// SubscriptionID和EventID用于SDK本地关联校验；只把后三个字段发送到服务端。 +type EventProof struct { + SubscriptionID string `json:"subscription_id"` + EventID string `json:"event_id"` + DeliveryID string `json:"delivery_id"` + Epoch uint64 `json:"epoch,string"` + LeaseSecret string `json:"lease_secret"` +} + +func (g EventGrant) Proof() EventProof { + return EventProof{SubscriptionID: g.Delivery.SubscriptionID, EventID: g.Event.ID, DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret} +} + +type EventAckResult struct { + DeliveryID string `json:"delivery_id"` + EventID string `json:"event_id"` + Epoch uint64 `json:"epoch,string"` + AcknowledgedAt time.Time `json:"acknowledged_at"` + Deduplicated bool `json:"deduplicated"` +} + +// NewEventMessageID 生成控制消息关联ID；ACK的持久化身份是交付ID、代次和证明。 +func NewEventMessageID() (string, error) { return NewWorkerMessageID() } +func (g EventGrant) String() string { + return fmt.Sprintf("EventGrant{EventID:%s DeliveryID:%s data:redacted}", g.Event.ID, g.Delivery.ID) +} +func (g EventGrant) GoString() string { return g.String() } +func (p EventProof) String() string { + return fmt.Sprintf("EventProof{DeliveryID:%s Epoch:%d secrets:redacted}", p.DeliveryID, p.Epoch) +} +func (p EventProof) GoString() string { return p.String() } diff --git a/sdk/go/examples/event-protocol/main.go b/sdk/go/examples/event-protocol/main.go new file mode 100644 index 0000000..a754a44 --- /dev/null +++ b/sdk/go/examples/event-protocol/main.go @@ -0,0 +1,112 @@ +// event-protocol 验证冻结结果、连接切换与ACK重报；不执行业务写入。 +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 err + } + var credential struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if json.Unmarshal(raw, &credential) != nil { + return errors.New("invalid credential file") + } + if token == "" { + token = credential.Token + } + if queue == "" { + queue = credential.QueueID + } + } + if token == "" || queue == "" { + return errors.New("event-enabled token and isolated example queue are required") + } + api, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + if err != nil { + return err + } + ctx, stop := context.WithTimeout(context.Background(), 15*time.Second) + defer stop() + id, err := orbit.NewEventMessageID() + if err != nil { + return err + } + created, err := api.CreateSubscription(ctx, "sub-"+id, orbit.CreateSubscriptionRequest{Name: "protocol-" + id, QueueID: queue, Types: []string{"task.cancelled"}, MaxPending: 10, MaxAttempts: 3}) + if err != nil { + return err + } + sub := created.Subscription + task, err := api.SubmitTask(ctx, "task-"+id, orbit.SubmitTaskRequest{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}) + if err != nil { + return err + } + if _, err = api.CancelTask(ctx, task.Task.ID, "cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "event protocol example"}); err != nil { + return err + } + first, err := api.DialEvents(ctx, orbit.EventRegistration{SubscriptionID: sub.ID, Capacity: 1}) + if err != nil { + return err + } + defer first.Close() + grant, err := first.Next(ctx) + if err != nil { + return err + } + if grant.Event.Snapshot.TaskID != task.Task.ID || grant.Event.Snapshot.State != orbit.TaskCancelled { + return errors.New("unexpected example queue event") + } + // 本例只读取并验证结果，没有业务数据库写入；业务处理必须先提交Inbox事务再ACK。 + ackID, err := orbit.NewEventMessageID() + if err != nil { + return err + } + confirmed, err := first.Ack(ctx, ackID, grant.Proof()) + if err != nil { + return err + } + first.Close() + second, err := api.DialEvents(ctx, orbit.EventRegistration{SubscriptionID: sub.ID, Capacity: 1}) + if err != nil { + return err + } + defer second.Close() + replayID, err := orbit.NewEventMessageID() + if err != nil { + return err + } + replayed, err := second.Ack(ctx, replayID, grant.Proof()) + if err != nil { + return err + } + if !replayed.Deduplicated || !replayed.AcknowledgedAt.Equal(confirmed.AcknowledgedAt) { + return errors.New("acknowledgement receipt changed across connections") + } + paused, err := api.ConfigureSubscription(ctx, sub.ID, "pause-"+id, orbit.ConfigureSubscriptionRequest{ExpectedRevision: sub.Revision, Enabled: false, Reason: "protocol example completed"}) + if err != nil { + return err + } + fmt.Printf("event=%s task=%s state=%s\n", grant.Event.ID, task.Task.ID, grant.Event.Snapshot.State) + fmt.Printf("ack_replayed=%t same_receipt=true pending=%d subscription_enabled=%t\n", replayed.Deduplicated, paused.Subscription.PendingCount, paused.Subscription.Enabled) + return nil +} diff --git a/sdk/go/worker_integration_test.go b/sdk/go/worker_integration_test.go index c829382..75956e2 100644 --- a/sdk/go/worker_integration_test.go +++ b/sdk/go/worker_integration_test.go @@ -48,18 +48,23 @@ func integrationWorkerClient(t *testing.T) (*Client, string, string) { // workerFaultGateway 在实际服务端确认后丢弃指定 ACK；SDK 到网关使用真实 TLS。 // 只丢响应，不伪造任务状态，测试凭据不进入 URL 或输出。 type workerFaultGateway struct { - drop atomic.Value - target atomic.Value - dropOffline atomic.Bool - dropCancel atomic.Bool - offline atomic.Bool - mu sync.Mutex - active map[*websocket.Conn]struct{} - dropped chan string - server *httptest.Server + confirmedEventPings atomic.Uint64 + drop atomic.Value + target atomic.Value + dropOffline atomic.Bool + dropCancel atomic.Bool + offline atomic.Bool + mu sync.Mutex + active map[*websocket.Conn]struct{} + dropped chan string + server *httptest.Server } func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { + return newProtocolFaultGateway(t, upstream, "/v1/worker/connect", workerSubprotocol) +} + +func newProtocolFaultGateway(t *testing.T, upstream, path, protocol string) *workerFaultGateway { t.Helper() g := &workerFaultGateway{dropped: make(chan string, 4), active: make(map[*websocket.Conn]struct{})} g.drop.Store("") @@ -77,14 +82,14 @@ func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { defer stop() header := http.Header{} header.Set("Authorization", r.Header.Get("Authorization")) - up, _, err := websocket.Dial(bridge, g.target.Load().(string)+"/v1/worker/connect", &websocket.DialOptions{HTTPHeader: header, Subprotocols: []string{workerSubprotocol}}) + up, _, err := websocket.Dial(bridge, g.target.Load().(string)+path, &websocket.DialOptions{HTTPHeader: header, Subprotocols: []string{protocol}}) if err != nil { http.Error(w, "upstream unavailable", 503) return } defer up.CloseNow() up.SetReadLimit(512 << 10) - down, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{workerSubprotocol}}) + down, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{protocol}}) if err != nil { return } @@ -138,6 +143,9 @@ func newWorkerFaultGateway(t *testing.T, upstream string) *workerFaultGateway { operation := operations[f.ReplyTo] delete(operations, f.ReplyTo) mu.Unlock() + if f.Type == "ack" && operation == "events.ping" { + g.confirmedEventPings.Add(1) + } if f.Type == "task.cancel" && g.dropCancel.CompareAndSwap(true, false) { g.dropped <- "task.cancel" return