diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index cb035d9..87d5d5b 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -59,10 +59,10 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( ### P6 首批实施顺序 1. 已补[首批可复现故障工具](reports/phase-6-fault-boundaries.md)：A04“派发已提交但消息尚未送达”、W01/W13事务提交结果不明，以及W16最终回执/旧PENDING结果重放/下一轮竞争。迁移器同时提供当前版本无外键新装基线；每个切点必须有实际数据库或进程证据，不能用响应错误推断事务是否已提交，也不能用普通SIGNAL竞争替代最终查询场景。 -2. 已实现[基础运行观测](api/observability.md)及[有界业务状态与事件采样](reports/phase-6-business-observability.md)：HTTP/WS、数据库连接池、后台批次、固定任务/交付/流程/Inbox状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供，运维人员也可用 `orbit-admin observe snapshot` 获取同一有界单节点快照。仍需端到端外部效果延迟、全局持久吞吐、Inbox从事件产生到推进完成的精确阶段延迟、全局发送队列、多节点聚合及生产采集/通知验证；SDK受控重连风暴已补[边界回归](reports/phase-6-reconnect-storm.md)，真实慢消费者和跨节点重连仍待目标拓扑。 +2. 已实现[基础运行观测](api/observability.md)及[有界业务状态与事件采样](reports/phase-6-business-observability.md)：HTTP/WS、数据库连接池、后台批次、固定任务/交付/流程/Inbox状态计数、最近事件类别、终态耗时样本、固定事件阶段耗时、最早年龄、截断/新鲜度指标和可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签；发布前只读结构核验由[`orbit-admin verify`](api/operations.md)提供，运维人员也可用 `orbit-admin observe snapshot` 获取同一有界单节点快照。仍需端到端外部效果延迟、全局持久吞吐、Inbox从事件产生到推进完成的精确阶段延迟、全局发送队列、多节点聚合及生产采集/通知验证；SDK受控重连风暴和真实慢消费者背压已补[边界回归](reports/phase-6-reconnect-storm.md)与[慢消费者报告](reports/phase-6-slow-consumer.md)，跨节点重连仍待目标拓扑。 3. 已补齐受保护管理环境中的固定凭据操作文件、轮换、撤销、幂等回执和脱敏管理审计；轮换保留授权绑定并要求新令牌重新建立会话，旧 Worker 会话被排空且立即过期，撤销允许在停用范围内执行且不改写历史记录。已用一次性 MySQL 定向验证旧令牌失效、新令牌重绑和零外键；仍需补齐提交不明故障切点、多节点安全重绑演练、凭据最小权限配置核对和生产密钥托管/审计演练。 4. 已补[`orbit-admin retention inventory`](api/retention.md)只读盘点和[阶段报告](reports/phase-6-retention-inventory.md)，能区分过期幂等、确定终态、未知效果、未终结流程、已确认/未确认交付、已消费/未消费Inbox、事件和审计规模；它不执行归档或删除。后续仍需评审各类保留周期、归档格式、受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 -5. 已补[`orbit-load`](api/load.md)有界受理负载工具及24次/1000次隔离样本，见[受理报告](reports/phase-6-load.md)。新增独立[`orbit-exec-load`](api/execution-load.md)，通过真实Worker、任务结果和尝试历史核对，提供短/混合/中长任务的受理、排队、授权执行及完成耗时分布；具体环境、原始JSON和失败分母见[执行测量报告](reports/phase-6-execution-load.md)。新增[`orbit-workflow-load`](api/workflow-load.md)，用本次发布版本的双分支并行流程，逐个核对不同任务、唯一尝试、来源、精确输出与本地调用；24/100实例及24实例截止负例已重新构建回归并绑定源码/二进制哈希，见[流程测量报告](reports/phase-6-workflow-load.md)。新增[`orbit-schedule-load`](api/schedule-load.md)，已完成24/128次定时集中触发与24次计划量的截止负例，保存实际分母、唯一任务/尝试、来源与暂停收尾证据，见[定时测量报告](reports/phase-6-schedule-load.md)。标准Go 1.27.1全量仍待复跑，已补[长期积压与资源曲线样本](reports/phase-6-backlog.md)：32×8 共256个触发点，在1个Worker/容量1和180秒截止下保留173个成功、83个未核验任务，并以独立数据库读取记录等待积压曲线；该单节点样本不替代目标拓扑容量。下一步补指标采集对照、长期存储增长、慢消费者/重连风暴和目标数据库拓扑容量。 +5. 已补[`orbit-load`](api/load.md)有界受理负载工具及24次/1000次隔离样本，见[受理报告](reports/phase-6-load.md)。新增独立[`orbit-exec-load`](api/execution-load.md)，通过真实Worker、任务结果和尝试历史核对，提供短/混合/中长任务的受理、排队、授权执行及完成耗时分布；具体环境、原始JSON和失败分母见[执行测量报告](reports/phase-6-execution-load.md)。新增[`orbit-workflow-load`](api/workflow-load.md)，用本次发布版本的双分支并行流程，逐个核对不同任务、唯一尝试、来源、精确输出与本地调用；24/100实例及24实例截止负例已重新构建回归并绑定源码/二进制哈希，见[流程测量报告](reports/phase-6-workflow-load.md)。新增[`orbit-schedule-load`](api/schedule-load.md)，已完成24/128次定时集中触发与24次计划量的截止负例，保存实际分母、唯一任务/尝试、来源与暂停收尾证据，见[定时测量报告](reports/phase-6-schedule-load.md)。标准Go 1.27.1全量仍待复跑，已补[长期积压与资源曲线样本](reports/phase-6-backlog.md)：32×8 共256个触发点，在1个Worker/容量1和180秒截止下保留173个成功、83个未核验任务，并以独立数据库读取记录等待积压曲线；该单节点样本不替代目标拓扑容量。SDK受控重连风暴及[真实慢消费者背压](reports/phase-6-slow-consumer.md)已补边界回归；下一步补指标采集对照、长期存储增长、跨节点重连与目标数据库拓扑容量。 6. 已补[隔离备份恢复演练](api/backup-restore.md)：源库和恢复库均为一次性 MySQL 容器，逻辑备份恢复后核对迁移账本、核心记录数量、固定索引和零外键；证据见[恢复报告](reports/phase-6-backup-restore.md)。该演练不替代生产备份介质、复制切换、跨版本恢复或 RPO/RTO 验收。 7. 已补[A18 数据库短暂不可用恢复演练](api/db-recovery.md)：readiness 在数据库停止期间返回 503，同卷恢复后返回 200，并用原权限范围继续受理任务；证据见[数据库恢复报告](reports/phase-6-db-recovery.md)。该演练不替代主从切换、复制确认、跨故障域或 RPO/RTO 验收。 diff --git a/docs/release-checklist.md b/docs/release-checklist.md index f4c7b2b..59102d9 100644 --- a/docs/release-checklist.md +++ b/docs/release-checklist.md @@ -41,7 +41,7 @@ | 受理、派发、流程推进和恢复吞吐 | 固定硬件、数据量、任务/流程混合模型，p50/p95/p99和资源曲线 | 已有受理样本、[真实Worker短/混合/中长执行样本](reports/phase-6-execution-load.md)、[Weave双分支流程样本](reports/phase-6-workflow-load.md)及[定时集中触发/积压曲线样本](reports/phase-6-backlog.md)；目标硬件资源曲线、生产容量和HA仍未验收 | | 100万+企业规模推导 | 租户、队列、连接、事件、索引和保存周期的容量模型 | 未验收，不能由租户数直接推导 | | MySQL 复制、切换、RPO/RTO | 目标拓扑上的实际切换、复制确认、恢复点和恢复时间 | 未验收 | -| 慢消费者、重连风暴、磁盘和锁等待 | 受控故障负载及积压清空证据 | SDK 受控连续断线边界已回归（见[重连报告](reports/phase-6-reconnect-storm.md)）；真实慢消费者、跨节点风暴、磁盘和锁等待未验收 | +| 慢消费者、重连风暴、磁盘和锁等待 | 受控故障负载及积压清空证据 | SDK 受控连续断线和真实 HTTP/WebSocket 容量1慢消费者已回归（见[重连报告](reports/phase-6-reconnect-storm.md)、[慢消费者报告](reports/phase-6-slow-consumer.md)）；跨节点风暴、磁盘和锁等待未验收 | | 跨版本 SDK/服务端 | 至少一个正式旧标签与候选版本的双向矩阵 | 当前仓库无正式标签，未验收 | 发布候选只有在所有“待目标环境/未验收”项得到对应环境证据后，才能进入生产审批。当前代码和阶段回归不等同于生产高可用、容量、RPO/RTO或百万企业承诺。 diff --git a/docs/reports/phase-6-slow-consumer.md b/docs/reports/phase-6-slow-consumer.md new file mode 100644 index 0000000..b93b4e8 --- /dev/null +++ b/docs/reports/phase-6-slow-consumer.md @@ -0,0 +1,39 @@ +# P6 真实慢消费者背压回归 + +日期：2026-09-20。本批在一次性 MySQL 和真实 Orbit HTTP/WebSocket 服务上验证 SDK 消费器容量为 1 时的慢处理器背压。它补充 SDK 连接层重连边界，不代表跨节点重连、全局积压容量、磁盘/锁等待或生产 RPO/RTO 验收。 + +## 场景和结果 + +`TestIntegrationEventConsumerSlowHandlerBackpressure` 先投递一条取消结果，启动正式 `EventConsumer`，让第一个处理器停在业务处理阶段；处理器容量固定为 1。随后再产生第二条结果，测试通过真实订阅读取确认两条交付均已持久化，但在释放第一个处理器之前没有启动第二次处理，也没有提交 ACK。释放后两条交付都完成并各自只尝试一次。 + +| 证据 | 结果 | +| --- | ---: | +| 消费器本地容量 | 1 | +| 持久化交付 | 2 | +| 处理器调用（释放前） | 1 | +| ACK（释放前） | 0 | +| 处理器调用（最终） | 2 | +| ACK（最终） | 2 | +| 每条交付 attempts | 1、1 | +| 重复处理 | 0 | + +同批定向真实服务回归还运行了 ACK 提交回执丢失和自然租约补发：三项合计 52.645 秒，均通过；慢消费者用例本身 3.15 秒。环境为 Docker Go 1.25.14 Linux arm64、一次性 MySQL 8.4.10，服务端和 SDK 使用临时 modfile，仓库声明的 Go 1.27.1 未修改。`go test -race` 通过，数据库外键为 0。 + +## 结论边界 + +该用例证明容量窗口会把慢处理器期间的后续结果留在服务端，释放后按原交付身份继续消费，不会在 SDK 内存中并行启动超出容量的处理器。业务处理器仍必须响应 context 取消，并在业务 Inbox 去重与业务效果同一事务提交后才返回成功。 + +当前仍未验证多个节点同时连接、跨节点慢消费者公平性、连接风暴叠加积压、租约到期期间的业务 Inbox 去重、数据库锁/连接池/磁盘曲线和生产容量。上述项目必须在目标拓扑中另行执行，不能由本次单节点样本推导。 + +## 复现 + +在一次性 MySQL、迁移完成、带 `worker-access` 与 `event-access` 的凭据以及真实 Orbit 服务后，执行： + +```sh +cd sdk/go +ORBIT_TEST_WORKER_URL=http://127.0.0.1:$PORT \ +ORBIT_TEST_WORKER_CREDENTIALS_FILE=/tmp/orbit-event-test/credentials.json \ +go test -race -count=1 -run '^TestIntegrationEventConsumer(HeartbeatAndLostAck|FailureRedeliversNaturally|SlowHandlerBackpressure)$' -v . +``` + +缺少隔离服务或凭据时测试会跳过，跳过不能记作通过。标准 `make integration` 会在下一次完整回归中自动包含该用例。 diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index 60d2727..191cd8f 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -23,12 +23,13 @@ | 真实定时集中触发测量 | `orbit-schedule-load`；4×6及32×4触发，2×12截止负例，见[定时测量报告](phase-6-schedule-load.md) | 正常152项逐项核验；负例18个未观察点、4个未核验任务，非零退出；38个计划均暂停，实际库外键0。SDK全模块竞态/vet/build通过；生产容量未验收 | | 长期积压与曲线 | `make schedule-backlog`；32×8共256个触发点，1 Worker/容量1，独立连接每2秒读取状态 | 最高等待249，180秒截止时173成功/83未核验，计划全部暂停，实际库外键0；目标硬件资源与生产容量未验收，见[积压报告](phase-6-backlog.md) | | SDK 重连边界 | `TestEventConsumerReconnectStormUsesBoundedBackoff`，真实 TLS 测试服务器，4次连续断线 | 第5次连接恢复，重连4次、处理器调用1次、ACK1次；真实慢消费者/跨节点重连仍未验收，见[重连报告](phase-6-reconnect-storm.md) | +| 真实慢消费者背压 | MySQL 8.4.10 一次性库、真实 Orbit HTTP/WebSocket、`TestIntegrationEventConsumerSlowHandlerBackpressure` | 容量1阻塞首个处理器时第二条交付保留在服务端；释放后2次处理/2次ACK、每条attempts=1；跨节点公平、资源曲线和生产容量仍未验收，见[慢消费者报告](phase-6-slow-consumer.md) | ## 当前环境仍未覆盖 此前使用容器网络别名运行存储集成时，两组 COMMIT 故障测试曾因自身 `127.0.0.1` 明文连接前置条件被拒绝；本轮已改用共享网络命名空间单独复跑并通过，因此不再把它们列为未验证场景。共享网络命名空间只解决测试进程的 loopback 观察条件，不改变生产部署约束。 -当前仍未完成仓库声明 Go 1.27.1 下的无过滤标准全量回归，以及 SDK/服务端跨标签矩阵；新增长期积压样本见[报告](phase-6-backlog.md)，仍未升级为生产容量结论。上述 Docker Go 1.25 共享网络结果补齐了当前容器环境的进程故障证据，但不替代目标发布环境的工具链、拓扑和版本兼容门禁。 +当前仍未完成仓库声明 Go 1.27.1 下的无过滤标准全量回归，以及 SDK/服务端跨标签矩阵；新增长期积压和慢消费者样本见[报告](phase-6-backlog.md)与[慢消费者报告](phase-6-slow-consumer.md)，仍未升级为生产容量结论。上述 Docker Go 1.25 共享网络结果补齐了当前容器环境的进程故障证据，但不替代目标发布环境的工具链、拓扑和版本兼容门禁。 历史 macOS arm64 + Go 1.27.1 + loopback MySQL 的完整切点记录见[故障边界报告](phase-6-fault-boundaries.md)。当前 Docker Go 1.25 结果仍不升级为仓库声明 Go 1.27.1 的正式全量结论。 diff --git a/sdk/go/event_integration_test.go b/sdk/go/event_integration_test.go index b75661c..67f6cef 100644 --- a/sdk/go/event_integration_test.go +++ b/sdk/go/event_integration_test.go @@ -149,6 +149,79 @@ func TestIntegrationEventConsumerHeartbeatAndLostAck(t *testing.T) { } t.Log("real 15-second heartbeat and committed ACK loss recovered; handler and delivery attempt each remained one") } +func TestIntegrationEventConsumerSlowHandlerBackpressure(t *testing.T) { + api, _, _ := integrationWorkerClient(t) + ctx, stop := context.WithTimeout(context.Background(), 25*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, api, 3) + _ = integrationCancelledEvent(t, ctx, api, sub) + started := make(chan struct{}) + release := make(chan struct{}) + var calls atomic.Int64 + consumer, err := NewEventConsumer(api, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, _ ResultEvent) error { + if calls.Add(1) == 1 { + close(started) + select { + case <-release: + case <-work.Done(): + return work.Err() + } + } + return nil + }}) + if err != nil { + t.Fatal(err) + } + run, cancel := context.WithCancel(ctx) + defer cancel() + done := make(chan error, 1) + go func() { done <- consumer.Run(run) }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("first slow handler did not start") + case err := <-done: + t.Fatal("consumer stopped", err) + } + _ = integrationCancelledEvent(t, ctx, api, sub) + deadline := time.Now().Add(2 * time.Second) + for { + current, getErr := api.GetSubscription(ctx, sub.ID) + if getErr != nil { + t.Fatal(getErr) + } + if current.PendingCount == 2 { + break + } + if time.Now().After(deadline) { + t.Fatalf("slow handler did not retain second delivery: pending=%d", current.PendingCount) + } + time.Sleep(20 * time.Millisecond) + } + time.Sleep(150 * time.Millisecond) + if calls.Load() != 1 || consumer.Stats().Acknowledged != 0 { + t.Fatalf("capacity-one consumer started or acknowledged work while handler was blocked: calls=%d stats=%+v", calls.Load(), consumer.Stats()) + } + close(release) + for consumer.Stats().Acknowledged < 2 { + select { + case <-ctx.Done(): + t.Fatal("slow consumer did not drain both deliveries") + case err := <-done: + t.Fatal("consumer stopped", err) + case <-time.After(20 * time.Millisecond): + } + } + cancel() + if err := awaitConsumerStop(t, done); err != nil { + t.Fatal(err) + } + page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked}) + if err != nil || len(page.Items) != 2 || page.Items[0].Attempts != 1 || page.Items[1].Attempts != 1 || calls.Load() != 2 { + t.Fatalf("slow-consumer backpressure or delivery accounting failed: err=%v deliveries=%+v calls=%d", err, page.Items, calls.Load()) + } + t.Log("capacity-one slow handler held the second delivery until release; both durable deliveries were acknowledged once") +} func TestIntegrationEventConsumerFailureRedeliversNaturally(t *testing.T) { api, _, _ := integrationWorkerClient(t) ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)