From a8a8dcfe1ee7d6d2d4c5b103564d702c1b14d6e1 Mon Sep 17 00:00:00 2001 From: k2safe Date: Thu, 17 Sep 2026 20:49:51 +0800 Subject: [PATCH] feat: add durable business inbox example and consumer crash recovery --- AGENTS.md | 2 +- Makefile | 4 + README.md | 2 +- docs/01-functional-spec.md | 2 +- docs/07-data-model.md | 10 +- docs/08-development-plan.md | 19 +- docs/09-result-subscription-core.md | 6 +- docs/api/retries.md | 2 +- docs/api/subscriptions.md | 2 +- docs/examples/README.md | 10 +- docs/examples/event-protocol.md | 2 +- docs/examples/result-inbox.md | 109 ++++++ docs/reports/phase-3-result-inbox.md | 83 +++++ docs/sdk/README.md | 2 +- docs/sdk/event-consumer.md | 2 +- examples/result-inbox/README.md | 21 ++ examples/result-inbox/cmd/receiver/main.go | 210 ++++++++++++ examples/result-inbox/go.mod | 15 + examples/result-inbox/go.sum | 6 + .../inbox/commit_loss_integration_test.go | 184 ++++++++++ .../consumer_process_integration_test.go | 298 +++++++++++++++++ examples/result-inbox/internal/inbox/model.go | 28 ++ .../result-inbox/internal/inbox/schema.sql | 32 ++ examples/result-inbox/internal/inbox/store.go | 314 ++++++++++++++++++ .../internal/inbox/store_integration_test.go | 187 +++++++++++ .../result-inbox/internal/inbox/store_test.go | 59 ++++ scripts/test-mysql.sh | 7 + 27 files changed, 1601 insertions(+), 17 deletions(-) create mode 100644 docs/examples/result-inbox.md create mode 100644 docs/reports/phase-3-result-inbox.md create mode 100644 examples/result-inbox/README.md create mode 100644 examples/result-inbox/cmd/receiver/main.go create mode 100644 examples/result-inbox/go.mod create mode 100644 examples/result-inbox/go.sum create mode 100644 examples/result-inbox/internal/inbox/commit_loss_integration_test.go create mode 100644 examples/result-inbox/internal/inbox/consumer_process_integration_test.go create mode 100644 examples/result-inbox/internal/inbox/model.go create mode 100644 examples/result-inbox/internal/inbox/schema.sql create mode 100644 examples/result-inbox/internal/inbox/store.go create mode 100644 examples/result-inbox/internal/inbox/store_integration_test.go create mode 100644 examples/result-inbox/internal/inbox/store_test.go diff --git a/AGENTS.md b/AGENTS.md index 8ac6bd3..22c40b7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,4 +14,4 @@ - 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 - 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 -- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK和节点进程故障回归；业务侧持久化Inbox示例及消费者进程退出后的事务去重回归完成前，不标记P3订阅能力完成。 +- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。P3仍需补齐批量受理、独立任务核查/人工重试、任务查询筛选和事务Outbox接入；不能因订阅完成而跳过原功能范围。 diff --git a/Makefile b/Makefile index 9f32501..5949659 100644 --- a/Makefile +++ b/Makefile @@ -4,19 +4,23 @@ GO ?= go fmt: $(GO) fmt ./... cd sdk/go && $(GO) fmt ./... + cd examples/result-inbox && GOWORK=off $(GO) fmt ./... test: $(GO) test -race ./... cd sdk/go && $(GO) test -race ./... + cd examples/result-inbox && GOWORK=off $(GO) test -race ./... vet: $(GO) vet ./... cd sdk/go && $(GO) vet ./... + cd examples/result-inbox && GOWORK=off $(GO) vet ./... build: $(GO) build -o bin/orbit ./cmd/orbit $(GO) build -o bin/orbit-migrate ./cmd/orbit-migrate $(GO) build -o bin/orbit-admin ./cmd/orbit-admin + cd examples/result-inbox && GOWORK=off $(GO) build -o ../../bin/orbit-inbox-example ./cmd/receiver integration: ./scripts/test-mysql.sh diff --git a/README.md b/README.md index 52de16a..5b390cd 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](docs/sdk/event-consumer.md) 已实现，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) 和 [独立 MySQL Inbox 示例](docs/examples/result-inbox.md) 已实现；剩余 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/01-functional-spec.md b/docs/01-functional-spec.md index 149897a..c1969fa 100644 --- a/docs/01-functional-spec.md +++ b/docs/01-functional-spec.md @@ -1,6 +1,6 @@ # Orbit 功能说明：Pulse 与公共能力 -> 版本 0.2 · 2026-09-17。Pulse 与 Weave 首版功能范围已确认；标注“建议”的细节待评审。Weave 具体范围另见流程设计。基础骨架已实现，任务业务能力仍待开发；实现状态见持续开发计划。 +> 版本 0.2 · 2026-09-17。Pulse 与 Weave 首版功能范围已确认；标注“建议”的细节待评审。Weave 具体范围另见流程设计。基础、任务执行、调度、资源控制和可靠订阅已实现；批量受理、独立任务核查/人工重试及查询筛选等继续开发，实现状态见持续开发计划。 ## 1. 目标与责任边界 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 74f7249..5005ab4 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -1,6 +1,6 @@ # 数据模型与实现基线 -更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现，Inbox 场景及流程编排继续开发。 +更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅核心、HTTP/WSS、管理与消费 SDK 已实现，独立 Inbox 场景已实现，流程编排继续开发。 ## 1. 模块与依赖 @@ -95,7 +95,7 @@ 禁止“先查后另开事务写入”。需要防止并发状态改变的关联行在事务内持锁，统一锁顺序并保留必要索引。当前不提供硬删除；将来增加清理时，必须核验在途任务、引用、幂等恢复窗口与审计保留期，按有限批次处理，不能依赖数据库级联。直接绕过服务写 SQL 不受应用校验保护。 -P1 已完成动作/队列授权、幂等摘要与原回执、受理事件及取消 CAS/审计事务；P3 扩展到运行中取消。订阅已实现服务层及持久化核心，包含事务内队列、所有者和交付归属校验；HTTP/WSS、管理与消费 SDK 已接入，业务 Inbox 示例待完成。流程目前仍只具备模型，后续引擎也必须补齐事务内父资源归属验证。 +P1 已完成动作/队列授权、幂等摘要与原回执、受理事件及取消 CAS/审计事务；P3 扩展到运行中取消。订阅已实现服务层及持久化核心，包含事务内队列、所有者和交付归属校验；HTTP/WSS、管理与消费 SDK 已接入，业务 Inbox 示例已独立交付。流程目前仍只具备模型，后续引擎也必须补齐事务内父资源归属验证。 P2 服务端已新增会话隔离、两步派发、租约、并发许可、结果回执/证据和恢复扫描，见 [执行协议](api/worker.md)。底层及自动 Worker SDK、尝试/证据查询、Schema 与进程故障恢复已通过 [P2 阶段验收](reports/phase-2.md)。 @@ -151,8 +151,12 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 ## 13. 可靠结果订阅核心 -000013_result_subscriptions 新增 EventStream、ResultEvent，扩展 Subscription 和 Delivery。289 个直接声明字段均有中文注释。生产事务在独立流行锁下分配序号，与任务、原事件和结果快照一起提交；消费事务不锁流行、任务、队列或应用额度。交付记录和扫描位置一起落库，每条交付单独持有租约和证明。重复确认不重复扣减积压，耗尽保留记录，人工追加预算不重置历史。详见 [订阅核心](09-result-subscription-core.md)。对外 HTTP/WSS、管理与消费 SDK 已实现，持久化 Inbox 完整场景继续推进。 +000013_result_subscriptions 新增 EventStream、ResultEvent，扩展 Subscription 和 Delivery。289 个直接声明字段均有中文注释。生产事务在独立流行锁下分配序号，与任务、原事件和结果快照一起提交；消费事务不锁流行、任务、队列或应用额度。交付记录和扫描位置一起落库，每条交付单独持有租约和证明。重复确认不重复扣减积压，耗尽保留记录，人工追加预算不重置历史。详见 [订阅核心](09-result-subscription-core.md)。对外 HTTP/WSS、管理与消费 SDK 已实现，持久化 Inbox 完整场景已交付。 ## 14. 交付历史查询索引 000014_delivery_history 增加按订阅及可选状态过滤的创建时间分页索引，领取、确认和重试不改变排序键。不增加模型字段或外键，仍为35个实体、289个中文注释字段。查询按凭据所有者、命名空间和订阅校验，游标绑定过滤条件；详情按来源事件ID读取不可变快照且不返回交付秘密。见 [订阅查询接口](api/subscriptions.md)。 + +## 15. 独立接收方示例模型 + +[result-inbox示例](examples/result-inbox.md) 在业务方自己的数据库建立receiver_inbox和receiver_task_views，2个实体、17个Go/SQL中文注释字段，无外键。它们属于独立示例模块，不计入平台35个实体，也不加入平台迁移。Inbox去重记录与业务视图同事务提交；视图仅按更高任务版本推进，重复和迟到事件不能重复更新或回退业务结果。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 5d7f0b1..e812d4a 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/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景、COMMIT响应丢失及消费者进程故障回归已完成；批量受理、独立任务核查/人工重试、业务键/时间筛选与事务Outbox接入待完成 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -35,6 +35,20 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最后才开始补文档。 +### P3 功能范围核对 + +可靠订阅链路完成后，重新对照 [原功能范围](01-functional-spec.md) 和 [事务衔接设计](04-reliability-and-data.md)，以下原定首版能力仍有缺口。不能只按已实现模块反推P3完成；先补齐并回归，再进入完整阶段验收。 + +| 剩余项 | 当前证据与后续交付 | +| --- | --- | +| 批量任务受理 | [任务API](api/tasks.md) 明确尚未提供；需有界逐项结果、独立幂等、部分失败补交、SDK与运行示例 | +| 独立任务人工核查与重试 | [重试API](api/retries.md) 仍未提供解除不确定性的核查接口；需证据、权限、版本、审计和历史保留，人工重试身份按关联新任务实现并冻结契约 | +| 任务查询筛选 | [当前ListTasks](../sdk/go/types.go) 仅队列/状态/分页；需按业务关联键及创建时间范围查询，游标绑定所有筛选，保持授权与有界索引扫描 | +| 业务到平台的事务Outbox接入 | [事务Outbox设计](04-reliability-and-data.md#61-业务系统到平台) 尚缺完整接入示例；需业务变更与待发请求同事务、稳定提交身份、响应不明核对和发送恢复，保持核心SDK无ORM/数据库依赖 | + +凭据安全重绑、历史清理、全量监控/容量与MySQL故障切换仍按P6执行；Weave的通用执行门控/内部事件接续随P4–P5落实，不将外部结果订阅当作内部流程引擎。 + + ## 4. SDK 与示例交付清单 | 场景 | 预期调用链 | 所属阶段 | @@ -48,7 +62,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 跨命名空间应用配额 | first 受理 → second 全局满额 → 释放 → 原键成功 → 恢复配置 | P3，已实现完整示例 | | 跨队列共享资源 | 两队列同键 → 一个有效许可 → 递增序号 → 资源端拒绝旧写入 | P3，已实现完整示例 | | 优先级与应用隔离 | 近同时到期按优先级 → 旧任务等待补偿 → 应用/队列轮转 | P3，已实现完整示例 | -| 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，核心、HTTP/WSS、管理 SDK/示例已实现；消费 SDK 已实现，持久化 Inbox 场景待完成 | +| 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，核心、HTTP/WSS、管理及消费 SDK、独立Inbox示例与真实进程故障回归已完成 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | | 审核与外部回执 | 提交流程 → 等待信号 → 授权信号注入 → 继续 | P4 | | 外部作业轮询 | 提交一次 → 有界查询处理中状态 → 回执或最终结果 → 超限核查 | P5 | @@ -79,6 +93,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [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) +- [P3 业务 Inbox 与消费者故障回归](reports/phase-3-result-inbox.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 6c0b6da..388ecf1 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 与自动消费器](sdk/event-consumer.md) 已实现，持久化 Inbox 示例继续开发；本页的存储方法属于服务端内部模块。 +状态：P3 的存储与服务层已实现并通过本地全量回归，见 [验证记录](reports/phase-3-result-subscription-core.md)。[HTTP/WSS、管理 SDK 与管理示例](api/subscriptions.md) 已接入。[事件连接 SDK 与自动消费器](sdk/event-consumer.md) 已实现，[持久化 Inbox 示例](examples/result-inbox.md) 及消费者强杀回归已完成；本页的存储方法属于服务端内部模块。 ## 1. 已实现范围 @@ -67,7 +67,7 @@ bootstrap 新增显式 --event-access，用于签发包含这三项动作的开 ACK 表示消费者已可靠保存或处理结果，不授予新业务执行权。当前最后一次秘密和代次即使过期，只要没有被新的交付替换，仍可用于迟到确认；新交付产生后，旧证明统一返回 DELIVERY_LEASE_LOST。确认和重新交付竞争通过订阅锁串行化，只会有一方有效。 -服务端不能保证消费者业务恰好执行一次。后续 SDK 示例需展示接收方以 event_id 建立唯一 Inbox，在本地事务中保存结果/业务变更，提交后才 ACK。断线和 ACK 响应丢失可能造成重复投递，调用方不得按“收到过一次”就直接丢弃尚未可靠保存的内容。 +服务端不能保证消费者业务恰好执行一次。[独立接收方示例](examples/result-inbox.md) 展示以 event_id 建立唯一 Inbox，在本地事务中保存结果/业务变更，提交后才 ACK。断线和 ACK 响应丢失可能造成重复投递，调用方不得按“收到过一次”就直接丢弃尚未可靠保存的内容。 每次交付授权、首次有效确认、订阅管理和人工重试均记录审计。交付审计包含事件标识、代次、次数、连接和租约截止，不记录交付秘密或结果正文。 @@ -91,4 +91,4 @@ retained_from_sequence 初始且目前固定为 1。未来实现清理后，如 ## 8. 下一步 -HTTP 创建/查询/分页、暂停/继续、交付与耗尽记录查询、人工重试，以及 orbit.events.v1 WSS、消息 Schema、管理 SDK 和管理示例已实现，见 [接口手册](api/subscriptions.md)。[事件连接 SDK 和自动消费器](sdk/event-consumer.md) 已实现；继续完成持久化 Inbox 去重示例与消费者进程故障回归，再验收 P3 的可靠订阅能力。 +HTTP 创建/查询/分页、暂停/继续、交付与耗尽记录查询、人工重试，以及 orbit.events.v1 WSS、消息 Schema、管理 SDK 和管理示例已实现，见 [接口手册](api/subscriptions.md)。[事件连接 SDK 和自动消费器](sdk/event-consumer.md) 已实现；[持久化 Inbox 去重示例](examples/result-inbox.md) 与消费者进程故障回归已完成。可靠订阅各层均有对应证据，P3其他功能缺口见开发计划。 diff --git a/docs/api/retries.md b/docs/api/retries.md index a6a7e0c..8d57582 100644 --- a/docs/api/retries.md +++ b/docs/api/retries.md @@ -1,6 +1,6 @@ # 安全重试 v1 -状态：P3 的安全重试已实现。使用现有任务提交与 Worker 结果接口，无独立的自动重试路由；[Cron/固定间隔计划](schedules.md) 和 [运行中取消](cancellation.md) 已接入；人工恢复与可靠结果订阅继续按 [计划](../08-development-plan.md) 开发。验证证据见 [本阶段报告](../reports/phase-3-safe-retry.md)。 +状态：P3 的安全重试已实现。使用现有任务提交与 Worker 结果接口，无独立的自动重试路由；[Cron/固定间隔计划](schedules.md) 和 [运行中取消](cancellation.md) 已接入；[可靠结果订阅](subscriptions.md) 与业务Inbox已实现；独立任务人工核查与重试继续按 [计划](../08-development-plan.md) 开发。验证证据见 [本阶段报告](../reports/phase-3-safe-retry.md)。 ## 1. 三种不同的重报 diff --git a/docs/api/subscriptions.md b/docs/api/subscriptions.md index 8c18446..8704858 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)。它演示管理与幂等，不启动结果消费者。正式 [事件连接 SDK 与自动消费器](../sdk/event-consumer.md) 已提供接收、ACK、心跳、重连与有界收尾；[完整协议示例](../examples/event-protocol.md) 验证跨连接重报原确认。持久化 Inbox 示例继续开发。 +[完整管理示例](../../sdk/go/examples/subscriptions/main.go) 的启动、输入输出、失败和保留规则见 [场景文档](../examples/subscriptions.md)。它演示管理与幂等，不启动结果消费者。正式 [事件连接 SDK 与自动消费器](../sdk/event-consumer.md) 已提供接收、ACK、心跳、重连与有界收尾；[完整协议示例](../examples/event-protocol.md) 验证跨连接重报原确认。[独立 MySQL Inbox 示例](../examples/result-inbox.md) 已实现并验证消费者进程退出后的去重。 机器可读消息契约见 [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 4d42d2c..48a85d1 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -114,7 +114,7 @@ - 步骤：创建 → 同键核对 → 暂停 → 恢复 → 查询与分页 → 最终暂停。 - 预期：duplicate=true、resumed_revision=3、final_enabled=false、revision=4、pending=0。 - 保留：每次运行留下一个已暂停订阅及审计；没有硬删除，受64个订阅上限约束。 -- 范围：本例只演示管理SDK；自动消费 SDK 已实现，持久化Inbox例子继续开发。 +- 范围：本例只演示管理SDK；自动消费 SDK 已实现，[持久化Inbox例子](result-inbox.md) 已独立交付。 ## 已可运行：结果消费协议与跨连接确认 @@ -122,3 +122,11 @@ - 操作：创建订阅 → 提交并取消任务 → 接收冻结结果 → ACK → 重连重报原证明 → 核对原确认时间 → 暂停订阅。 - 输出：ack_replayed=true、same_receipt=true、pending=0、subscription_enabled=false。 - 范围：本例只做读取验证，不写业务数据；[自动消费器](../sdk/event-consumer.md) 的生产接入须提供事务Inbox，不能把示例当作业务恰好一次保证。 + +## 已可运行：独立业务 MySQL Inbox + +- 独立项目：[result-inbox](../../examples/result-inbox/README.md)，[完整前置、命令及故障说明](result-inbox.md)。 +- 操作：接收冻结结果 → Inbox与业务视图同事务提交 → 重复核对 → ACK → 暂停本次演示订阅。 +- 输出：inbox_duplicate=true、applied_events=1、ack_confirmed=1、pending=0。 +- 覆盖：并发重复、旧版本保护、回滚、真实MySQL COMMIT响应丢失、业务提交后/ACK前消费者SIGKILL及新进程去重。 +- 边界：只覆盖接收方本地SQL效果；核心SDK不依赖数据库，外部副作用需要接收方自己的幂等/Outbox设计。 diff --git a/docs/examples/event-protocol.md b/docs/examples/event-protocol.md index 1b604c2..50360c1 100644 --- a/docs/examples/event-protocol.md +++ b/docs/examples/event-protocol.md @@ -2,7 +2,7 @@ [完整代码](../../sdk/go/examples/event-protocol/main.go) 使用正式 Go SDK 创建取消结果订阅，提交并取消一个待执行任务，接收并验证冻结结果，然后 ACK；关闭连接后在新连接中重报原证明，验证原确认时间和去重回执，最后暂停本次订阅。 -本例只读取并验证结果，没有业务数据库写入。生产业务应使用 [自动消费器及事务去重约定](../sdk/event-consumer.md)。持久化 Inbox 完整示例尚待交付。 +本例只读取并验证结果，没有业务数据库写入。生产业务应使用 [自动消费器及事务去重约定](../sdk/event-consumer.md)。持久化接入见 [独立 MySQL Inbox 示例](result-inbox.md)。 ## 启动 diff --git a/docs/examples/result-inbox.md b/docs/examples/result-inbox.md new file mode 100644 index 0000000..6832a49 --- /dev/null +++ b/docs/examples/result-inbox.md @@ -0,0 +1,109 @@ +# 独立业务数据库的结果 Inbox + +[完整示例项目](../../examples/result-inbox/README.md) 演示：结果通过正式消费 SDK 到达接收方后，在自己的 MySQL 中将事件去重记录与任务结果视图一起提交，再允许 SDK ACK。重复事件不会再次推进视图，旧版本不会覆盖新版本。 + +示例不连接平台内部存储，不依赖业务 ORM，不创建外键或级联删除。它演示的是本地SQL事务中的去重，不保证任意外部HTTP副作用恰好一次。源码、schema与模型都在独立的 examples/result-inbox 模块内，核心 sdk/go/go.mod 没有新增数据库驱动。 + +## 1. 前置与初始化 + +需要 Go1.27.1、已启动的 Orbit，以及接收方自己的 MySQL8.4+ 专用空数据库，例如 receiver_example。不要把接收方表建到平台业务库。先用接收方管理员创建数据库并通过安全环境配置注入 ORBIT_RECEIVER_DSN；DSN必须显式包含库名。生产使用受信任TLS及自己的数据库持久化、备份策略。 + +初始化所需权限为CREATE TABLE，运行账号只需要这两张接收方表的SELECT/INSERT/UPDATE。应用运行不需要DELETE或修改平台库。示例初始化只执行初始DDL，已有表会报错；不自动修复部分DDL或覆盖已有结构，正式升级应采用接收方自己的前向迁移。 + +从仓库根目录准备一个专用示例队列和事件凭据（平台 ORBIT_MYSQL_DSN 已按初始化手册配置）： + +```sh +mkdir -p .local +chmod 700 .local +go run ./cmd/orbit-admin bootstrap \ + --application inbox-example --namespace default --environment development \ + --queue inbox-demo --event-access --output .local/inbox.credentials.json +``` + +随后从仓库根目录进入示例项目，ORBIT_RECEIVER_DSN由环境管理器提供： + +```sh +cd examples/result-inbox +GOWORK=off go run ./cmd/receiver init +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/inbox.credentials.json \ + GOWORK=off go run ./cmd/receiver demo +``` + +本地示例用回环HTTP；生产连接使用HTTPS/WSS。GOWORK=off让此独立示例模块不依赖根workspace。首次构建按go.mod/go.sum解析依赖。 + +## 2. 演示链路与输出 + +demo会创建一个只关心task.cancelled的订阅，提交并取消一个待执行任务，再启动EventConsumer。收到本次任务的事件后，连续两次调用相同业务事务：第一次推进结果视图，第二次核对相同已提交的Inbox记录。处理器返回nil后SDK确认交付，程序检查当前业务视图，再暂停本次订阅。 + +```text +receiver_schema_initialized=true foreign_keys=0 +task=<本次任务ID> state=CANCELLED inbox_duplicate=true applied_events=1 +ack_confirmed=1 pending=0 subscription_enabled=false +``` + +第一行来自init，后两行来自demo。demo总时限20秒，消费器收尾最多5秒。它只接收本次任务，需用专用隔离队列；队列中另一个程序的取消结果会被拒绝处理，避免演示混用业务数据。 + +demo失败时非零退出，保留已受理任务、订阅、交付和业务记录供核查。成功时仅暂停订阅，不硬删历史。每次运行新建身份，命名空间订阅上限64，重复演示宜使用可重建的测试环境。服务端故障和强杀不由demo命令自行注入，专门回归见下文。 + +## 3. 持续消费已有订阅 + +使用与订阅创建时相同的凭据，提供订阅ID。配置队列必须和该订阅一致；程序启动时先查询并核对，防止配置错队列后不断耗费投递预算。 + +```sh +ORBIT_URL=http://127.0.0.1:8080 \ + ORBIT_CREDENTIALS_FILE=../../.local/inbox.credentials.json \ + ORBIT_SUBSCRIPTION_ID='<已有订阅的32位ID>' \ + RECEIVER_NAME=task-result-view \ + GOWORK=off go run ./cmd/receiver consume +``` + +ORBIT_RECEIVER_DSN仍需由环境提供。ORBIT_TOKEN、ORBIT_QUEUE_ID可覆盖凭据文件对应值。持续消费需要events:read、events:consume及订阅绑定队列权限；demo另外需要events:manage、tasks:submit和tasks:cancel。程序输出稳定事件分类、事件ID与脱敏存储错误，不输出凭据、结果正文或租约证明。 + +RECEIVER_NAME默认task-result-view，最多64个ASCII字符，以字母或数字开头，其余允许字母、数字、点、下划线和连字符。它表示固定的接收用途；重启、扩容、多个实例必须复用同一名称和队列。不同用途是不同业务视图，不能每次启动随机换名来规避重复。 + +持续消费并发4，收到SIGINT/SIGTERM时取消在途处理器并最多收尾10秒。只关闭本进程，不暂停共享订阅；其他消费者仍可继续。单次业务事务最多5秒。死锁、数据库不可用、提交不明等返回错误，SDK不ACK，按服务端交付租约及预算恢复。 + +## 4. 数据模型与事务 + +[schema.sql](../../examples/result-inbox/internal/inbox/schema.sql) 与 [中文模型](../../examples/result-inbox/internal/inbox/model.go) 定义两个实体、17个字段；不计入平台本身的35个实体。 + +| 表 | 身份和用途 | +| --- | --- | +| receiver_inbox | `(consumer_name, queue_id, event_id)`唯一；保存冻结事件、SHA256摘要、是否推进视图和UTC受理时间 | +| receiver_task_views | `(consumer_name, queue_id, task_id)`唯一；保存最大任务版本、最后事件/状态/结果、实际推进次数及UTC时间 | + +Apply的事务顺序： + +1. 校验来源队列、事件/任务/版本/结果状态关联；规范化完整事件JSON并计算摘要。对象键和空白顺序不影响摘要；数字使用json.Number保持精度及原十进制表达，时间统一UTC。重复键、非法Unicode、无效JSON、超过512KiB的事件被拒绝。 +2. 普通INSERT写Inbox唯一键。只对MySQL 1062重复键进入核对；其他错误都返回失败。唯一键竞争等待胜出事务提交或回滚，不把未提交占位当作成功。 +3. 重复时结束本事务，再查询已提交记录并比较摘要。同身份同内容返回Duplicate=true；不同内容报冲突，不ACK。Result.Applied返回原事务是否推进过视图，不表示本次又执行一次。 +4. 首次接收时原子建立或锁定单任务视图。事务内占位版本为0，正常提交后不留下占位状态。只接受更高任务版本；旧事件记录为applied=false，保留历史；同任务同版本但不同事件ID视为冲突并整体回滚。 +5. 更新视图及Inbox的applied标志，在一个READ COMMITTED事务中COMMIT。业务版本、结果及推进次数与Inbox同时提交；成功后Handle才返回nil。 + +TaskView是接收方只读结果视图，不是平台权威任务状态。NEEDS_REVIEW必须保留为待核查，不能转译成成功。原事件快照不从后来的GetTask重新拼装。未提供result时保存SQL NULL，JSON大整数不会经float64转换。 + +## 5. 故障与核查 + +| 窗口 | 行为 | +| --- | --- | +| 业务更新失败 | 事务回滚，Inbox与视图都不残留；原事件仍可重试 | +| COMMIT响应未知 | 返回ErrCommitUnconfirmed，不ACK；之后按原事件身份查Inbox，已提交则去重，未提交则重新处理 | +| 业务提交后消费者被强杀 | 原交付未确认；新进程收到补发后使用相同消费者范围命中Inbox，再确认新交付证明 | +| 平台ACK回应丢失 | SDK保留原证明重报；进程也退出时，后续补发仍通过Inbox去重 | +| 同ID不同内容 / 同任务同版本不同ID | 返回冲突，不自动覆盖或确认；检查来源数据与接收范围，保留原记录 | +| 连续失败耗尽预算 | 平台保留EXPIRED并占用窗口；修复根因后通过管理API追加有限预算，不能换消费者名掩盖问题 | + +Inbox保留期必须覆盖平台可补发、人工重试及备份恢复窗口。本例不清理或归档记录；删除去重记录后再次补发可能重新产生业务效果。生产还需要接收方权限、加密、备份恢复及容量治理。 + +本地SQL事务无法原子包含外部HTTP、邮件等效果。需要外部副作用时，应在同事务写接收方Outbox，再以稳定业务身份执行和核查；不能在持锁期间调用外部系统。此示例没有冒充通用Outbox框架。 + +## 6. 实际回归 + +根目录 `make integration` 创建临时MySQL，平台使用orbit_test，接收方使用独立orbit_receiver_test；结束清理容器。示例普通race/vet/build也纳入根Makefile。 + +- [存储回归](../../examples/result-inbox/internal/inbox/store_integration_test.go)：8个并发相同事件只推进1次、内容冲突、旧版本与范围隔离、业务更新失败整笔回滚、两表17个中文列注释及零外键。 +- [COMMIT响应丢失](../../examples/result-inbox/internal/inbox/commit_loss_integration_test.go)：本地协议代理收到MySQL对COMMIT的真实OK后丢弃并断连，调用方报告提交不明；另一连接验证实际数据已提交，新的Store按相同身份核对，不重复更新。 +- [消费者真实进程故障](../../examples/result-inbox/internal/inbox/consumer_process_integration_test.go)：读取数据库证明业务已提交、平台仍LEASED且未ACK，再SIGKILL消费者并Wait确认退出；新进程在真实30秒租约到期后去重并确认，同一交付attempts=2，业务applied_events仍为1。 + +故障屏障仅在测试二进制中提供，正常CLI没有故障开关。COMMIT代理仅处理本地无TLS测试连接，不作为生产组件。测试没有模拟外部业务副作用，也不等同于跨地域或MySQL主从故障切换验收。 diff --git a/docs/reports/phase-3-result-inbox.md b/docs/reports/phase-3-result-inbox.md new file mode 100644 index 0000000..0975b7e --- /dev/null +++ b/docs/reports/phase-3-result-inbox.md @@ -0,0 +1,83 @@ +# P3 业务 Inbox 与消费者故障回归 + +验证日期：2026-09-17。交付 [独立 MySQL Inbox 示例](../examples/result-inbox.md)。可靠订阅已具备生产者事务日志、HTTP/WSS、管理及消费 SDK、事务接收示例与对应故障证据。P3仍需补齐原功能范围中的批量受理、独立任务核查/人工重试、查询筛选和业务Outbox接入，持续目标保持进行中。 + +## 1. 交付与模块边界 + +- [独立示例模块](../../examples/result-inbox/go.mod) 引用正式SDK和MySQL驱动，不依赖平台内部模型/存储，不给核心SDK增加数据库依赖。GOWORK=off可单独构建、测试和运行。 +- [业务存储](../../examples/result-inbox/internal/inbox/store.go) 以固定消费者用途、队列、事件身份去重；冻结文档摘要冲突拒绝确认。Inbox与任务结果视图同事务提交，视图仅接受更高任务版本。 +- [接收方模型](../../examples/result-inbox/internal/inbox/model.go) 与 [独立SQL](../../examples/result-inbox/internal/inbox/schema.sql) 包含2个实体、17个字段，Go与SQL均有中文注释，无外键和级联删除。平台本身仍是35个实体、289个直接声明持久化字段，未增加平台迁移。 +- [可运行CLI](../../examples/result-inbox/cmd/receiver/main.go) 提供init、demo、consume；初始化不覆盖已有表，持续消费核对订阅队列，演示核对业务重复、ACK及订阅暂停。 +- Makefile加入独立示例的fmt/race/vet/build，统一集成脚本单独创建orbit_receiver_test数据库，与平台orbit_test隔离。 + +## 2. 验证环境与结果 + +macOS arm64、Go1.27.1、Docker MySQL8.4.10、一次性数据库、回环HTTP/WSS、真实Orbit和消费者子进程。数据库代理只用于本地无TLS的COMMIT故障注入，测试不会连接既有业务库；脚本结束清理自己的容器和临时凭据。 + +```sh +make fmt test vet build +make integration +cd examples/result-inbox +GOWORK=off go test -race ./... +GOWORK=off go vet ./... +``` + +| 分组 | 本次结果 | +| --- | --- | +| 全量平台MySQL集成 | PASS，91.331秒 | +| SDK Worker连接/自动运行器 | PASS，87.413秒 | +| SDK计划 / 队列 / 订阅管理 / 应用额度 | PASS，5.637 / 4.449 / 33.392 / 1.395秒 | +| SDK事件连接与自动消费 | PASS，50.959秒 | +| 新业务接收方六项集成 | PASS，35.512秒 | +| 其中真实COMMIT响应丢失 | PASS，0.02秒 | +| 其中消费者提交后/ACK前SIGKILL | PASS，34.20秒；使用真实30秒交付租约 | +| 原九项真实进程故障 | PASS，60.953秒；加上新消费者故障共十项 | +| 完整接入示例 | 15个全部通过，包括新的独立Inbox demo | +| 根及独立示例fmt/race/vet/build | PASS | + +第一次专项启动脚本使用了不存在的测试名，导致未建平台表而失败；修正为实际的TestIntegrationSchemaAndConstraints后，专项及上表全量回归通过。没有把“无测试运行”的输出当成验证证据。 + +实际demo输出： + +```text +receiver_schema_initialized=true foreign_keys=0 +task=<本次任务ID> state=CANCELLED inbox_duplicate=true applied_events=1 +ack_confirmed=1 pending=0 subscription_enabled=false +``` + +## 3. 正确性与数据证据 + +[真实存储测试](../../examples/result-inbox/internal/inbox/store_integration_test.go) 以8个并发事务处理同一事件：1次首次处理、7次已提交重复，Inbox只有1行，业务applied_events=1。相同事件ID改动内容返回冲突，原结果及推进次数保持不变。 + +迟到的低版本事件保存Inbox历史但不回退视图；同任务同版本不同事件ID整体回滚。不同消费者用途有独立去重范围，错队列被拒绝。测试用MySQL触发器使业务更新实际失败，验证Inbox及占位视图一起回滚，移除注入后原身份可重新提交。 + +[文档身份测试](../../examples/result-inbox/internal/inbox/store_test.go) 验证对象键顺序、空白和同一时刻的时区表达不改变摘要；业务数字变化会改变摘要。保留9007199254740993和HTML字符，拒绝重复JSON键、无效JSON/UTF-8和未配对UTF-16转义。 + +真实information_schema断言接收方2张表、17个中文列注释、0个外键。平台原v7升级回归继续确认70个历史外键最终为0、325项原索引列定义保留。 + +## 4. MySQL COMMIT响应丢失 + +[代理测试](../../examples/result-inbox/internal/inbox/commit_loss_integration_test.go) 转发真实MySQL握手、事务和语句。收到COMMIT对应的成功OK包后，代理丢弃该包并断开连接；没有伪造成功响应，也没有直接修改表制造已提交状态。 + +Apply返回ErrCommitUnconfirmed而非成功。测试用另一条直连查询确认Inbox及业务视图确实已经提交，再用新建Store处理同一冻结事件：命中相同摘要，返回Duplicate=true，业务推进次数保持1。测试日志中的MySQL unexpected EOF是这次有意断连的结果。 + +这证明本地事务提交不明时按稳定事件身份核对的路径，不是MySQL主从切换、磁盘故障或跨地域复制验收。 + +## 5. 消费者真实进程强杀 + +[消费者子进程测试](../../examples/result-inbox/internal/inbox/consumer_process_integration_test.go) 使用正式EventConsumer与Store： + +1. 首个进程在Apply成功后停在测试屏障，尚未从处理器返回nil，因此SDK未发送ACK。 +2. 父进程查询业务数据库，确认Inbox与视图均已提交、applied_events=1；查询平台，确认交付仍LEASED且acknowledged_at为空。 +3. SIGKILL消费者，用Wait确认其确实因信号退出。重启时不会复用原进程内存、证明或去重集合。 +4. 第二个全新进程使用相同消费者用途、队列和订阅，在原交付真实租约到期后收到补发。它读取业务数据库的原Inbox，Duplicate=true，然后ACK新证明并正常退出。 +5. 最终event_id、delivery_id保持不变，平台attempts=2并ACKED，业务Inbox仍1行、applied_events仍1。 + +测试屏障只存在于测试二进制；正常示例CLI不提供故障开关。已经提交的业务结果不因强杀而回滚，恢复依靠持久化Inbox，而非把新进程视为从未执行过。 + +## 6. 剩余边界与计划 + +- Inbox保证本例SQL事务内的原子去重与版本推进，不涵盖邮件、外部HTTP等业务副作用；需要相应幂等和接收方Outbox设计。 +- 核心SDK仍只有网络协议依赖，业务表不进入平台迁移。正式接收方需要自己的DDL升级、权限、加密、容量、保留及备份恢复方案。 +- 历史清理、凭据重绑、全量监控告警、MySQL故障切换和生产容量/RPO/RTO仍按P6实施。 +- 已重新核对[原功能范围](../01-functional-spec.md)与[事务衔接要求](../04-reliability-and-data.md)，将尚缺的批量受理、独立任务核查/人工重试、业务键/时间筛选与业务Outbox接入明确列回[持续计划](../08-development-plan.md)。P3及整个平台均未宣称最终完成。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 6d3bf3f..983a651 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -188,4 +188,4 @@ P2 已提供底层协议、[自动 Worker 运行器及示例](worker-runtime.md) ## 订阅管理示例 -[完整运行方法](../examples/subscriptions.md) 使用显式 --event-access 凭据，执行创建/同键核对/暂停/恢复/查询，再暂停本次订阅并保留审计。源码见 [examples/subscriptions](../../sdk/go/examples/subscriptions/main.go)。WSS 连接 SDK 与自动消费器见 [消费手册](event-consumer.md)，可运行 [协议示例](../examples/event-protocol.md) 验证跨连接 ACK；持久化 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 示例](../examples/result-inbox.md) 已独立交付，核心SDK不新增数据库驱动。 diff --git a/docs/sdk/event-consumer.md b/docs/sdk/event-consumer.md index 37151eb..f797018 100644 --- a/docs/sdk/event-consumer.md +++ b/docs/sdk/event-consumer.md @@ -2,7 +2,7 @@ 对应 [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 内存状态作为持久化去重。 +已提供 [EventConnection](../../sdk/go/event_connection.go)、[EventConsumer](../../sdk/go/event_consumer.go)、[公开类型](../../sdk/go/event_types.go) 与 [可运行协议示例](../examples/event-protocol.md)。[独立 MySQL Inbox 示例](../examples/result-inbox.md) 已提供业务事务去重，并验证消费者提交后、ACK前被强杀的恢复；SDK内存状态仍不能代替Inbox。 ## 1. 自动消费 diff --git a/examples/result-inbox/README.md b/examples/result-inbox/README.md new file mode 100644 index 0000000..fef88e6 --- /dev/null +++ b/examples/result-inbox/README.md @@ -0,0 +1,21 @@ +# 独立 MySQL Inbox 示例 + +这是接收方应用示例，使用正式 Orbit Go SDK 和自己的 MySQL 数据库。核心 SDK 不依赖数据库驱动；示例是独立 Go module，通过本地 replace 引用 SDK。 + +完整前置、调用、输出、故障处理与事务边界见 [运行手册](../../docs/examples/result-inbox.md)。 + +从本目录执行： + +```sh +GOWORK=off go run ./cmd/receiver init +GOWORK=off go run ./cmd/receiver demo +GOWORK=off go run ./cmd/receiver consume +``` + +- `init`：使用 ORBIT_RECEIVER_DSN 初始化专用空数据库，不自动升级已有表。 +- `demo`：使用事件凭据和隔离队列产生结果，验证两次业务处理只推进一次，再确认并暂停本次订阅。 +- `consume`：消费 ORBIT_SUBSCRIPTION_ID 指定的已有订阅，RECEIVER_NAME 在重启和多实例之间保持固定。 + +[业务事务](internal/inbox/store.go) 同时写入 [Inbox 与视图](internal/inbox/schema.sql)，两张表、17个中文字段注释、零外键。模型见 [model.go](internal/inbox/model.go)。 + +根目录 `make test/vet/build` 包含本模块；`make integration` 创建独立临时数据库，运行数据库、COMMIT响应丢失、真实消费者强杀回归和demo，结束清理自建容器。未设置集成环境时普通go test跳过外部测试，不表示这些故障已经验证。 diff --git a/examples/result-inbox/cmd/receiver/main.go b/examples/result-inbox/cmd/receiver/main.go new file mode 100644 index 0000000..377834f --- /dev/null +++ b/examples/result-inbox/cmd/receiver/main.go @@ -0,0 +1,210 @@ +// receiver 使用独立业务数据库演示结果Inbox与版本化业务视图。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "os/signal" + "syscall" + "time" + + "git.yuntuo.hk/k2safe/orbit/examples/result-inbox/internal/inbox" + 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 { + if len(os.Args) != 2 || (os.Args[1] != "init" && os.Args[1] != "consume" && os.Args[1] != "demo") { + return errors.New("usage: receiver init|consume|demo") + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + open, cancel := context.WithTimeout(ctx, 10*time.Second) + db, err := inbox.Open(open, os.Getenv("ORBIT_RECEIVER_DSN")) + cancel() + if err != nil { + return err + } + defer db.Close() + if os.Args[1] == "init" { + bounded, stop := context.WithTimeout(ctx, 10*time.Second) + defer stop() + if err = inbox.Init(bounded, db); err != nil { + return err + } + fmt.Println("receiver_schema_initialized=true foreign_keys=0") + return nil + } + api, queue, err := client() + if err != nil { + return err + } + name := os.Getenv("RECEIVER_NAME") + if name == "" { + name = "task-result-view" + } + store, err := inbox.New(db, name, queue) + if err != nil { + return err + } + if os.Args[1] == "demo" { + return demo(ctx, api, store, queue) + } + check, checkStop := context.WithTimeout(ctx, 5*time.Second) + subscription, err := api.GetSubscription(check, os.Getenv("ORBIT_SUBSCRIPTION_ID")) + checkStop() + if err != nil { + return err + } + if subscription.QueueID != queue { + return errors.New("subscription does not match configured receiver queue") + } + consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: subscription.ID, Capacity: 4, Handle: func(work context.Context, event orbit.ResultEvent) error { + err := store.Handle(work, event) + if err != nil { + fmt.Fprintf(os.Stderr, "receiver_event=%s error=%v\n", event.ID, err) + } + return err + }, ShutdownTimeout: 10 * time.Second}) + if err != nil { + return err + } + observed := make(chan struct{}) + go func() { + defer close(observed) + for event := range consumer.Events() { + fmt.Printf("consumer_event=%s event_id=%s code=%s\n", event.Type, event.EventID, event.Code) + } + }() + err = consumer.Run(ctx) + <-observed + return err +} +func client() (*orbit.Client, string, 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 nil, "", errors.New("cannot read credential file") + } + var c struct { + Token string `json:"token"` + QueueID string `json:"queue_id"` + } + if json.Unmarshal(raw, &c) != nil { + return nil, "", errors.New("invalid credential file") + } + if token == "" { + token = c.Token + } + if queue == "" { + queue = c.QueueID + } + } + if token == "" || queue == "" { + return nil, "", errors.New("event-enabled credential and queue are required") + } + c, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + return c, queue, err +} +func demo(parent context.Context, api *orbit.Client, store *inbox.Store, queue string) error { + ctx, stop := context.WithTimeout(parent, 20*time.Second) + defer stop() + id, err := orbit.NewEventMessageID() + if err != nil { + return err + } + created, err := api.CreateSubscription(ctx, "inbox-sub-"+id, orbit.CreateSubscriptionRequest{Name: "inbox-" + id, QueueID: queue, Types: []string{"task.cancelled"}, MaxPending: 10, MaxAttempts: 3}) + if err != nil { + return err + } + task, err := api.SubmitTask(ctx, "inbox-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, "inbox-cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "durable inbox example"}); err != nil { + return err + } + duplicate := make(chan bool, 1) + consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: created.Subscription.ID, Capacity: 1, ShutdownTimeout: 5 * time.Second, Handle: func(work context.Context, event orbit.ResultEvent) error { + if event.Snapshot.TaskID != task.Task.ID { + return inbox.ErrInvalidEvent + } + if _, err := store.Apply(work, event); err != nil { + return err + } + // 再用同一冻结事件进入业务事务，证明重复不会再次推进视图；真实进程故障另有回归。 + again, err := store.Apply(work, event) + if err != nil { + return err + } + if !again.Duplicate { + return errors.New("receiver duplicate was not recognized") + } + select { + case duplicate <- true: + default: + } + return nil + }}) + if err != nil { + return err + } + run, cancel := context.WithCancel(ctx) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(run) }() + confirmed := false + for !confirmed { + select { + case <-ctx.Done(): + cancel() + <-done + return errors.New("inbox demo timed out before confirmed acknowledgement") + case err := <-done: + if err == nil { + err = errors.New("consumer stopped before acknowledgement") + } + return err + case event, ok := <-consumer.Events(): + if !ok { + if err := <-done; err != nil { + return err + } + return errors.New("consumer stopped before acknowledgement") + } + confirmed = event.Type == "acknowledged" + } + } + cancel() + if err = <-done; err != nil { + return err + } + select { + case <-duplicate: + default: + return errors.New("duplicate verification missing") + } + view, err := store.View(ctx, task.Task.ID) + if err != nil { + return err + } + if view.AppliedEvents != 1 || view.LastState != "CANCELLED" { + return errors.New("duplicate changed business projection") + } + paused, err := api.ConfigureSubscription(ctx, created.Subscription.ID, "inbox-pause-"+id, orbit.ConfigureSubscriptionRequest{ExpectedRevision: created.Subscription.Revision, Enabled: false, Reason: "durable inbox example completed"}) + if err != nil { + return err + } + fmt.Printf("task=%s state=%s inbox_duplicate=true applied_events=%d\n", task.Task.ID, view.LastState, view.AppliedEvents) + fmt.Printf("ack_confirmed=%d pending=%d subscription_enabled=%t\n", consumer.Stats().Acknowledged, paused.Subscription.PendingCount, paused.Subscription.Enabled) + return nil +} diff --git a/examples/result-inbox/go.mod b/examples/result-inbox/go.mod new file mode 100644 index 0000000..593ffac --- /dev/null +++ b/examples/result-inbox/go.mod @@ -0,0 +1,15 @@ +module git.yuntuo.hk/k2safe/orbit/examples/result-inbox + +go 1.27.1 + +require ( + git.yuntuo.hk/k2safe/orbit/sdk/go v0.0.0 + github.com/go-sql-driver/mysql v1.10.1 +) + +require ( + filippo.io/edwards25519 v1.2.0 // indirect + github.com/coder/websocket v1.8.15 // indirect +) + +replace git.yuntuo.hk/k2safe/orbit/sdk/go => ../../sdk/go diff --git a/examples/result-inbox/go.sum b/examples/result-inbox/go.sum new file mode 100644 index 0000000..f808e3b --- /dev/null +++ b/examples/result-inbox/go.sum @@ -0,0 +1,6 @@ +filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo= +filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc= +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/go-sql-driver/mysql v1.10.1 h1:arlSnNLq6a5yxGxV7qg9lF4j0C+KwD6NbQyKr9QL6ME= +github.com/go-sql-driver/mysql v1.10.1/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk= diff --git a/examples/result-inbox/internal/inbox/commit_loss_integration_test.go b/examples/result-inbox/internal/inbox/commit_loss_integration_test.go new file mode 100644 index 0000000..3c48980 --- /dev/null +++ b/examples/result-inbox/internal/inbox/commit_loss_integration_test.go @@ -0,0 +1,184 @@ +package inbox + +import ( + "context" + "encoding/binary" + "errors" + "io" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/go-sql-driver/mysql" +) + +// commitLossGateway 只用于无TLS的隔离MySQL回归；捕获COMMIT的真实成功包后丢弃并断连。 +// 不伪造SQL结果，也不修改数据库中的提交状态。 +type commitLossGateway struct { + listener net.Listener + dropped chan struct{} + armed atomic.Bool + mu sync.Mutex + closing bool + connections map[net.Conn]struct{} + wg sync.WaitGroup + accepted chan struct{} +} + +func newCommitLossGateway(t *testing.T, target string) *commitLossGateway { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + g := &commitLossGateway{listener: listener, dropped: make(chan struct{}, 1), connections: map[net.Conn]struct{}{}, accepted: make(chan struct{})} + go func() { + defer close(g.accepted) + for { + down, err := listener.Accept() + if err != nil { + return + } + g.wg.Add(1) + go func() { + defer g.wg.Done() + defer down.Close() + up, err := net.DialTimeout("tcp", target, 3*time.Second) + if err != nil { + return + } + defer up.Close() + g.mu.Lock() + if g.closing { + g.mu.Unlock() + return + } + g.connections[down] = struct{}{} + g.connections[up] = struct{}{} + g.mu.Unlock() + defer func() { g.mu.Lock(); delete(g.connections, down); delete(g.connections, up); g.mu.Unlock() }() + var committing atomic.Bool + done := make(chan struct{}, 2) + go func() { + defer func() { done <- struct{}{} }() + for { + packet, err := readMySQLPacket(down) + if err != nil { + return + } + if packet[3] == 0 && len(packet) > 4 && packet[4] == 3 && strings.EqualFold(strings.TrimSpace(string(packet[5:])), "COMMIT") { + committing.Store(true) + } + if writeMySQLPacket(up, packet) != nil { + return + } + } + }() + go func() { + defer func() { done <- struct{}{} }() + for { + packet, err := readMySQLPacket(up) + if err != nil { + return + } + if committing.Load() && packet[3] == 1 && len(packet) > 4 && packet[4] == 0 && g.armed.CompareAndSwap(true, false) { + g.dropped <- struct{}{} + return + } + if writeMySQLPacket(down, packet) != nil { + return + } + } + }() + <-done + down.Close() + up.Close() + <-done + }() + } + }() + t.Cleanup(func() { + listener.Close() + <-g.accepted + g.mu.Lock() + g.closing = true + for c := range g.connections { + c.Close() + } + g.mu.Unlock() + g.wg.Wait() + }) + return g +} +func readMySQLPacket(r io.Reader) ([]byte, error) { + var header [4]byte + if _, err := io.ReadFull(r, header[:]); err != nil { + return nil, err + } + size := int(binary.LittleEndian.Uint32(header[:]) & 0x00ffffff) + packet := make([]byte, 4+size) + copy(packet, header[:]) + _, err := io.ReadFull(r, packet[4:]) + return packet, err +} +func writeMySQLPacket(w io.Writer, b []byte) error { + for len(b) > 0 { + n, err := w.Write(b) + if err != nil { + return err + } + if n == 0 { + return io.ErrShortWrite + } + b = b[n:] + } + return nil +} + +func TestIntegrationReceiverCommitResponseLoss(t *testing.T) { + direct, dsn := integrationDB(t) + cfg, err := mysql.ParseDSN(dsn) + if err != nil || cfg.TLSConfig != "" { + t.Fatal("commit fault requires isolated plaintext MySQL") + } + gateway := newCommitLossGateway(t, cfg.Addr) + cfg.Addr = gateway.listener.Addr().String() + ctx, stop := context.WithTimeout(context.Background(), 15*time.Second) + defer stop() + proxied, err := Open(ctx, cfg.FormatDSN()) + if err != nil { + t.Fatal(err) + } + defer proxied.Close() + s := testStore(t, proxied) + event := testEvent(t, s.queue) + gateway.armed.Store(true) + if _, err = s.Apply(ctx, event); !errors.Is(err, ErrCommitUnconfirmed) { + t.Fatal("lost COMMIT response acknowledged", err) + } + select { + case <-gateway.dropped: + case <-ctx.Done(): + t.Fatal("real COMMIT success was not intercepted") + } + restarted, err := New(direct, s.consumer, s.queue) + if err != nil { + t.Fatal(err) + } + view, err := restarted.View(ctx, event.AggregateID) + if err != nil || view.AppliedEvents != 1 || inboxCount(t, ctx, restarted) != 1 { + t.Fatal("expected actual commit before response loss", err) + } + result, err := restarted.Apply(ctx, event) + if err != nil || !result.Duplicate || !result.Applied { + t.Fatal("unknown commit not reconciled from durable identity", err) + } + after, err := restarted.View(ctx, event.AggregateID) + if err != nil || after.AppliedEvents != 1 { + t.Fatal("unknown commit repeated business effect", err) + } + t.Log("proxy observed actual MySQL COMMIT OK and discarded it; caller returned unconfirmed, fresh store deduplicated the already committed event") +} diff --git a/examples/result-inbox/internal/inbox/consumer_process_integration_test.go b/examples/result-inbox/internal/inbox/consumer_process_integration_test.go new file mode 100644 index 0000000..6980099 --- /dev/null +++ b/examples/result-inbox/internal/inbox/consumer_process_integration_test.go @@ -0,0 +1,298 @@ +package inbox + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "strings" + "testing" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +type receiverMarker struct { + Phase string `json:"phase"` + EventID string `json:"event_id"` + Duplicate bool `json:"duplicate"` +} + +func receiverClient(t *testing.T) *orbit.Client { + t.Helper() + base, path := os.Getenv("ORBIT_TEST_RECEIVER_URL"), os.Getenv("ORBIT_TEST_RECEIVER_CREDENTIALS_FILE") + if base == "" || path == "" { + t.Skip("requires isolated real Orbit process and receiver credential") + } + if !strings.HasPrefix(base, "http://127.0.0.1:") { + t.Fatal("receiver process integration requires loopback Orbit") + } + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal("cannot read test credential") + } + var credentials struct { + Token string `json:"token"` + } + if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" { + t.Fatal("invalid test credential") + } + c, err := orbit.NewClient(base, orbit.Options{Token: credentials.Token}) + if err != nil { + t.Fatal(err) + } + return c +} + +// 此入口只存在于测试二进制。处理器使用正式Store和Consumer，故障屏障位于Apply提交后、返回nil前。 +func TestReceiverProcessHelper(t *testing.T) { + mode := os.Getenv("ORBIT_TEST_RECEIVER_CHILD") + if mode == "" { + t.Skip("subprocess helper") + } + if mode != "hold_after_commit" && mode != "confirm_duplicate" { + t.Fatal("invalid receiver child mode") + } + db, _ := integrationDB(t) + api := receiverClient(t) + s, err := New(db, os.Getenv("ORBIT_TEST_RECEIVER_NAME"), os.Getenv("ORBIT_TEST_RECEIVER_QUEUE")) + if err != nil { + t.Fatal(err) + } + ctx, stop := context.WithTimeout(context.Background(), 45*time.Second) + defer stop() + consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: os.Getenv("ORBIT_TEST_RECEIVER_SUBSCRIPTION"), Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, event orbit.ResultEvent) error { + result, err := s.Apply(work, event) + if err != nil { + return err + } + marker, _ := json.Marshal(receiverMarker{Phase: "committed", EventID: event.ID, Duplicate: result.Duplicate}) + fmt.Println("ORBIT_RECEIVER_TEST " + string(marker)) + if mode == "hold_after_commit" { + <-work.Done() + return work.Err() + } + if !result.Duplicate { + return errors.New("restarted receiver did not find committed inbox") + } + 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 { + select { + case <-ctx.Done(): + cancel() + <-done + t.Fatal("receiver helper timed out") + case err := <-done: + if err != nil { + t.Fatal(err) + } + return + case event, ok := <-consumer.Events(): + if !ok { + if err := <-done; err != nil { + t.Fatal(err) + } + return + } + if event.Type == "acknowledged" { + marker, _ := json.Marshal(receiverMarker{Phase: "acknowledged", EventID: event.EventID}) + fmt.Println("ORBIT_RECEIVER_TEST " + string(marker)) + cancel() + if err := <-done; err != nil { + t.Fatal(err) + } + return + } + } + } +} + +type receiverChild struct { + cmd *exec.Cmd + done chan struct{} + waitErr error + markers chan receiverMarker +} + +func receiverEnv(values map[string]string) []string { + out := []string{} + for _, entry := range os.Environ() { + key, _, _ := strings.Cut(entry, "=") + if _, found := values[key]; !found { + out = append(out, entry) + } + } + for key, value := range values { + out = append(out, key+"="+value) + } + return out +} +func startReceiverChild(t *testing.T, s *Store, sub, mode string) *receiverChild { + t.Helper() + binary, err := os.Executable() + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(binary, "-test.run=^TestReceiverProcessHelper$") + cmd.Env = receiverEnv(map[string]string{"ORBIT_TEST_RECEIVER_CHILD": mode, "ORBIT_TEST_RECEIVER_NAME": s.consumer, "ORBIT_TEST_RECEIVER_QUEUE": s.queue, "ORBIT_TEST_RECEIVER_SUBSCRIPTION": sub}) + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + cmd.Stderr = io.Discard + p := &receiverChild{cmd: cmd, done: make(chan struct{}), markers: make(chan receiverMarker, 8)} + if err = cmd.Start(); err != nil { + t.Fatal(err) + } + readDone := make(chan struct{}) + go func() { + defer close(readDone) + scanner := bufio.NewScanner(stdout) + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "ORBIT_RECEIVER_TEST ") { + continue + } + var marker receiverMarker + if json.Unmarshal([]byte(strings.TrimPrefix(line, "ORBIT_RECEIVER_TEST ")), &marker) == nil { + select { + case p.markers <- marker: + default: + } + } + } + }() + go func() { <-readDone; p.waitErr = cmd.Wait(); close(p.done) }() + t.Cleanup(func() { + select { + case <-p.done: + return + default: + } + _ = cmd.Process.Kill() + select { + case <-p.done: + case <-time.After(5 * time.Second): + t.Error("owned receiver child failed to exit") + } + }) + return p +} +func (p *receiverChild) marker(t *testing.T, ctx context.Context, phase string) receiverMarker { + t.Helper() + for { + select { + case marker := <-p.markers: + if marker.Phase == phase { + return marker + } + case <-p.done: + select { + case marker := <-p.markers: + if marker.Phase == phase { + return marker + } + default: + } + t.Fatal("receiver child exited before expected boundary", p.waitErr) + case <-ctx.Done(): + t.Fatal("receiver child did not reach " + phase) + } + } +} +func (p *receiverChild) kill(t *testing.T) { + t.Helper() + select { + case <-p.done: + t.Fatal("receiver exited before fault injection") + default: + } + if err := p.cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + select { + case <-p.done: + var exit *exec.ExitError + if !errors.As(p.waitErr, &exit) || exit.ExitCode() != -1 { + t.Fatal("receiver did not terminate by signal") + } + case <-time.After(5 * time.Second): + t.Fatal("receiver SIGKILL was not confirmed") + } +} + +func TestIntegrationReceiverProcessCommitBeforeAck(t *testing.T) { + db, _ := integrationDB(t) + api := receiverClient(t) + ctx, stop := context.WithTimeout(context.Background(), 45*time.Second) + defer stop() + id := newID(t) + queue, err := api.CreateQueue(ctx, "queue-"+id, orbit.CreateQueueRequest{Name: "inbox-process-" + id, MaxConcurrency: 1, RatePerSecond: 10}) + if err != nil { + t.Fatal(err) + } + s, err := New(db, "restart-"+id, queue.Queue.ID) + if err != nil { + t.Fatal(err) + } + created, err := api.CreateSubscription(ctx, "sub-"+id, orbit.CreateSubscriptionRequest{Name: "inbox-process-" + id, QueueID: queue.Queue.ID, Types: []string{"task.cancelled"}, MaxPending: 1, MaxAttempts: 3}) + if err != nil { + t.Fatal(err) + } + task, err := api.SubmitTask(ctx, "task-"+id, orbit.SubmitTaskRequest{QueueID: queue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}) + if err != nil { + t.Fatal(err) + } + if _, err = api.CancelTask(ctx, task.Task.ID, "cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "receiver process fault"}); err != nil { + t.Fatal(err) + } + first := startReceiverChild(t, s, created.Subscription.ID, "hold_after_commit") + committed := first.marker(t, ctx, "committed") + if committed.Duplicate { + t.Fatal("first receiver unexpectedly deduplicated") + } + view, err := s.View(ctx, task.Task.ID) + if err != nil || view.AppliedEvents != 1 || inboxCount(t, ctx, s) != 1 { + t.Fatal("receiver barrier occurred before durable commit", err) + } + before, err := api.ListEventDeliveries(ctx, created.Subscription.ID, orbit.DeliveryPageRequest{}) + if err != nil || len(before.Items) != 1 || before.Items[0].State != orbit.DeliveryLeased || before.Items[0].AcknowledgedAt != nil { + t.Fatal("fault boundary is not before ACK", err) + } + first.kill(t) + second := startReceiverChild(t, s, created.Subscription.ID, "confirm_duplicate") + duplicate := second.marker(t, ctx, "committed") + ack := second.marker(t, ctx, "acknowledged") + if !duplicate.Duplicate || duplicate.EventID != committed.EventID || ack.EventID != committed.EventID { + t.Fatal("new process did not reconcile original event") + } + select { + case <-second.done: + if second.waitErr != nil { + t.Fatal("replacement receiver failed", second.waitErr) + } + case <-ctx.Done(): + t.Fatal("replacement receiver did not exit cleanly") + } + after, err := s.View(ctx, task.Task.ID) + if err != nil || after.AppliedEvents != 1 || inboxCount(t, ctx, s) != 1 { + t.Fatal("business effect repeated after consumer restart", err) + } + confirmed, err := api.ListEventDeliveries(ctx, created.Subscription.ID, orbit.DeliveryPageRequest{State: orbit.DeliveryAcked}) + if err != nil || len(confirmed.Items) != 1 || confirmed.Items[0].Attempts != 2 || confirmed.Items[0].ID != before.Items[0].ID { + t.Fatal("delivery recovery or acknowledgement history wrong", err) + } + t.Log("business commit verified before ACK; consumer SIGKILL confirmed, new process used durable Inbox after real lease expiry; one business effect and two delivery attempts") +} diff --git a/examples/result-inbox/internal/inbox/model.go b/examples/result-inbox/internal/inbox/model.go new file mode 100644 index 0000000..2f708cc --- /dev/null +++ b/examples/result-inbox/internal/inbox/model.go @@ -0,0 +1,28 @@ +package inbox + +import "time" + +// Record 对应receiver_inbox，与业务视图同事务提交；存在记录才表示该事件处理已提交。 +type Record struct { + ConsumerName string // ConsumerName 固定用途名；进程重启及多实例必须复用。 + QueueID string // QueueID 固定来源队列，不能从连接会话推导。 + EventID string // EventID 稳定事件身份，不使用交付ID或代次。 + EventDigest []byte // EventDigest 32字节SHA256，防止同事件身份被换成不同内容。 + EventJSON []byte // EventJSON 规范化冻结事件，含业务数据、不含租约秘密。 + Applied bool // Applied 该事件是否推进业务视图；旧版本为false但仍可靠保存。 + ReceivedAt time.Time // ReceivedAt 数据库UTC受理时间，微秒精度。 +} + +// TaskView 对应receiver_task_views；这是接收方业务数据，不是Orbit任务权威状态。 +type TaskView struct { + ConsumerName string // ConsumerName 与Inbox相同的固定用途。 + QueueID string // QueueID 与Inbox相同的绑定队列。 + TaskID string // TaskID 原任务标识，业务视图主键的一部分。 + LastRevision uint64 // LastRevision 已应用的最大任务版本；0只允许在未提交事务的占位行中。 + LastEventID *string // LastEventID 最后推进视图的事件；仅占位状态可为空。 + LastState string // LastState 冻结结果状态；NEEDS_REVIEW不能作为成功或最终失败。 + ResultJSON []byte // ResultJSON 当前结果，nil表示来源未提供；数值保持精确。 + AppliedEvents uint64 // AppliedEvents 实际推进次数，重复及迟到旧版本不增加。 + CreatedAt time.Time // CreatedAt 视图首次创建的UTC时间，微秒精度。 + UpdatedAt time.Time // UpdatedAt 最后推进的UTC时间，微秒精度。 +} diff --git a/examples/result-inbox/internal/inbox/schema.sql b/examples/result-inbox/internal/inbox/schema.sql new file mode 100644 index 0000000..a2ca50f --- /dev/null +++ b/examples/result-inbox/internal/inbox/schema.sql @@ -0,0 +1,32 @@ +-- 接收方独立数据库的初始示例表；不属于Orbit平台迁移，不创建外键或级联删除。 +CREATE TABLE receiver_inbox ( + consumer_name VARCHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '固定消费者用途；同一用途重启或多实例必须复用', + queue_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '接收方绑定的队列标识；隔离不同来源范围', + event_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '原事件标识；跨连接及交付代次保持不变', + event_digest BINARY(32) NOT NULL COMMENT '规范化冻结事件JSON的SHA256；同身份不同内容拒绝确认', + event_json MEDIUMTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '冻结事件及业务结果；包含业务数据须限制访问，不含租约证明', + applied BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否推进业务视图；迟到的旧版本只保留Inbox历史', + received_at DATETIME(6) NOT NULL COMMENT '本次业务事务受理事件的UTC时间，微秒精度', + PRIMARY KEY (consumer_name,queue_id,event_id), + CONSTRAINT ck_receiver_event_json CHECK (JSON_VALID(event_json)), + CONSTRAINT ck_receiver_applied CHECK (applied IN (0,1)) +) ENGINE=InnoDB COMMENT='事务Inbox；去重记录与业务视图一起提交，禁止独立预写成功标志'; + +CREATE TABLE receiver_task_views ( + consumer_name VARCHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '固定消费者用途；与Inbox使用相同范围', + queue_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '固定队列来源；由处理器事务前验证归属', + task_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '原任务标识；每个任务建立一份版本化业务视图', + last_revision BIGINT UNSIGNED NOT NULL COMMENT '已应用最大任务版本；0仅为同事务中尚未更新的占位状态', + last_event_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NULL COMMENT '推进当前视图的事件；占位行为空，提交后的有效视图非空', + last_state VARCHAR(24) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '冻结任务结果状态；UNINITIALIZED只用于同事务内占位', + result_json MEDIUMTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '当前结果JSON；来源未提供结果时为空，保留精确数字', + applied_events BIGINT UNSIGNED NOT NULL COMMENT '实际推进视图的事件次数；重复和旧版本不增加', + created_at DATETIME(6) NOT NULL COMMENT '视图首次建立的UTC时间，微秒精度', + updated_at DATETIME(6) NOT NULL COMMENT '视图最后推进的UTC时间，微秒精度', + PRIMARY KEY (consumer_name,queue_id,task_id), + CONSTRAINT ck_receiver_revision CHECK ( + (last_revision=0 AND last_event_id IS NULL AND last_state='UNINITIALIZED' AND applied_events=0) + OR (last_revision>0 AND last_event_id IS NOT NULL AND applied_events>0 AND last_state IN ('SUCCEEDED','FAILED','CANCELLED','NEEDS_REVIEW')) + ), + CONSTRAINT ck_receiver_result CHECK (result_json IS NULL OR JSON_VALID(result_json)) +) ENGINE=InnoDB COMMENT='业务侧任务结果视图；锁定单任务行后仅接受更高版本，防止迟到结果倒退'; diff --git a/examples/result-inbox/internal/inbox/store.go b/examples/result-inbox/internal/inbox/store.go new file mode 100644 index 0000000..cb6dbf2 --- /dev/null +++ b/examples/result-inbox/internal/inbox/store.go @@ -0,0 +1,314 @@ +// Package inbox 演示业务端同事务去重和版本化结果视图，不是通用业务恰好一次框架。 +package inbox + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + _ "embed" + "encoding/json" + "errors" + "io" + "regexp" + "strconv" + "strings" + "time" + "unicode/utf8" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "github.com/go-sql-driver/mysql" +) + +//go:embed schema.sql +var schema string + +var ( + ErrInvalidEvent = errors.New("receiver event or scope is invalid") + ErrEventConflict = errors.New("receiver event identity has conflicting content; investigate before acknowledgement") + ErrRevisionConflict = errors.New("receiver task revision has conflicting event identities") + ErrStorage = errors.New("receiver storage operation failed; event remains unacknowledged") + ErrCommitUnconfirmed = errors.New("receiver transaction commit is unconfirmed; reconcile original event on redelivery") + namePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`) + idPattern = regexp.MustCompile(`^[a-f0-9]{32}$`) +) + +type Store struct { + db *sql.DB + consumer, queue string +} +type Result struct { + Duplicate bool // Duplicate 当前事件已在相同消费者范围内提交，内容摘要也相同。 + Applied bool // Applied 原事务是否推进视图；重复命中时返回历史值，不表示再次推进。 +} + +// Open 仅连接接收方数据库；密码和原始DSN不进入错误输出，不自动创建或修改表。 +func Open(ctx context.Context, dsn string) (*sql.DB, error) { + cfg, err := mysql.ParseDSN(dsn) + if err != nil || cfg.DBName == "" { + return nil, errors.New("valid receiver MySQL DSN with explicit database is required") + } + cfg.ParseTime = true + cfg.Loc = time.UTC + cfg.MultiStatements = false + cfg.AllowAllFiles = false + cfg.Timeout = 5 * time.Second + cfg.ReadTimeout = 8 * time.Second + cfg.WriteTimeout = 8 * time.Second + if cfg.Params == nil { + cfg.Params = map[string]string{} + } + cfg.Params["time_zone"] = "'+00:00'" + connector, err := mysql.NewConnector(cfg) + if err != nil { + return nil, ErrStorage + } + db := sql.OpenDB(connector) + db.SetMaxOpenConns(8) + db.SetMaxIdleConns(4) + db.SetConnMaxLifetime(5 * time.Minute) + if err = db.PingContext(ctx); err != nil { + db.Close() + return nil, ErrStorage + } + return db, nil +} + +// Init 初始化专用空数据库。已有表会报错；升级已有业务数据库应使用接收方自己的迁移。 +func Init(ctx context.Context, db *sql.DB) error { + for _, statement := range strings.Split(schema, ";") { + if strings.TrimSpace(statement) == "" { + continue + } + if _, err := db.ExecContext(ctx, statement); err != nil { + return ErrStorage + } + } + return nil +} +func New(db *sql.DB, consumer, queue string) (*Store, error) { + if db == nil || !namePattern.MatchString(consumer) || !idPattern.MatchString(queue) { + return nil, ErrInvalidEvent + } + return &Store{db: db, consumer: consumer, queue: queue}, nil +} + +// Handle 可直接作为EventConsumer处理器；只有事务确认提交或命中相同已提交记录才返回nil。 +func (s *Store) Handle(ctx context.Context, event orbit.ResultEvent) error { + _, err := s.Apply(ctx, event) + return err +} + +// Apply 不含网络副作用。事务失败或COMMIT响应未知时不允许SDK发送ACK。 +func (s *Store) Apply(parent context.Context, event orbit.ResultEvent) (Result, error) { + ctx, stop := context.WithTimeout(parent, 5*time.Second) + defer stop() + raw, digest, err := s.document(event) + if err != nil { + return Result{}, err + } + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return Result{}, ErrStorage + } + defer tx.Rollback() + _, err = tx.ExecContext(ctx, `INSERT INTO receiver_inbox(consumer_name,queue_id,event_id,event_digest,event_json,applied,received_at) VALUES(?,?,?,?,?,FALSE,UTC_TIMESTAMP(6))`, s.consumer, s.queue, event.ID, digest[:], raw) + if err != nil { + var duplicate *mysql.MySQLError + if !errors.As(err, &duplicate) || duplicate.Number != 1062 { + return Result{}, ErrStorage + } + // 唯一键冲突已等待胜出事务完成。结束本事务，再核对已提交记录；不能忽略其他SQL错误。 + if err = tx.Rollback(); err != nil { + return Result{}, ErrStorage + } + var saved []byte + var applied bool + if err = s.db.QueryRowContext(ctx, `SELECT event_digest,applied FROM receiver_inbox WHERE consumer_name=? AND queue_id=? AND event_id=?`, s.consumer, s.queue, event.ID).Scan(&saved, &applied); err != nil { + return Result{}, ErrStorage + } + if !bytes.Equal(saved, digest[:]) { + return Result{}, ErrEventConflict + } + return Result{Duplicate: true, Applied: applied}, nil + } + // 原子建立或取得单任务行锁，不依赖客户端RowsAffected选项判断新旧。 + _, err = tx.ExecContext(ctx, `INSERT INTO receiver_task_views(consumer_name,queue_id,task_id,last_revision,last_state,applied_events,created_at,updated_at) VALUES(?,?,?,0,'UNINITIALIZED',0,UTC_TIMESTAMP(6),UTC_TIMESTAMP(6)) ON DUPLICATE KEY UPDATE task_id=receiver_task_views.task_id`, s.consumer, s.queue, event.AggregateID) + if err != nil { + return Result{}, ErrStorage + } + var revision uint64 + if err = tx.QueryRowContext(ctx, `SELECT last_revision FROM receiver_task_views WHERE consumer_name=? AND queue_id=? AND task_id=? FOR UPDATE`, s.consumer, s.queue, event.AggregateID).Scan(&revision); err != nil { + return Result{}, ErrStorage + } + if event.AggregateRevision == revision { + return Result{}, ErrRevisionConflict + } + applied := event.AggregateRevision > revision + if applied { + var result any + if len(event.Snapshot.Result) > 0 { + result = []byte(event.Snapshot.Result) + } + _, err = tx.ExecContext(ctx, `UPDATE receiver_task_views SET last_revision=?,last_event_id=?,last_state=?,result_json=?,applied_events=applied_events+1,updated_at=UTC_TIMESTAMP(6) WHERE consumer_name=? AND queue_id=? AND task_id=?`, event.AggregateRevision, event.ID, string(event.Snapshot.State), result, s.consumer, s.queue, event.AggregateID) + if err != nil { + return Result{}, ErrStorage + } + if _, err = tx.ExecContext(ctx, `UPDATE receiver_inbox SET applied=TRUE WHERE consumer_name=? AND queue_id=? AND event_id=?`, s.consumer, s.queue, event.ID); err != nil { + return Result{}, ErrStorage + } + } + if err = tx.Commit(); err != nil { + return Result{}, ErrCommitUnconfirmed + } + return Result{Applied: applied}, nil +} + +func (s *Store) document(event orbit.ResultEvent) ([]byte, [32]byte, error) { + var none [32]byte + snapshot := event.Snapshot + if len(event.Details)+len(snapshot.Result) > 512<<10 { + return nil, none, ErrInvalidEvent + } + if !idPattern.MatchString(event.ID) || event.Sequence == 0 || event.AggregateKind != "task" || !idPattern.MatchString(event.AggregateID) || snapshot.TaskID != event.AggregateID || snapshot.QueueID != s.queue || event.AggregateRevision == 0 || event.AggregateRevision != snapshot.Revision || event.OccurredAt.IsZero() || snapshot.SchemaVersion != 1 || snapshot.CurrentAttemptID != nil && !idPattern.MatchString(*snapshot.CurrentAttemptID) { + return nil, none, ErrInvalidEvent + } + states := map[string]orbit.TaskState{"task.succeeded": orbit.TaskSucceeded, "task.failed": orbit.TaskFailed, "task.cancelled": orbit.TaskCancelled, "task.needs_review": orbit.TaskNeedsReview} + if state, ok := states[event.Type]; !ok || state != snapshot.State { + return nil, none, ErrInvalidEvent + } + if len(bytes.TrimSpace(event.Details)) == 0 || bytes.TrimSpace(event.Details)[0] != '{' { + return nil, none, ErrInvalidEvent + } + event.OccurredAt = event.OccurredAt.UTC() + var buffer bytes.Buffer + encoder := json.NewEncoder(&buffer) + encoder.SetEscapeHTML(false) + if encoder.Encode(event) != nil || buffer.Len() > 512<<10 { + return nil, none, ErrInvalidEvent + } + // 对象键与空白规范化，数字使用json.Number保留原精度及十进制表达。 + canonical, err := canonicalJSON(buffer.Bytes()) + if err != nil { + return nil, none, err + } + return canonical, sha256.Sum256(canonical), nil +} +func canonicalJSON(raw []byte) ([]byte, error) { + if !utf8.Valid(raw) || !json.Valid(raw) || !pairedUnicode(raw) { + return nil, ErrInvalidEvent + } + d := json.NewDecoder(bytes.NewReader(raw)) + d.UseNumber() + var value func(int) (any, error) + value = func(depth int) (any, error) { + if depth > 32 { + return nil, ErrInvalidEvent + } + token, err := d.Token() + if err != nil { + return nil, ErrInvalidEvent + } + switch token { + case json.Delim('{'): + out := map[string]any{} + for d.More() { + key, err := d.Token() + if err != nil { + return nil, ErrInvalidEvent + } + name, ok := key.(string) + if !ok { + return nil, ErrInvalidEvent + } + if _, found := out[name]; found { + return nil, ErrInvalidEvent + } + out[name], err = value(depth + 1) + if err != nil { + return nil, err + } + } + _, err = d.Token() + return out, err + case json.Delim('['): + out := []any{} + for d.More() { + item, err := value(depth + 1) + if err != nil { + return nil, err + } + out = append(out, item) + } + _, err = d.Token() + return out, err + default: + return token, nil + } + } + doc, err := value(0) + if err != nil { + return nil, ErrInvalidEvent + } + if _, err = d.Token(); err != io.EOF { + return nil, ErrInvalidEvent + } + var out bytes.Buffer + e := json.NewEncoder(&out) + e.SetEscapeHTML(false) + if e.Encode(doc) != nil { + return nil, ErrInvalidEvent + } + return bytes.TrimSuffix(out.Bytes(), []byte{'\n'}), nil +} + +// 不将未配对UTF-16转义静默转换成替代字符，避免摘要与实际业务文本失真。 +func pairedUnicode(raw []byte) bool { + inside := false + for i := 0; i < len(raw); i++ { + if raw[i] == '"' { + inside = !inside + continue + } + if !inside || raw[i] != '\\' { + continue + } + i++ + if i >= len(raw) { + return false + } + if raw[i] != 'u' { + continue + } + if i+4 >= len(raw) { + return false + } + n, err := strconv.ParseUint(string(raw[i+1:i+5]), 16, 16) + if err != nil { + return false + } + i += 4 + if n < 0xd800 || n > 0xdfff { + continue + } + if n > 0xdbff || i+6 >= len(raw) || raw[i+1] != '\\' || raw[i+2] != 'u' { + return false + } + low, err := strconv.ParseUint(string(raw[i+3:i+7]), 16, 16) + if err != nil || low < 0xdc00 || low > 0xdfff { + return false + } + i += 6 + } + return !inside +} + +// View 返回业务侧当前视图；不会从Orbit重新拼装结果。 +func (s *Store) View(ctx context.Context, task string) (TaskView, error) { + out := TaskView{ConsumerName: s.consumer, QueueID: s.queue, TaskID: task} + err := s.db.QueryRowContext(ctx, `SELECT last_revision,last_event_id,last_state,result_json,applied_events,created_at,updated_at FROM receiver_task_views WHERE consumer_name=? AND queue_id=? AND task_id=?`, s.consumer, s.queue, task).Scan(&out.LastRevision, &out.LastEventID, &out.LastState, &out.ResultJSON, &out.AppliedEvents, &out.CreatedAt, &out.UpdatedAt) + if err != nil { + return TaskView{}, ErrStorage + } + return out, nil +} diff --git a/examples/result-inbox/internal/inbox/store_integration_test.go b/examples/result-inbox/internal/inbox/store_integration_test.go new file mode 100644 index 0000000..abedc7e --- /dev/null +++ b/examples/result-inbox/internal/inbox/store_integration_test.go @@ -0,0 +1,187 @@ +package inbox + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "os" + "strings" + "sync" + "testing" + "time" + "unicode" + + "github.com/go-sql-driver/mysql" +) + +func integrationDB(t *testing.T) (*sql.DB, string) { + t.Helper() + dsn := os.Getenv("ORBIT_TEST_RECEIVER_DSN") + if dsn == "" { + t.Skip("requires dedicated disposable receiver database") + } + cfg, err := mysql.ParseDSN(dsn) + if err != nil || cfg.DBName != "orbit_receiver_test" || !strings.HasPrefix(cfg.Addr, "127.0.0.1:") { + t.Fatal("receiver integration must use dedicated disposable loopback database") + } + ctx, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + db, err := Open(ctx, dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + return db, dsn +} +func testStore(t *testing.T, db *sql.DB) *Store { + t.Helper() + s, err := New(db, "test-"+newID(t), newID(t)) + if err != nil { + t.Fatal(err) + } + return s +} +func inboxCount(t *testing.T, ctx context.Context, s *Store) int { + t.Helper() + var count int + if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM receiver_inbox WHERE consumer_name=? AND queue_id=?`, s.consumer, s.queue).Scan(&count); err != nil { + t.Fatal(err) + } + return count +} +func TestIntegrationReceiverSchema(t *testing.T) { + db, _ := integrationDB(t) + ctx, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + var fk int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fk); err != nil || fk != 0 { + t.Fatal("receiver foreign keys", fk, err) + } + rows, err := db.QueryContext(ctx, `SELECT COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN ('receiver_inbox','receiver_task_views')`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + count := 0 + for rows.Next() { + var comment string + if err = rows.Scan(&comment); err != nil { + t.Fatal(err) + } + if !strings.ContainsFunc(comment, func(r rune) bool { return unicode.Is(unicode.Han, r) }) { + t.Fatal("receiver column missing Chinese comment") + } + count++ + } + if rows.Err() != nil || count != 17 { + t.Fatal("receiver model column count", count, rows.Err()) + } + t.Log("receiver schema has two tables, 17 Chinese column comments and zero foreign keys") +} +func TestIntegrationReceiverConcurrentDuplicateAndConflict(t *testing.T) { + db, _ := integrationDB(t) + s := testStore(t, db) + ctx, stop := context.WithTimeout(context.Background(), 15*time.Second) + defer stop() + event := testEvent(t, s.queue) + type outcome struct { + result Result + err error + } + results := make(chan outcome, 8) + start := make(chan struct{}) + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { defer wg.Done(); <-start; result, err := s.Apply(ctx, event); results <- outcome{result, err} }() + } + close(start) + wg.Wait() + close(results) + duplicates := 0 + for result := range results { + if result.err != nil { + t.Fatal(result.err) + } + if result.result.Duplicate { + duplicates++ + } + } + view, err := s.View(ctx, event.AggregateID) + if err != nil || duplicates != 7 || view.AppliedEvents != 1 || view.LastRevision != event.AggregateRevision || inboxCount(t, ctx, s) != 1 { + t.Fatal("concurrent duplicate changed business effect", duplicates, err) + } + event.Snapshot.Result = json.RawMessage(`{"tampered":true}`) + if _, err = s.Apply(ctx, event); !errors.Is(err, ErrEventConflict) { + t.Fatal("same identity different content acknowledged", err) + } + unchanged, err := s.View(ctx, event.AggregateID) + if err != nil || string(unchanged.ResultJSON) != string(view.ResultJSON) || unchanged.AppliedEvents != 1 { + t.Fatal("conflicting event changed view", err) + } +} +func TestIntegrationReceiverLateVersionsAndIsolation(t *testing.T) { + db, _ := integrationDB(t) + s := testStore(t, db) + ctx, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + event := testEvent(t, s.queue) + if _, err := s.Apply(ctx, event); err != nil { + t.Fatal(err) + } + old := event + old.ID = newID(t) + old.AggregateRevision-- + old.Snapshot.Revision-- + old.Snapshot.Result = json.RawMessage(`{"older":true}`) + result, err := s.Apply(ctx, old) + if err != nil || result.Applied || result.Duplicate { + t.Fatal("old version incorrectly applied", err) + } + view, err := s.View(ctx, event.AggregateID) + if err != nil || view.LastRevision != event.AggregateRevision || view.AppliedEvents != 1 || inboxCount(t, ctx, s) != 2 { + t.Fatal("late event reverted business state", err) + } + conflict := event + conflict.ID = newID(t) + if _, err = s.Apply(ctx, conflict); !errors.Is(err, ErrRevisionConflict) || inboxCount(t, ctx, s) != 2 { + t.Fatal("same task revision conflict persisted", err) + } + other, err := New(db, s.consumer+"-other", s.queue) + if err != nil { + t.Fatal(err) + } + if result, err = other.Apply(ctx, event); err != nil || result.Duplicate { + t.Fatal("different consumer purpose shared dedup scope", err) + } + wrongQueue := testStore(t, db) + if _, err = wrongQueue.Apply(ctx, event); !errors.Is(err, ErrInvalidEvent) { + t.Fatal("cross-queue event accepted", err) + } +} +func TestIntegrationReceiverTransactionRollback(t *testing.T) { + db, _ := integrationDB(t) + s := testStore(t, db) + ctx, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + event := testEvent(t, s.queue) + _, err := db.ExecContext(ctx, `CREATE TRIGGER receiver_test_rollback BEFORE UPDATE ON receiver_task_views FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected receiver projection rollback'`) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _, _ = db.Exec(`DROP TRIGGER IF EXISTS receiver_test_rollback`) }) + if _, err = s.Apply(ctx, event); !errors.Is(err, ErrStorage) { + t.Fatal("injected update did not fail", err) + } + var views int + if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM receiver_task_views WHERE consumer_name=? AND queue_id=?`, s.consumer, s.queue).Scan(&views); err != nil || views != 0 || inboxCount(t, ctx, s) != 0 { + t.Fatal("partial transaction persisted", err) + } + if _, err = db.ExecContext(ctx, `DROP TRIGGER receiver_test_rollback`); err != nil { + t.Fatal(err) + } + if result, err := s.Apply(ctx, event); err != nil || !result.Applied || result.Duplicate { + t.Fatal("rollback poisoned event identity", err) + } +} diff --git a/examples/result-inbox/internal/inbox/store_test.go b/examples/result-inbox/internal/inbox/store_test.go new file mode 100644 index 0000000..545a2cc --- /dev/null +++ b/examples/result-inbox/internal/inbox/store_test.go @@ -0,0 +1,59 @@ +package inbox + +import ( + "bytes" + "encoding/json" + "errors" + "testing" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func newID(t *testing.T) string { + t.Helper() + id, err := orbit.NewEventMessageID() + if err != nil { + t.Fatal(err) + } + return id +} +func testEvent(t *testing.T, queue string) orbit.ResultEvent { + t.Helper() + task := newID(t) + return orbit.ResultEvent{ID: newID(t), Sequence: 7, Type: "task.succeeded", AggregateKind: "task", AggregateID: task, AggregateRevision: 9007199254740993, OccurredAt: time.Now().UTC().Truncate(time.Microsecond), Details: json.RawMessage(`{"origin":"worker"}`), Snapshot: orbit.TaskResultSnapshot{SchemaVersion: 1, TaskID: task, QueueID: queue, State: orbit.TaskSucceeded, Revision: 9007199254740993, Result: json.RawMessage(`{"n":9007199254740993,"text":"<&>"}`)}} +} +func TestFrozenEventIdentity(t *testing.T) { + s := &Store{consumer: "result-view", queue: newID(t)} + e := testEvent(t, s.queue) + raw, a, err := s.document(e) + if err != nil || !bytes.Contains(raw, []byte("9007199254740993")) || !bytes.Contains(raw, []byte("<&>")) { + t.Fatal("precision changed", err) + } + e.Snapshot.Result = json.RawMessage(`{ "text": "<&>", "n": 9007199254740993 }`) + _, b, err := s.document(e) + if err != nil || a != b { + t.Fatal("equivalent formatting changed identity", err) + } + e.OccurredAt = e.OccurredAt.In(time.FixedZone("offset", 8*3600)) + _, b, err = s.document(e) + if err != nil || a != b { + t.Fatal("timezone representation changed event identity", err) + } + e.Snapshot.Result = json.RawMessage(`{"n":9007199254740994,"text":"<&>"}`) + _, b, err = s.document(e) + if err != nil || a == b { + t.Fatal("changed result not detected", err) + } + e.Snapshot.QueueID = newID(t) + if _, _, err = s.document(e); !errors.Is(err, ErrInvalidEvent) { + t.Fatal("wrong receiver scope accepted", err) + } +} +func TestCanonicalRejectsDuplicateAndInvalidJSON(t *testing.T) { + for _, raw := range []string{`{"a":1,"a":2}`, `{"a":null,"a":2}`, `{"a":1} {}`, `{"a":`, `{"a":"\ud800"}`, `{"a":"\udfff"}`, "\xff"} { + if _, err := canonicalJSON([]byte(raw)); !errors.Is(err, ErrInvalidEvent) { + t.Fatal("invalid document accepted") + } + } +} diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index e00b396..9e490d7 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -66,6 +66,13 @@ 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/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) +# 接收方业务数据独立建库；数据库驱动仅属于此示例模块，不加入核心SDK。 +"$TASK_DOCKER" exec "$TASK_TEST_NAME" sh -c 'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot --protocol=tcp -e "CREATE DATABASE orbit_receiver_test CHARACTER SET utf8mb4 COLLATE utf8mb4_bin"' +export ORBIT_TEST_RECEIVER_DSN="root:$TASK_DB_PASSWORD@tcp(127.0.0.1:$TASK_DB_PORT)/orbit_receiver_test" +(cd examples/result-inbox && GOWORK=off ORBIT_RECEIVER_DSN="$ORBIT_TEST_RECEIVER_DSN" "$TASK_GO" run ./cmd/receiver init) +(cd examples/result-inbox && GOWORK=off ORBIT_TEST_RECEIVER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_RECEIVER_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" test -race -count=1 -run Integration -v ./...) +(cd examples/result-inbox && GOWORK=off ORBIT_RECEIVER_DSN="$ORBIT_TEST_RECEIVER_DSN" ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/worker.credentials.json" "$TASK_GO" run ./cmd/receiver demo) + # 应用配额场景使用独立应用与两个命名空间，普通 Worker 凭据不扩权。 ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap --application sdk-app-resources --namespace first --environment test --queue first --resource-access --output "$TASK_TEST_DIR/app.credentials.json"