diff --git a/AGENTS.md b/AGENTS.md index 7094503..2c1825f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,7 +14,7 @@ - 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 - 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 -- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，任务查询已支持业务键、创建时间及直接来源（普通/人工重试/计划固定版本），P3仍需补齐事务Outbox接入；不能因订阅完成而跳过原功能范围。 +- 可靠订阅已完成核心、HTTP/WSS、管理及消费SDK、独立业务Inbox示例、节点及消费者进程故障回归。批量受理已提供逐项幂等与未知结果分类。独立任务核查/人工重试已实现，任务查询已支持业务键、创建时间及直接来源（普通/人工重试/计划固定版本），业务事务Outbox独立示例已实现并通过全量链路回归；P3仍需原功能范围与验收证据逐项核对，不能因局部功能完成而跳过原范围。 - 批量受理最多32项、512 KiB；每项复用单项身份与事务，未知结果必须原键核对。任务提交按认证→应用额度→提交幂等身份→队列持锁，禁止恢复为先抢提交唯一键后等待额度锁；原回执仍须先于配额上限检查返回。 @@ -24,3 +24,5 @@ - 任务列表队列必填，business_key精确匹配（包含尾部空格），created_from包含/created_before不包含。v2游标绑定所有筛选，不能因网络或游标错误删筛选降级；旧游标仅兼容旧查询。四种固定索引覆盖状态/业务键组合，禁止全量取回后内存过滤。业务键仅Go空字符串映射NULL。 - 任务来源通过独立只读路由查询，tasks:read仅授予当前任务的有限来源引用；计划定义仍需schedules:read。显式核对命名空间、队列、计划、触发与固定版本，悬空/错误/未实现所有者不得默认为DIRECT；不依赖外键，也不向派发路径加入来源查询。 + +- 业务事务Outbox示例使用独立module和业务数据库，核心SDK不得引入SQL驱动。业务记录/待发命令同事务；原请求、操作ID、目标及提交键冻结，HTTP在事务外，确认按最新租约代次与业务关联同事务。24小时/100次预算内原键核对，禁止换键、重置预算或把ACCEPTED解释成业务成功；REVIEW保留待核查数据。 diff --git a/Makefile b/Makefile index 5949659..34bb203 100644 --- a/Makefile +++ b/Makefile @@ -5,22 +5,26 @@ fmt: $(GO) fmt ./... cd sdk/go && $(GO) fmt ./... cd examples/result-inbox && GOWORK=off $(GO) fmt ./... + cd examples/task-outbox && GOWORK=off $(GO) fmt ./... test: $(GO) test -race ./... cd sdk/go && $(GO) test -race ./... cd examples/result-inbox && GOWORK=off $(GO) test -race ./... + cd examples/task-outbox && GOWORK=off $(GO) test -race ./... vet: $(GO) vet ./... cd sdk/go && $(GO) vet ./... cd examples/result-inbox && GOWORK=off $(GO) vet ./... + cd examples/task-outbox && 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 + cd examples/task-outbox && GOWORK=off $(GO) build -o ../../bin/orbit-outbox-example ./cmd/sender integration: ./scripts/test-mysql.sh diff --git a/README.md b/README.md index 82dc3c6..b9db9cc 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) 和 [独立 MySQL Inbox 示例](docs/examples/result-inbox.md) 已实现；[批量任务受理](docs/api/batch-tasks.md) 已提供逐项幂等、失败分类与 SDK 示例；[人工核查与重试](docs/api/task-recovery.md) 已提供带证据的管理接口、关联新任务、审计查询及运行示例；[业务键与时间范围查询](docs/api/task-queries.md) 已提供过滤、分页与SDK场景；[任务来源查询](docs/api/task-source.md) 已提供人工重试父任务、计划触发记录与固定版本；剩余 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) 已实现；[批量任务受理](docs/api/batch-tasks.md) 已提供逐项幂等、失败分类与 SDK 示例；[人工核查与重试](docs/api/task-recovery.md) 已提供带证据的管理接口、关联新任务、审计查询及运行示例；[业务键与时间范围查询](docs/api/task-queries.md) 已提供过滤、分页与SDK场景；[任务来源查询](docs/api/task-source.md) 已提供人工重试父任务、计划触发记录与固定版本；[业务事务Outbox](docs/examples/task-outbox.md) 已提供独立业务库、稳定提交身份、发送恢复与CLI；剩余 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 83d256c..376ddec 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 具体范围另见流程设计。基础、任务执行、调度、资源控制和可靠订阅已实现；批量受理、独立任务核查/人工重试、业务键/创建时间查询及直接来源追踪已实现，事务Outbox接入等继续开发，实现状态见持续开发计划。 +> 版本 0.2 · 2026-09-17。Pulse 与 Weave 首版功能范围已确认；标注“建议”的细节待评审。Weave 具体范围另见流程设计。基础、任务执行、调度、资源控制和可靠订阅已实现；批量受理、独立任务核查/人工重试、业务键/创建时间查询及直接来源追踪已实现，业务事务Outbox已提供完整接入示例；整体阶段验收和后续能力见持续开发计划。 ## 1. 目标与责任边界 diff --git a/docs/04-reliability-and-data.md b/docs/04-reliability-and-data.md index 9a19f1a..5dc2b7e 100644 --- a/docs/04-reliability-and-data.md +++ b/docs/04-reliability-and-data.md @@ -134,6 +134,8 @@ stateDiagram-v2 Orbit 数据库事务不能自动包含 接入方的业务事务。SDK 的事务接入组件只辅助接入方在自己的数据库写入待发请求，核心 HTTP / WS 客户端不绑定业务 ORM。[事务 Outbox 参考](https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html) +当前[任务Outbox示例](examples/task-outbox.md)已提供独立Go module、业务写入与待发请求同事务、有限租约/退避/预算、原身份核对及CLI。发送前核对命名空间和队列，HTTP在事务外，受理确认与业务任务关联同事务保存；流程启动/信号命令类型随Weave实现。已实现的是业务适配参考，核心SDK不引入MySQL驱动。 + ### 6.2 平台到结果接收方 结果事件与终态同事务保存，投递状态独立跟踪。接收方可靠处理后确认；确认丢失则重复交付，由接收方去重。 diff --git a/docs/07-data-model.md b/docs/07-data-model.md index fcdb787..821c6f3 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -178,3 +178,7 @@ HTTP/SDK创建时间范围使用包含下界、不包含上界；v2游标以固 ### 任务直接来源视图 [来源查询](api/task-source.md)复用Task.OwnerKind/OwnerID/RetryOf、Schedule、ScheduleVersion与ScheduleOccurrence，沿已有(namespace_id,task_id)唯一索引及计划版本键定位，不增加表、持久化字段、索引或迁移。只读响应模型的每个字段也有中文注释。任务、计划、触发版本与队列归属在认证事务中显式验证，缺失关联不被默认为普通任务；流程来源随Weave实现。 + +### 业务事务Outbox示例模型 + +[独立task-outbox](examples/task-outbox.md)包含sender_outbox和sender_jobs，2个业务实体、22个Go/SQL中文注释字段，无外键；不计入平台35个实体和289字段。Outbox保存固定命名空间/队列、业务操作、原提交键/请求/摘要、状态/次数/代次/到期时间及原受理回执，业务行同事务创建和关联。命名空间与队列范围内按wake_at索引领取，终结状态wake_at为空；发送次数与总发送期限有界。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index bf67cdd..efd89fc 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/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景、COMMIT响应丢失及消费者进程故障回归已完成；批量受理、SDK、接口与部分补交示例已完成全量回归；独立任务核查/人工重试、审计查询、两次真实HTTP回执丢失及完整示例已通过全量回归；业务键/创建时间筛选、SDK与示例已通过全量回归；任务直接来源（普通/人工重试/计划固定版本）及SDK示例已通过全量回归；事务Outbox接入待完成 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅、批量受理、独立任务核查/人工重试、查询筛选与事务Outbox接入 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅核心、HTTP/WSS、管理 SDK 和管理示例已实现并回归；消费 SDK、协议示例及两项节点强杀回归已完成；业务 Inbox 场景、COMMIT响应丢失及消费者进程故障回归已完成；批量受理、SDK、接口与部分补交示例已完成全量回归；独立任务核查/人工重试、审计查询、两次真实HTTP回执丢失及完整示例已通过全量回归；业务键/创建时间筛选、SDK与示例已通过全量回归；任务直接来源（普通/人工重试/计划固定版本）及SDK示例已通过全量回归；事务Outbox独立业务示例、CLI及故障恢复已实现并通过全量链路回归，待完成P3整体范围验收 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -45,13 +45,13 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 独立任务人工核查与重试 | 已实现[管理API及SDK](api/task-recovery.md)、专门权限、显式安全声明、证据/版本/审计和关联新任务；[完整场景](examples/recovery.md)及[回归报告](reports/phase-3-task-recovery.md)同步交付 | | 任务查询筛选 | 已实现[业务键及创建时间查询](api/task-queries.md)，v2游标绑定全部筛选、旧游标受限兼容、四种索引路径与[完整场景](examples/task-query.md)；[回归报告](reports/phase-3-task-queries.md)同步记录 | | 任务详情来源追踪 | 已实现[独立来源接口](api/task-source.md)、SDK和普通/计划/人工恢复场景，返回计划及触发固定版本；同命名空间归属、当前权限和来源缺失拒绝已通过[全量回归](reports/phase-3-task-source.md)，流程来源随Weave接入 | -| 业务到平台的事务Outbox接入 | [事务Outbox设计](04-reliability-and-data.md#61-业务系统到平台) 尚缺完整接入示例；需业务变更与待发请求同事务、稳定提交身份、响应不明核对和发送恢复，保持核心SDK无ORM/数据库依赖 | +| 业务到平台的事务Outbox接入 | 已实现[独立业务模块及完整接入](examples/task-outbox.md)，覆盖业务变更/待发请求同事务、有限发送恢复、CLI及真实HTTP/MySQL/进程故障，见[报告](reports/phase-3-task-outbox.md)；核心SDK保持无ORM/数据库依赖 | 凭据安全重绑、历史清理、全量监控/容量与MySQL故障切换仍按P6执行；Weave的通用执行门控/内部事件接续随P4–P5落实，不将外部结果订阅当作内部流程引擎。 -### P3 事务Outbox接入的剩余验收 +### P3 事务Outbox接入的验收范围 -以下为下一项实现与测试要求，目前尚未交付： +以下内容已由独立业务示例实现并进入回归，实际证据见[Outbox报告](reports/phase-3-task-outbox.md)。P3仍需逐项对照原功能范围完成整体验收，再推进Weave；不能仅依据本表无待开发行即宣布阶段完成： 1. 独立业务示例module，以业务记录和待发请求同一个MySQL事务提交；核心SDK保持无ORM/数据库依赖，业务表示例也禁止外键，字段提供中文注释。 2. 同时冻结目标命名空间/队列、原幂等键、完整请求及内容摘要。相同业务操作重报不新建待发项，内容不同必须冲突；凭据原文不能进入Outbox或日志。 @@ -77,6 +77,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 优先级与应用隔离 | 近同时到期按优先级 → 旧任务等待补偿 → 应用/队列轮转 | P3，已实现完整示例 | | 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，核心、HTTP/WSS、管理及消费 SDK、独立Inbox示例与真实进程故障回归已完成 | | 批量文件作业受理 | 逐项身份 → 部分结果 → 同键核对 → 释放额度 → 只补交失败项 | P3，已实现完整示例 | +| 业务事务Outbox | 业务记录/请求同事务 → 原键发送 → 受理/业务关联同事务 → 故障恢复 | P3，已实现独立模块与完整示例 | | 业务操作定位 | 精确业务键 → 创建区间 → 游标分页 → 查询详情 | P3，已实现完整示例 | | 独立任务人工恢复 | 未知结果 → 证据核查 → 保留原任务 → 关联新任务 → 审计查询 | P3，已实现完整示例 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | @@ -114,6 +115,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 - [P3 人工核查与重试回归](reports/phase-3-task-recovery.md) - [P3 业务键与时间查询回归](reports/phase-3-task-queries.md) - [P3 任务直接来源回归](reports/phase-3-task-source.md) +- [P3 业务事务Outbox回归](reports/phase-3-task-outbox.md) - [接口手册](api/README.md) - [SDK 接入](sdk/README.md) - [通用场景示例](examples/README.md) diff --git a/docs/api/tasks.md b/docs/api/tasks.md index 7f29c60..79f5091 100644 --- a/docs/api/tasks.md +++ b/docs/api/tasks.md @@ -106,6 +106,8 @@ curl -i "$ORBIT_URL/v1/tasks" \ SDK：`Client.SubmitTask(ctx, key, request)`。 +需要将业务变更与提交可靠衔接时，使用[事务Outbox接入示例](../examples/task-outbox.md)：先在业务事务里冻结原操作和请求，再在事务外调用此接口；新增示例不改变本HTTP契约。 + ## GET /v1/tasks/{task_id} 权限：tasks:read 和任务所在队列范围。200 的 data 直接为上方 task 对象，返回当前状态和完整参数/已保存结果。不存在、跨命名空间或不在该凭据队列范围时返回 404。 diff --git a/docs/examples/README.md b/docs/examples/README.md index debe034..1cde70f 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -2,6 +2,14 @@ 示例只使用通用的作业、对象、文件、资源与审核信号，不包含接入方项目名称或行业规则。 +## 已可运行：业务事务与任务提交Outbox + +- 独立模块：[task-outbox](../../examples/task-outbox/README.md)，[完整前置、命令、Go方法及故障处理](task-outbox.md)。 +- 操作：业务记录和待发请求同事务 → 持久化领取 → 原幂等键HTTP提交 → 原回执与业务关联同事务确认。 +- 预期：local_duplicate=true、state=ACCEPTED、sends=1；ACCEPTED仅表示平台受理，演示随后取消任务并保留历史。 +- 覆盖：8个并发调用/20项领取、业务回滚、旧代次拒绝、HTTP回执丢失、三处MySQL COMMIT回执丢失及发送进程SIGKILL恢复。 +- 边界：业务侧两张表均无外键，22个中文注释字段；固定发送预算，REVIEW保留原记录供业务管理核查，核心SDK不引入数据库依赖。 + ## 已可运行：健康检查 - 源码：[SDK 健康检查](../../sdk/go/examples/health/main.go)。 diff --git a/docs/examples/task-outbox.md b/docs/examples/task-outbox.md new file mode 100644 index 0000000..eb2d26b --- /dev/null +++ b/docs/examples/task-outbox.md @@ -0,0 +1,174 @@ +# 业务事务与任务提交 Outbox + +完整项目：[task-outbox](../../examples/task-outbox/README.md)。它演示接入方保存对象处理请求的业务记录，同时保存向Orbit提交任务的待发命令。平台受理后，再在业务库里同时确认Outbox与任务关联。核心SDK继续独立，不依赖MySQL或ORM。 + +## 1. 数据与状态 + +[中文模型](../../examples/task-outbox/internal/outbox/model.go)和[SQL](../../examples/task-outbox/internal/outbox/schema.sql)包含2张业务表、22个字段，全部有中文说明，外键为零，不进入Orbit平台迁移。 + +| 表 | 职责与关键约束 | +| --- | --- | +| sender_jobs | 业务操作、对象引用及已确认任务ID；主键(namespace_id,operation_id) | +| sender_outbox | 固定命名空间/队列、原提交键、原请求、SHA256、发送状态、次数/代次、截止及回执；主键与业务操作相同 | + +operation_id由调用方先持久化，再作为参数传入，为32位小写十六进制ID。source_ref要求有效UTF-8、非全空白、最多512个Unicode字符且不含控制字符。提交键固定为`outbox-task-`，同命名空间内唯一。业务源引用、目标队列或正文不同的重复操作返回ErrConflict；相同对象可以有不同业务操作，幂等范围由业务操作语义决定。 + +```mermaid +flowchart LR + B[业务记录与待发命令同事务] --> P[PENDING] + P -->|领取并提交租约| S[SENDING] + S -->|平台受理回执| A[ACCEPTED与业务关联同事务] + S -->|未知响应或背压| P + S -->|租约过期后原身份接管| S + S -->|内容、权限或预算需核查| R[REVIEW] + P -->|发送截止或次数耗尽| R +``` + +| 本地状态 | 含义 | +| --- | --- | +| PENDING | 待下一次有限发送；可能曾有未确认的平台受理 | +| SENDING | 已领取30秒租约；HTTP在事务外，进程退出后可按数据库时间接管 | +| ACCEPTED | 原平台回执和业务任务关联均已提交；任务执行结果另行GetTask或订阅获取 | +| REVIEW | 自动发送停止，保留原身份、请求及稳定错误分类供核查 | + +TaskID为空只说明本地尚未确认。业务提交前不发HTTP，发送失败不回滚已提交业务记录。真正的对象处理仍由业务Worker完成；本例演示提交后取消，不调用外部对象服务。 + +## 2. 初始化与完整演示 + +需要Go1.27.1、MySQL8.4+及当前Orbit迁移/服务。分别配置平台库`ORBIT_MYSQL_DSN`和独立业务库`ORBIT_SENDER_DSN`，通过环境注入凭据，不把真实密码写入文档、命令历史或Git。 + +业务库由业务方先创建，示意SQL在自己的开发MySQL执行： + +```sql +CREATE DATABASE orbit_sender_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin; +``` + +从仓库根目录初始化Orbit，输出凭据文件必须不存在： + +```sh +go run ./cmd/orbit-migrate up +mkdir -p .local +chmod 700 .local +go run ./cmd/orbit-admin bootstrap \ + --application outbox-demo --namespace default --environment development \ + --queue outbox --output .local/outbox.credentials.json +go run ./cmd/orbit +``` + +另开终端，设置指向空业务库的ORBIT_SENDER_DSN后，从仓库根目录执行： + +```sh +cd examples/task-outbox +export ORBIT_URL=http://127.0.0.1:8080 +export ORBIT_CREDENTIALS_FILE=../../.local/outbox.credentials.json +GOWORK=off go run ./cmd/sender init +GOWORK=off go run ./cmd/sender demo +``` + +init只创建本模块两张表，已有表会报错。生产已有业务库应使用自身迁移，不直接运行此初始化。demo使用独立操作ID，先输出ID，再创建一条业务记录、重复核对、发送并取消受理任务。运行上限20秒，任务从平台受理起有60秒总时限；原历史和审计保留。 + +```text +sender_schema_initialized=true foreign_keys=0 +demo_operation=<本次操作ID> +operation=<本次操作ID> state=ACCEPTED sends=1 code=ACCEPTED task=<任务ID> +local_duplicate=true accepted_once=true execution_success_not_implied=true +``` + +每次demo新建操作，失败后应保留打印的ID并查询已有记录，不能再次运行demo当成原操作恢复。演示队列保持专用且不启动object.prepare处理器，避免取消示例与真实业务执行竞争。 + +## 3. 分开提交、发送和恢复 + +以下命令沿用上面的环境。`SENDER_OPERATION_ID`应替换为业务系统已持久化的32位操作ID，重报继续使用同一个值。 + +```sh +export SENDER_OPERATION_ID=11111111111111111111111111111111 +GOWORK=off go run ./cmd/sender enqueue "$SENDER_OPERATION_ID" objects/input.txt +GOWORK=off go run ./cmd/sender status "$SENDER_OPERATION_ID" +GOWORK=off go run ./cmd/sender send +``` + +enqueue只提交本地事务，不等待网络。首次输出local_committed=true、duplicate=false；同内容再次调用返回duplicate=true，即使Outbox后来已受理也不再次创建业务记录。输入固定handler=object.prepare、版本v1、business_key=操作ID，payload包含对象引用和精确整数9007199254740993。业务引用与完整请求在首次事务中冻结。 + +send持续处理当前命名空间与队列的待发项，无任务时每500毫秒等待；存储等错误后等待2秒。可运行多个发送进程；Ctrl-C/SIGTERM停止后，用同一环境重新启动send即可继续。数据库租约使用索引范围查询及SKIP LOCKED领取，不依赖进程内队列。 + +另开相同环境的终端查看： + +```sh +GOWORK=off go run ./cmd/sender status "$SENDER_OPERATION_ID" +``` + +预期由PENDING/SENDING变成ACCEPTED，包含任务ID。运行演示失败、COMMIT未知、服务中断或进程重启后都先查看同一个ID。status只读，不触发发送或变更预算。REVIEW不自动恢复；本例没有重置预算、换键或强制确认命令。 + +凭据文件提供token、namespace_id、queue_id；ORBIT_TOKEN、ORBIT_NAMESPACE_ID、ORBIT_QUEUE_ID可覆盖对应字段。发送器需要queues:read、tasks:submit及固定队列范围；demo还需tasks:read和tasks:cancel。换凭据后发送前再次GetQueue核对命名空间与队列；配置的目标地址必须仍指向原Orbit部署，不能将已有Outbox切换到独立的新部署。 + +## 4. Go接入接口 + +以下方法属于示例的内部业务适配包，完整源码见[store.go](../../examples/task-outbox/internal/outbox/store.go)及[sender.go](../../examples/task-outbox/internal/outbox/sender.go)。接入自己的业务模块时保留事务顺序与协议，替换sender_jobs为实际业务表；核心Orbit客户端仍使用独立SDK。 + +```go +func Open(ctx context.Context, dsn string) (*sql.DB, error) +func Init(ctx context.Context, db *sql.DB) error +func New(db *sql.DB, namespaceID, queueID string) (*Store, error) +func (s *Store) Enqueue(ctx context.Context, operationID, sourceRef string, + request orbit.SubmitTaskRequest) (duplicate bool, err error) +func (s *Store) SendOnce(ctx context.Context, api *orbit.Client) (Record, error) +func (s *Store) Status(ctx context.Context, operationID string) (Record, error) +``` + +已有db、命名空间、队列与客户端时，调用片段如下；完整imports/main/上下文和错误处理见[CLI源码](../../examples/task-outbox/cmd/sender/main.go)。 + +```go +store, err := outbox.New(db, namespaceID, queueID) +if err != nil { return err } +duplicate, err := store.Enqueue(ctx, operationID, "objects/input.txt", + orbit.SubmitTaskRequest{ + QueueID: queueID, Handler: "object.prepare", + HandlerVersion: "v1", PayloadVersion: "v1", + Payload: json.RawMessage(`{"source_ref":"objects/input.txt"}`), + BusinessKey: operationID, TaskTimeoutMS: 60000, + }) +if err != nil { return err } +fmt.Println("本地已提交", duplicate) +record, err := store.SendOnce(ctx, client) +if errors.Is(err, outbox.ErrIdle) { return nil } +if err != nil { return err } +fmt.Println(record.OperationID, record.State) +``` + +Enqueue先拒绝非法UTF-8字符串，避免JSON编码静默替换业务标识，再冻结当前v1默认时限、恢复等级、尝试策略和UTC时间；SHA256校验冻结文档的字节，不采用平台内部摘要算法。不对业务JSON对象键或数值表达做语义等价推断，因此同操作重报应保留原输入；完整平台字段规则仍由[任务API](../api/tasks.md)校验。升级业务适配层或SDK前应对已有待发文档做兼容回归，不能静默更新请求。 + +业务事务固定先锁Outbox操作身份，再写业务行；平台确认事务也先锁Outbox再写业务关联。适配已有业务事务时需统一全局锁顺序。所有网络交换均在这些事务之外，禁止先向平台提交再保存本地业务记录。 + +## 5. 预算、错误和不确定结果 + +| 规则 | 当前示例值与处理 | +| --- | --- | +| 单次本地事务 | 最多5秒；失败不盲目重放整个事务 | +| 领取租约 | 30秒，数据库UTC时间；每次领取send_count/lease_epoch同时递增 | +| 网络工作期限 | 从领取前开始最多15秒；提交HTTP再限制10秒及数据库剩余租约/截止 | +| 发送次数 | 最多100次领取，包含领取后未发HTTP便退出的情况 | +| 总发送期限 | 首次本地入库后24小时，短于平台90天幂等保留窗口；期满停止新发送 | +| 退避 | 指数1、2、4…最多60秒，加0–999毫秒抖动；429遵守有效Retry-After，整体不超过1小时 | +| 确认 | 只接受最近租约代次；尚未被替代的过期代次仍可保存真实受理回执 | + +| 错误或分类 | 行为 | +| --- | --- | +| ErrConflict | 同业务操作内容不同；核查原记录，不换ID掩盖冲突 | +| ErrCommitUnconfirmed | 本地提交结果未知；查询或用原操作重报Enqueue，发送器按已提交状态继续 | +| ErrStorage | 本地存储失败；保留原身份并恢复连接，不能推断远端未受理 | +| ErrLeaseLost | 已有新代次或新状态；旧发送器停止写回 | +| UNCONFIRMED / BACKPRESSURE | 网络、超时、5xx或429后原键有限重报，次数/截止耗尽后转REVIEW | +| AUTHORIZATION / HTTP_REJECTED | 权限或明确HTTP拒绝，转REVIEW；可能已有历史未知受理，不能直接当作未创建 | +| IDEMPOTENCY_CONFLICT | 原键已对应其他内容，停止并核查 | +| INVALID_DOCUMENT / INVALID_RECEIPT / TARGET_MISMATCH | 请求摘要、回执归属或目标不可信，转REVIEW，保留证据 | +| SEND_BUDGET_EXHAUSTED / SEND_DEADLINE | 不再自动发送，保留原请求供人工核查 | + +send_count统计本地发送领取，与Pulse的执行尝试次数分别计数；原键重复提交不会新建任务或执行尝试。自动发送期限是恢复安全边界，不代表任务执行期限；任务总时限从平台受理时开始。带绝对available_at/start_deadline_at的命令在积压后可能不再合法，不能为了通过校验修改原键对应的时间字段。迟到的合法成功回执仍可确认，已被新代次替换的发送器只能查询新状态。 + +## 6. 故障验证与生产边界 + +`make integration`验证本地并发重复、业务回滚、确认原子性、多发送者领取、旧代次拒绝、背压/权限/预算、精确数字、真实HTTP受理回执丢失、三个阶段真实MySQL COMMIT成功包丢失，以及平台已提交后发送器SIGKILL/新进程自然租约接管。CLI的初始化、enqueue重复、status、send、正常停止和demo也进入回归。详细证据见[验证报告](../reports/phase-3-task-outbox.md)。 + +本例的业务效果仅为本地SQL记录及平台受理，不提供外部服务恰好一次执行。业务Worker继续承担业务幂等；执行结果需要[可靠结果订阅与Inbox](result-inbox.md)等衔接。敏感请求和回执留在业务数据库，应纳入业务访问控制、审计、备份及保留策略；日志只输出操作ID和稳定状态，不输出token或请求正文。 + +REVIEW后的人工核查、证据审批及受控恢复由接入方的业务管理流程承担，示例不提供无证据清空错误/重置预算接口。不要直接修改SQL状态制造成功，也不要在无法确认的情况下换业务操作ID。生产MySQL故障切换、长期归档、跨部署迁移和容量仍按P6验收；本例不启动流程或发送Weave信号，相应Outbox命令类型随流程API演进。 diff --git a/docs/reports/phase-3-task-outbox.md b/docs/reports/phase-3-task-outbox.md new file mode 100644 index 0000000..789cfcd --- /dev/null +++ b/docs/reports/phase-3-task-outbox.md @@ -0,0 +1,75 @@ +# P3 业务事务 Outbox 回归 + +验证日期：2026-09-17。独立业务示例、发送恢复、CLI和接入文档已实现；定向真实数据库、全量链路及最终输入边界补充回归均已正常退出，状态码0。P3最终范围核对、Weave及生产部署验收继续按计划推进。 + +## 1. 实现证据 + +- [独立模块](../../examples/task-outbox/README.md)只依赖独立Orbit SDK与业务侧MySQL驱动，核心SDK没有新增依赖。 +- [模型](../../examples/task-outbox/internal/outbox/model.go)与[建表](../../examples/task-outbox/internal/outbox/schema.sql)：2个业务实体、22个Go/SQL中文注释字段，零外键；不改变平台35个实体和289个持久化字段。 +- [业务事务](../../examples/task-outbox/internal/outbox/store.go)：业务操作与待发命令原子创建，稳定操作/提交身份、冻结请求、摘要、有限次数、租约代次及原回执；确认与业务任务关联再次同事务。 +- [发送器](../../examples/task-outbox/internal/outbox/sender.go)：事务外HTTP、当前凭据核对目标、原键重报、背压退避、未知结果分类、过期接管与旧代次拒绝。 +- [CLI](../../examples/task-outbox/cmd/sender/main.go)及[完整命令/方法/恢复手册](../examples/task-outbox.md)：init、enqueue、send、status和demo。 +- [Makefile](../../Makefile)和[隔离脚本](../../scripts/test-mysql.sh)纳入独立模块，使用单独业务数据库和专用平台凭据，不触碰现有开发数据库。 + +## 2. 定向数据库与故障证据 + +[数据库回归](../../examples/task-outbox/internal/outbox/store_integration_test.go)验证： + +1. 8个相同业务操作并发入库只有一条业务记录/Outbox，7次重复核对；修改正文、源引用或目标队列不能复用原操作。 +2. 实际业务INSERT失败时Outbox一起回滚；业务关联UPDATE失败时ACCEPTED一起回滚，恢复后可保存原回执。 +3. 8个领取者处理20条待发项，每项初次只被一个有效代次领取；过期接管沿用提交键，旧代次确认拒绝。 +4. 错误目标、损坏文档、权限拒绝、幂等冲突和无效回执停止自动发送；背压遵守Retry-After，次数耗尽和截止产生REVIEW，最后一次未知响应不能绕过上限。 +5. 实际Schema查询确认22个中文列注释、外键0，通过实际违规写入验证JSON、状态、次数和截止CHECK生效，领取查询使用ix_sender_due且无filesort；大整数9007199254740993在冻结请求与真实回执中保持精确。 + +[COMMIT故障代理](../../examples/task-outbox/internal/outbox/commit_loss_integration_test.go)在业务入库、领取和本地确认三处分别截获真实MySQL COMMIT OK包后丢弃。调用方得到ErrCommitUnconfirmed，新Store分别核对已存在业务记录、已提交租约和已确认关联，不假定COMMIT失败即回滚。 + +[HTTP回执丢失](../../examples/task-outbox/internal/outbox/http_integration_test.go)先取得真实平台受理结果，再向客户端发送不完整响应并断开；Outbox保持PENDING，原键重报后同一任务ACCEPTED、deduplicated=true、发送次数2，队列只有一项任务。错误持久化命名空间在提交前被拒绝。 + +[发送进程故障](../../examples/task-outbox/internal/outbox/process_integration_test.go)等待真实平台受理后、本地确认前的屏障，再SIGKILL并确认旧进程已退出。新进程等待数据库30秒租约自然到期，以原键取得原回执；最终只存在一个平台任务和一条本地业务记录。测试没有直接改时间来替代这项进程接管。 + +定向脚本第三轮正常退出，状态码0：Outbox集成33.009秒，其中发送进程故障31.19秒；init/demo均正常完成。随后增加的[真实CLI测试](../../examples/task-outbox/internal/outbox/cli_integration_test.go)进入本轮全量回归，验证编译后的enqueue重复、status、send及正常信号停止。 + +## 3. 发现与修正 + +- 初次故障注入复用了上一笔COMMIT标记，误把后续事务准备响应当作COMMIT成功；错误标记未消费导致代理等待。通过当前进程和数据库状态核实后，对归属本轮的测试进程取得SIGQUIT堆栈，确认问题，再修正逐命令标记及非阻塞故障通知；原测试终态失败后才重新运行。 +- 进程恢复的首个断言统计了整个命名空间的业务记录，混入之前HTTP测试的其他队列；改为沿Outbox关联按命名空间和队列核对。平台任务及操作身份没有发生重复。 +- HTTP故障先发送成功头与不完整正文，确保客户端看到响应不完整，不把连接层透明重放当成Outbox自身恢复证据。 +- 最终复查增加非法UTF-8字符串的入库前拒绝，防止Go JSON编码静默替换业务标识；单元和真实数据库验证均确认业务行与Outbox未写入。此修改后重新完成基础检查和Outbox完整独立链路。 + +这些修正属于测试注入和验证范围；报告不把早期失败记为通过。 + +## 4. 环境、复现与全量结果 + +macOS、Go1.27.1、Docker单实例MySQL8.4.10，临时数据库、随机凭据与回环端口。业务表与平台库分离，但本轮使用同一个隔离MySQL进程，不代表生产跨故障域验收。 + +```sh +make fmt test vet build +make integration +``` + +全量脚本和最终Outbox补充脚本均已观察到终态退出码0。没有配置数据库时go test跳过Integration，不能据此声称已覆盖真实故障。 + +| 验证组 | 本轮实际结果 | +| --- | --- | +| fmt / race / vet / build | PASS；最终UTF-8修改后再次全部通过 | +| 全量MySQL平台组 | PASS，121.429秒；历史外键70→0、323个未替换索引组件保留 | +| SDK Worker | PASS，87.148秒 | +| SDK计划 / 队列 / 批量 / 查询 | PASS，5.607 / 4.957 / 1.407 / 1.362秒 | +| SDK订阅管理 / 消费 | PASS，34.081 / 51.750秒 | +| SDK人工核查与重试 | PASS，3.375秒 | +| Inbox数据库与消费者故障 | PASS，34.989秒 | +| Outbox完整组 | PASS，34.327秒；含CLI 1.34秒、真实发送进程恢复31.16秒 | +| SDK应用额度 | PASS，1.425秒 | +| 九项平台节点/Worker故障 | PASS，62.786秒；加上Inbox与Outbox进程，共十一项进程故障 | +| 完整可运行示例 | 17个核心SDK程序 + Inbox + Outbox，共19个通过 | +| 最终UTF-8增强后的Outbox独立复测 | PASS，33.808秒；真实MySQL、HTTP、CLI、三个COMMIT丢响应与自然租约进程恢复全部重跑，init/demo通过 | + +全量链路先通过，随后仅增强独立Outbox模块的非法UTF-8入库校验，并补充完整模块复测；平台、核心SDK和协议代码未改变。文档链接/JSON块、SDK无服务端或MySQL依赖检查通过。耗时是这次正确性回归记录，不是容量或吞吐指标。 + +## 5. 范围与后续 + +- 本例证明本地业务事务与单项任务受理的稳定身份衔接，业务实际副作用仍由Worker幂等承担，结果回写使用订阅/Inbox等链路。 +- 固定24小时发送窗口、100次领取、30秒租约、原请求重报；不提供无证据重置预算、自动换键或无限恢复。 +- REVIEW保留原数据，由接入方带证据和审计的管理流程处置；本示例未实现通用业务审核系统或跨业务ORM适配。 +- 核心SDK、平台Schema和执行协议均未改变；新增业务示例不是已发布的兼容性SDK包。 +- 流程启动与信号Outbox类型需随Weave API实现；P6继续容量、数据库切换、备份恢复及运维验证。当前正确性与进程恢复证据不等同于生产高可用或百万规模容量结论。 diff --git a/docs/sdk/README.md b/docs/sdk/README.md index 2a11ebf..63f251f 100644 --- a/docs/sdk/README.md +++ b/docs/sdk/README.md @@ -80,6 +80,8 @@ fmt.Println(current.ID, current.State, current.Revision) 受理回执不是业务执行结果。同键重报返回原始快照；当前状态应使用 GetTask。业务侧创建操作与平台提交之间存在双写边界，接入方应在自身事务中保存操作记录/Outbox，再使用稳定身份提交，不能依靠随机新键补发来“修复”未知结果。 +[业务事务Outbox完整场景](../examples/task-outbox.md)已提供独立业务数据库、原子业务写入、持久化发送器、原身份恢复及CLI；核心SDK保持无MySQL/ORM依赖。业务适配包的Enqueue/SendOnce/Status与实际运行命令均在该手册说明。 + ## 3. 取消与错误处理 ```go diff --git a/examples/task-outbox/README.md b/examples/task-outbox/README.md new file mode 100644 index 0000000..b209bf4 --- /dev/null +++ b/examples/task-outbox/README.md @@ -0,0 +1,19 @@ +# 业务事务 Outbox 示例 + +独立Go module，使用业务方自己的MySQL保存业务记录与待发请求，再通过独立Orbit SDK提交任务。平台与业务库分别提交，通过原幂等身份衔接恢复。 + +- [完整初始化、运行、恢复与接口说明](../../docs/examples/task-outbox.md) +- [命令行程序](cmd/sender/main.go)：init、enqueue、send、status、demo。 +- [业务事务](internal/outbox/store.go)、[发送器](internal/outbox/sender.go)、[中文模型](internal/outbox/model.go)与[无外键表结构](internal/outbox/schema.sql)。 + +本模块是可运行的业务接入参考。核心SDK不引入MySQL驱动；生产业务应将本地事务接入自身业务模型、权限、审计与迁移，固定SDK版本并验证兼容性。本例的internal包不作为已发布通用SDK包承诺兼容。 + +从本目录运行基础检查： + +```sh +GOWORK=off go test -race ./... +GOWORK=off go vet ./... +GOWORK=off go build ./cmd/sender +``` + +真实数据库、HTTP、COMMIT响应丢失、CLI及发送进程故障由仓库根目录的`make integration`统一执行，使用专门的一次性数据库。没有测试环境变量时Integration会跳过。 diff --git a/examples/task-outbox/cmd/sender/main.go b/examples/task-outbox/cmd/sender/main.go new file mode 100644 index 0000000..8db5b53 --- /dev/null +++ b/examples/task-outbox/cmd/sender/main.go @@ -0,0 +1,192 @@ +// sender 演示业务事务Outbox与可重启的有限发送器。 +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "os/signal" + "syscall" + "time" + + "git.yuntuo.hk/k2safe/orbit/examples/task-outbox/internal/outbox" + 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 { + return errors.New("usage: sender init|enqueue |send|status |demo") + } + command := os.Args[1] + if !(command == "init" && len(os.Args) == 2 || command == "send" && len(os.Args) == 2 || command == "demo" && len(os.Args) == 2 || command == "enqueue" && len(os.Args) == 4 || command == "status" && len(os.Args) == 3) { + return errors.New("invalid sender command or arguments") + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + open, cancel := context.WithTimeout(ctx, 10*time.Second) + db, err := outbox.Open(open, os.Getenv("ORBIT_SENDER_DSN")) + cancel() + if err != nil { + return err + } + defer db.Close() + if command == "init" { + bounded, done := context.WithTimeout(ctx, 10*time.Second) + defer done() + if err = outbox.Init(bounded, db); err != nil { + return err + } + fmt.Println("sender_schema_initialized=true foreign_keys=0") + return nil + } + api, ns, queue, err := client() + if err != nil { + return err + } + store, err := outbox.New(db, ns, queue) + if err != nil { + return err + } + switch command { + case "enqueue": + id, source := os.Args[2], os.Args[3] + request := request(queue, id, source) + duplicate, err := store.Enqueue(ctx, id, source, request) + if err != nil { + return err + } + fmt.Printf("operation=%s local_committed=true duplicate=%t\n", id, duplicate) + return nil + case "status": + r, err := store.Status(ctx, os.Args[2]) + if err != nil { + return err + } + printRecord(r) + return nil + case "demo": + return demo(ctx, store, api, queue) + case "send": + for ctx.Err() == nil { + r, err := store.SendOnce(ctx, api) + wait := 500 * time.Millisecond + if err == nil { + printRecord(r) + } else if !errors.Is(err, outbox.ErrIdle) { + fmt.Fprintln(os.Stderr, err) + wait = 2 * time.Second + } + timer := time.NewTimer(wait) + select { + case <-ctx.Done(): + timer.Stop() + case <-timer.C: + } + } + return nil + } + return nil +} +func client() (*orbit.Client, string, string, error) { + token, ns, queue := os.Getenv("ORBIT_TOKEN"), os.Getenv("ORBIT_NAMESPACE_ID"), 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 sender credential file") + } + var c struct { + Token string `json:"token"` + NamespaceID string `json:"namespace_id"` + QueueID string `json:"queue_id"` + } + if json.Unmarshal(raw, &c) != nil { + return nil, "", "", errors.New("invalid sender credential file") + } + if token == "" { + token = c.Token + } + if ns == "" { + ns = c.NamespaceID + } + if queue == "" { + queue = c.QueueID + } + } + if token == "" { + return nil, "", "", errors.New("sender credential is required") + } + c, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token}) + return c, ns, queue, err +} +func request(queue, id, source string) orbit.SubmitTaskRequest { + payload, _ := json.Marshal(struct { + SourceRef string `json:"source_ref"` + Sequence uint64 `json:"sequence"` + }{source, 9007199254740993}) + return orbit.SubmitTaskRequest{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, BusinessKey: id, TaskTimeoutMS: 60000} +} +func printRecord(r outbox.Record) { + task := "" + if r.TaskID != nil { + task = *r.TaskID + } + fmt.Printf("operation=%s state=%s sends=%d code=%s task=%s\n", r.OperationID, r.State, r.SendCount, r.LastCode, task) +} +func demo(parent context.Context, s *outbox.Store, api *orbit.Client, queue string) error { + ctx, stop := context.WithTimeout(parent, 20*time.Second) + defer stop() + id, err := orbit.NewEventMessageID() + if err != nil { + return err + } + // 演示每次新建一个操作；真实接入必须先由业务方保存ID再调用Enqueue。 + fmt.Printf("demo_operation=%s\n", id) + in := request(queue, id, "objects/outbox-demo.txt") + if _, err = s.Enqueue(ctx, id, "objects/outbox-demo.txt", in); err != nil { + return err + } + duplicate, err := s.Enqueue(ctx, id, "objects/outbox-demo.txt", in) + if err != nil || !duplicate { + return errors.New("outbox local duplicate was not recovered") + } + for { + r, err := s.Status(ctx, id) + if err != nil { + return err + } + if r.State == "REVIEW" { + return errors.New("outbox demo needs review using its printed operation identity") + } + if r.State == "ACCEPTED" { + task, err := api.GetTask(ctx, *r.TaskID) + if err != nil { + return err + } + if _, err = api.CancelTask(ctx, task.ID, "outbox-demo-cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Revision, Reason: "outbox submission demo completed"}); err != nil { + return err + } + printRecord(r) + fmt.Println("local_duplicate=true accepted_once=true execution_success_not_implied=true") + return nil + } + _, err = s.SendOnce(ctx, api) + if err != nil && !errors.Is(err, outbox.ErrIdle) { + return err + } + timer := time.NewTimer(100 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } +} diff --git a/examples/task-outbox/go.mod b/examples/task-outbox/go.mod new file mode 100644 index 0000000..7fcb21b --- /dev/null +++ b/examples/task-outbox/go.mod @@ -0,0 +1,15 @@ +module git.yuntuo.hk/k2safe/orbit/examples/task-outbox + +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/task-outbox/go.sum b/examples/task-outbox/go.sum new file mode 100644 index 0000000..f808e3b --- /dev/null +++ b/examples/task-outbox/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/task-outbox/internal/outbox/cli_integration_test.go b/examples/task-outbox/internal/outbox/cli_integration_test.go new file mode 100644 index 0000000..fea22fc --- /dev/null +++ b/examples/task-outbox/internal/outbox/cli_integration_test.go @@ -0,0 +1,105 @@ +package outbox + +import ( + "context" + "io" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "testing" + "time" +) + +func TestIntegrationSenderCLI(t *testing.T) { + s, api, base, _ := realStore(t) + ctx, stop := context.WithTimeout(context.Background(), 30*time.Second) + defer stop() + binary := filepath.Join(t.TempDir(), "sender") + build := exec.CommandContext(ctx, filepath.Join(runtime.GOROOT(), "bin", "go"), "build", "-o", binary, "../../cmd/sender") + if raw, err := build.CombinedOutput(); err != nil { + t.Fatalf("build sender: %v %s", err, raw) + } + values := map[string]string{"ORBIT_SENDER_DSN": os.Getenv("ORBIT_TEST_SENDER_DSN"), "ORBIT_URL": base, "ORBIT_CREDENTIALS_FILE": os.Getenv("ORBIT_TEST_SENDER_CREDENTIALS_FILE"), "ORBIT_NAMESPACE_ID": s.namespace, "ORBIT_QUEUE_ID": s.queue, "ORBIT_TOKEN": ""} + env := []string{} + for _, entry := range os.Environ() { + key, _, _ := strings.Cut(entry, "=") + if _, ok := values[key]; !ok { + env = append(env, entry) + } + } + for key, value := range values { + env = append(env, key+"="+value) + } + call := func(args ...string) string { + t.Helper() + cmd := exec.CommandContext(ctx, binary, args...) + cmd.Env = env + raw, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("sender command failed: %v %s", err, raw) + } + return string(raw) + } + id := testID(t) + if !strings.Contains(call("enqueue", id, "objects/cli.txt"), "duplicate=false") || !strings.Contains(call("enqueue", id, "objects/cli.txt"), "duplicate=true") || !strings.Contains(call("status", id), "state=PENDING") { + t.Fatal("CLI local transaction mismatch") + } + cmd := exec.CommandContext(ctx, binary, "send") + cmd.Env = env + cmd.Stdout = io.Discard + cmd.Stderr = io.Discard + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + var waitErr error + go func() { waitErr = cmd.Wait(); close(done) }() + t.Cleanup(func() { + select { + case <-done: + return + default: + } + cmd.Process.Kill() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("owned sender CLI did not exit") + } + }) + var r Record + for { + r = status(t, s, id) + if r.State == "ACCEPTED" { + break + } + if r.State == "REVIEW" { + t.Fatal("CLI needs review", r.LastCode) + } + select { + case <-done: + t.Fatal("sender exited before acceptance", waitErr) + case <-ctx.Done(): + t.Fatal("CLI sender timeout") + case <-time.After(100 * time.Millisecond): + } + } + if err := cmd.Process.Signal(os.Interrupt); err != nil { + t.Fatal(err) + } + select { + case <-done: + if waitErr != nil { + t.Fatal("sender did not stop gracefully", waitErr) + } + case <-ctx.Done(): + t.Fatal("sender shutdown timeout") + } + if !strings.Contains(call("status", id), "state=ACCEPTED") || countRows(t, s, "sender_jobs") != 1 || r.SendCount != 1 { + t.Fatal("CLI acceptance did not persist") + } + cancelAccepted(t, api, *r.TaskID) + t.Log("built CLI enqueue duplicate/status/send/graceful stop exercised against real business DB and Orbit") +} diff --git a/examples/task-outbox/internal/outbox/commit_loss_integration_test.go b/examples/task-outbox/internal/outbox/commit_loss_integration_test.go new file mode 100644 index 0000000..63a2fbc --- /dev/null +++ b/examples/task-outbox/internal/outbox/commit_loss_integration_test.go @@ -0,0 +1,226 @@ +package outbox + +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 { + committing.Store(strings.EqualFold(strings.TrimSpace(string(packet[5:])), "COMMIT")) + } + if writeMySQLPacket(up, packet) != nil { + return + } + } + }() + go func() { + defer func() { done <- struct{}{} }() + for { + packet, err := readMySQLPacket(up) + if err != nil { + return + } + if packet[3] == 1 && committing.Swap(false) && len(packet) > 4 && packet[4] == 0 && g.armed.CompareAndSwap(true, false) { + select { + case g.dropped <- struct{}{}: + default: + t.Error("unconsumed COMMIT fault marker") + } + 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 TestIntegrationSenderCommitResponseLoss(t *testing.T) { + direct, dsn := integrationDB(t) + cfg, err := mysql.ParseDSN(dsn) + if err != nil || cfg.TLSConfig != "" { + t.Fatal("fault requires isolated plaintext MySQL") + } + gateway := newCommitLossGateway(t, cfg.Addr) + cfg.Addr = gateway.listener.Addr().String() + ctx, stop := context.WithTimeout(context.Background(), 30*time.Second) + defer stop() + proxied, err := Open(ctx, cfg.FormatDSN()) + if err != nil { + t.Fatal(err) + } + defer proxied.Close() + for _, phase := range []string{"enqueue", "claim", "confirm"} { + t.Run(phase, func(t *testing.T) { + s := testStore(t, proxied) + fresh, _ := New(direct, s.namespace, s.queue) + id := testID(t) + in := testRequest(s.queue) + if phase == "enqueue" { + gateway.armed.Store(true) + if _, err = s.Enqueue(ctx, id, "objects/test", in); !errors.Is(err, ErrCommitUnconfirmed) { + t.Fatal("enqueue reported confirmed", err) + } + dup, e := fresh.Enqueue(ctx, id, "objects/test", in) + if e != nil || !dup || countRows(t, fresh, "sender_jobs") != 1 { + t.Fatal("local commit not reconciled", e) + } + } else { + if _, err = s.Enqueue(ctx, id, "objects/test", in); err != nil { + t.Fatal(err) + } + if phase == "claim" { + gateway.armed.Store(true) + } + claim, e := s.claim(ctx) + if phase == "claim" { + if !errors.Is(e, ErrCommitUnconfirmed) { + t.Fatal("claim reported confirmed", e) + } + r := status(t, fresh, id) + if r.State != "SENDING" || r.SendCount != 1 { + t.Fatal("claim was not actually committed") + } + execSQL(t, direct, `UPDATE sender_outbox SET wake_at=TIMESTAMPADD(SECOND,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=? AND operation_id=?`, s.namespace, id) + r, e = fresh.claim(ctx) + if e != nil || r.SendCount != 2 || r.IdempotencyKey != "outbox-task-"+id { + t.Fatal("unknown claim did not recover", e) + } + } else { + if e != nil { + t.Fatal(e) + } + receipt := fakeReceipt(t, s) + gateway.armed.Store(true) + if e = s.finish(ctx, claim, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(e, ErrCommitUnconfirmed) { + t.Fatal("confirmation reported confirmed", e) + } + r := status(t, fresh, id) + if r.State != "ACCEPTED" || r.TaskID == nil || *r.TaskID != receipt.Task.ID { + t.Fatal("confirmed commit not visible") + } + if _, e = fresh.claim(ctx); !errors.Is(e, ErrIdle) { + t.Fatal("confirmed item eligible for resend", e) + } + var task string + if e = direct.QueryRowContext(ctx, `SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=?`, s.namespace, id).Scan(&task); e != nil || task != receipt.Task.ID { + t.Fatal("business confirmation not atomic", e) + } + } + } + select { + case <-gateway.dropped: + case <-ctx.Done(): + t.Fatal("actual COMMIT OK not intercepted") + } + }) + } + t.Log("real COMMIT OK discarded for business enqueue, sender claim and local acceptance; fresh stores reconciled persisted identity and state") +} diff --git a/examples/task-outbox/internal/outbox/http_integration_test.go b/examples/task-outbox/internal/outbox/http_integration_test.go new file mode 100644 index 0000000..c9108ab --- /dev/null +++ b/examples/task-outbox/internal/outbox/http_integration_test.go @@ -0,0 +1,176 @@ +package outbox + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func realClient(t *testing.T) (*orbit.Client, string, string) { + t.Helper() + base, path := os.Getenv("ORBIT_TEST_SENDER_URL"), os.Getenv("ORBIT_TEST_SENDER_CREDENTIALS_FILE") + if base == "" || path == "" { + t.Skip("requires isolated Orbit process") + } + if !strings.HasPrefix(base, "http://127.0.0.1:") { + t.Fatal("sender HTTP test requires loopback") + } + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal("cannot read sender credential") + } + var cred struct { + Token string `json:"token"` + } + if json.Unmarshal(raw, &cred) != nil || cred.Token == "" { + t.Fatal("invalid sender credential") + } + api, err := orbit.NewClient(base, orbit.Options{Token: cred.Token}) + if err != nil { + t.Fatal(err) + } + return api, base, cred.Token +} +func realStore(t *testing.T) (*Store, *orbit.Client, string, string) { + t.Helper() + db, _ := integrationDB(t) + api, base, token := realClient(t) + ctx, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + id := testID(t) + queue, err := api.CreateQueue(ctx, "outbox-test-"+id, orbit.CreateQueueRequest{Name: "outbox-" + id, MaxConcurrency: 1, RatePerSecond: 10}) + if err != nil { + t.Fatal(err) + } + s, err := New(db, queue.Queue.NamespaceID, queue.Queue.ID) + if err != nil { + t.Fatal(err) + } + return s, api, base, token +} +func cancelAccepted(t *testing.T, api *orbit.Client, id string) { + t.Helper() + ctx, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + task, err := api.GetTask(ctx, id) + if err != nil { + t.Fatal(err) + } + if _, err = api.CancelTask(ctx, id, "outbox-test-cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Revision, Reason: "outbox regression completed"}); err != nil { + t.Fatal(err) + } +} + +func TestIntegrationSenderHTTPReceiptLoss(t *testing.T) { + s, api, base, token := realStore(t) + id := enqueue(t, s) + target, _ := url.Parse(base) + var drop atomic.Bool + drop.Store(true) + var posts atomic.Int32 + committed := make(chan orbit.SubmitTaskResult, 1) + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + up := r.Clone(r.Context()) + up.URL.Scheme = target.Scheme + up.URL.Host = target.Host + up.Host = target.Host + up.RequestURI = "" + resp, err := http.DefaultTransport.RoundTrip(up) + if err != nil { + http.Error(w, "upstream unavailable", 502) + return + } + defer resp.Body.Close() + raw, err := io.ReadAll(io.LimitReader(resp.Body, (1<<20)+1)) + if err != nil { + http.Error(w, "upstream read failed", 502) + return + } + if r.Method == "POST" && r.URL.Path == "/v1/tasks" { + posts.Add(1) + if r.Header.Get("Idempotency-Key") != "outbox-task-"+id { + t.Error("original key changed") + } + if resp.StatusCode >= 200 && resp.StatusCode < 300 && drop.CompareAndSwap(true, false) { + var envelope struct { + Data orbit.SubmitTaskResult `json:"data"` + } + if json.Unmarshal(raw, &envelope) != nil || !idPattern.MatchString(envelope.Data.Task.ID) { + t.Error("invalid real receipt") + http.Error(w, "bad receipt", 502) + return + } + committed <- envelope.Data + // 先发送成功响应头与不完整正文，避免Transport把无响应头的断线当成可透明重放。 + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Length", strconv.Itoa(len(raw))) + w.WriteHeader(resp.StatusCode) + w.Write(raw[:1]) + w.(http.Flusher).Flush() + conn, _, err := w.(http.Hijacker).Hijack() + if err == nil { + conn.Close() + } + return + } + } + for key, values := range resp.Header { + w.Header()[key] = values + } + w.WriteHeader(resp.StatusCode) + w.Write(raw) + })) + defer proxy.Close() + through, err := orbit.NewClient(proxy.URL, orbit.Options{Token: token}) + if err != nil { + t.Fatal(err) + } + ctx, stop := context.WithTimeout(context.Background(), 20*time.Second) + defer stop() + r, err := s.SendOnce(ctx, through) + if err != nil || r.State != "PENDING" || r.LastCode != "UNCONFIRMED" { + t.Fatal("unknown remote commit classified as done", r.State, err) + } + var original orbit.SubmitTaskResult + select { + case original = <-committed: + case <-ctx.Done(): + t.Fatal("real accepted receipt not intercepted") + } + if _, err = api.GetTask(ctx, original.Task.ID); err != nil { + t.Fatal("remote was not actually committed", err) + } + execSQL(t, s.db, `UPDATE sender_outbox SET wake_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, s.namespace, id) + r, err = s.SendOnce(ctx, through) + if err != nil || r.State != "ACCEPTED" || r.TaskID == nil || *r.TaskID != original.Task.ID || r.SendCount != 2 { + t.Fatal("original identity not recovered", r.State, err) + } + var receipt orbit.SubmitTaskResult + if json.Unmarshal(r.ReceiptJSON, &receipt) != nil || !receipt.Deduplicated || !strings.Contains(string(receipt.Task.Payload), "9007199254740993") { + t.Fatal("replayed receipt or precision lost") + } + page, err := api.ListTasks(ctx, orbit.ListTasksRequest{QueueID: s.queue}) + if err != nil || len(page.Items) != 1 || posts.Load() != 2 { + t.Fatal("remote duplicate task", len(page.Items), err) + } + // 相同凭据搭配错误的持久化目标命名空间，必须在提交前停止。 + wrong, _ := New(s.db, testID(t), s.queue) + wrongID := enqueue(t, wrong) + r, err = wrong.SendOnce(ctx, through) + if err != nil || r.OperationID != wrongID || r.State != "REVIEW" || r.LastCode != "TARGET_MISMATCH" || posts.Load() != 2 { + t.Fatal("wrong namespace submitted", r.State, err) + } + cancelAccepted(t, api, original.Task.ID) + t.Log("real HTTP receipt discarded after Orbit commit; original key replay recovered exactly one task; wrong namespace never posted") +} diff --git a/examples/task-outbox/internal/outbox/model.go b/examples/task-outbox/internal/outbox/model.go new file mode 100644 index 0000000..a6358ae --- /dev/null +++ b/examples/task-outbox/internal/outbox/model.go @@ -0,0 +1,33 @@ +package outbox + +import "time" + +// Record 是业务数据库中的待发命令；只确认平台受理，不能表示业务执行成功。 +type Record struct { + NamespaceID string // NamespaceID 固定目标命名空间，换凭据不能改变归属。 + QueueID string // QueueID 固定目标队列，发送前通过当前凭据核对归属。 + OperationID string // OperationID 调用方持久化的业务操作ID，同命名空间内唯一。 + IdempotencyKey string // IdempotencyKey 由操作ID派生的固定提交键，重报禁止换键。 + RequestJSON []byte // RequestJSON 冻结的任务提交正文，保留数字精度，不含凭据。 + RequestDigest []byte // RequestDigest 冻结正文的SHA256，发送前重新校验，非平台语义摘要。 + State string // State PENDING待发、SENDING领取中、ACCEPTED已受理、REVIEW需核查。 + SendCount uint32 // SendCount 持久化领取次数，包含领取后未发HTTP即退出的次数，上限100。 + LeaseEpoch uint64 // LeaseEpoch 每次领取递增；旧代次不能更新新代次的本地结果。 + WakeAt *time.Time // WakeAt PENDING的下次领取时间或SENDING的租约期限；终结状态为空，UTC微秒。 + SendDeadlineAt time.Time // SendDeadlineAt 首次入库后24小时停止新发送，短于平台90天幂等窗口。 + LastCode string // LastCode 最近一次稳定分类，不保存HTTP错误正文、凭据或业务秘密。 + TaskID *string // TaskID 仅ACCEPTED记录平台任务ID，其余状态为空。 + ReceiptJSON []byte // ReceiptJSON 仅ACCEPTED保存原受理回执；不是当前执行状态。 + CreatedAt time.Time // CreatedAt 业务事务首次提交命令的UTC时间，微秒精度。 + UpdatedAt time.Time // UpdatedAt 最近状态变化的UTC时间，不能替代租约代次。 +} + +// Job 演示接入方业务记录，与Record原子创建，受理确认时再原子关联任务。 +type Job struct { + NamespaceID string // NamespaceID 与待发命令使用相同命名空间范围。 + OperationID string // OperationID 业务操作标识，与同事务Outbox一一对应。 + SourceRef string // SourceRef 待处理对象引用，不含对象正文；首次创建后固定。 + TaskID *string // TaskID 已确认的平台受理ID；为空不证明平台尚未受理。 + CreatedAt time.Time // CreatedAt 业务记录首次创建UTC时间，微秒精度。 + UpdatedAt time.Time // UpdatedAt 关联平台任务时的UTC时间，微秒精度。 +} diff --git a/examples/task-outbox/internal/outbox/process_integration_test.go b/examples/task-outbox/internal/outbox/process_integration_test.go new file mode 100644 index 0000000..d62fce0 --- /dev/null +++ b/examples/task-outbox/internal/outbox/process_integration_test.go @@ -0,0 +1,239 @@ +package outbox + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "strings" + "syscall" + "testing" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +type roundTripper func(*http.Request) (*http.Response, error) + +func (f roundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +type senderMarker struct { + Phase string `json:"phase"` + TaskID string `json:"task_id"` + Sends uint32 `json:"sends"` + Duplicate bool `json:"duplicate"` +} + +func emitMarker(m senderMarker) { + raw, _ := json.Marshal(m) + fmt.Println("ORBIT_SENDER_TEST " + string(raw)) +} + +// 只存在于测试二进制，使用正式Store和真实HTTP；屏障位于平台提交后、本地确认前。 +func TestSenderProcessHelper(t *testing.T) { + mode := os.Getenv("ORBIT_TEST_SENDER_CHILD") + if mode == "" { + t.Skip("subprocess helper") + } + if mode != "hold_after_remote_commit" && mode != "recover" { + t.Fatal("invalid child mode") + } + db, _ := integrationDB(t) + _, base, token := realClient(t) + s, err := New(db, os.Getenv("ORBIT_TEST_SENDER_NAMESPACE"), os.Getenv("ORBIT_TEST_SENDER_QUEUE")) + if err != nil { + t.Fatal(err) + } + options := orbit.Options{Token: token} + if mode == "hold_after_remote_commit" { + options.HTTPClient = &http.Client{Timeout: 10 * time.Second, Transport: roundTripper(func(r *http.Request) (*http.Response, error) { + resp, err := http.DefaultTransport.RoundTrip(r) + if err != nil { + return nil, err + } + if r.Method != "POST" || r.URL.Path != "/v1/tasks" { + return resp, nil + } + defer resp.Body.Close() + raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if err != nil { + return nil, err + } + var envelope struct { + Data orbit.SubmitTaskResult `json:"data"` + } + if resp.StatusCode != 201 || json.Unmarshal(raw, &envelope) != nil || !idPattern.MatchString(envelope.Data.Task.ID) { + return nil, errors.New("child did not receive actual first acceptance") + } + emitMarker(senderMarker{Phase: "remote_committed", TaskID: envelope.Data.Task.ID}) + <-r.Context().Done() + return nil, r.Context().Err() + })} + } + api, err := orbit.NewClient(base, options) + if err != nil { + t.Fatal(err) + } + ctx, stop := context.WithTimeout(context.Background(), 45*time.Second) + defer stop() + for ctx.Err() == nil { + r, err := s.SendOnce(ctx, api) + if err == nil { + if r.State != "ACCEPTED" { + t.Fatal("child did not accept", r.State) + } + var receipt orbit.SubmitTaskResult + if json.Unmarshal(r.ReceiptJSON, &receipt) != nil || !receipt.Deduplicated { + t.Fatal("child did not recover original receipt") + } + emitMarker(senderMarker{Phase: "accepted", TaskID: *r.TaskID, Sends: r.SendCount, Duplicate: receipt.Deduplicated}) + return + } + if !errors.Is(err, ErrIdle) { + t.Fatal(err) + } + timer := time.NewTimer(100 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + case <-timer.C: + } + } + t.Fatal("child sender timed out") +} + +type senderChild struct { + cmd *exec.Cmd + done chan struct{} + err error + markers chan senderMarker +} + +func startSender(t *testing.T, s *Store, mode string) *senderChild { + t.Helper() + binary, err := os.Executable() + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(binary, "-test.run=^TestSenderProcessHelper$") + values := map[string]string{"ORBIT_TEST_SENDER_CHILD": mode, "ORBIT_TEST_SENDER_NAMESPACE": s.namespace, "ORBIT_TEST_SENDER_QUEUE": s.queue} + for _, entry := range os.Environ() { + key, _, _ := strings.Cut(entry, "=") + if _, ok := values[key]; !ok { + cmd.Env = append(cmd.Env, entry) + } + } + for key, value := range values { + cmd.Env = append(cmd.Env, key+"="+value) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + cmd.Stderr = io.Discard + p := &senderChild{cmd: cmd, done: make(chan struct{}), markers: make(chan senderMarker, 4)} + if err = cmd.Start(); err != nil { + t.Fatal(err) + } + go func() { + scanner := bufio.NewScanner(stdout) + for scanner.Scan() { + if raw, ok := strings.CutPrefix(scanner.Text(), "ORBIT_SENDER_TEST "); ok { + var m senderMarker + if json.Unmarshal([]byte(raw), &m) == nil { + p.markers <- m + } + } + } + p.err = 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 sender child did not exit") + } + }) + return p +} +func (p *senderChild) marker(t *testing.T, ctx context.Context, phase string) senderMarker { + t.Helper() + for { + select { + case m := <-p.markers: + if m.Phase == phase { + return m + } + case <-p.done: + select { + case m := <-p.markers: + if m.Phase == phase { + return m + } + default: + } + t.Fatal("sender exited without expected marker", p.err) + case <-ctx.Done(): + t.Fatal("sender marker timeout") + } + } +} +func TestIntegrationSenderProcessCommitRecovery(t *testing.T) { + s, api, _, _ := realStore(t) + id := enqueue(t, s) + ctx, stop := context.WithTimeout(context.Background(), 50*time.Second) + defer stop() + first := startSender(t, s, "hold_after_remote_commit") + committed := first.marker(t, ctx, "remote_committed") + if _, err := api.GetTask(ctx, committed.TaskID); err != nil { + t.Fatal("remote commit not confirmed", err) + } + r := status(t, s, id) + if r.State != "SENDING" || r.TaskID != nil || r.SendCount != 1 { + t.Fatal("local receipt already confirmed") + } + if err := first.cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + select { + case <-first.done: + case <-ctx.Done(): + t.Fatal("sender kill not observed") + } + state, ok := first.cmd.ProcessState.Sys().(syscall.WaitStatus) + if !ok || !state.Signaled() || state.Signal() != syscall.SIGKILL { + t.Fatal("sender was not killed", first.err) + } + restarted := startSender(t, s, "recover") + result := restarted.marker(t, ctx, "accepted") + select { + case <-restarted.done: + case <-ctx.Done(): + t.Fatal("replacement sender not exited") + } + if restarted.err != nil || result.TaskID != committed.TaskID || result.Sends != 2 || !result.Duplicate { + t.Fatal("replacement did not recover original receipt", restarted.err) + } + r = status(t, s, id) + if r.State != "ACCEPTED" || r.TaskID == nil || *r.TaskID != committed.TaskID { + t.Fatal("durable local acceptance missing") + } + page, err := api.ListTasks(ctx, orbit.ListTasksRequest{QueueID: s.queue}) + if err != nil || len(page.Items) != 1 || countRows(t, s, "sender_jobs") != 1 { + t.Fatal("restart duplicated remote task or business record", err) + } + cancelAccepted(t, api, committed.TaskID) + t.Log("sender SIGKILL observed after actual Orbit commit, before local confirmation; new process waited for natural lease expiry and recovered one task using original key") +} diff --git a/examples/task-outbox/internal/outbox/schema.sql b/examples/task-outbox/internal/outbox/schema.sql new file mode 100644 index 0000000..616830e --- /dev/null +++ b/examples/task-outbox/internal/outbox/schema.sql @@ -0,0 +1,42 @@ +-- 业务方独立数据库的示例表，不属于平台迁移，不创建外键或级联删除。 +CREATE TABLE sender_outbox ( + namespace_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '固定目标命名空间，重启及换凭据不得改变', + queue_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '固定目标队列，发送前核对当前授权及命名空间', + operation_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '业务操作标识，同命名空间内唯一，调用方持久化后复用', + idempotency_key VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '冻结平台提交键，由业务操作标识派生，禁止换键补发', + request_json MEDIUMTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT '冻结任务提交正文，保留精确数字，不存认证凭据', + request_digest BINARY(32) NOT NULL COMMENT '冻结正文SHA256，防止同身份不同内容，非平台摘要算法', + state VARCHAR(16) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT 'PENDING待发，SENDING领取中，ACCEPTED已受理，REVIEW需核查', + send_count INT UNSIGNED NOT NULL COMMENT '领取次数，包含领取后崩溃的次数，最多100', + lease_epoch BIGINT UNSIGNED NOT NULL COMMENT '领取代次递增，旧发送器不能覆盖新代次结果', + wake_at DATETIME(6) NULL COMMENT '待发下次时间或发送租约期限，终结后为空，UTC微秒', + send_deadline_at DATETIME(6) NOT NULL COMMENT '首次入库24小时后的发送截止，短于平台幂等保留窗口', + last_code VARCHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '最近稳定分类，不保存服务端错误正文或秘密', + task_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NULL COMMENT '已确认的平台任务标识，只在ACCEPTED非空', + receipt_json MEDIUMTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '原受理回执，仅ACCEPTED非空，不代表当前执行结果', + created_at DATETIME(6) NOT NULL COMMENT '业务事务首次创建命令UTC时间，微秒精度', + updated_at DATETIME(6) NOT NULL COMMENT '本地状态最近变化UTC时间，微秒精度', + PRIMARY KEY (namespace_id,operation_id), + UNIQUE KEY uq_sender_identity(namespace_id,idempotency_key), + KEY ix_sender_due(namespace_id,queue_id,wake_at,operation_id), + CONSTRAINT ck_sender_document CHECK (JSON_VALID(request_json)), + CONSTRAINT ck_sender_receipt CHECK (receipt_json IS NULL OR JSON_VALID(receipt_json)), + CONSTRAINT ck_sender_count CHECK (send_count<=100 AND lease_epoch=send_count), + CONSTRAINT ck_sender_state CHECK ( + (state IN ('PENDING','SENDING') AND wake_at IS NOT NULL AND task_id IS NULL AND receipt_json IS NULL) + OR (state='REVIEW' AND wake_at IS NULL AND task_id IS NULL AND receipt_json IS NULL) + OR (state='ACCEPTED' AND wake_at IS NULL AND task_id IS NOT NULL AND receipt_json IS NOT NULL) + ), + CONSTRAINT ck_sender_lease CHECK (state<>'SENDING' OR lease_epoch>0), + CONSTRAINT ck_sender_deadline CHECK (send_deadline_at>created_at AND send_deadline_at<=TIMESTAMPADD(HOUR,24,created_at)) +) ENGINE=InnoDB COMMENT='业务事务Outbox，持久化原请求与有限重发状态，不承担业务恰好一次'; + +CREATE TABLE sender_jobs ( + namespace_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '与Outbox相同的命名空间范围', + operation_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '本地业务操作标识，与Outbox同事务创建', + source_ref VARCHAR(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NOT NULL COMMENT '待处理对象引用，首次入库后固定，不存对象正文或凭据', + task_id CHAR(32) CHARACTER SET ascii COLLATE ascii_bin NULL COMMENT '已确认的平台受理任务，为空不证明远端未受理', + created_at DATETIME(6) NOT NULL COMMENT '业务记录创建UTC时间，微秒精度', + updated_at DATETIME(6) NOT NULL COMMENT '关联任务最近变化UTC时间，微秒精度', + PRIMARY KEY (namespace_id,operation_id) +) ENGINE=InnoDB COMMENT='接入方示例业务记录，与待发请求及受理关联分别同事务保存'; diff --git a/examples/task-outbox/internal/outbox/sender.go b/examples/task-outbox/internal/outbox/sender.go new file mode 100644 index 0000000..dfde242 --- /dev/null +++ b/examples/task-outbox/internal/outbox/sender.go @@ -0,0 +1,107 @@ +package outbox + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "math/rand/v2" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +// SendOnce 有界领取并发送一项；网络交换在数据库事务外，ErrIdle表示本次无到期项。 +// 返回记录只包含本地发送状态；ACCEPTED不是业务执行成功，REVIEW必须人工核查原身份。 +func (s *Store) SendOnce(parent context.Context, api *orbit.Client) (Record, error) { + if api == nil { + return Record{}, ErrInvalid + } + // 整个网络阶段从领取前即受单调时钟15秒约束，短于30秒租约；暂停进程后不会获得新发送预算。 + work, stop := context.WithTimeout(parent, 15*time.Second) + defer stop() + r, err := s.claim(work) + if err != nil { + return Record{}, err + } + if r.State == "REVIEW" { + return s.Status(parent, r.OperationID) + } + state, code, delay, receipt := s.deliver(work, api, r) + // 网络超时不能阻止保存未知结果分类，收尾仍服从调用方总上下文且最多5秒。 + if err = s.finish(parent, r, state, code, delay, receipt); err != nil { + return Record{}, err + } + return s.Status(parent, r.OperationID) +} + +func (s *Store) deliver(ctx context.Context, api *orbit.Client, r Record) (string, string, time.Duration, *orbit.SubmitTaskResult) { + digest := sha256.Sum256(r.RequestJSON) + if !bytes.Equal(digest[:], r.RequestDigest) || r.NamespaceID != s.namespace || r.QueueID != s.queue || r.IdempotencyKey != "outbox-task-"+r.OperationID { + return "REVIEW", "INVALID_DOCUMENT", 0, nil + } + var in orbit.SubmitTaskRequest + decoder := json.NewDecoder(bytes.NewReader(r.RequestJSON)) + decoder.DisallowUnknownFields() + if decoder.Decode(&in) != nil || in.QueueID != s.queue { + return "REVIEW", "INVALID_DOCUMENT", 0, nil + } + // 先通过当前凭据读取固定队列；不靠配置声称命名空间正确，不向错误环境盲发命令。 + queue, err := api.GetQueue(ctx, s.queue) + if err != nil { + state, code, delay := classify(err, r.SendCount) + return state, code, delay, nil + } + if queue.ID != s.queue || queue.NamespaceID != s.namespace { + return "REVIEW", "TARGET_MISMATCH", 0, nil + } + // 数据库时间决定剩余截止预算；额外查询不授予新租约，换代或临近截止时不再开始HTTP。 + var remaining int64 + err = s.db.QueryRowContext(ctx, `SELECT TIMESTAMPDIFF(MICROSECOND,UTC_TIMESTAMP(6),LEAST(wake_at,send_deadline_at)) FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=? AND state='SENDING' AND lease_epoch=?`, s.namespace, s.queue, r.OperationID, r.LeaseEpoch).Scan(&remaining) + if err != nil || remaining <= 0 { + return "PENDING", "UNCONFIRMED", time.Second, nil + } + send, stop := context.WithTimeout(ctx, min(10*time.Second, time.Duration(remaining)*time.Microsecond)) + defer stop() + result, err := api.SubmitTask(send, r.IdempotencyKey, in) + if err != nil { + state, code, delay := classify(err, r.SendCount) + return state, code, delay, nil + } + if !validReceipt(r, result) { + return "REVIEW", "INVALID_RECEIPT", 0, nil + } + return "ACCEPTED", "ACCEPTED", 0, &result +} + +func validReceipt(r Record, receipt orbit.SubmitTaskResult) bool { + t := receipt.Task + return idPattern.MatchString(t.ID) && t.NamespaceID == r.NamespaceID && t.QueueID == r.QueueID && t.Revision > 0 +} + +func classify(err error, count uint32) (state, code string, delay time.Duration) { + state, code = "PENDING", "UNCONFIRMED" + delay = min(time.Minute, time.Second*time.Duration(uint64(1)< 256<<10 || !utf8.Valid(in.Payload) || !json.Valid(in.Payload) || bytes.Equal(bytes.TrimSpace(in.Payload), []byte("null")) { + return nil, ErrInvalid + } + if in.TaskTimeoutMS == 0 { + in.TaskTimeoutMS = 86400000 + } + if in.RecoverySafety == "" { + in.RecoverySafety = orbit.RecoveryReconcileOnly + } + if in.Retry.MaxAttempts == 0 { + in.Retry.MaxAttempts = 1 + } + if in.Retry.AttemptTimeoutMS == 0 { + in.Retry.AttemptTimeoutMS = 30000 + } + if in.Retry.MaxAttempts > 1 { + if in.Retry.InitialDelayMS == 0 { + in.Retry.InitialDelayMS = 1000 + } + if in.Retry.MaxDelayMS == 0 { + in.Retry.MaxDelayMS = 30000 + } + } + if in.AvailableAt != nil { + at := in.AvailableAt.UTC() + in.AvailableAt = &at + } + if in.StartDeadlineAt != nil { + at := in.StartDeadlineAt.UTC() + in.StartDeadlineAt = &at + } + raw, err := json.Marshal(in) + if err != nil || len(raw) > 512<<10 { + return nil, ErrInvalid + } + return raw, nil +} + +// Enqueue 原子创建业务记录及Outbox。duplicate只表示该本地事务已经提交，不表示平台受理。 +// 调用方在第一次调用前持久化operationID；COMMIT回执未知时仍用原ID、原内容再次调用。 +func (s *Store) Enqueue(parent context.Context, operationID, sourceRef string, in orbit.SubmitTaskRequest) (duplicate bool, err error) { + if !idPattern.MatchString(operationID) || !utf8.ValidString(sourceRef) || strings.TrimSpace(sourceRef) == "" || utf8.RuneCountInString(sourceRef) > 512 || strings.ContainsFunc(sourceRef, unicode.IsControl) { + return false, ErrInvalid + } + raw, err := s.document(in) + if err != nil { + return false, err + } + digest := sha256.Sum256(raw) + ctx, stop := context.WithTimeout(parent, 5*time.Second) + defer stop() + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return false, ErrStorage + } + defer tx.Rollback() + _, err = tx.ExecContext(ctx, `INSERT INTO sender_outbox(namespace_id,queue_id,operation_id,idempotency_key,request_json,request_digest,state,send_count,lease_epoch,wake_at,send_deadline_at,last_code,created_at,updated_at) VALUES(?,?,?,?,?,?,'PENDING',0,0,UTC_TIMESTAMP(6),TIMESTAMPADD(HOUR,24,UTC_TIMESTAMP(6)),'',UTC_TIMESTAMP(6),UTC_TIMESTAMP(6))`, s.namespace, s.queue, operationID, "outbox-task-"+operationID, raw, digest[:]) + if err != nil { + var mysqlErr *mysql.MySQLError + if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1062 { + return false, ErrStorage + } + if tx.Rollback() != nil { + return false, ErrStorage + } + var saved []byte + var queue, source string + err = s.db.QueryRowContext(ctx, `SELECT o.request_digest,o.queue_id,j.source_ref FROM sender_outbox o JOIN sender_jobs j ON j.namespace_id=o.namespace_id AND j.operation_id=o.operation_id WHERE o.namespace_id=? AND o.operation_id=?`, s.namespace, operationID).Scan(&saved, &queue, &source) + if err != nil { + return false, ErrStorage + } + if queue != s.queue || source != sourceRef || !bytes.Equal(saved, digest[:]) { + return false, ErrConflict + } + return true, nil + } + // 锁序统一为Outbox→业务行，不能在业务行持锁时反向调用发送确认。 + if _, err = tx.ExecContext(ctx, `INSERT INTO sender_jobs(namespace_id,operation_id,source_ref,created_at,updated_at) VALUES(?,?,?,UTC_TIMESTAMP(6),UTC_TIMESTAMP(6))`, s.namespace, operationID, sourceRef); err != nil { + return false, ErrStorage + } + if tx.Commit() != nil { + return false, ErrCommitUnconfirmed + } + return false, nil +} + +type scanner interface{ Scan(...any) error } + +func scanRecord(row scanner, r *Record, extra ...any) error { + args := []any{&r.NamespaceID, &r.QueueID, &r.OperationID, &r.IdempotencyKey, &r.RequestJSON, &r.RequestDigest, &r.State, &r.SendCount, &r.LeaseEpoch, &r.WakeAt, &r.SendDeadlineAt, &r.LastCode, &r.TaskID, &r.ReceiptJSON, &r.CreatedAt, &r.UpdatedAt} + return row.Scan(append(args, extra...)...) +} +func (s *Store) Status(parent context.Context, id string) (Record, error) { + if !idPattern.MatchString(id) { + return Record{}, ErrInvalid + } + ctx, stop := context.WithTimeout(parent, 5*time.Second) + defer stop() + var r Record + err := scanRecord(s.db.QueryRowContext(ctx, `SELECT `+recordColumns+` FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=?`, s.namespace, s.queue, id), &r) + if errors.Is(err, sql.ErrNoRows) { + return Record{}, ErrNotFound + } + if err != nil { + return Record{}, ErrStorage + } + return r, nil +} + +func (s *Store) claim(parent context.Context) (Record, error) { + ctx, stop := context.WithTimeout(parent, 5*time.Second) + defer stop() + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return Record{}, ErrStorage + } + defer tx.Rollback() + var r Record + var now time.Time + err = scanRecord(tx.QueryRowContext(ctx, `SELECT `+recordColumns+`,UTC_TIMESTAMP(6) FROM sender_outbox FORCE INDEX(ix_sender_due) WHERE namespace_id=? AND queue_id=? AND wake_at<=UTC_TIMESTAMP(6) ORDER BY wake_at,operation_id LIMIT 1 FOR UPDATE SKIP LOCKED`, s.namespace, s.queue), &r, &now) + if errors.Is(err, sql.ErrNoRows) { + return Record{}, ErrIdle + } + if err != nil { + return Record{}, ErrStorage + } + if r.State != "PENDING" && r.State != "SENDING" { + return Record{}, ErrStorage + } + if !now.Before(r.SendDeadlineAt) || r.SendCount >= maxSends { + r.State = "REVIEW" + r.WakeAt = nil + r.LastCode = "SEND_BUDGET_EXHAUSTED" + if !now.Before(r.SendDeadlineAt) { + r.LastCode = "SEND_DEADLINE" + } + } else { + r.State = "SENDING" + r.SendCount++ + r.LeaseEpoch++ + wake := now.Add(30 * time.Second) + if wake.After(r.SendDeadlineAt) { + wake = r.SendDeadlineAt + } + r.WakeAt = &wake + } + _, err = tx.ExecContext(ctx, `UPDATE sender_outbox SET state=?,send_count=?,lease_epoch=?,wake_at=?,last_code=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, r.State, r.SendCount, r.LeaseEpoch, r.WakeAt, r.LastCode, s.namespace, r.OperationID) + if err != nil { + return Record{}, ErrStorage + } + if tx.Commit() != nil { + return Record{}, ErrCommitUnconfirmed + } + return r, nil +} + +// finish 只接受最近代次；过期但尚未被替代的成功回执可落库，重复HTTP只复用原键。 +func (s *Store) finish(parent context.Context, claim Record, state, code string, delay time.Duration, receipt *orbit.SubmitTaskResult) error { + ctx, stop := context.WithTimeout(parent, 5*time.Second) + defer stop() + tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return ErrStorage + } + defer tx.Rollback() + var current Record + var now time.Time + err = scanRecord(tx.QueryRowContext(ctx, `SELECT `+recordColumns+`,UTC_TIMESTAMP(6) FROM sender_outbox WHERE namespace_id=? AND queue_id=? AND operation_id=? FOR UPDATE`, s.namespace, s.queue, claim.OperationID), ¤t, &now) + if err != nil { + return ErrStorage + } + if current.State != "SENDING" || current.LeaseEpoch != claim.LeaseEpoch { + return ErrLeaseLost + } + var wake *time.Time + var task *string + var raw []byte + if state == "ACCEPTED" { + if receipt == nil || !validReceipt(current, *receipt) { + return ErrInvalid + } + raw, err = json.Marshal(receipt) + if err != nil || len(raw) > 1<<20 { + return ErrInvalid + } + task = &receipt.Task.ID + var existing *string + if err = tx.QueryRowContext(ctx, `SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=? FOR UPDATE`, s.namespace, current.OperationID).Scan(&existing); err != nil { + return ErrStorage + } + if existing != nil && *existing != *task { + return ErrConflict + } + if _, err = tx.ExecContext(ctx, `UPDATE sender_jobs SET task_id=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, *task, s.namespace, current.OperationID); err != nil { + return ErrStorage + } + } else if state == "PENDING" { + if delay < time.Second || delay > time.Hour { + return ErrInvalid + } + if !now.Before(current.SendDeadlineAt) || current.SendCount >= maxSends { + state = "REVIEW" + code = "SEND_BUDGET_EXHAUSTED" + } else { + next := now.Add(delay) + if next.After(current.SendDeadlineAt) { + next = current.SendDeadlineAt + } + wake = &next + } + } else if state != "REVIEW" { + return ErrInvalid + } + if _, err = tx.ExecContext(ctx, `UPDATE sender_outbox SET state=?,last_code=?,wake_at=?,task_id=?,receipt_json=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, state, code, wake, task, raw, s.namespace, current.OperationID); err != nil { + return ErrStorage + } + if tx.Commit() != nil { + return ErrCommitUnconfirmed + } + return nil +} diff --git a/examples/task-outbox/internal/outbox/store_integration_test.go b/examples/task-outbox/internal/outbox/store_integration_test.go new file mode 100644 index 0000000..c0aed42 --- /dev/null +++ b/examples/task-outbox/internal/outbox/store_integration_test.go @@ -0,0 +1,361 @@ +package outbox + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + "unicode" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" + "github.com/go-sql-driver/mysql" +) + +func integrationDB(t *testing.T) (*sql.DB, string) { + t.Helper() + dsn := os.Getenv("ORBIT_TEST_SENDER_DSN") + if dsn == "" { + t.Skip("requires isolated sender database") + } + cfg, err := mysql.ParseDSN(dsn) + if err != nil || cfg.DBName != "orbit_sender_test" || !strings.HasPrefix(cfg.Addr, "127.0.0.1:") { + t.Fatal("sender tests require 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, testID(t), testID(t)) + if err != nil { + t.Fatal(err) + } + return s +} +func enqueue(t *testing.T, s *Store) string { + t.Helper() + id := testID(t) + if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); err != nil { + t.Fatal(err) + } + return id +} +func status(t *testing.T, s *Store, id string) Record { + t.Helper() + r, err := s.Status(context.Background(), id) + if err != nil { + t.Fatal(err) + } + return r +} +func execSQL(t *testing.T, db *sql.DB, q string, args ...any) { + t.Helper() + ctx, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + if _, err := db.ExecContext(ctx, q, args...); err != nil { + t.Fatal(err) + } +} +func countRows(t *testing.T, s *Store, table string) int { + t.Helper() + if table != "sender_jobs" && table != "sender_outbox" { + t.Fatal("invalid test table") + } + var n int + query := `SELECT COUNT(*) FROM sender_outbox WHERE namespace_id=? AND queue_id=?` + if table == "sender_jobs" { + query = `SELECT COUNT(*) FROM sender_jobs j JOIN sender_outbox o ON o.namespace_id=j.namespace_id AND o.operation_id=j.operation_id WHERE o.namespace_id=? AND o.queue_id=?` + } + if err := s.db.QueryRow(query, s.namespace, s.queue).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} +func fakeReceipt(t *testing.T, s *Store) orbit.SubmitTaskResult { + return orbit.SubmitTaskResult{Task: orbit.Task{ID: testID(t), NamespaceID: s.namespace, QueueID: s.queue, Revision: 1, State: orbit.TaskWaiting}} +} + +func TestIntegrationSenderSchema(t *testing.T) { + db, _ := integrationDB(t) + var fk int + if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fk); err != nil || fk != 0 { + t.Fatal(fk, err) + } + rows, err := db.Query(`SELECT COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN ('sender_outbox','sender_jobs')`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + n := 0 + for rows.Next() { + var c string + if rows.Scan(&c) != nil || !strings.ContainsFunc(c, func(r rune) bool { return unicode.Is(unicode.Han, r) }) { + t.Fatal("missing Chinese column comment") + } + n++ + } + if rows.Err() != nil || n != 22 { + t.Fatal(n, rows.Err()) + } + rows.Close() + s := testStore(t, db) + id := enqueue(t, s) + for _, assignment := range []string{"state='ACCEPTED'", "send_count=101,lease_epoch=101", "request_json='not json'", "send_deadline_at=TIMESTAMPADD(HOUR,48,created_at)"} { + _, err = db.Exec(`UPDATE sender_outbox SET `+assignment+` WHERE namespace_id=? AND operation_id=?`, s.namespace, id) + var rejected *mysql.MySQLError + if !errors.As(err, &rejected) || rejected.Number != 3819 { + t.Fatal("invalid persisted state did not violate CHECK", assignment, err) + } + } + var plan string + if err = db.QueryRow(`EXPLAIN FORMAT=JSON SELECT operation_id FROM sender_outbox FORCE INDEX(ix_sender_due) WHERE namespace_id=? AND queue_id=? AND wake_at<=UTC_TIMESTAMP(6) ORDER BY wake_at,operation_id LIMIT 1 FOR UPDATE SKIP LOCKED`, s.namespace, s.queue).Scan(&plan); err != nil { + t.Fatal(err) + } + if !strings.Contains(plan, `"key": "ix_sender_due"`) || strings.Contains(plan, `"using_filesort": true`) { + t.Fatal("sender due query lost ordered index", plan) + } + t.Log("two business tables, 22 Chinese-commented columns, zero foreign keys") +} +func TestIntegrationSenderConcurrentEnqueueAndConflict(t *testing.T) { + db, _ := integrationDB(t) + s := testStore(t, db) + id := testID(t) + in := testRequest(s.queue) + invalid := in + invalid.BusinessKey = "operation-\xff" + if _, err := s.Enqueue(context.Background(), id, "objects/test", invalid); !errors.Is(err, ErrInvalid) || countRows(t, s, "sender_jobs") != 0 || countRows(t, s, "sender_outbox") != 0 { + t.Fatal("invalid UTF-8 changed business state", err) + } + type outcome struct { + duplicate bool + err error + } + results := make(chan outcome, 8) + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + dup, err := s.Enqueue(context.Background(), id, "objects/test", in) + results <- outcome{dup, err} + }) + } + wg.Wait() + close(results) + duplicates := 0 + for r := range results { + if r.err != nil { + t.Fatal(r.err) + } + if r.duplicate { + duplicates++ + } + } + if duplicates != 7 || countRows(t, s, "sender_jobs") != 1 || countRows(t, s, "sender_outbox") != 1 { + t.Fatal("duplicate business transaction", duplicates) + } + if _, err := s.Enqueue(context.Background(), id, "objects/other", in); !errors.Is(err, ErrConflict) { + t.Fatal("business identity conflict lost", err) + } + in.Payload = json.RawMessage(`{"sequence":9007199254740992}`) + if _, err := s.Enqueue(context.Background(), id, "objects/test", in); !errors.Is(err, ErrConflict) { + t.Fatal("request content conflict lost", err) + } + other, _ := New(db, s.namespace, testID(t)) + if _, err := other.Status(context.Background(), id); !errors.Is(err, ErrNotFound) { + t.Fatal("cross queue record returned", err) + } + if _, err := other.Enqueue(context.Background(), id, "objects/test", testRequest(other.queue)); !errors.Is(err, ErrConflict) { + t.Fatal("operation was moved to another queue", err) + } + if !strings.Contains(string(status(t, s, id).RequestJSON), "9007199254740993") { + t.Fatal("large integer rounded") + } +} +func TestIntegrationSenderBusinessRollbackAndConfirmationAtomicity(t *testing.T) { + db, _ := integrationDB(t) + s := testStore(t, db) + id := testID(t) + execSQL(t, db, `CREATE TRIGGER sender_insert_fail BEFORE INSERT ON sender_jobs FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected business insert failure'`) + t.Cleanup(func() { + db.Exec(`DROP TRIGGER IF EXISTS sender_insert_fail`) + db.Exec(`DROP TRIGGER IF EXISTS sender_update_fail`) + }) + if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); !errors.Is(err, ErrStorage) { + t.Fatal(err) + } + if countRows(t, s, "sender_outbox") != 0 || countRows(t, s, "sender_jobs") != 0 { + t.Fatal("business rollback left a request") + } + execSQL(t, db, `DROP TRIGGER sender_insert_fail`) + if _, err := s.Enqueue(context.Background(), id, "objects/test", testRequest(s.queue)); err != nil { + t.Fatal(err) + } + claim, err := s.claim(context.Background()) + if err != nil { + t.Fatal(err) + } + receipt := fakeReceipt(t, s) + execSQL(t, db, `CREATE TRIGGER sender_update_fail BEFORE UPDATE ON sender_jobs FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected business confirmation failure'`) + if err = s.finish(context.Background(), claim, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(err, ErrStorage) { + t.Fatal(err) + } + if r := status(t, s, id); r.State != "SENDING" || r.TaskID != nil { + t.Fatal("partial confirmation committed") + } + execSQL(t, db, `DROP TRIGGER sender_update_fail`) + if err = s.finish(context.Background(), claim, "ACCEPTED", "ACCEPTED", 0, &receipt); err != nil { + t.Fatal(err) + } + var task string + if err = db.QueryRow(`SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=?`, s.namespace, id).Scan(&task); err != nil || task != receipt.Task.ID { + t.Fatal("business confirmation missing", err) + } +} +func TestIntegrationSenderConcurrentClaimsAndStaleGeneration(t *testing.T) { + db, _ := integrationDB(t) + s := testStore(t, db) + for range 20 { + enqueue(t, s) + } + claims := make(chan Record, 20) + errs := make(chan error, 8) + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + for { + r, err := s.claim(context.Background()) + if errors.Is(err, ErrIdle) { + return + } + if err != nil { + errs <- err + return + } + claims <- r + } + }) + } + wg.Wait() + close(claims) + close(errs) + for err := range errs { + t.Fatal(err) + } + seen := map[string]bool{} + var first Record + for r := range claims { + if seen[r.OperationID] || r.SendCount != 1 { + t.Fatal("duplicate claim") + } + seen[r.OperationID] = true + first = r + } + if len(seen) != 20 { + t.Fatal("unclaimed records", len(seen)) + } + execSQL(t, db, `UPDATE sender_outbox SET wake_at=TIMESTAMPADD(SECOND,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=? AND operation_id=?`, s.namespace, first.OperationID) + replacement, err := s.claim(context.Background()) + if err != nil || replacement.LeaseEpoch != 2 || replacement.IdempotencyKey != first.IdempotencyKey { + t.Fatal(replacement, err) + } + receipt := fakeReceipt(t, s) + if err = s.finish(context.Background(), first, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(err, ErrLeaseLost) { + t.Fatal("stale sender overwrote result", err) + } + if err = s.finish(context.Background(), replacement, "ACCEPTED", "ACCEPTED", 0, &receipt); err != nil { + t.Fatal(err) + } +} +func TestIntegrationSenderHTTPClassificationAndBudgets(t *testing.T) { + db, _ := integrationDB(t) + for _, mode := range []string{"success", "backpressure", "forbidden", "conflict", "unknown", "wrong_target", "invalid_receipt", "corrupt_request", "exhausted", "last_attempt", "deadline"} { + t.Run(mode, func(t *testing.T) { + s := testStore(t, db) + id := enqueue(t, s) + var posts atomic.Int32 + if mode == "corrupt_request" { + execSQL(t, db, `UPDATE sender_outbox SET request_json='{}' WHERE namespace_id=?`, s.namespace) + } + if mode == "exhausted" { + execSQL(t, db, `UPDATE sender_outbox SET send_count=100,lease_epoch=100 WHERE namespace_id=?`, s.namespace) + } + if mode == "last_attempt" { + execSQL(t, db, `UPDATE sender_outbox SET send_count=99,lease_epoch=99 WHERE namespace_id=?`, s.namespace) + } + if mode == "deadline" { + execSQL(t, db, `UPDATE sender_outbox SET send_deadline_at=TIMESTAMPADD(MICROSECOND,1,created_at) WHERE namespace_id=?`, s.namespace) + } + receipt := fakeReceipt(t, s) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == "GET" { + ns := s.namespace + if mode == "wrong_target" { + ns = testID(t) + } + json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": orbit.Queue{ID: s.queue, NamespaceID: ns}}) + return + } + posts.Add(1) + if r.Header.Get("Idempotency-Key") != "outbox-task-"+id { + t.Error("wrong submission identity") + } + switch mode { + case "backpressure": + w.Header().Set("Retry-After", "30") + w.WriteHeader(429) + json.NewEncoder(w).Encode(map[string]any{"request_id": id, "error": map[string]string{"code": "QUEUE_FULL", "message": "full"}}) + case "forbidden", "conflict": + code, status := "PERMISSION_DENIED", 403 + if mode == "conflict" { + code, status = "IDEMPOTENCY_CONFLICT", 409 + } + w.WriteHeader(status) + json.NewEncoder(w).Encode(map[string]any{"request_id": id, "error": map[string]string{"code": code, "message": "rejected"}}) + case "unknown", "last_attempt": + w.WriteHeader(502) + default: + reply := receipt + if mode == "invalid_receipt" { + reply.Task.NamespaceID = testID(t) + } + json.NewEncoder(w).Encode(map[string]any{"request_id": id, "data": reply}) + } + })) + defer server.Close() + api, _ := orbit.NewClient(server.URL, orbit.Options{Token: "test-token"}) + r, err := s.SendOnce(context.Background(), api) + if err != nil { + t.Fatal(err) + } + want := "REVIEW" + if mode == "success" { + want = "ACCEPTED" + } + if mode == "backpressure" || mode == "unknown" { + want = "PENDING" + } + if r.State != want { + t.Fatal(mode, r.State, r.LastCode) + } + if (mode == "wrong_target" || mode == "corrupt_request" || mode == "exhausted" || mode == "deadline") && posts.Load() != 0 { + t.Fatal("unsafe HTTP sent") + } + if mode == "backpressure" && (r.WakeAt == nil || time.Until(*r.WakeAt) < 29*time.Second) { + t.Fatal("Retry-After was ignored") + } + }) + } +} diff --git a/examples/task-outbox/internal/outbox/store_test.go b/examples/task-outbox/internal/outbox/store_test.go new file mode 100644 index 0000000..885365b --- /dev/null +++ b/examples/task-outbox/internal/outbox/store_test.go @@ -0,0 +1,63 @@ +package outbox + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "testing" + "time" + + orbit "git.yuntuo.hk/k2safe/orbit/sdk/go" +) + +func testID(t *testing.T) string { + t.Helper() + id, err := orbit.NewEventMessageID() + if err != nil { + t.Fatal(err) + } + return id +} +func testRequest(queue string) orbit.SubmitTaskRequest { + return orbit.SubmitTaskRequest{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"sequence":9007199254740993,"ref":"objects/test"}`), TaskTimeoutMS: 60000} +} +func TestFrozenDocumentAndFailureClassification(t *testing.T) { + s := &Store{namespace: testID(t), queue: testID(t)} + in := testRequest(s.queue) + in.TaskTimeoutMS = 0 + raw, err := s.document(in) + if err != nil { + t.Fatal(err) + } + var frozen orbit.SubmitTaskRequest + if json.Unmarshal(raw, &frozen) != nil || frozen.TaskTimeoutMS != 86400000 || frozen.Retry.MaxAttempts != 1 || frozen.Retry.AttemptTimeoutMS != 30000 || !bytes.Contains(frozen.Payload, []byte("9007199254740993")) { + t.Fatal("frozen defaults or precision changed") + } + in.QueueID = testID(t) + if _, err = s.document(in); !errors.Is(err, ErrInvalid) { + t.Fatal("cross queue request frozen") + } + in = testRequest(s.queue) + in.BusinessKey = "operation-\xff" + if _, err = s.document(in); !errors.Is(err, ErrInvalid) { + t.Fatal("invalid UTF-8 was silently rewritten") + } + for _, test := range []struct { + err error + state, code string + }{ + {context.DeadlineExceeded, "PENDING", "UNCONFIRMED"}, + {&orbit.APIError{StatusCode: 503, Code: "UNAVAILABLE"}, "PENDING", "UNCONFIRMED"}, + {&orbit.APIError{StatusCode: 429, RetryAfter: 30 * time.Second}, "PENDING", "BACKPRESSURE"}, + {&orbit.APIError{StatusCode: 403}, "REVIEW", "AUTHORIZATION"}, + {&orbit.APIError{StatusCode: 409, Code: "IDEMPOTENCY_CONFLICT"}, "REVIEW", "IDEMPOTENCY_CONFLICT"}, + {&orbit.HTTPError{StatusCode: 302}, "REVIEW", "HTTP_REJECTED"}, + {&orbit.HTTPError{StatusCode: 502}, "PENDING", "UNCONFIRMED"}, + } { + state, code, delay := classify(test.err, 100) + if state != test.state || code != test.code || delay < time.Second || delay > time.Hour { + t.Fatal(state, code, delay) + } + } +} diff --git a/scripts/test-mysql.sh b/scripts/test-mysql.sh index 5cd0d66..fcb57d3 100755 --- a/scripts/test-mysql.sh +++ b/scripts/test-mysql.sh @@ -83,6 +83,14 @@ export ORBIT_TEST_RECEIVER_DSN="root:$TASK_DB_PASSWORD@tcp(127.0.0.1:$TASK_DB_PO (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) +# 发送方业务表独立建库；示例不向核心SDK加入数据库依赖。 +ORBIT_MYSQL_DSN="$ORBIT_TEST_MYSQL_DSN" "$TASK_TEST_DIR/orbit-admin" bootstrap --application sdk-sender --namespace default --environment test --queue outbox --output "$TASK_TEST_DIR/sender.credentials.json" +"$TASK_DOCKER" exec "$TASK_TEST_NAME" sh -c 'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot --protocol=tcp -e "CREATE DATABASE orbit_sender_test CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin"' +export ORBIT_TEST_SENDER_DSN="root:$TASK_DB_PASSWORD@tcp(127.0.0.1:$TASK_DB_PORT)/orbit_sender_test" +(cd examples/task-outbox && GOWORK=off ORBIT_SENDER_DSN="$ORBIT_TEST_SENDER_DSN" "$TASK_GO" run ./cmd/sender init) +(cd examples/task-outbox && GOWORK=off ORBIT_TEST_SENDER_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_TEST_SENDER_CREDENTIALS_FILE="$TASK_TEST_DIR/sender.credentials.json" "$TASK_GO" test -race -count=1 -run Integration -v ./...) +(cd examples/task-outbox && GOWORK=off ORBIT_SENDER_DSN="$ORBIT_TEST_SENDER_DSN" ORBIT_URL="http://127.0.0.1:$TASK_HTTP_PORT" ORBIT_CREDENTIALS_FILE="$TASK_TEST_DIR/sender.credentials.json" "$TASK_GO" run ./cmd/sender 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" TASK_APP_ID=$(sed -n 's/.*"application_id":"\([^"]*\)".*/\1/p' "$TASK_TEST_DIR/app.credentials.json")