diff --git a/docs/reports/phase-6-slow-consumer.md b/docs/reports/phase-6-slow-consumer.md index b93b4e8..ca761b5 100644 --- a/docs/reports/phase-6-slow-consumer.md +++ b/docs/reports/phase-6-slow-consumer.md @@ -17,13 +17,15 @@ | 每条交付 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。 +`TestIntegrationEventProcessSlowConsumerPeerProgress` 另外启动两个真实 Orbit 进程和两个独立 WebSocket 消费器:节点 A 持有第一条交付并阻塞,节点 B 在同一订阅上领取并确认第二条交付;释放 A 后第一条再确认。两个节点身份不同,最终两条交付仍各一次尝试。 + +同批定向真实服务回归还运行了 ACK 提交回执丢失和自然租约补发:三项合计 52.645 秒,均通过;单节点慢消费者用例本身 3.15 秒。双进程用例另以两个真实 Orbit 进程运行 3.125 秒并通过。环境为 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 去重、数据库锁/连接池/磁盘曲线和生产容量。上述项目必须在目标拓扑中另行执行,不能由本次单节点样本推导。 +当前仍未验证连接风暴叠加积压、租约到期期间的业务 Inbox 去重、数据库锁/连接池/磁盘曲线和目标拓扑生产容量;本次双进程只证明一个慢消费者不会阻塞同一订阅的对等节点,不证明跨故障域切换、全局公平或容量上限。上述项目必须在目标拓扑中另行执行,不能由本次样本推导。 ## 复现 diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index 191cd8f..27369a4 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -23,7 +23,7 @@ | 真实定时集中触发测量 | `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) | +| 真实慢消费者背压 | MySQL 8.4.10 一次性库、真实 Orbit HTTP/WebSocket;单节点 `TestIntegrationEventConsumerSlowHandlerBackpressure` 与双进程 `TestIntegrationEventProcessSlowConsumerPeerProgress` | 容量1阻塞首个处理器时第二条交付保留在服务端;对等节点仍可推进,释放后2次处理/2次ACK、每条attempts=1;连接风暴、资源曲线和生产容量仍未验收,见[慢消费者报告](phase-6-slow-consumer.md) | ## 当前环境仍未覆盖 diff --git a/sdk/go/event_process_integration_test.go b/sdk/go/event_process_integration_test.go index de68874..ec3d7ca 100644 --- a/sdk/go/event_process_integration_test.go +++ b/sdk/go/event_process_integration_test.go @@ -101,3 +101,95 @@ func TestIntegrationEventProcessAckCommitResponseLoss(t *testing.T) { } t.Log("ACK commit verified before response loss and SIGKILL; replacement process returned original receipt without another handler or delivery attempt") } + +func TestIntegrationEventProcessSlowConsumerPeerProgress(t *testing.T) { + a := startOrbitProcess(t) + _, _, token := integrationWorkerClient(t) + apiA := processClient(t, a, token) + b := startOrbitProcess(t) + apiB := processClient(t, b, token) + if a.nodeID == b.nodeID { + t.Fatal("server processes reused node identity") + } + ctx, stop := context.WithTimeout(context.Background(), 45*time.Second) + defer stop() + sub := integrationEventSubscription(t, ctx, apiA, 3) + _ = integrationCancelledEvent(t, ctx, apiA, sub) + started := make(chan struct{}) + release := make(chan struct{}) + var callsA, callsB atomic.Int64 + consumerA, err := NewEventConsumer(apiA, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, _ ResultEvent) error { + if callsA.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() + doneA := make(chan error, 1) + go func() { doneA <- consumerA.Run(run) }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("node A slow handler did not start") + case err := <-doneA: + t.Fatal("node A consumer stopped", err) + } + consumerB, err := NewEventConsumer(apiB, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(context.Context, ResultEvent) error { + callsB.Add(1) + return nil + }}) + if err != nil { + t.Fatal(err) + } + doneB := make(chan error, 1) + go func() { doneB <- consumerB.Run(run) }() + awaitConsumerEvent(t, consumerB, "connected") + _ = integrationCancelledEvent(t, ctx, apiA, sub) + deadline := time.Now().Add(5 * time.Second) + for consumerB.Stats().Acknowledged < 1 { + select { + case <-ctx.Done(): + t.Fatal("node B did not progress while node A was slow") + case err := <-doneB: + t.Fatal("node B consumer stopped", err) + case <-time.After(20 * time.Millisecond): + } + if time.Now().After(deadline) { + t.Fatal("node B did not acknowledge peer delivery before deadline") + } + } + if callsA.Load() != 1 || callsB.Load() != 1 || consumerA.Stats().Acknowledged != 0 { + t.Fatalf("peer progress violated slow consumer isolation: callsA=%d callsB=%d statsA=%+v statsB=%+v", callsA.Load(), callsB.Load(), consumerA.Stats(), consumerB.Stats()) + } + close(release) + for consumerA.Stats().Acknowledged < 1 { + select { + case <-ctx.Done(): + t.Fatal("node A did not drain its held delivery") + case err := <-doneA: + t.Fatal("node A consumer stopped", err) + case <-time.After(20 * time.Millisecond): + } + } + cancel() + if err := awaitConsumerStop(t, doneA); err != nil { + t.Fatal(err) + } + if err := awaitConsumerStop(t, doneB); err != nil { + t.Fatal(err) + } + page, err := apiB.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked}) + if err != nil || len(page.Items) != 2 || page.Items[0].Attempts != 1 || page.Items[1].Attempts != 1 { + t.Fatalf("peer delivery accounting failed: err=%v deliveries=%+v", err, page.Items) + } + t.Log("node B acknowledged a second delivery while node A's capacity-one handler was blocked; both nodes preserved one attempt per delivery") +}