diff --git a/AGENTS.md b/AGENTS.md index f6aef7b..af9031a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,3 +11,7 @@ - 研发任务持续更新 docs/08-development-plan.md。接口、SDK、可运行示例和对应回归随功能同步交付。 - 模型/迁移变更运行真实 MySQL 集成测试；执行和恢复变更覆盖竞争、重复、迟到及故障场景。以实际测试报告判断完成，不把设计文档当作已实现能力。 - 编译和基础回归入口是 make test、make vet、make build；隔离数据库端到端回归入口是 make integration。 + +- 结果日志的序号在独立命名空间流行锁下分配，与任务状态及原事件同事务提交；禁止用最大事件ID或时间戳推断提交水位。生产者不得锁订阅行。 +- 创建订阅先锁队列再锁流行；消费按订阅→交付持锁，不能再取得任务、队列、额度或流行锁。ACK只验证最近代次及证明，过期未被替代的证明仍可确认；重复确认不能再次扣减pending_count。 +- 可靠订阅当前只完成持久化核心；HTTP/WSS、独立SDK、可运行Inbox示例和真实进程故障回归完成前，不标记P3订阅能力完成。 diff --git a/README.md b/README.md index f894b5f..63795d5 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Orbit 是面向多个业务项目的通用任务调度与流程编排平台， ## 当前进度 -已完成 P0–P2 阶段验收：33 个数据模型、版本化迁移、凭据及范围校验、队列与任务 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，剩余 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/09-result-subscription-core.md) 已实现持久化核心，HTTP/WSS 和 SDK 接入待完成；剩余 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/cmd/orbit-admin/main.go b/cmd/orbit-admin/main.go index 686b144..b1ae5ef 100644 --- a/cmd/orbit-admin/main.go +++ b/cmd/orbit-admin/main.go @@ -40,6 +40,7 @@ func run() error { flags.StringVar(&output, "output", "", "new private credentials JSON file") flags.BoolVar(&in.WorkerAccess, "worker-access", false, "also grant Worker connect and consume actions for development") flags.BoolVar(&in.ScheduleAccess, "schedule-access", false, "also grant schedule read and manage actions for development") + flags.BoolVar(&in.EventAccess, "event-access", false, "also grant result subscription read, manage and consume actions") if err := flags.Parse(os.Args[2:]); err != nil { return err } diff --git a/docs/07-data-model.md b/docs/07-data-model.md index 48ea143..743f124 100644 --- a/docs/07-data-model.md +++ b/docs/07-data-model.md @@ -1,6 +1,6 @@ # 数据模型与实现基线 -更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；订阅及流程编排继续开发。 +更新：2026-09-17。数据库统一禁止外键。当前已实现任务执行、安全重试、运行中取消、计划调度、队列限速/背压和应用配额；优先级及节点公平领取已实现；可靠订阅持久化核心已实现，HTTP/WSS 与 SDK 接入及流程编排继续开发。 ## 1. 模块与依赖 @@ -35,7 +35,7 @@ ## 3. 实体清单 -当前共 33 个实体：初始迁移包含 30 个，000003 增加进程会话头和迟到证据，000010 增加应用共享额度。迁移自身另有 `orbit_schema_migrations` 记录。 +当前共 35 个实体：初始迁移包含 30 个，000003 增加进程会话头和迟到证据，000010 增加应用共享额度，000013 增加结果流与不可变结果快照。迁移自身另有 `orbit_schema_migrations` 记录。 | 模块 | Go 模型 | 数据表 | | --- | --- | --- | @@ -45,6 +45,8 @@ | platform | `Credential` | `orbit_credentials` | | platform | `IdempotencyKey` | `orbit_idempotency_keys` | | platform | `Event` | `orbit_events` | +| platform | `EventStream` | `orbit_event_streams` | +| platform | `ResultEvent` | `orbit_result_events` | | platform | `Subscription` | `orbit_subscriptions` | | platform | `Delivery` | `orbit_deliveries` | | platform | `AuditEvent` | `orbit_audit_events` | @@ -73,7 +75,7 @@ | weave | `InboxEvent` | `weave_inbox` | | weave | `PendingAdvance` | `weave_pending_advances` | -设计文档中的 `task_events` / `result_events` 当前合并为 `orbit_events`，通过聚合类型、事件类型和状态版本区分；对外投递使用 `orbit_deliveries`，内部流程接续使用 `weave_inbox`。`weave_pending_advances` 保存暂停或分批推进后的续接检查点。 +`orbit_events` 保存原始状态事件，`orbit_result_events` 使用相同事件 ID 保存队列、提交序号和冻结快照，`orbit_event_streams` 串行分配命名空间内的结果序号；对外投递使用 `orbit_deliveries`，内部流程接续预留 `weave_inbox`。`weave_pending_advances` 保存暂停或分批推进后的续接检查点。 ## 4. 已落实的数据库约束 @@ -93,7 +95,7 @@ 禁止“先查后另开事务写入”。需要防止并发状态改变的关联行在事务内持锁，统一锁顺序并保留必要索引。当前不提供硬删除；将来增加清理时，必须核验在途任务、引用、幂等恢复窗口与审计保留期，按有限批次处理，不能依赖数据库级联。直接绕过服务写 SQL 不受应用校验保护。 -P1 已完成动作/队列授权、幂等摘要与原回执、受理事件及取消 CAS/审计事务；P3 扩展到运行中取消。流程及订阅目前只具备模型，后续实现也必须补齐事务内父资源归属验证，不能把模型存在当作已完成引擎。 +P1 已完成动作/队列授权、幂等摘要与原回执、受理事件及取消 CAS/审计事务；P3 扩展到运行中取消。订阅已实现服务层及持久化核心，包含事务内队列、所有者和交付归属校验；HTTP/WSS 和 SDK 尚未接入。流程目前仍只具备模型，后续引擎也必须补齐事务内父资源归属验证。 P2 服务端已新增会话隔离、两步派发、租约、并发许可、结果回执/证据和恢复扫描，见 [执行协议](api/worker.md)。底层及自动 Worker SDK、尝试/证据查询、Schema 与进程故障恢复已通过 [P2 阶段验收](reports/phase-2.md)。 @@ -146,3 +148,7 @@ P2 服务端已新增会话隔离、两步派发、租约、并发许可、结 000012_dispatch_priority 开放既有 Priority，范围 0–9，并增加对应 CHECK。新增只读生成列 DispatchAt（AvailableAt 减 Priority 乘30秒）及 `(namespace_id,queue_id,state,dispatch_at,available_at,id)` 排序索引；历史默认优先级 0 的排序时间等于 AvailableAt。Generated 字段随退避和恢复自动重新计算，不允许独立写入。33 个实体、274 个直接声明的持久化字段全部保留中文注释，新增 SQL 字段和 Priority 的 COMMENT 同步更新，无外键。 派发先检查实际到期、截止、能力、门控和资源键，再按排序时间选择。节点公平领取池根据认证后的 ApplicationID 分组，再按命名空间/队列轮转；内存队列和短期空探测缓存可在重启后重建，不保存授权事实。详细约束、SDK 和升级步骤见 [优先级与公平派发](api/priority-and-fairness.md)。 + +## 13. 可靠结果订阅核心 + +000013_result_subscriptions 新增 EventStream、ResultEvent，扩展 Subscription 和 Delivery。289 个直接声明字段均有中文注释。生产事务在独立流行锁下分配序号，与任务、原事件和结果快照一起提交；消费事务不锁流行、任务、队列或应用额度。交付记录和扫描位置一起落库，每条交付单独持有租约和证明。重复确认不重复扣减积压，耗尽保留记录，人工追加预算不重置历史。详见 [订阅核心](09-result-subscription-core.md)。对外 HTTP/WSS 与 SDK 尚未实现，P3 继续推进。 diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 539e5da..046d2c7 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 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅待开发 | +| P3 Pulse 完整能力 | 安全重试、延时、Cron/固定间隔、漏跑策略、计划版本、运行中取消、并发许可、配额/背压、可靠结果订阅 | 多实例领取/触发竞争、重试上限、取消竞争、队列隔离、积压恢复、事件重复交付 | 进行中：安全重试、计划调度、运行中取消、队列控制/限速/积压背压、跨命名空间应用配额、显式资源键许可及 SDK 场景已实现并通过全量回归；优先级及节点公平派发已实现并通过全量回归；可靠订阅持久化核心已实现，HTTP/WSS、SDK 及断线故障回归继续开发 | | P4 Weave 主流程 | SDK 定义构建与发布校验、实例、顺序/分支/结构化并行汇合、信号/定时等待、版本绑定 | 重启接续、早到/重复信号、并行竞争、内部事件重放、发布版本兼容；端到端通用流程 | 待开发 | | P5 Weave 恢复与查询 | 有界查询、暂停继续、取消、人工核查恢复、控制门控与审计 | 轮次与尝试独立预算；暂停/开始/回执/超时竞争；恢复保留预算；未知结果不盲目重复 | 待开发 | | P6 生产验证与运维 | 完整监控、凭据轮换、备份恢复、滚动升级、运行手册、可复现负载和故障工具 | 对照生产验收清单执行，报告环境/指标/遗留项；容量、RPO/RTO 按真实拓扑验证 | 待开发 | @@ -48,6 +48,7 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 | 跨命名空间应用配额 | first 受理 → second 全局满额 → 释放 → 原键成功 → 恢复配置 | P3，已实现完整示例 | | 跨队列共享资源 | 两队列同键 → 一个有效许可 → 递增序号 → 资源端拒绝旧写入 | P3，已实现完整示例 | | 优先级与应用隔离 | 近同时到期按优先级 → 旧任务等待补偿 → 应用/队列轮转 | P3，已实现完整示例 | +| 可靠结果订阅 | 冻结结果 → 持久化交付 → Inbox 去重 → 确认/补发 → 耗尽后人工重试 | P3，已实现服务与存储核心；HTTP/WSS 和 SDK 待接入 | | 并行生成结果 | 发布流程 → 并行执行 → 汇合 → 完成 | P4 | | 审核与外部回执 | 提交流程 → 等待信号 → 授权信号注入 → 继续 | P4 | | 外部作业轮询 | 提交一次 → 有界查询处理中状态 → 回执或最终结果 → 超限核查 | P5 | @@ -74,6 +75,8 @@ SDK、文档与示例随 P1–P5 同步开发，P7 是全量核对，不是最 ## 7. 文档入口 - [数据模型](07-data-model.md) +- [可靠结果订阅核心](09-result-subscription-core.md) +- [P3 订阅核心回归](reports/phase-3-result-subscription-core.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 new file mode 100644 index 0000000..41377dd --- /dev/null +++ b/docs/09-result-subscription-core.md @@ -0,0 +1,94 @@ +# 可靠结果订阅：持久化核心 + +状态：P3 的存储与服务层已实现并通过本地全量回归，见 [验证记录](reports/phase-3-result-subscription-core.md)。HTTP 管理接口、事件 WSS、独立 SDK 和可运行订阅示例尚未接入；不能把本页的方法当作已经发布的 SDK。对外接口接通后再冻结消息 Schema 和完整调用手册。 + +## 1. 已实现范围 + +结果来源限定为 task.succeeded、task.failed、task.cancelled、task.needs_review。安全重试产生的 task.requeued 不表示最终结果，不进入这个结果流。needs_review 表示需要核查，可能以后再产生权威结果，接入方不能把它当作成功或最终失败。 + +来源任务状态、原状态事件、不可变结果快照、结果流序号、原执行回执/审计在同一数据库事务中提交。任何一项失败都会回滚，Worker 不会收到“结果确认成功但没有可靠通知记录”的回执。 + +`orbit_events` 继续保存状态事件；新增 `orbit_result_events` 使用相同事件 ID 保存队列、提交序号和快照。快照包含 schema_version=1、task_id、queue_id、state、revision、result、effect_uncertain 和可空 current_attempt_id。不包含任务输入和租约秘密。结果使用 JSON 文本保存，保留大整数的原始数值文本。来源任务之后发生改变时，不会重新生成已发布的快照。 + +## 2. 为什么不能按最大事件 ID 推进 + +自增 ID、随机 ID 和事件创建时间都不能证明事务提交顺序。如果较小 ID 的事务还没有提交，消费者先看到较大的 ID 并移动游标，就可能永久漏掉较小的事件。 + +`orbit_event_streams` 为每个命名空间保存 last_sequence。生产事务持有独立流行锁，分配下一个序号并一起提交。后来事务在前一个提交或回滚前，不能获得更大的可见序号。回滚不消耗序号。 + +消费者读取已经提交的流头，并且只扫描不超过这个位置的日志。读取流头不加锁；订阅创建需要锁流行，以确定“从现在开始”的准确边界。该方案存在命名空间内的短暂串行分配热点，其实际吞吐和锁等待仍须在 P6 验证。 + +锁顺序： + +- 任务路径沿用认证/应用额度/队列/任务/尝试/许可的既定顺序，保存结果时最后取得流行锁；不访问订阅行。 +- 创建订阅按认证 → 幂等身份 → 队列共享锁 → 流行锁 → 新订阅/审计执行。 +- 消费、确认和管理按认证 → 必要的幂等身份 → 订阅 → 交付执行；不再取得任务、队列、应用额度或流行锁。 + +## 3. 订阅和授权 + +每个订阅绑定创建凭据和一个队列，事件类型集合在创建时冻结。名字限 1–128 个 ASCII 字符，以字母或数字开头，其余允许字母、数字、点、下划线和连字符；命名空间内唯一。首版每命名空间最多 64 个订阅，没有公开删除或更换所属凭据的能力。 + +| 字段 | 语义 | +| --- | --- | +| from_sequence | 订阅此序号之后的结果；省略时使用创建事务确定的最新提交序号；显式 0 表示从现存结果日志开始 | +| types | 四类结果的非空子集；省略或空数组使用全部四类，重复或其他类型被拒绝 | +| max_pending | 已物化但尚未确认的交付上限，省略或 0 使用 100，自定义范围 1–1000 | +| max_attempts | 每条新交付初始可发送次数，省略或 0 使用 20，自定义范围 1–100 | +| cursor_sequence | 已扫描并可靠建立交付记录的位置，**不是消费者已确认位置** | +| pending_count | PENDING、LEASED、EXPIRED 的总数；每次有效确认只扣减一次 | +| revision | 配置并发版本；物化和积压数量变化不增加该版本 | + +新动作独立授权：events:manage、events:read、events:consume。每个操作同时检查凭据有效性、所属应用/命名空间启用、队列范围和订阅所有者。同一命名空间的另一张凭据，也不能读取、领取、确认或管理该订阅。管理权限不能隐含消费权限，任务读取权限不能隐含订阅权限。 + +bootstrap 新增显式 --event-access，用于签发包含这三项动作的开发凭据；不会给既有凭据自动扩权。轮换后的安全重新绑定列入 P6，当前不能将旧凭据所属订阅自动转移。 + +创建、暂停/继续、人工重试都有持久化幂等身份及原回执。身份作用域包含所属凭据与目标资源，复合身份使用 SHA-256 编码到 64 字符，避免跨凭据回执泄漏或数据库截断。 + +## 4. 从结果日志到交付记录 + +每次领取在订阅锁下最多扫描 100 条本队列已提交结果，再按冻结类型过滤。选中的事件先创建唯一的 `(namespace_id, subscription_id, event_id)` 交付，随后推进 cursor_sequence，两者同事务提交。 + +达到 max_pending 后停止物化。未物化的结果保留在来源日志；任务生产者不锁订阅、也不等待消费者腾出交付窗口。有效确认释放空间后，后续扫描继续追赶。每条交付独立确认，处理顺序可以不同，不靠单个最大 ACK 序号跳过中间事件。 + +这是有界交付窗口，不是“所有存储都已限制大小”。首版结果日志和已确认交付不自动删除；持久化积压仍占数据库空间。清理、水位推进、容量告警及运维治理需要后续 P6 完成。 + +## 5. 租约、确认与恢复 + +| 状态 | 含义与转换 | +| --- | --- | +| PENDING | 等待就绪；领取时持久化租约后转 LEASED | +| LEASED | 已生成一次发送授权；有效确认转 ACKED，超时后按预算转 PENDING 或 EXPIRED | +| ACKED | 已可靠确认；重复确认返回原确认时间，不再次扣减 pending_count，也不重复写确认审计 | +| EXPIRED | 已耗尽交付预算，仍保留事件和积压占用；允许带原因的人工追加预算，或接受尚未被替换的最近证明的迟到确认 | + +租约固定 30 秒，每次领取增加 lease_epoch 和 attempts，并产生新的 32 字节随机秘密。数据库只保存摘要；确认时使用常量时间摘要比较。当前内部领取按服务端连接 ID 计算容量，单连接容量 1–32。连接 ID 由后续传输层生成，不能接受客户端自报来绕过容量。 + +下一次启用状态下的领取，会先回收该订阅的过期交付。未耗尽时使用 `min(2^(attempts-1), 30)` 秒退避；耗尽时转 EXPIRED，不无限自动发送。当前没有独立订阅清扫进程：无人消费或订阅暂停期间，租约状态可能保持 LEASED，直至下一次领取执行恢复。 + +ACK 表示消费者已可靠保存或处理结果，不授予新业务执行权。当前最后一次秘密和代次即使过期，只要没有被新的交付替换，仍可用于迟到确认；新交付产生后，旧证明统一返回 DELIVERY_LEASE_LOST。确认和重新交付竞争通过订阅锁串行化，只会有一方有效。 + +服务端不能保证消费者业务恰好执行一次。后续 SDK 示例需展示接收方以 event_id 建立唯一 Inbox，在本地事务中保存结果/业务变更，提交后才 ACK。断线和 ACK 响应丢失可能造成重复投递，调用方不得按“收到过一次”就直接丢弃尚未可靠保存的内容。 + +每次交付授权、首次有效确认、订阅管理和人工重试均记录审计。交付审计包含事件标识、代次、次数、连接和租约截止，不记录交付秘密或结果正文。 + +## 6. 管理与人工重试 + +暂停只停止新的物化和领取，不撤回已经发出的结果。暂停期间仍接受有效 ACK。恢复后继续处理原有积压，事件和交付 ID 不变。 + +人工重试仅适用于 EXPIRED，必须提供 expected_revision、非空 reason 和 additional_attempts（1–100）。追加后的累计预算最多 1000000；已使用次数、当前代次和原事件身份均不归零。新领取才产生新代次。相同幂等键重报返回原响应，不重复加预算。 + +当前服务层方法见 [platform/events/service.go](../platform/events/service.go)，持久化实现见 [订阅](../internal/storage/mysql/event_subscriptions.go)、[交付](../internal/storage/mysql/event_deliveries.go) 和 [结果日志](../internal/storage/mysql/result_journal.go)。这些是服务端内部模块，独立 SDK 不应直接依赖它们。 + +## 7. 迁移与历史边界 + +[000013](../internal/storage/mysql/migrations/000013_result_subscriptions.up.sql) 新增两个实体，并扩展订阅和交付字段。当前共 35 个实体、289 个直接声明的持久化字段，全部中文注释；新增或调整的 SQL 列提供中文 COMMENT。没有新增外键或级联删除。 + +升级必须停止旧版本写入节点，应用迁移，再启动支持新结果日志的版本。旧的运行中写入代码不会补写结果日志，不能在启用订阅后继续混跑。迁移沿用显式版本、校验和、dirty 标记和就绪检查，不自动修改历史迁移。 + +迁移前的 orbit_events 原样保留，不补造无法证明正确的旧结果快照。from_sequence=0 只覆盖迁移启用后的结果日志，不代表系统建立以来全部任务。历史占位订阅缺少可靠队列绑定，升级后保留但禁用，不对外消费；历史交付保留原次数和代次，预算至少等于已用次数。 + +retained_from_sequence 初始且目前固定为 1。未来实现清理后，如果订阅游标落在保留边界以前，必须返回 EVENT_HISTORY_EXPIRED，不得静默跳到最新结果。请求未来序号也会被拒绝。 + +## 8. 下一步 + +接入 HTTP 创建/查询/分页、暂停/继续、交付与耗尽记录查询、人工重试；实现 orbit.events.v1 WSS 的认证、hello、交付、ACK、心跳、连接上限、背压和退出；补独立 Go SDK、Inbox 去重示例、严格消息 Schema 和完整接口手册。随后做断路、ACK 响应丢失、节点进程强杀及 SDK 端到端回归，再验收 P3 的可靠订阅能力。 diff --git a/docs/api/README.md b/docs/api/README.md index e166756..4619c6e 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -20,7 +20,7 @@ | 任务冻结策略与 Worker `retryable` | P3 安全重试已实现 | [安全重试](retries.md) | | `/v1/tasks/{task_id}:cancel` 与 `task.cancel` 通知 | P3 运行中取消已实现 | [取消及 SDK 场景](cancellation.md) | | `/v1/schedules`、不可变版本、暂停/恢复及触发查询 | P3 已实现 | [计划接口与 SDK 场景](schedules.md) | -| `/v1/events/connect` | 计划 P3，尚未实现 | [协议草案](../03-http-websocket-protocol.md) | +| `/v1/events/connect` 及订阅管理 | 持久化核心已实现；HTTP/WSS 和 SDK 尚未接入 | [订阅核心及边界](../09-result-subscription-core.md) | | `/v1/workflows` 与流程管理 | 计划 P4–P5，尚未实现 | [流程接口规划](workflows.md) | 未实现路由返回 404。健康检查就绪不表示任务执行、定时调度或流程引擎已经可用。 diff --git a/docs/api/authentication.md b/docs/api/authentication.md index ebfe4bc..e7cb53a 100644 --- a/docs/api/authentication.md +++ b/docs/api/authentication.md @@ -57,3 +57,5 @@ P3 增加 schedules:read 和 schedules:manage。bootstrap 显式传 --schedule-a P2 增加 workers:connect 和 workers:consume。bootstrap 只有显式提供 --worker-access 时才同时授予这两项开发权限；现有凭据不自动变化。Worker 还需独立的进程恢复秘密和尝试秘密，见 [Worker 契约](worker.md)。 P3 应用额度管理新增 `application:resources:read` 和 `application:resources:manage`。只有显式 `--resource-access` 才授予，作用域是所属应用的全部命名空间，普通队列管理不能替代。查询返回聚合计数，不开放其他命名空间的任务数据。`bootstrap --application-id <现有应用ID>` 可由数据库管理员增加命名空间，与 `--application` 二选一；验证应用启用及额度记录存在后，同事务创建新资源，所有秘密交付约定保持不变。应用默认未终结上限 1000000、并发 10000、开始速率 100000/秒；详见 [应用共享配额](application-resources.md)。 + +P3 结果订阅核心新增 `events:read`、`events:manage` 和 `events:consume`，bootstrap 仅显式 `--event-access` 时授予。订阅绑定创建凭据与单个队列，不能用同命名空间的另一张凭据消费或管理。当前 HTTP/WSS 和独立 SDK 尚未接入，详见 [订阅核心边界](../09-result-subscription-core.md)。 diff --git a/docs/reports/phase-3-result-subscription-core.md b/docs/reports/phase-3-result-subscription-core.md new file mode 100644 index 0000000..d3219c6 --- /dev/null +++ b/docs/reports/phase-3-result-subscription-core.md @@ -0,0 +1,80 @@ +# P3 可靠结果订阅持久化核心回归 + +验证日期：2026-09-17。本阶段完成结果日志、订阅和交付的服务/存储核心；HTTP/WSS、独立 SDK、Inbox 示例和订阅进程故障用例尚未交付。P3 与持续开发目标继续进行，不视为完整订阅能力验收。 + +## 1. 实现内容 + +[核心设计与边界](../09-result-subscription-core.md) 对应以下实现： + +- [迁移 000013](../../internal/storage/mysql/migrations/000013_result_subscriptions.up.sql) 增加命名空间结果流和不可变结果快照，扩展订阅窗口、游标、交付预算及证明摘要。最终 Schema 无外键，模型和 SQL 字段注释同步。 +- [结果日志](../../internal/storage/mysql/result_journal.go) 接入原状态事件事务；成功、已知失败、取消和需要核查四类结果同事务追加快照。每个命名空间独立串行分配序号，不按随机 ID 或最大自增 ID 推断提交顺序。 +- [订阅服务](../../platform/events/service.go) 和 [订阅存储](../../internal/storage/mysql/event_subscriptions.go) 实现范围权限、凭据所有者、单队列冻结过滤、历史边界、幂等创建、暂停/继续、64 个订阅上限。 +- [交付存储](../../internal/storage/mysql/event_deliveries.go) 实现有限批次物化、未确认窗口、30 秒租约、连接容量、到期退避、次数耗尽保留、最新证明隔离、重复/迟到确认和人工追加预算。每次授权及管理确认有审计，秘密只保存摘要。 +- 数据模型目前共 35 个实体、289 个直接声明的持久化字段；逐字段中文注释检查无遗漏。SQL 模型映射由真实数据库回归校验。 + +## 2. 环境与命令 + +macOS 本机、Go 1.27.1、Docker MySQL 8.4.10、一次性数据库、回环 HTTP/WSS。使用随机测试凭据，测试清理自己的容器；不连接既有业务数据库。 + +```sh +make fmt test vet build +make integration +cd sdk/go +GOWORK=off go test -race ./... +``` + +上述检查均通过。模型、接口、SDK、示例与故障检查分别执行；没有数据库环境的普通 go test 会跳过集成，不能替代本次真实数据库验证。 + +| 分组 | 本次结果 | +| --- | --- | +| 全部 MySQL 集成 | PASS，85.486 秒 | +| 新增订阅核心 15 组场景 | PASS，3.49 秒 | +| 新增提交顺序/创建边界 3 组 | PASS，0.63 秒 | +| v12 订阅迁移升级 | PASS，3.79 秒 | +| SDK Worker 协议及运行器 | PASS，89.166 秒 | +| SDK 计划生命周期 | PASS，5.626 秒 | +| SDK 队列 / 应用资源 | PASS，4.501 / 1.386 秒 | +| 既有七项真实进程故障 | PASS，28.528 秒 | +| 既有十二个完整 SDK 示例 | 全部通过 | +| fmt / race / vet / build / 独立 SDK 模块 | 全部通过 | + +耗时仅记录本地回归，不代表生产吞吐、SLO、容量或高可用结论。 + +## 3. 新增正确性和竞争证据 + +[订阅数据库用例](../../internal/storage/mysql/event_subscriptions_integration_test.go) 验证： + +1. 来源重复回报不会重复追加结果；省略游标从创建边界之后开始，显式 0 可回放现存结果日志。类型过滤会安全推进扫描位置。 +2. 结果 9007199254740993 保留原数值文本，`<&>` 不被无谓扩大；后续改动任务结果不会改变已保存快照。数据库拒绝 UPDATE 结果日志。 +3. 不同命名空间、同命名空间另一凭据、移除队列范围、缺少独立动作及凭据撤销均被阻止。同一个幂等键不会跨所有者泄漏原回执。 +4. 未确认窗口为 1 时，仍可提交后续 3 条结果；确认后按日志逐条追赶，无跳过或重复扣减。持有订阅行锁期间，独立任务取消仍在限定时间内提交结果。 +5. 12 个消费者争抢同一条交付，只有一个获得有效租约。12 次并发确认中只有一次首次生效，所有响应保留相同确认时间，只有一条确认审计。 +6. 丢弃存储服务对象后，重新创建服务仍从数据库恢复相同交付和事件身份。租约到期后退避，新的发送增加次数和代次并更换秘密；旧证明失效。 +7. 次数耗尽后保留 EXPIRED 及积压占用。最新证明未被新代次替换时允许迟到确认，重复确认返回原回执。 +8. 人工重试增加预算而不重置次数、代次或事件身份；相同管理身份重报不重复追加。错误状态或版本被拒绝。 +9. 暂停阻止新发送，但允许在途 ACK；恢复后继续接收后续结果。单连接容量在数据库事务内执行，确认后释放容量。 +10. 12 次迟到 ACK 与新领取竞争，最终只存在一个有效确认，旧证明不能确认新代次，也不会泄漏窗口额度。 +11. 注入交付 INSERT 失败，交付记录、扫描位置和 pending_count 一起回滚；注入审计 INSERT 失败，ACK、确认时间和计数一起回滚。 +12. 注入结果日志 INSERT 失败，任务状态、原状态事件、流序号及应用额度一起回滚；原请求身份可以重报成功。 +13. 安全重试不产生最终结果通知；最终失败、UNKNOWN 及执行失联恢复的核查结果进入结果日志。 +14. 未来游标、缺失历史和订阅数量超限分别返回明确错误。秘密原文不进入确认审计。 + +提交顺序用例显式保持第一个生产事务未提交，同时启动另一个事务。消费者看不到越过未提交事务的较大序号，且不等待生产者流锁。首事务提交时依次获得 1、2；首事务回滚时后一个获得 1，无非事务性空洞。创建“从现在开始”的订阅必须等前一个生产事务完成，并正确捕获最终提交边界。 + +回归修正了首次物化后的时钟边界及复合幂等作用域超过既有 64 字符的问题，最终完整套件重新通过。 + +## 4. 升级和无外键证据 + +[v12 专项升级](../../internal/storage/mysql/result_subscription_upgrade_integration_test.go) 重建真实旧 Schema，保留已有任务事件及数值精度；旧占位订阅禁用且 queue_id 为空，不自动取得消费权限。旧交付已使用 25 次、代次 25，升级后原值不变，预算至少覆盖已用次数。重复迁移安全，未凭空生成历史结果快照。 + +[v7 无外键升级](../../internal/storage/mysql/no_foreign_keys_integration_test.go) 继续验证原有 70 个外键移除后为 0，325 项原索引列定义保留、任务数据不变，手工引入外键时启动检查拒绝。新结果表和交付扫描索引作为新增内容单独排除于“原索引不变”比较；没有跳过原索引的校验。 + +v4 安全重试及 v9 应用配额回填和在途释放回归仍通过。当前迁移新增结果日志，需要停止旧写入节点后升级；不支持旧二进制持续写入但遗漏新日志的混跑部署。 + +## 5. 后续验收边界 + +- 尚未开放订阅 HTTP/WSS 路由，未声称新订阅能由独立 SDK 调用。本次十二个示例及七项强杀故障覆盖的是既有能力。 +- 本次“恢复服务对象”只验证交付权威状态不在内存，不能替代订阅 WSS 断路、ACK 响应丢失或真实节点进程强杀；这些将在传输层和 SDK 接入后验证。 +- 首版按领取触发该订阅的租约恢复，没有独立订阅后台清扫。暂停或没有消费者时，过期 LEASED 可能延迟转为 EXPIRED。 +- 未确认交付窗口有限，来源日志和已确认历史目前无自动清理，积压仍消耗存储。生产容量告警、保留/清理、监控及凭据轮换属于 P6。 +- 命名空间流锁热点、结果大小和数据库写放大，以及 MySQL 故障切换、跨故障域部署、RPO/RTO 仍需真实目标环境测试。 diff --git a/internal/storage/mysql/application_resources_upgrade_integration_test.go b/internal/storage/mysql/application_resources_upgrade_integration_test.go index 6b4dffb..3e84394 100644 --- a/internal/storage/mysql/application_resources_upgrade_integration_test.go +++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go @@ -65,6 +65,9 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) { assertApplicationCounts(t, f, 4, 2, 1) // 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。 f.exec(`DROP TABLE orbit_application_resources`) + f.exec(`DROP TABLE orbit_result_events,orbit_event_streams`) + f.exec(`ALTER TABLE orbit_subscriptions DROP CHECK ck_subscription_window, DROP COLUMN queue_id, DROP COLUMN cursor_sequence, DROP COLUMN pending_count, DROP COLUMN max_pending, DROP COLUMN max_attempts`) + f.exec(`ALTER TABLE orbit_deliveries DROP INDEX ix_delivery_subscription_due, DROP INDEX ix_delivery_subscription_lease, DROP CHECK ck_delivery_budget, DROP COLUMN attempt_budget, DROP COLUMN token_digest`) f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_priority_dispatch, DROP CHECK ck_task_priority, DROP COLUMN dispatch_at`) f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_concurrency_key, DROP COLUMN concurrency_key, DROP COLUMN concurrency_key_digest`) f.exec(`ALTER TABLE pulse_attempts DROP COLUMN concurrency_token`) diff --git a/internal/storage/mysql/bootstrap.go b/internal/storage/mysql/bootstrap.go index ec671e7..497b38f 100644 --- a/internal/storage/mysql/bootstrap.go +++ b/internal/storage/mysql/bootstrap.go @@ -14,6 +14,7 @@ import ( // BootstrapInput 仅供持有数据库管理权限的初始化命令使用，不暴露为公共 HTTP。 type BootstrapInput struct { + EventAccess bool // EventAccess 显式增加结果订阅查询、管理和消费权限。 Application string // Application 新应用名称，与 ExistingApplicationID 二选一。 Namespace string // Namespace 命名空间名称。 Environment string // Environment 环境名称，必须显式指定。 @@ -80,6 +81,9 @@ func Bootstrap(ctx context.Context, db *sql.DB, in BootstrapInput, deliver func( if in.ResourceAccess { permissions.Actions = append(permissions.Actions, "application:resources:read", "application:resources:manage") } + if in.EventAccess { + permissions.Actions = append(permissions.Actions, "events:read", "events:manage", "events:consume") + } raw, err := json.Marshal(permissions) if err != nil { return err diff --git a/internal/storage/mysql/event_deliveries.go b/internal/storage/mysql/event_deliveries.go new file mode 100644 index 0000000..3568288 --- /dev/null +++ b/internal/storage/mysql/event_deliveries.go @@ -0,0 +1,311 @@ +package mysql + +import ( + "context" + "crypto/subtle" + "database/sql" + "encoding/json" + "errors" + "fmt" + "math" + "slices" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/platform/events" + platform "git.yuntuo.hk/k2safe/orbit/platform/model" +) + +// materializeResults 持有订阅锁，流和日志只做已提交读，不反向锁生产者。 +// 先保存独立交付记录再推进扫描序号；未确认窗口满时停止扫描，保留来源日志。 +func materializeResults(ctx context.Context, tx *sql.Tx, ns core.ID, sub *events.Subscription) error { + if sub.PendingCount >= sub.MaxPending { + return nil + } + var head, retained uint64 + if err := tx.QueryRowContext(ctx, `SELECT last_sequence,retained_from_sequence FROM orbit_event_streams WHERE namespace_id=?`, ns).Scan(&head, &retained); err != nil { + return err + } + if sub.CursorSequence < retained-1 { + return events.HistoryExpired + } + limit := min(uint32(100), sub.MaxPending-sub.PendingCount) + rows, err := tx.QueryContext(ctx, `SELECT r.id,r.sequence,e.type FROM orbit_result_events r LEFT JOIN orbit_events e ON e.namespace_id=r.namespace_id AND e.id=r.id WHERE r.namespace_id=? AND r.queue_id=? AND r.sequence>? AND r.sequence<=? ORDER BY r.sequence LIMIT ?`, ns, sub.QueueID, sub.CursorSequence, head, limit) + if err != nil { + return err + } + type candidate struct { + id core.ID + sequence uint64 + eventType string + } + batch := make([]candidate, 0, limit) + for rows.Next() { + var c candidate + if err = rows.Scan(&c.id, &c.sequence, &c.eventType); err != nil { + rows.Close() + return err + } + batch = append(batch, c) + } + err = rows.Err() + rows.Close() + if err != nil { + return err + } + cursor := sub.CursorSequence + pending := sub.PendingCount + for _, c := range batch { + cursor = c.sequence + if !slices.Contains(sub.Types, c.eventType) { + continue + } + id, err := core.NewID() + if err != nil { + return err + } + if _, err = tx.ExecContext(ctx, `INSERT INTO orbit_deliveries(namespace_id,id,subscription_id,event_id,available_at,attempt_budget) VALUES (?,?,?,?,UTC_TIMESTAMP(6),?)`, ns, id, sub.ID, c.id, sub.MaxAttempts); err != nil { + return err + } + pending++ + } + if len(batch) < int(limit) { + cursor = head + } + if cursor == sub.CursorSequence && pending == sub.PendingCount { + return nil + } + if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET cursor_sequence=?,pending_count=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, cursor, pending, ns, sub.ID); err != nil { + return err + } + sub.CursorSequence, sub.PendingCount = cursor, pending + return nil +} + +// recoverDeliveries 在新领取前有界回收过期租约。耗尽预算仍占积压，禁止静默丢弃。 +// 最后一次证明保留；迟到确认只在还没有新交付代次时有效。 +func recoverDeliveries(ctx context.Context, tx *sql.Tx, ns, sub core.ID, now time.Time) error { + _, err := tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state=IF(attempts>=attempt_budget,'EXPIRED','PENDING'),available_at=TIMESTAMPADD(SECOND,LEAST(30,POW(2,LEAST(attempts-1,5))),?),lease_owner=NULL,lease_until=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND state='LEASED' AND lease_until<=?`, now, now, ns, sub, now) + return err +} + +type storedDelivery struct { + view events.Delivery + digest []byte +} + +func readDelivery(ctx context.Context, tx *sql.Tx, ns, sub, id core.ID, lock bool) (storedDelivery, error) { + var out storedDelivery + q := `SELECT id,subscription_id,event_id,revision,state,available_at,lease_epoch,lease_until,attempts,attempt_budget,acknowledged_at,token_digest FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND id=?` + if lock { + q += ` FOR UPDATE` + } + v := &out.view + err := tx.QueryRowContext(ctx, q, ns, sub, id).Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &out.digest) + if errors.Is(err, sql.ErrNoRows) { + return out, fault.NotFound + } + if err != nil { + return out, err + } + err = tx.QueryRowContext(ctx, `SELECT sequence FROM orbit_result_events WHERE namespace_id=? AND id=?`, ns, v.EventID).Scan(&v.Sequence) + return out, err +} +func readResultEvent(ctx context.Context, tx *sql.Tx, ns, id core.ID) (events.Event, error) { + var out events.Event + var details, snapshot []byte + err := tx.QueryRowContext(ctx, `SELECT r.id,r.sequence,e.type,e.aggregate_kind,e.aggregate_id,e.aggregate_revision,e.occurred_at,e.payload,r.snapshot FROM orbit_result_events r LEFT JOIN orbit_events e ON e.namespace_id=r.namespace_id AND e.id=r.id WHERE r.namespace_id=? AND r.id=?`, ns, id).Scan(&out.ID, &out.Sequence, &out.Type, &out.AggregateKind, &out.AggregateID, &out.AggregateRevision, &out.OccurredAt, &details, &snapshot) + if err != nil { + return out, err + } + out.Details = details + err = json.Unmarshal(snapshot, &out.Snapshot) + return out, err +} +func (s *EventStore) GetDelivery(ctx context.Context, t auth.Token, sub, id core.ID) (events.Delivery, error) { + var out events.Delivery + err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error { + if _, err := readSubscription(ctx, tx, t, p, sub, false); err != nil { + return err + } + d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, false) + out = d.view + return err + }) + return out, err +} +func (s *EventStore) ClaimDelivery(ctx context.Context, t auth.Token, id core.ID, in events.ClaimInput) (*events.Grant, error) { + var out *events.Grant + err := s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error { + sub, err := readSubscription(ctx, tx, t, p, id, true) + if err != nil { + return err + } + if !sub.Enabled { + return nil + } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + if err = recoverDeliveries(ctx, tx, t.NamespaceID, id, now); err != nil { + return err + } + if err = materializeResults(ctx, tx, t.NamespaceID, &sub); err != nil { + return err + } + // 物化记录的 available_at 来自写入时的数据库时钟；不能继续用物化前的时间领取。 + now, err = dbNow(ctx, tx) + if err != nil { + return err + } + var active uint32 + if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state='LEASED' AND lease_owner=?`, t.NamespaceID, id, in.ConnectionID).Scan(&active); err != nil { + return err + } + if active >= in.Capacity { + return nil + } + var deliveryID core.ID + err = tx.QueryRowContext(ctx, `SELECT id FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state='PENDING' AND available_at<=? ORDER BY available_at,id LIMIT 1 FOR UPDATE`, t.NamespaceID, id, now).Scan(&deliveryID) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return err + } + stored, err := readDelivery(ctx, tx, t.NamespaceID, id, deliveryID, false) + if err != nil { + return err + } + v := stored.view + if v.Attempts >= v.AttemptBudget || v.LeaseEpoch == math.MaxUint64 || v.Revision == math.MaxUint64 { + return fault.Conflict + } + secret, digest, err := events.NewSecret() + if err != nil { + return err + } + until := now.Add(events.LeaseDuration) + if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='LEASED',lease_owner=?,lease_epoch=lease_epoch+1,lease_until=?,token_digest=?,attempts=attempts+1,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND id=?`, in.ConnectionID, until, digest, now, t.NamespaceID, id, deliveryID); err != nil { + return err + } + v.State = platform.DeliveryLeased + v.LeaseEpoch++ + v.LeaseUntil = &until + v.Attempts++ + v.Revision++ + event, err := readResultEvent(ctx, tx, t.NamespaceID, v.EventID) + if err != nil { + return err + } + operation, err := core.NewID() + if err != nil { + return err + } + if err = appendAudit(ctx, tx, t, operation, v.ID, "delivery", "delivery.lease", "grant durable result delivery", nil, map[string]any{"event_id": v.EventID, "epoch": fmt.Sprint(v.LeaseEpoch), "attempts": v.Attempts, "lease_until": until, "connection_id": in.ConnectionID}); err != nil { + return err + } + out = &events.Grant{Delivery: v, Event: event, LeaseSecret: secret} + return nil + }) + if err != nil { + return nil, err + } + return out, nil +} +func (s *EventStore) AckDelivery(ctx context.Context, t auth.Token, id core.ID, in events.AckInput) (events.AckResult, error) { + var out events.AckResult + digest, err := events.SecretDigest(in.LeaseSecret) + if err != nil { + return out, err + } + err = s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error { + sub, err := readSubscription(ctx, tx, t, p, id, true) + if err != nil { + return err + } + d, err := readDelivery(ctx, tx, t.NamespaceID, id, in.DeliveryID, true) + if err != nil { + return err + } + v := d.view + if v.LeaseEpoch != in.Epoch || subtle.ConstantTimeCompare(d.digest, digest[:]) != 1 { + return events.LeaseLost + } + out = events.AckResult{DeliveryID: v.ID, EventID: v.EventID, Epoch: in.Epoch} + if v.State == platform.DeliveryAcked { + if v.AcknowledgedAt == nil { + return errors.New("acked delivery lacks confirmation timestamp") + } + out.AcknowledgedAt = *v.AcknowledgedAt + out.Deduplicated = true + return nil + } + if sub.PendingCount == 0 || v.Revision == math.MaxUint64 { + return errors.New("delivery confirmation invariant failed") + } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='ACKED',lease_owner=NULL,lease_until=NULL,acknowledged_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND id=?`, now, now, t.NamespaceID, id, v.ID); err != nil { + return err + } + if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET pending_count=pending_count-1,updated_at=? WHERE namespace_id=? AND id=?`, now, t.NamespaceID, id); err != nil { + return err + } + operation, err := core.NewID() + if err != nil { + return err + } + if err = appendAudit(ctx, tx, t, operation, v.ID, "delivery", "delivery.ack", "consumer confirmed durable processing", map[string]any{"state": v.State}, map[string]any{"state": platform.DeliveryAcked, "event_id": v.EventID, "epoch": fmt.Sprint(in.Epoch)}); err != nil { + return err + } + out.AcknowledgedAt = now + return nil + }) + return out, err +} +func (s *EventStore) RetryDelivery(ctx context.Context, t auth.Token, sub, id core.ID, key string, in events.RetryInput) (events.RetryResult, error) { + var out events.RetryResult + err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error { + identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "delivery.retry", eventScope(t.CredentialID, sub, id), key, "delivery", id, in) + if err != nil { + return err + } + if _, err = readSubscription(ctx, tx, t, p, sub, true); err != nil { + return err + } + if cached != nil { + if err = json.Unmarshal(cached, &out); err != nil { + return err + } + out.Deduplicated = true + return nil + } + d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, true) + if err != nil { + return err + } + v := d.view + if v.State != platform.DeliveryExpired || v.Revision != in.ExpectedRevision || v.Revision == math.MaxUint64 || uint64(v.AttemptBudget)+uint64(in.AdditionalAttempts) > 1000000 { + return fault.Conflict + } + if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='PENDING',attempt_budget=attempt_budget+?,available_at=UTC_TIMESTAMP(6),revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND subscription_id=? AND id=?`, in.AdditionalAttempts, t.NamespaceID, sub, id); err != nil { + return err + } + d, err = readDelivery(ctx, tx, t.NamespaceID, sub, id, false) + if err != nil { + return err + } + out.Delivery = d.view + if err = appendAudit(ctx, tx, t, identity, id, "delivery", "delivery.retry", in.Reason, v, out.Delivery); err != nil { + return err + } + return saveIdentity(ctx, tx, t.NamespaceID, identity, out) + }) + return out, err +} diff --git a/internal/storage/mysql/event_subscriptions.go b/internal/storage/mysql/event_subscriptions.go new file mode 100644 index 0000000..12070f9 --- /dev/null +++ b/internal/storage/mysql/event_subscriptions.go @@ -0,0 +1,177 @@ +package mysql + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "math" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/platform/events" + driver "github.com/go-sql-driver/mysql" +) + +// EventStore 管理独立结果订阅；消费者事务不锁任务、队列、应用额度或结果流行。 +// 创建时按认证→请求身份→队列共享锁→结果流锁顺序确定提交边界。 +type EventStore struct{ *TaskStore } + +func NewEventStore(db *sql.DB) *EventStore { return &EventStore{NewTaskStore(db)} } + +var _ events.Store = (*EventStore)(nil) + +const subscriptionColumns = `id,queue_id,credential_id,name,revision,enabled,filter,cursor_sequence,pending_count,max_pending,max_attempts,created_at,updated_at` + +func readSubscription(ctx context.Context, tx *sql.Tx, t auth.Token, p auth.Permissions, id core.ID, lock bool) (events.Subscription, error) { + var out events.Subscription + var raw []byte + q := `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND id=? AND credential_id=? AND queue_id IS NOT NULL` + if lock { + q += ` FOR UPDATE` + } + err := tx.QueryRowContext(ctx, q, t.NamespaceID, id, t.CredentialID).Scan(&out.ID, &out.QueueID, &out.CredentialID, &out.Name, &out.Revision, &out.Enabled, &raw, &out.CursorSequence, &out.PendingCount, &out.MaxPending, &out.MaxAttempts, &out.CreatedAt, &out.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return out, fault.NotFound + } + if err != nil { + return out, err + } + if !p.AllowsQueue(out.QueueID) { + return events.Subscription{}, fault.NotFound + } + if err = json.Unmarshal(raw, &out.Types); err != nil { + return out, err + } + return out, nil +} + +func (s *EventStore) CreateSubscription(ctx context.Context, t auth.Token, key string, in events.CreateInput) (events.SubscriptionResult, error) { + var out events.SubscriptionResult + err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error { + if !p.AllowsQueue(in.QueueID) { + return fault.Forbidden + } + id, err := core.NewID() + if err != nil { + return err + } + identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "subscription.create", string(t.CredentialID), key, "subscription", id, in) + if err != nil { + return err + } + if cached != nil { + if err = json.Unmarshal(cached, &out); err != nil { + return err + } + out.Deduplicated = true + return nil + } + var queue core.ID + if err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_queues WHERE namespace_id=? AND id=? FOR SHARE`, t.NamespaceID, in.QueueID).Scan(&queue); errors.Is(err, sql.ErrNoRows) { + return fault.NotFound + } else if err != nil { + return err + } + head, retained, err := lockEventStream(ctx, tx, t.NamespaceID) + if err != nil { + return err + } + cursor := head + if in.FromSequence != nil { + cursor = *in.FromSequence + } + if cursor > head { + return fault.Invalid("from_sequence is ahead of committed history") + } + if cursor < retained-1 { + return events.HistoryExpired + } + var count int + if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_subscriptions WHERE namespace_id=?`, t.NamespaceID).Scan(&count); err != nil { + return err + } + if count >= events.MaxSubscriptions { + return events.SubscriptionLimit + } + filter, err := json.Marshal(in.Types) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO orbit_subscriptions(namespace_id,id,name,credential_id,filter,queue_id,cursor_sequence,max_pending,max_attempts) VALUES (?,?,?,?,?,?,?,?,?)`, t.NamespaceID, id, in.Name, t.CredentialID, string(filter), in.QueueID, cursor, in.MaxPending, in.MaxAttempts) + var me *driver.MySQLError + if errors.As(err, &me) && me.Number == 1062 { + return fault.Conflict + } + if err != nil { + return err + } + out.Subscription, err = readSubscription(ctx, tx, t, p, id, false) + if err != nil { + return err + } + if err = appendAudit(ctx, tx, t, identity, id, "subscription", "subscription.create", "create durable result subscription", nil, out.Subscription); err != nil { + return err + } + return saveIdentity(ctx, tx, t.NamespaceID, identity, out) + }) + return out, err +} +func (s *EventStore) GetSubscription(ctx context.Context, t auth.Token, id core.ID) (events.Subscription, error) { + var out events.Subscription + err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error { + var err error + out, err = readSubscription(ctx, tx, t, p, id, false) + return err + }) + return out, err +} +func (s *EventStore) ControlSubscription(ctx context.Context, t auth.Token, id core.ID, key string, in events.ControlInput) (events.SubscriptionResult, error) { + var out events.SubscriptionResult + err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error { + identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "subscription.control", eventScope(t.CredentialID, id), key, "subscription", id, in) + if err != nil { + return err + } + before, err := readSubscription(ctx, tx, t, p, id, true) + if err != nil { + return err + } + if cached != nil { + if err = json.Unmarshal(cached, &out); err != nil { + return err + } + out.Deduplicated = true + return nil + } + if before.Revision != in.ExpectedRevision || before.Revision == math.MaxUint64 { + return fault.Conflict + } + if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET enabled=?,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, in.Enabled, t.NamespaceID, id); err != nil { + return err + } + out.Subscription, err = readSubscription(ctx, tx, t, p, id, false) + if err != nil { + return err + } + if err = appendAudit(ctx, tx, t, identity, id, "subscription", "subscription.control", in.Reason, before, out.Subscription); err != nil { + return err + } + return saveIdentity(ctx, tx, t.NamespaceID, identity, out) + }) + return out, err +} + +// eventScope 将带域隔离的复合资源身份压缩到已有64字符幂等作用域，避免截断或串联歧义。 +func eventScope(ids ...core.ID) string { + h := sha256.New() + h.Write([]byte("orbit-event-scope-v1")) + for _, id := range ids { + h.Write([]byte{0}) + h.Write([]byte(id)) + } + return hex.EncodeToString(h.Sum(nil)) +} diff --git a/internal/storage/mysql/event_subscriptions_integration_test.go b/internal/storage/mysql/event_subscriptions_integration_test.go new file mode 100644 index 0000000..dbf9db2 --- /dev/null +++ b/internal/storage/mysql/event_subscriptions_integration_test.go @@ -0,0 +1,709 @@ +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "os" + "strings" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/platform/events" + platform "git.yuntuo.hk/k2safe/orbit/platform/model" + pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" + "git.yuntuo.hk/k2safe/orbit/pulse/service" + "git.yuntuo.hk/k2safe/orbit/pulse/worker" +) + +type eventFixture struct { + *executionFixture + events *events.Service + store *EventStore +} + +func newEventFixture(t *testing.T, ctx context.Context, db *sql.DB) *eventFixture { + f := newExecutionFixture(t, ctx, db) + var p auth.Permissions + var raw []byte + if err := db.QueryRowContext(ctx, `SELECT permissions FROM orbit_credentials WHERE namespace_id=? AND id=?`, f.token.NamespaceID, f.token.CredentialID).Scan(&raw); err != nil { + t.Fatal(err) + } + if err := json.Unmarshal(raw, &p); err != nil { + t.Fatal(err) + } + p.Actions = append(p.Actions, "events:manage", "events:read", "events:consume") + raw, _ = json.Marshal(p) + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.token.NamespaceID, f.token.CredentialID) + store := NewEventStore(db) + return &eventFixture{f, events.New(store), store} +} +func (f *eventFixture) create(in events.CreateInput) events.Subscription { + f.t.Helper() + if in.Name == "" { + in.Name = "result-" + string(executionID(f.t)) + } + if in.QueueID == "" { + in.QueueID = f.resources.QueueID + } + out, err := f.events.Create(f.ctx, f.token, string(executionID(f.t)), in) + if err != nil { + f.t.Fatal(err) + } + return out.Subscription +} +func (f *eventFixture) cancelResult() service.TaskView { + f.t.Helper() + task := f.submit(1) + out, err := f.tasks.CancelTask(f.ctx, f.token, task.ID, string(executionID(f.t)), service.CancelInput{ExpectedRevision: task.Revision, Reason: "create result notification"}) + if err != nil { + f.t.Fatal(err) + } + return out.Task +} +func (f *eventFixture) grant(sub core.ID) *events.Grant { + f.t.Helper() + v, err := f.events.Claim(f.ctx, f.token, sub, events.ClaimInput{ConnectionID: executionID(f.t), Capacity: 1}) + if err != nil || v == nil { + f.t.Fatalf("grant: %v", err) + } + return v +} +func (f *eventFixture) ack(sub core.ID, g *events.Grant) events.AckResult { + f.t.Helper() + v, err := f.events.Ack(f.ctx, f.token, sub, events.AckInput{DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret}) + if err != nil { + f.t.Fatal(err) + } + return v +} +func (f *eventFixture) subscription(id core.ID) events.Subscription { + f.t.Helper() + v, err := f.events.Get(f.ctx, f.token, id) + if err != nil { + f.t.Fatal(err) + } + return v +} +func (f *eventFixture) expireDelivery(g *events.Grant) { + f.exec(`UPDATE orbit_deliveries SET lease_until=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, g.Delivery.ID) +} +func noEventGrant(t *testing.T, f *eventFixture, sub core.ID) { + t.Helper() + g, err := f.events.Claim(f.ctx, f.token, sub, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1}) + if err != nil || g != nil { + t.Fatalf("unexpected grant: %v %v", g, err) + } +} + +func TestIntegrationResultSubscriptions(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 32, 8) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + zero := uint64(0) + t.Run("explicit_boundary_filters_and_frozen_result_precision", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + task := f.submit(1) + ref := retryStart(t, f.executionFixture) + raw := json.RawMessage(`{"n":9007199254740993,"text":"<&>"}`) + in := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: raw} + message := executionID(t) + if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil { + t.Fatal(err) + } + if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil { + t.Fatal(err) + } + if f.count("orbit_result_events", "1=1") != 1 { + t.Fatal("duplicate result journal entry") + } + recent := f.create(events.CreateInput{}) + if recent.CursorSequence != 1 { + t.Fatal("new subscription did not capture committed head") + } + noEventGrant(t, f, recent.ID) + history := f.create(events.CreateInput{FromSequence: &zero, Types: []string{"task.succeeded"}}) + g := f.grant(history.ID) + if g.Event.ID != g.Delivery.EventID || g.Event.Sequence != 1 || g.Event.Snapshot.TaskID != task.ID || string(g.Event.Snapshot.Result) != string(raw) || g.Event.Snapshot.CurrentAttemptID == nil || *g.Event.Snapshot.CurrentAttemptID != ref.AttemptID { + t.Fatalf("bad snapshot: %+v", g.Event) + } + f.exec(`UPDATE pulse_tasks SET result='{"changed":true}' WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID) + f.ack(history.ID, g) + replay := f.create(events.CreateInput{FromSequence: &zero}) + g = f.grant(replay.ID) + if string(g.Event.Snapshot.Result) != string(raw) { + t.Fatal("snapshot read mutable task result") + } + f.ack(replay.ID, g) + f.cancelResult() + noEventGrant(t, f, history.ID) + if f.subscription(history.ID).CursorSequence != 2 { + t.Fatal("filtered event did not advance cursor") + } + g = f.grant(recent.ID) + if g.Event.Type != "task.cancelled" || g.Event.Sequence != 2 { + t.Fatal("new subscription missed future result") + } + _, err := db.ExecContext(ctx, `UPDATE orbit_result_events SET snapshot='{}' WHERE namespace_id=? AND id=?`, f.token.NamespaceID, g.Event.ID) + if err == nil { + t.Fatal("journal is mutable") + } + }) + t.Run("create_idempotency_queue_ownership_and_permission_isolation", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + other := newEventFixture(t, ctx, db) + in := events.CreateInput{Name: "stable-result", QueueID: f.resources.QueueID, FromSequence: &zero} + a, err := f.events.Create(ctx, f.token, "stable", in) + if err != nil { + t.Fatal(err) + } + b, err := f.events.Create(ctx, f.token, "stable", in) + if err != nil || !b.Deduplicated || a.Subscription.ID != b.Subscription.ID { + t.Fatal("unstable create receipt", err) + } + in.MaxPending = 2 + _, err = f.events.Create(ctx, f.token, "stable", in) + expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT") + _, err = f.events.Get(ctx, other.token, a.Subscription.ID) + expectWorkerCode(t, err, "NOT_FOUND") + _, err = f.events.Create(ctx, f.token, "cross-queue", events.CreateInput{Name: "cross", QueueID: other.resources.QueueID}) + expectWorkerCode(t, err, "NOT_FOUND") + secondID := executionID(t) + secret, digest, err := auth.Issue(f.token.NamespaceID, secondID) + if err != nil { + t.Fatal(err) + } + p, _ := json.Marshal(auth.Permissions{Actions: []string{"events:read", "events:consume", "events:manage"}, Queues: []string{"*"}}) + f.exec(`INSERT INTO orbit_credentials(namespace_id,id,name,secret_digest,permissions) VALUES (?,?,?, ?,?)`, f.token.NamespaceID, secondID, "second", digest, string(p)) + second, err := auth.Parse(secret) + if err != nil { + t.Fatal(err) + } + _, err = f.events.Get(ctx, second, a.Subscription.ID) + expectWorkerCode(t, err, "NOT_FOUND") + own, err := f.events.Create(ctx, second, "stable", events.CreateInput{Name: "second", QueueID: f.resources.QueueID, FromSequence: &zero}) + if err != nil || own.Subscription.ID == a.Subscription.ID { + t.Fatal("create receipt leaked across owners", err) + } + p, _ = json.Marshal(auth.Permissions{Actions: []string{"events:read", "events:consume", "events:manage"}, Queues: []string{string(executionID(t))}}) + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(p), f.token.NamespaceID, secondID) + _, err = f.events.Get(ctx, second, own.Subscription.ID) + expectWorkerCode(t, err, "NOT_FOUND") + p, _ = json.Marshal(auth.Permissions{Actions: []string{"tasks:read"}, Queues: []string{"*"}}) + f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(p), f.token.NamespaceID, f.token.CredentialID) + _, err = f.events.Claim(ctx, f.token, a.Subscription.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1}) + expectWorkerCode(t, err, "PERMISSION_DENIED") + f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, f.token.CredentialID) + _, err = f.events.Get(ctx, f.token, a.Subscription.ID) + expectWorkerCode(t, err, "UNAUTHENTICATED") + }) + t.Run("backpressure_does_not_block_result_commit_or_skip_history", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{MaxPending: 1}) + f.cancelResult() + first := f.grant(sub.ID) + for range 3 { + f.cancelResult() + } + noEventGrant(t, f, sub.ID) + got := f.subscription(sub.ID) + if got.PendingCount != 1 || got.CursorSequence != 1 || f.count("orbit_result_events", "1=1") != 4 { + t.Fatal("window violated or producer blocked") + } + f.ack(sub.ID, first) + for seq := uint64(2); seq <= 4; seq++ { + g := f.grant(sub.ID) + if g.Event.Sequence != seq { + t.Fatal("backlog skipped") + } + f.ack(sub.ID, g) + } + if got = f.subscription(sub.ID); got.PendingCount != 0 || got.CursorSequence != 4 || got.Revision != 1 { + t.Fatal("invalid drained window or configuration revision changed") + } + }) + t.Run("competing_consumers_and_duplicate_acks", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + f.cancelResult() + type answer struct { + g *events.Grant + err error + } + results := make(chan answer, 12) + for range 12 { + connection := executionID(t) + go func() { + g, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: connection, Capacity: 1}) + results <- answer{g, err} + }() + } + var winner *events.Grant + for range 12 { + r := <-results + if r.err != nil { + t.Fatal(r.err) + } + if r.g != nil { + if winner != nil { + t.Fatal("two grants for one unexpired delivery") + } + winner = r.g + } + } + if winner == nil { + t.Fatal("no consumer won") + } + acks := make(chan events.AckResult, 12) + errs := make(chan error, 12) + for range 12 { + go func() { + v, err := f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: winner.Delivery.ID, Epoch: winner.Delivery.LeaseEpoch, LeaseSecret: winner.LeaseSecret}) + acks <- v + errs <- err + }() + } + fresh := 0 + var ackTime time.Time + for range 12 { + v := <-acks + if err := <-errs; err != nil { + t.Fatal(err) + } + if !v.Deduplicated { + fresh++ + } + if ackTime.IsZero() { + ackTime = v.AcknowledgedAt + } else if !ackTime.Equal(v.AcknowledgedAt) { + t.Fatal("ack receipt changed") + } + } + if fresh != 1 || f.subscription(sub.ID).PendingCount != 0 || f.count("orbit_audit_events", "resource_id=? AND action='delivery.ack'", winner.Delivery.ID) != 1 { + t.Fatal("duplicate ack side effects") + } + }) + t.Run("lease_expiry_backoff_latest_proof_and_node_restart", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{MaxAttempts: 2}) + f.cancelResult() + first := f.grant(sub.ID) + f.expireDelivery(first) + // 新建存储实例模拟所有连接内存丢失；权威交付依然来自数据库。 + f.events = events.New(NewEventStore(db)) + noEventGrant(t, f, sub.ID) + v, err := f.events.Delivery(ctx, f.token, sub.ID, first.Delivery.ID) + if err != nil || v.State != platform.DeliveryPending || v.Attempts != 1 { + t.Fatal("lease was not recovered", err) + } + f.exec(`UPDATE orbit_deliveries SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, v.ID) + second := f.grant(sub.ID) + if second.Event.ID != first.Event.ID || second.Delivery.ID != first.Delivery.ID || second.Delivery.LeaseEpoch != 2 || second.Delivery.Attempts != 2 || second.LeaseSecret == first.LeaseSecret { + t.Fatal("redelivery identity/proof incorrect") + } + _, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: first.Delivery.LeaseEpoch, LeaseSecret: first.LeaseSecret}) + expectWorkerCode(t, err, "DELIVERY_LEASE_LOST") + f.expireDelivery(second) + noEventGrant(t, f, sub.ID) + v, err = f.events.Delivery(ctx, f.token, sub.ID, second.Delivery.ID) + if err != nil || v.State != platform.DeliveryExpired || f.subscription(sub.ID).PendingCount != 1 { + t.Fatal("exhaustion discarded event", err) + } + // 确认本身不是新的执行授权；最近证明在未被新代次替代时允许迟到确认。 + f.ack(sub.ID, second) + if !f.ack(sub.ID, second).Deduplicated { + t.Fatal("late ack replay lost") + } + }) + t.Run("manual_retry_adds_budget_without_resetting_attempt_or_epoch", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{MaxAttempts: 1}) + f.cancelResult() + first := f.grant(sub.ID) + f.expireDelivery(first) + noEventGrant(t, f, sub.ID) + v, err := f.events.Delivery(ctx, f.token, sub.ID, first.Delivery.ID) + if err != nil { + t.Fatal(err) + } + in := events.RetryInput{ExpectedRevision: v.Revision, Reason: "receiver fixed", AdditionalAttempts: 2} + out, err := f.events.Retry(ctx, f.token, sub.ID, v.ID, "retry", in) + if err != nil || out.Delivery.Attempts != 1 || out.Delivery.AttemptBudget != 3 || out.Delivery.LeaseEpoch != 1 { + t.Fatal("budget reset", err) + } + second := f.grant(sub.ID) + if second.Delivery.Attempts != 2 || second.Delivery.LeaseEpoch != 2 { + t.Fatal("retry reused attempt") + } + again, err := f.events.Retry(ctx, f.token, sub.ID, v.ID, "retry", in) + if err != nil || !again.Deduplicated || again.Delivery.Revision != out.Delivery.Revision { + t.Fatal("retry replay changed receipt", err) + } + _, err = f.events.Retry(ctx, f.token, sub.ID, v.ID, "another", in) + expectWorkerCode(t, err, "STATE_CONFLICT") + _, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: 1, LeaseSecret: first.LeaseSecret}) + expectWorkerCode(t, err, "DELIVERY_LEASE_LOST") + f.ack(sub.ID, second) + if f.count("orbit_audit_events", "resource_id=? AND action='delivery.retry'", v.ID) != 1 { + t.Fatal("manual retry audit duplicated") + } + }) + t.Run("pause_blocks_new_delivery_but_allows_inflight_ack", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + f.cancelResult() + g := f.grant(sub.ID) + paused, err := f.events.Control(ctx, f.token, sub.ID, "pause", events.ControlInput{ExpectedRevision: 1, Enabled: false, Reason: "receiver maintenance"}) + if err != nil || paused.Subscription.Enabled { + t.Fatal(err) + } + f.cancelResult() + noEventGrant(t, f, sub.ID) + f.ack(sub.ID, g) + _, err = f.events.Control(ctx, f.token, sub.ID, "resume", events.ControlInput{ExpectedRevision: 2, Enabled: true, Reason: "receiver ready"}) + if err != nil { + t.Fatal(err) + } + resumed := f.grant(sub.ID) + if resumed.Event.Sequence != 2 { + t.Fatal("resume lost result") + } + }) + t.Run("ack_and_materialization_are_transactional_with_audit", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + f.cancelResult() + f.exec(`CREATE TRIGGER test_delivery_reject BEFORE INSERT ON orbit_deliveries FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected delivery failure'`) + t.Cleanup(func() { + f.exec(`DROP TRIGGER IF EXISTS test_delivery_reject`) + f.exec(`DROP TRIGGER IF EXISTS test_delivery_ack_reject`) + }) + _, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1}) + if err == nil { + t.Fatal("expected delivery insertion failure") + } + f.exec(`DROP TRIGGER test_delivery_reject`) + v := f.subscription(sub.ID) + if v.CursorSequence != 0 || v.PendingCount != 0 || f.count("orbit_deliveries", "subscription_id=?", sub.ID) != 0 { + t.Fatal("partial materialization escaped rollback") + } + g := f.grant(sub.ID) + f.exec(`CREATE TRIGGER test_delivery_ack_reject BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected audit failure'`) + _, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret}) + if err == nil { + t.Fatal("ack escaped audit failure") + } + f.exec(`DROP TRIGGER test_delivery_ack_reject`) + d, err := f.events.Delivery(ctx, f.token, sub.ID, g.Delivery.ID) + if err != nil || d.State != platform.DeliveryLeased || d.AcknowledgedAt != nil || f.subscription(sub.ID).PendingCount != 1 { + t.Fatal("ack partial commit", err) + } + f.ack(sub.ID, g) + }) + t.Run("producer_journal_failure_rolls_back_task_event_and_quota", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + task := f.submit(1) + f.exec(`CREATE TRIGGER test_result_reject BEFORE INSERT ON orbit_result_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected journal failure'`) + t.Cleanup(func() { f.exec(`DROP TRIGGER IF EXISTS test_result_reject`) }) + _, err := f.tasks.CancelTask(ctx, f.token, task.ID, "cancel", service.CancelInput{ExpectedRevision: task.Revision, Reason: "verify atomic result"}) + if err == nil { + t.Fatal("expected journal failure") + } + f.exec(`DROP TRIGGER test_result_reject`) + if f.get(task.ID).State != pulse.TaskWaiting || f.count("orbit_events", "aggregate_id=?", task.ID) != 1 || f.count("orbit_event_streams", "1=1") != 0 { + t.Fatal("result partial commit") + } + assertApplicationCounts(t, f.executionFixture, 1, 0, 0) + if _, err = f.tasks.CancelTask(ctx, f.token, task.ID, "cancel", service.CancelInput{ExpectedRevision: task.Revision, Reason: "verify atomic result"}); err != nil { + t.Fatal(err) + } + }) + t.Run("all_terminal_and_unknown_paths_emit_but_safe_requeue_does_not", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + f.submit(2) + first := retryStart(t, f.executionFixture) + retryFinish(t, f.executionFixture, first, "FAILED", true) + if f.count("orbit_result_events", "1=1") != 0 { + t.Fatal("safe retry published final result") + } + makeRetryDue(f.executionFixture, first.TaskID) + second := retryStart(t, f.executionFixture) + retryFinish(t, f.executionFixture, second, "FAILED", false) + g := f.grant(sub.ID) + if g.Event.Type != "task.failed" { + t.Fatal("missing failure result") + } + f.ack(sub.ID, g) + f.submit(1) + unknown := retryStart(t, f.executionFixture) + retryFinish(t, f.executionFixture, unknown, "UNKNOWN", false) + g = f.grant(sub.ID) + if g.Event.Type != "task.needs_review" || !g.Event.Snapshot.EffectUncertain { + t.Fatal("missing unknown result") + } + f.ack(sub.ID, g) + f.submit(1) + lost := retryStart(t, f.executionFixture) + f.expire(lost) + if _, err := f.storeB.Recover(ctx, 100); err != nil { + t.Fatal(err) + } + g = f.grant(sub.ID) + if g.Event.Type != "task.needs_review" { + t.Fatal("recovery failed to publish result") + } + }) + t.Run("namespace_limit_future_cursor_and_history_gap_are_explicit", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + future := uint64(1) + _, err := f.events.Create(ctx, f.token, "future", events.CreateInput{Name: "future", QueueID: f.resources.QueueID, FromSequence: &future}) + expectWorkerCode(t, err, "INVALID_ARGUMENT") + sub := f.create(events.CreateInput{}) + f.cancelResult() + f.exec(`UPDATE orbit_event_streams SET retained_from_sequence=2 WHERE namespace_id=?`, f.token.NamespaceID) + _, err = f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1}) + expectWorkerCode(t, err, "EVENT_HISTORY_EXPIRED") + _, err = f.events.Create(ctx, f.token, "expired", events.CreateInput{Name: "expired", QueueID: f.resources.QueueID, FromSequence: &zero}) + expectWorkerCode(t, err, "EVENT_HISTORY_EXPIRED") + for i := 1; i < events.MaxSubscriptions; i++ { + f.create(events.CreateInput{Name: fmt.Sprintf("limit-%d", i)}) + } + _, err = f.events.Create(ctx, f.token, "overflow", events.CreateInput{Name: "overflow", QueueID: f.resources.QueueID}) + expectWorkerCode(t, err, "SUBSCRIPTION_LIMIT") + }) + + t.Run("locked_subscriber_cannot_block_task_result_commit", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + task := f.submit(1) + tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + var id core.ID + if err = tx.QueryRowContext(ctx, `SELECT id FROM orbit_subscriptions WHERE namespace_id=? AND id=? FOR UPDATE`, f.token.NamespaceID, sub.ID).Scan(&id); err != nil { + t.Fatal(err) + } + bounded, stop := context.WithTimeout(ctx, 2*time.Second) + defer stop() + if _, err = f.tasks.CancelTask(bounded, f.token, task.ID, "while-locked", service.CancelInput{ExpectedRevision: task.Revision, Reason: "subscriber is slow"}); err != nil { + t.Fatal("producer waited for subscriber lock", err) + } + if err = tx.Rollback(); err != nil { + t.Fatal(err) + } + g := f.grant(sub.ID) + if g.Event.Snapshot.TaskID != task.ID { + t.Fatal("result lost while subscriber blocked") + } + }) + t.Run("connection_capacity_is_database_enforced", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + f.cancelResult() + f.cancelResult() + claim := events.ClaimInput{ConnectionID: executionID(t), Capacity: 1} + first, err := f.events.Claim(ctx, f.token, sub.ID, claim) + if err != nil || first == nil { + t.Fatal(err) + } + if g, err := f.events.Claim(ctx, f.token, sub.ID, claim); err != nil || g != nil { + t.Fatal("connection capacity exceeded", err) + } + f.ack(sub.ID, first) + next, err := f.events.Claim(ctx, f.token, sub.ID, claim) + if err != nil || next == nil || next.Event.ID == first.Event.ID { + t.Fatal("capacity not released on confirmation", err) + } + }) + t.Run("late_ack_races_new_delivery_without_double_confirmation", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{MaxPending: 1, MaxAttempts: 2}) + for range 12 { + f.cancelResult() + first := f.grant(sub.ID) + f.expireDelivery(first) + noEventGrant(t, f, sub.ID) + f.exec(`UPDATE orbit_deliveries SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, first.Delivery.ID) + type claimResult struct { + grant *events.Grant + err error + } + claims := make(chan claimResult, 1) + acks := make(chan error, 1) + start := make(chan struct{}) + connection := executionID(t) + go func() { + <-start + g, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: connection, Capacity: 1}) + claims <- claimResult{g, err} + }() + go func() { + <-start + _, err := f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: first.Delivery.LeaseEpoch, LeaseSecret: first.LeaseSecret}) + acks <- err + }() + close(start) + c := <-claims + ackErr := <-acks + if c.err != nil { + t.Fatal(c.err) + } + if c.grant == nil { + if ackErr != nil { + t.Fatal("neither ack nor redelivery won", ackErr) + } + } else { + expectWorkerCode(t, ackErr, "DELIVERY_LEASE_LOST") + f.ack(sub.ID, c.grant) + } + if f.subscription(sub.ID).PendingCount != 0 || f.count("orbit_audit_events", "resource_id=? AND action='delivery.ack'", first.Delivery.ID) != 1 { + t.Fatal("race duplicated confirmation or leaked window") + } + } + }) + t.Run("proof_is_not_stored_in_audit_or_plaintext_column", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + f.cancelResult() + g := f.grant(sub.ID) + f.ack(sub.ID, g) + var before, after string + if err := db.QueryRowContext(ctx, "SELECT `before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND resource_id=? AND action='delivery.ack'", f.token.NamespaceID, g.Delivery.ID).Scan(&before, &after); err != nil { + t.Fatal(err) + } + if strings.Contains(before+after, g.LeaseSecret) { + t.Fatal("delivery proof leaked to audit") + } + }) +} + +func TestIntegrationResultCommitOrder(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 16, 4) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + for _, rollback := range []bool{false, true} { + t.Run(fmt.Sprintf("first_rollback_%t", rollback), func(t *testing.T) { + f := newEventFixture(t, ctx, db) + sub := f.create(events.CreateInput{}) + first, second := f.submit(1), f.submit(1) + finish := func(tx *sql.Tx, task service.TaskView) error { + if _, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='CANCELLED',completed_at=UTC_TIMESTAMP(6),revision=revision+1 WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID); err != nil { + return err + } + return emitEvent(ctx, tx, f.token.NamespaceID, task.ID, task.Revision+1, "task", "task.cancelled", map[string]any{"state": "CANCELLED"}) + } + tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + if err = finish(tx, first); err != nil { + t.Fatal(err) + } + started := make(chan struct{}) + result := make(chan error, 1) + go func() { + result <- WithTx(ctx, db, func(other *sql.Tx) error { close(started); return finish(other, second) }) + }() + <-started + select { + case err := <-result: + t.Fatalf("later transaction crossed uncommitted stream: %v", err) + case <-time.After(100 * time.Millisecond): + } + // 已提交读在第一事务未提交时看不到较大的序号；消费也不等待生产者流锁。 + noEventGrant(t, f, sub.ID) + if rollback { + err = tx.Rollback() + } else { + err = tx.Commit() + } + if err != nil { + t.Fatal(err) + } + if err = <-result; err != nil { + t.Fatal(err) + } + want := uint64(2) + if rollback { + want = 1 + } + var head uint64 + if err = db.QueryRowContext(ctx, `SELECT last_sequence FROM orbit_event_streams WHERE namespace_id=?`, f.token.NamespaceID).Scan(&head); err != nil || head != want { + t.Fatal("nontransactional sequence", err) + } + for seq := uint64(1); seq <= want; seq++ { + g := f.grant(sub.ID) + if g.Event.Sequence != seq { + t.Fatal("commit order skipped result") + } + f.ack(sub.ID, g) + } + }) + } + t.Run("create_new_boundary_waits_for_earlier_producer", func(t *testing.T) { + f := newEventFixture(t, ctx, db) + task := f.submit(1) + tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + if _, err = tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='CANCELLED',completed_at=UTC_TIMESTAMP(6),revision=2 WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID); err != nil { + t.Fatal(err) + } + if err = emitEvent(ctx, tx, f.token.NamespaceID, task.ID, 2, "task", "task.cancelled", map[string]any{}); err != nil { + t.Fatal(err) + } + type result struct { + v events.SubscriptionResult + err error + } + ch := make(chan result, 1) + go func() { + v, err := f.events.Create(ctx, f.token, "boundary", events.CreateInput{Name: "boundary", QueueID: f.resources.QueueID}) + ch <- result{v, err} + }() + select { + case <-ch: + t.Fatal("subscription creation crossed stream boundary") + case <-time.After(100 * time.Millisecond): + } + if err = tx.Commit(); err != nil { + t.Fatal(err) + } + r := <-ch + if r.err != nil || r.v.Subscription.CursorSequence != 1 { + t.Fatal("boundary did not use committed head", r.err) + } + noEventGrant(t, f, r.v.Subscription.ID) + }) +} diff --git a/internal/storage/mysql/integration_test.go b/internal/storage/mysql/integration_test.go index 3b26566..9b7da40 100644 --- a/internal/storage/mysql/integration_test.go +++ b/internal/storage/mysql/integration_test.go @@ -189,6 +189,8 @@ func checkModelColumns(t *testing.T, ctx context.Context, db *sql.DB) { platform.Credential{}, platform.IdempotencyKey{}, platform.Event{}, + platform.EventStream{}, + platform.ResultEvent{}, platform.Subscription{}, platform.Delivery{}, platform.AuditEvent{}, diff --git a/internal/storage/mysql/migrations/000013_result_subscriptions.up.sql b/internal/storage/mysql/migrations/000013_result_subscriptions.up.sql new file mode 100644 index 0000000..7e8ba89 --- /dev/null +++ b/internal/storage/mysql/migrations/000013_result_subscriptions.up.sql @@ -0,0 +1,51 @@ +-- 结果日志按命名空间串行分配序号，与来源状态事件同事务提交；不使用自增 ID 推断提交顺序。 +-- 禁止外键；来源事件、队列和订阅归属由服务端事务校验。 +CREATE TABLE orbit_event_streams ( + namespace_id BINARY(16) NOT NULL COMMENT '结果流所属命名空间；主键同时作为该流序号分配的事务锁', + last_sequence BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已分配结果序号；从1开始，与结果日志同事务提交，回滚不消耗序号', + retained_from_sequence BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '最早保留序号；首版不清理结果日志，固定为1，供以后安全清理校验', + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '创建时间，统一UTC，精度微秒', + updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '最后分配序号时间，统一UTC，不代替序号', + PRIMARY KEY(namespace_id), + CONSTRAINT ck_event_stream_range CHECK (retained_from_sequence>0 AND retained_from_sequence-1<=last_sequence) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='命名空间结果流的提交序号和保留边界'; + +CREATE TABLE orbit_result_events ( + namespace_id BINARY(16) NOT NULL COMMENT '结果所属命名空间，参与主键与服务端关联校验', + id BINARY(16) NOT NULL COMMENT '与来源orbit_events记录使用相同标识；跨订阅消费去重依据', + revision BIGINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '不可变记录版本，固定为1', + created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '创建时间，统一UTC，精度微秒', + updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) COMMENT '不可变记录的初始写入时间，统一UTC', + sequence BIGINT UNSIGNED NOT NULL COMMENT '命名空间内事务分配的结果序号；较小序号必先提交或整体回滚', + queue_id BINARY(16) NOT NULL COMMENT '来源任务的队列标识；订阅过滤和授权按该队列检查', + snapshot MEDIUMTEXT NOT NULL COMMENT '版本化结果快照JSON文本，保留数值精度，不含任务输入和租约秘密', + PRIMARY KEY(namespace_id,id), + UNIQUE KEY uq_result_sequence(namespace_id,sequence), + KEY ix_result_queue_sequence(namespace_id,queue_id,sequence), + CONSTRAINT ck_result_event CHECK (revision=1 AND sequence>0 AND JSON_VALID(snapshot)) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_as_cs COMMENT='不可变任务结果日志；与任务状态及来源事件一起提交'; + +CREATE TRIGGER orbit_result_events_immutable BEFORE UPDATE ON orbit_result_events FOR EACH ROW + SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='result events are immutable'; + +ALTER TABLE orbit_subscriptions + MODIFY COLUMN filter JSON NOT NULL COMMENT '创建时冻结的结果事件类型数组JSON，不允许修改以免漏消费', + MODIFY COLUMN enabled BOOLEAN NOT NULL DEFAULT TRUE COMMENT '是否启用新交付；暂停保留积压和在途结果，仍可接受有效确认', + ADD COLUMN queue_id BINARY(16) NULL COMMENT '冻结订阅队列；历史占位配置为空且不可消费，须通过新接口创建有效订阅', + ADD COLUMN cursor_sequence BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已扫描并可靠创建交付记录的结果序号；不是消费者确认位置', + ADD COLUMN pending_count INT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已物化但未确认数量，包含等待、租约中和耗尽次数的交付', + ADD COLUMN max_pending INT UNSIGNED NOT NULL DEFAULT 100 COMMENT '未确认交付上限，1至1000；达到后暂停物化但保留来源结果', + ADD COLUMN max_attempts INT UNSIGNED NOT NULL DEFAULT 20 COMMENT '每条新交付的初始发送次数上限，1至100，不包含任务执行业务次数', + ADD CONSTRAINT ck_subscription_window CHECK (max_pending BETWEEN 1 AND 1000 AND max_attempts BETWEEN 1 AND 100 AND pending_count<=max_pending); + +-- 旧配置没有公开消费协议和队列授权绑定，保留原记录但不自动开启消费。 +UPDATE orbit_subscriptions SET enabled=FALSE; + +ALTER TABLE orbit_deliveries + ADD COLUMN attempt_budget INT UNSIGNED NOT NULL DEFAULT 20 COMMENT '累计允许交付次数，人工追加预算不重置已使用次数和租约代次', + ADD COLUMN token_digest BINARY(32) NULL COMMENT '最近一次交付秘密的SHA256摘要；从未发出或历史占位记录为空，确认后保留以核对重复确认', + ADD KEY ix_delivery_subscription_due(namespace_id,subscription_id,state,available_at,id), + ADD KEY ix_delivery_subscription_lease(namespace_id,subscription_id,state,lease_until,id); + +UPDATE orbit_deliveries SET attempt_budget=GREATEST(attempt_budget,attempts); +ALTER TABLE orbit_deliveries ADD CONSTRAINT ck_delivery_budget CHECK (attempts<=attempt_budget AND attempt_budget>0); diff --git a/internal/storage/mysql/no_foreign_keys_integration_test.go b/internal/storage/mysql/no_foreign_keys_integration_test.go index b7542ac..1c6b648 100644 --- a/internal/storage/mysql/no_foreign_keys_integration_test.go +++ b/internal/storage/mysql/no_foreign_keys_integration_test.go @@ -88,7 +88,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) { } indexes := func() []string { t.Helper() - rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME<>'orbit_application_resources' AND INDEX_NAME<>'ix_task_priority_dispatch' ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) + rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`) if err != nil { t.Fatal(err) } diff --git a/internal/storage/mysql/result_journal.go b/internal/storage/mysql/result_journal.go new file mode 100644 index 0000000..e553830 --- /dev/null +++ b/internal/storage/mysql/result_journal.go @@ -0,0 +1,52 @@ +package mysql + +import ( + "context" + "database/sql" + "errors" + "math" + + "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/events" +) + +// appendResultEvent 必须在来源任务更新和事件写入之后、原事务提交之前调用。 +// 流行一直锁到提交；后来的事务无法提交更大序号并越过尚未提交的较小序号。 +// 不访问订阅或交付行：慢消费者不能反向阻塞任务的结果持久化。 +func appendResultEvent(ctx context.Context, tx *sql.Tx, ns, eventID, taskID core.ID, revision uint64, eventType string) error { + var snapshot events.TaskResult + var raw []byte + snapshot.SchemaVersion = 1 + err := tx.QueryRowContext(ctx, `SELECT id,queue_id,state,revision,result,effect_uncertain,current_attempt_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, taskID).Scan(&snapshot.TaskID, &snapshot.QueueID, &snapshot.State, &snapshot.Revision, &raw, &snapshot.EffectUncertain, &snapshot.CurrentAttemptID) + if err != nil { + return err + } + if snapshot.Revision != revision || eventType != "task."+map[string]string{"SUCCEEDED": "succeeded", "FAILED": "failed", "CANCELLED": "cancelled", "NEEDS_REVIEW": "needs_review"}[snapshot.State] { + return errors.New("task result event does not match persisted task state") + } + snapshot.Result = raw + body, err := jsonwire.Marshal(snapshot) + if err != nil { + return err + } + head, _, err := lockEventStream(ctx, tx, ns) + if err != nil { + return err + } + if head == math.MaxUint64 { + return errors.New("event sequence exhausted") + } + if _, err = tx.ExecContext(ctx, `INSERT INTO orbit_result_events(namespace_id,id,sequence,queue_id,snapshot) VALUES (?,?,?,?,?)`, ns, eventID, head+1, snapshot.QueueID, string(body)); err != nil { + return err + } + _, err = tx.ExecContext(ctx, `UPDATE orbit_event_streams SET last_sequence=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=?`, head+1, ns) + return err +} +func lockEventStream(ctx context.Context, tx *sql.Tx, ns core.ID) (head, retained uint64, err error) { + if _, err = tx.ExecContext(ctx, `INSERT INTO orbit_event_streams(namespace_id) VALUES (?) ON DUPLICATE KEY UPDATE namespace_id=orbit_event_streams.namespace_id`, ns); err != nil { + return + } + err = tx.QueryRowContext(ctx, `SELECT last_sequence,retained_from_sequence FROM orbit_event_streams WHERE namespace_id=? FOR UPDATE`, ns).Scan(&head, &retained) + return +} diff --git a/internal/storage/mysql/result_subscription_upgrade_integration_test.go b/internal/storage/mysql/result_subscription_upgrade_integration_test.go new file mode 100644 index 0000000..26b16d8 --- /dev/null +++ b/internal/storage/mysql/result_subscription_upgrade_integration_test.go @@ -0,0 +1,109 @@ +package mysql + +import ( + "context" + "os" + "testing" + "time" + + core "git.yuntuo.hk/k2safe/orbit/model" + mysqldriver "github.com/go-sql-driver/mysql" +) + +// 从真实v12结构升级，历史占位订阅不被错误地当成已获队列授权的有效消费端。 +func TestIntegrationResultSubscriptionsUpgrade(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("requires disposable MySQL") + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + admin, err := Open(ctx, dsn, 2, 1) + if err != nil { + t.Fatal(err) + } + defer admin.Close() + name := "orbit_result_upgrade_" + string(executionID(t)) + if _, err = admin.ExecContext(ctx, "CREATE DATABASE `"+name+"`"); err != nil { + t.Fatal(err) + } + defer func() { + cleanup, stop := context.WithTimeout(context.Background(), 10*time.Second) + defer stop() + if _, err := admin.ExecContext(cleanup, "DROP DATABASE `"+name+"`"); err != nil { + t.Error(err) + } + }() + cfg, err := mysqldriver.ParseDSN(dsn) + if err != nil { + t.Fatal("invalid test DSN") + } + cfg.DBName = name + db, err := Open(ctx, cfg.FormatDSN(), 4, 2) + if err != nil { + t.Fatal(err) + } + defer db.Close() + exec := func(q string, args ...any) { + t.Helper() + if _, err := db.ExecContext(ctx, q, args...); err != nil { + t.Fatal(err) + } + } + exec(`CREATE TABLE orbit_schema_migrations (version BIGINT UNSIGNED PRIMARY KEY,name VARCHAR(191) NOT NULL,checksum CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,dirty BOOLEAN NOT NULL,applied_at DATETIME(6) NULL) ENGINE=InnoDB`) + migrations, err := Migrations() + if err != nil { + t.Fatal(err) + } + for _, m := range migrations { + if m.Version > 12 { + break + } + statements, err := splitSQL(m.SQL) + if err != nil { + t.Fatal(err) + } + for _, q := range statements { + exec(q) + } + exec(`INSERT INTO orbit_schema_migrations(version,name,checksum,dirty,applied_at) VALUES (?,?,?,FALSE,UTC_TIMESTAMP(6))`, m.Version, m.Name, m.Checksum) + } + app, ns, queue, credential, task, event, sub, delivery := executionID(t), executionID(t), executionID(t), executionID(t), executionID(t), executionID(t), executionID(t), executionID(t) + exec(`INSERT INTO orbit_applications(id,name) VALUES (?,?)`, app, string(app)) + exec(`INSERT INTO orbit_namespaces(id,application_id,name,environment) VALUES (?,?,'default','test')`, ns, app) + exec(`INSERT INTO orbit_credentials(namespace_id,id,name,secret_digest,permissions) VALUES (?,?,'legacy',?, '{}')`, ns, credential, core.Digest{}) + exec(`INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,policy) VALUES (?,?,'legacy',1,1,'{}')`, ns, queue) + exec(`INSERT INTO pulse_tasks(namespace_id,id,queue_id,state,handler,handler_version,payload_version,payload,policy_snapshot,available_at,deadline_at,attempt_timeout_ms,max_attempts,completed_at) VALUES (?,?,?,'CANCELLED','object.prepare','v1','v1','{}','{}',UTC_TIMESTAMP(6),DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR),5000,1,UTC_TIMESTAMP(6))`, ns, task, queue) + exec(`INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,'task',?,1,'task.cancelled','{"n":9007199254740993}',UTC_TIMESTAMP(6))`, ns, event, task) + exec(`INSERT INTO orbit_subscriptions(namespace_id,id,name,credential_id,filter) VALUES (?,?,'legacy',?,'{}')`, ns, sub, credential) + exec(`INSERT INTO orbit_deliveries(namespace_id,id,subscription_id,event_id,state,available_at,attempts,lease_epoch) VALUES (?,?,?,?,'EXPIRED',UTC_TIMESTAMP(6),25,25)`, ns, delivery, sub, event) + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + if err = CheckSchema(ctx, db); err != nil { + t.Fatal(err) + } + var enabled bool + var queueID *core.ID + var filter string + if err = db.QueryRowContext(ctx, `SELECT enabled,queue_id,filter FROM orbit_subscriptions WHERE namespace_id=? AND id=?`, ns, sub).Scan(&enabled, &queueID, &filter); err != nil || enabled || queueID != nil || filter != "{}" { + t.Fatal("legacy subscription acquired unintended authority", err) + } + var attempts, budget, epoch uint64 + var secret []byte + if err = db.QueryRowContext(ctx, `SELECT attempts,attempt_budget,lease_epoch,token_digest FROM orbit_deliveries WHERE namespace_id=? AND id=?`, ns, delivery).Scan(&attempts, &budget, &epoch, &secret); err != nil || attempts != 25 || budget != 25 || epoch != 25 || secret != nil { + t.Fatal("legacy delivery history changed", err) + } + var count int + if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_result_events`).Scan(&count); err != nil || count != 0 { + t.Fatal("migration synthesized unverified historical snapshots", err) + } + var payload string + if err = db.QueryRowContext(ctx, `SELECT payload FROM orbit_events WHERE namespace_id=? AND id=?`, ns, event).Scan(&payload); err != nil || payload != `{"n":9007199254740993}` { + t.Fatal("original event changed", err) + } + if err = Migrate(ctx, db); err != nil { + t.Fatal(err) + } + t.Log("v12 upgrade preserved source history and cumulative attempts, disabled unscoped legacy subscriptions, introduced no foreign keys") +} diff --git a/internal/storage/mysql/task_store.go b/internal/storage/mysql/task_store.go index 88d0760..244e2be 100644 --- a/internal/storage/mysql/task_store.go +++ b/internal/storage/mysql/task_store.go @@ -15,6 +15,7 @@ import ( "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" core "git.yuntuo.hk/k2safe/orbit/model" "git.yuntuo.hk/k2safe/orbit/platform/auth" + "git.yuntuo.hk/k2safe/orbit/platform/events" pulse "git.yuntuo.hk/k2safe/orbit/pulse/model" "git.yuntuo.hk/k2safe/orbit/pulse/service" driver "github.com/go-sql-driver/mysql" @@ -125,6 +126,9 @@ func emitEvent(ctx context.Context, tx *sql.Tx, ns, resource core.ID, revision u return err } _, err = tx.ExecContext(ctx, `INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,?,?,?,?,?,UTC_TIMESTAMP(6))`, ns, id, kind, resource, revision, event, string(raw)) + if err == nil && kind == "task" && events.ResultType(event) { + return appendResultEvent(ctx, tx, ns, id, resource, revision, event) + } return err } func appendAudit(ctx context.Context, tx *sql.Tx, token auth.Token, operation, resource core.ID, kind, action, reason string, before, after any) error { diff --git a/platform/auth/token.go b/platform/auth/token.go index 8679b38..ead75a8 100644 --- a/platform/auth/token.go +++ b/platform/auth/token.go @@ -58,7 +58,7 @@ type Permissions struct { Queues []string `json:"queues"` } -var supported = map[string]bool{"application:resources:read": true, "application:resources:manage": true, "tasks:submit": true, "tasks:read": true, "tasks:cancel": true, "queues:read": true, "queues:manage": true, "workers:connect": true, "workers:consume": true, "schedules:read": true, "schedules:manage": true} +var supported = map[string]bool{"events:read": true, "events:manage": true, "events:consume": true, "application:resources:read": true, "application:resources:manage": true, "tasks:submit": true, "tasks:read": true, "tasks:cancel": true, "queues:read": true, "queues:manage": true, "workers:connect": true, "workers:consume": true, "schedules:read": true, "schedules:manage": true} func (p Permissions) Validate() error { if len(p.Actions) == 0 || len(p.Actions) > 32 || len(p.Queues) == 0 || len(p.Queues) > 128 { diff --git a/platform/events/service.go b/platform/events/service.go new file mode 100644 index 0000000..3f19aad --- /dev/null +++ b/platform/events/service.go @@ -0,0 +1,277 @@ +// Package events 定义可靠结果订阅契约；交付采用至少一次语义，接收方按事件ID去重。 +package events + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "regexp" + "slices" + "strings" + "time" + "unicode/utf8" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" + platform "git.yuntuo.hk/k2safe/orbit/platform/model" +) + +const LeaseDuration = 30 * time.Second +const MaxSubscriptions = 64 + +var ( + LeaseLost = fault.New("DELIVERY_LEASE_LOST", "delivery proof was superseded or is invalid") + HistoryExpired = fault.New("EVENT_HISTORY_EXPIRED", "requested event history is no longer retained") + SubscriptionLimit = fault.New("SUBSCRIPTION_LIMIT", "namespace subscription limit reached") +) + +func ResultType(s string) bool { + switch s { + case "task.succeeded", "task.failed", "task.cancelled", "task.needs_review": + return true + } + return false +} + +// TaskResult 是事件发生时冻结的快照，不从后续可变任务状态重新生成。 +type TaskResult struct { + SchemaVersion uint32 `json:"schema_version"` + TaskID core.ID `json:"task_id"` + QueueID core.ID `json:"queue_id"` + State string `json:"state"` + Revision uint64 `json:"revision,string"` + Result json.RawMessage `json:"result,omitempty"` + EffectUncertain bool `json:"effect_uncertain"` + CurrentAttemptID *core.ID `json:"current_attempt_id,omitempty"` +} + +type Event struct { + ID core.ID `json:"id"` + Sequence uint64 `json:"sequence,string"` + Type string `json:"type"` + AggregateKind string `json:"aggregate_kind"` + AggregateID core.ID `json:"aggregate_id"` + AggregateRevision uint64 `json:"aggregate_revision,string"` + OccurredAt time.Time `json:"occurred_at"` + // Details 是来源事件元信息，不包含租约秘密和任务输入。 + Details json.RawMessage `json:"details"` + Snapshot TaskResult `json:"snapshot"` +} + +type CreateInput struct { + Name string `json:"name"` + QueueID core.ID `json:"queue_id"` + Types []string `json:"types,omitempty"` + // FromSequence 之后的结果纳入订阅；省略为创建时已提交的最新序号，0为从现存日志开始。 + FromSequence *uint64 `json:"from_sequence,string,omitempty"` + MaxPending uint32 `json:"max_pending,omitempty"` + MaxAttempts uint32 `json:"max_attempts,omitempty"` +} + +type Subscription struct { + ID core.ID `json:"id"` + QueueID core.ID `json:"queue_id"` + CredentialID core.ID `json:"credential_id"` + Name string `json:"name"` + Revision uint64 `json:"revision,string"` + Enabled bool `json:"enabled"` + Types []string `json:"types"` + CursorSequence uint64 `json:"cursor_sequence,string"` + PendingCount uint32 `json:"pending_count"` + MaxPending uint32 `json:"max_pending"` + MaxAttempts uint32 `json:"max_attempts"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type SubscriptionResult struct { + Subscription Subscription `json:"subscription"` + Deduplicated bool `json:"deduplicated"` +} + +type ControlInput struct { + ExpectedRevision uint64 `json:"expected_revision,string"` + Enabled bool `json:"enabled"` + Reason string `json:"reason"` +} + +type Delivery struct { + ID core.ID `json:"id"` + SubscriptionID core.ID `json:"subscription_id"` + EventID core.ID `json:"event_id"` + Sequence uint64 `json:"sequence,string"` + Revision uint64 `json:"revision,string"` + State platform.DeliveryState `json:"state"` + AvailableAt time.Time `json:"available_at"` + LeaseEpoch uint64 `json:"lease_epoch,string"` + LeaseUntil *time.Time `json:"lease_until,omitempty"` + Attempts uint32 `json:"attempts"` + AttemptBudget uint32 `json:"attempt_budget"` + AcknowledgedAt *time.Time `json:"acknowledged_at,omitempty"` +} + +// ClaimInput 由服务端已认证的连接生成，ConnectionID不得直接使用客户端自报值。 +type ClaimInput struct { + ConnectionID core.ID + Capacity uint32 +} + +type Grant struct { + Delivery Delivery `json:"delivery"` + Event Event `json:"event"` + // LeaseSecret 仅交付给当前连接；持久层、日志和审计仅保存摘要。 + LeaseSecret string `json:"lease_secret"` +} + +type AckInput struct { + DeliveryID core.ID `json:"delivery_id"` + Epoch uint64 `json:"epoch,string"` + LeaseSecret string `json:"lease_secret"` +} + +type AckResult struct { + DeliveryID core.ID `json:"delivery_id"` + EventID core.ID `json:"event_id"` + Epoch uint64 `json:"epoch,string"` + AcknowledgedAt time.Time `json:"acknowledged_at"` + Deduplicated bool `json:"deduplicated"` +} + +type RetryInput struct { + ExpectedRevision uint64 `json:"expected_revision,string"` + Reason string `json:"reason"` + AdditionalAttempts uint32 `json:"additional_attempts"` +} + +type RetryResult struct { + Delivery Delivery `json:"delivery"` + Deduplicated bool `json:"deduplicated"` +} + +type Store interface { + CreateSubscription(context.Context, auth.Token, string, CreateInput) (SubscriptionResult, error) + GetSubscription(context.Context, auth.Token, core.ID) (Subscription, error) + ControlSubscription(context.Context, auth.Token, core.ID, string, ControlInput) (SubscriptionResult, error) + GetDelivery(context.Context, auth.Token, core.ID, core.ID) (Delivery, error) + ClaimDelivery(context.Context, auth.Token, core.ID, ClaimInput) (*Grant, error) + AckDelivery(context.Context, auth.Token, core.ID, AckInput) (AckResult, error) + RetryDelivery(context.Context, auth.Token, core.ID, core.ID, string, RetryInput) (RetryResult, error) +} + +type Service struct{ store Store } + +func New(s Store) *Service { return &Service{store: s} } + +var validName = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*$`) + +func keyValid(key string) error { + if len(key) < 1 || len(key) > 128 { + return fault.Invalid("invalid Idempotency-Key") + } + for _, c := range key { + if c < 33 || c > 126 { + return fault.Invalid("invalid Idempotency-Key") + } + } + return nil +} +func reasonValid(revision uint64, reason string) bool { + return revision > 0 && utf8.ValidString(reason) && strings.TrimSpace(reason) != "" && utf8.RuneCountInString(reason) <= 1024 +} +func (s *Service) Create(ctx context.Context, t auth.Token, key string, in CreateInput) (SubscriptionResult, error) { + if err := keyValid(key); err != nil { + return SubscriptionResult{}, err + } + if len(in.Name) > 128 || !validName.MatchString(in.Name) || in.QueueID.Validate() != nil { + return SubscriptionResult{}, fault.Invalid("invalid subscription name or queue_id") + } + if in.MaxPending == 0 { + in.MaxPending = 100 + } + if in.MaxAttempts == 0 { + in.MaxAttempts = 20 + } + if in.MaxPending > 1000 || in.MaxAttempts > 100 { + return SubscriptionResult{}, fault.Invalid("subscription limits are invalid") + } + if len(in.Types) == 0 { + in.Types = []string{"task.succeeded", "task.failed", "task.cancelled", "task.needs_review"} + } + if len(in.Types) > 4 { + return SubscriptionResult{}, fault.Invalid("too many event types") + } + in.Types = slices.Clone(in.Types) + slices.Sort(in.Types) + for i, v := range in.Types { + if !ResultType(v) || i > 0 && v == in.Types[i-1] { + return SubscriptionResult{}, fault.Invalid("unsupported or repeated event type") + } + } + return s.store.CreateSubscription(ctx, t, key, in) +} +func (s *Service) Get(ctx context.Context, t auth.Token, id core.ID) (Subscription, error) { + if id.Validate() != nil { + return Subscription{}, fault.Invalid("invalid subscription_id") + } + return s.store.GetSubscription(ctx, t, id) +} +func (s *Service) Control(ctx context.Context, t auth.Token, id core.ID, key string, in ControlInput) (SubscriptionResult, error) { + if err := keyValid(key); err != nil { + return SubscriptionResult{}, err + } + if id.Validate() != nil || !reasonValid(in.ExpectedRevision, in.Reason) { + return SubscriptionResult{}, fault.Invalid("subscription_id, revision and reason are required") + } + return s.store.ControlSubscription(ctx, t, id, key, in) +} +func (s *Service) Delivery(ctx context.Context, t auth.Token, id, delivery core.ID) (Delivery, error) { + if id.Validate() != nil || delivery.Validate() != nil { + return Delivery{}, fault.Invalid("invalid subscription_id or delivery_id") + } + return s.store.GetDelivery(ctx, t, id, delivery) +} +func (s *Service) Claim(ctx context.Context, t auth.Token, id core.ID, in ClaimInput) (*Grant, error) { + if id.Validate() != nil || in.ConnectionID.Validate() != nil || in.Capacity < 1 || in.Capacity > 32 { + return nil, fault.Invalid("invalid delivery claim") + } + return s.store.ClaimDelivery(ctx, t, id, in) +} +func (s *Service) Ack(ctx context.Context, t auth.Token, id core.ID, in AckInput) (AckResult, error) { + if id.Validate() != nil || in.DeliveryID.Validate() != nil || in.Epoch == 0 { + return AckResult{}, fault.Invalid("invalid delivery confirmation") + } + if _, err := SecretDigest(in.LeaseSecret); err != nil { + return AckResult{}, err + } + return s.store.AckDelivery(ctx, t, id, in) +} +func (s *Service) Retry(ctx context.Context, t auth.Token, id, delivery core.ID, key string, in RetryInput) (RetryResult, error) { + if err := keyValid(key); err != nil { + return RetryResult{}, err + } + if id.Validate() != nil || delivery.Validate() != nil || !reasonValid(in.ExpectedRevision, in.Reason) || in.AdditionalAttempts < 1 || in.AdditionalAttempts > 100 { + return RetryResult{}, fault.Invalid("invalid delivery retry") + } + return s.store.RetryDelivery(ctx, t, id, delivery, key, in) +} + +func SecretDigest(secret string) (core.Digest, error) { + if len(secret) != 43 { + return core.Digest{}, fault.Invalid("invalid delivery secret") + } + raw, err := base64.RawURLEncoding.Strict().DecodeString(secret) + if err != nil || len(raw) != 32 { + return core.Digest{}, fault.Invalid("invalid delivery secret") + } + return core.Digest(sha256.Sum256(raw)), nil +} +func NewSecret() (string, core.Digest, error) { + var raw [32]byte + if _, err := rand.Read(raw[:]); err != nil { + return "", core.Digest{}, err + } + return base64.RawURLEncoding.EncodeToString(raw[:]), core.Digest(sha256.Sum256(raw[:])), nil +} diff --git a/platform/events/service_test.go b/platform/events/service_test.go new file mode 100644 index 0000000..f725653 --- /dev/null +++ b/platform/events/service_test.go @@ -0,0 +1,84 @@ +package events + +import ( + "context" + "strings" + "testing" + + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +type recordingStore struct { + Store + input CreateInput + calls int +} + +func (s *recordingStore) CreateSubscription(_ context.Context, _ auth.Token, _ string, in CreateInput) (SubscriptionResult, error) { + s.input = in + s.calls++ + return SubscriptionResult{}, nil +} + +func TestSubscriptionNormalizationAndValidation(t *testing.T) { + store := &recordingStore{} + svc := New(store) + good := CreateInput{Name: "object-results", QueueID: core.ID(strings.Repeat("1", 32))} + if _, err := svc.Create(context.Background(), auth.Token{}, "create", good); err != nil { + t.Fatal(err) + } + if store.input.MaxPending != 100 || store.input.MaxAttempts != 20 || len(store.input.Types) != 4 || store.input.FromSequence != nil { + t.Fatal("invalid defaults") + } + // 排序仅作用于内部副本，不修改调用方重报保存的原输入。 + types := []string{"task.succeeded", "task.cancelled"} + good.Types = types + if _, err := svc.Create(context.Background(), auth.Token{}, "create", good); err != nil { + t.Fatal(err) + } + if types[0] != "task.succeeded" || store.input.Types[0] != "task.cancelled" { + t.Fatal("normalization modified caller") + } + cases := []func(*CreateInput){ + func(in *CreateInput) { in.Name = " spaced " }, + func(in *CreateInput) { in.QueueID = "missing" }, + func(in *CreateInput) { in.MaxPending = 1001 }, + func(in *CreateInput) { in.MaxAttempts = 101 }, + func(in *CreateInput) { in.Types = []string{"task.accepted"} }, + func(in *CreateInput) { in.Types = []string{"task.failed", "task.failed"} }, + } + for i, mutate := range cases { + in := good + mutate(&in) + if _, err := svc.Create(context.Background(), auth.Token{}, "create", in); err == nil { + t.Fatalf("accepted invalid case %d", i) + } + } + for _, key := range []string{"", "with space", strings.Repeat("a", 129)} { + if _, err := svc.Create(context.Background(), auth.Token{}, key, good); err == nil { + t.Fatal("accepted invalid identity") + } + } + if store.calls != 2 { + t.Fatal("invalid input reached storage") + } +} +func TestDeliveryProofStrictness(t *testing.T) { + secret, digest, err := NewSecret() + if err != nil { + t.Fatal(err) + } + actual, err := SecretDigest(secret) + if err != nil || actual != digest { + t.Fatal("proof roundtrip", err) + } + for _, s := range []string{"", secret + "=", secret + "\n", secret[:42], strings.Repeat("!", 43)} { + if _, err := SecretDigest(s); err == nil { + t.Fatal("accepted noncanonical proof") + } + } + if ResultType("task.requeued") || ResultType("task.accepted") || !ResultType("task.needs_review") { + t.Fatal("invalid result event scope") + } +} diff --git a/platform/model/delivery.go b/platform/model/delivery.go index 483372c..d0b9622 100644 --- a/platform/model/delivery.go +++ b/platform/model/delivery.go @@ -27,6 +27,10 @@ type Delivery struct { LeaseUntil *time.Time `db:"lease_until"` // Attempts 已产生的交付尝试次数，不等同于业务执行次数。 Attempts uint32 `db:"attempts"` + // AttemptBudget 累计允许发送次数；人工追加预算，不重置Attempts或LeaseEpoch。 + AttemptBudget uint32 `db:"attempt_budget"` + // TokenDigest 最近交付秘密的摘要；首次发送前为空，确认后保留以验证重复确认。 + TokenDigest *core.Digest `db:"token_digest"` // AcknowledgedAt 接收方可靠处理后的确认时间；未确认时为空。 AcknowledgedAt *time.Time `db:"acknowledged_at"` } diff --git a/platform/model/event_stream.go b/platform/model/event_stream.go new file mode 100644 index 0000000..7dc5db5 --- /dev/null +++ b/platform/model/event_stream.go @@ -0,0 +1,38 @@ +package model + +import ( + core "git.yuntuo.hk/k2safe/orbit/model" + "time" +) + +// EventStream 表示命名空间结果流；独立锁行避免升级认证身份共享锁。 +// 对应 orbit_event_streams；序号分配必须与任务状态和结果日志同事务提交。 +type EventStream struct { + // NamespaceID 结果流所属命名空间，也是序号分配锁的主键。 + NamespaceID core.ID `db:"namespace_id"` + // LastSequence 已分配序号，从1开始；回滚不消耗序号，0表示尚无结果。 + LastSequence uint64 `db:"last_sequence"` + // RetainedFromSequence 最早保留序号；首版不清理，固定为1，防止以后静默跨过缺失历史。 + RetainedFromSequence uint64 `db:"retained_from_sequence"` + // CreatedAt 结果流创建时间，统一UTC，精度微秒。 + CreatedAt time.Time `db:"created_at"` + // UpdatedAt 最后分配序号时间，不能代替结果序号。 + UpdatedAt time.Time `db:"updated_at"` +} + +func (EventStream) TableName() string { return "orbit_event_streams" } + +// ResultEvent 表示从迁移13启用的不可变任务结果快照；不补造历史任务事件。 +// 对应 orbit_result_events；Record.ID 与来源 Event.ID 相同，关联由事务校验。 +type ResultEvent struct { + // Record 包含所属命名空间、来源事件标识、固定版本1及UTC时间。 + core.Record + // Sequence 命名空间内提交有序的结果序号，首个为1。 + Sequence uint64 `db:"sequence"` + // QueueID 来源任务队列，订阅必须具备该队列的消费权限。 + QueueID core.ID `db:"queue_id"` + // Snapshot 版本化结果快照JSON，保留数值精度，不包含任务输入或租约秘密。 + Snapshot core.JSON `db:"snapshot"` +} + +func (ResultEvent) TableName() string { return "orbit_result_events" } diff --git a/platform/model/subscription.go b/platform/model/subscription.go index aecd947..fdabf3e 100644 --- a/platform/model/subscription.go +++ b/platform/model/subscription.go @@ -13,10 +13,20 @@ type Subscription struct { Name string `db:"name"` // CredentialID 关联的命名空间凭据标识。 CredentialID core.ID `db:"credential_id"` - // Filter 订阅过滤规则的 JSON 文档。 + // Filter 冻结的结果事件类型数组JSON；创建后不能改写，以免静默漏消费。 Filter core.JSON `db:"filter"` - // Enabled 是否启用；关闭后拒绝相应的新接入或新操作。 + // Enabled 是否启用新交付；暂停不撤销已发出的结果，也不拒绝有效确认。 Enabled bool `db:"enabled"` + // QueueID 创建时冻结的队列；历史占位记录为空且不允许消费。 + QueueID *core.ID `db:"queue_id"` + // CursorSequence 已扫描并物化交付的结果序号，不表示消费者已确认。 + CursorSequence uint64 `db:"cursor_sequence"` + // PendingCount 未确认交付数，包含次数耗尽记录；每次有效确认只减一次。 + PendingCount uint32 `db:"pending_count"` + // MaxPending 未确认交付上限，1至1000；达到后保留来源日志并暂停物化。 + MaxPending uint32 `db:"max_pending"` + // MaxAttempts 新交付的初始发送次数上限，1至100，不等同于任务执行次数。 + MaxAttempts uint32 `db:"max_attempts"` } func (Subscription) TableName() string { return "orbit_subscriptions" }