From 125740f58dc7be795811d9e69f1e755f4a3cbb9a Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 21:23:34 +0800 Subject: [PATCH] test: cover capacity reservation lifecycle --- docs/development/implementation-plan.md | 4 + docs/requirements/completion-audit.md | 3 +- docs/testing/test-strategy.md | 3 +- findings.md | 4 +- internal/domain/proxy/capacity_test.go | 191 ++++++++++++++++++++++++ progress.md | 3 + 6 files changed, 204 insertions(+), 4 deletions(-) create mode 100644 internal/domain/proxy/capacity_test.go diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 8f9cb33..a332bd9 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -84,6 +84,10 @@ test/{fixtures,integration,e2e,load}/ - [x] Prove with 1,000 concurrent goroutines that effective capacity is never exceeded. - [ ] Add race coverage and duplicate-release invariant metrics hook. +当前进度（2026-07-29）：固定 Max 下的每 Proxy 打包 CAS、Cancel/Commit/Release +生命周期、重复终结、错误顺序和同一 Reservation 并发终结已通过领域测试；动态 +降容契约、低基数不变量指标、Linux race 证据及短 TTL runtime 排空回收待完成。 + ## Task 4: Routing and Sequential Switching **Files:** `internal/domain/routing/*.go`, corresponding tests diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index a78df82..f0dda09 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -25,7 +25,8 @@ - `CFG-*`：YAML v4 未知字段拒绝、监听保护、引用/上限/认证边界校验，21 份 配置持续测试。 - `PROXY-* / CAP-*`：唯一键、TTL 优先级、状态迁移与 Active/Reserved 打包 - 原子计数；1,000 goroutine 不超卖测试。 + 原子计数；1,000 goroutine 不超卖，以及 Cancel、重复终结、错误顺序、并发 + Commit/Cancel/Release 计数守恒测试。 - `ROUTE-001 / ROUTE-004`：首条命中规则与进程内 Concurrent Sequential 单次 切换；策略运行时接线、持久化恢复和跨实例 CAS 尚未完成。 - `FETCH-005 / FETCH-006`：Valid、Empty、DuplicateOnly、Error 分类。 diff --git a/docs/testing/test-strategy.md b/docs/testing/test-strategy.md index 3cd7743..b7aa419 100644 --- a/docs/testing/test-strategy.md +++ b/docs/testing/test-strategy.md @@ -48,7 +48,8 @@ 1. **并发容量**：1000 协程争用同一 Proxy，始终满足 `active + reserved <= effectiveMaxConcurrency`。 2. **Reservation 生命周期**：Dial 成功/失败、超时、取消和重复 Release 均不 - 泄漏或产生负计数。 + 泄漏或产生负计数。领域层已覆盖 Cancel、重复终结、错误顺序和同一 + Reservation 并发终结；Gateway Handler 覆盖建连失败、重试和请求取消。 3. **singleflight**：100 个缺池信号只产生一个有效 Fetch 调度。 4. **Provider 限流**：requestInterval、maxInFlight、timeout、重试和 429 `Retry-After` 在虚拟时钟下准确。 diff --git a/findings.md b/findings.md index e7cb00c..3627011 100644 --- a/findings.md +++ b/findings.md @@ -122,8 +122,8 @@ Routing 自上而下匹配，首条命中停止；支持 Gateway 与 Extract 两 - 每个 Proxy ID 已有独立打包原子计数，固定 Max 下 1,000 并发不会超卖；这满足 当前 Gateway 热路径的基本预留不变量。 -- Reservation 缺少 Cancel、重复终结、错误顺序和并发 Commit/Cancel 的完整领域 - 测试，Gateway 也会忽略 Release/Cancel 错误，尚无低基数不变量观测 seam。 +- Reservation 已补齐 Cancel、重复终结、错误顺序和并发 Commit/Cancel/Release + 的领域测试；Gateway 仍会忽略 Release/Cancel 错误，尚无低基数不变量观测 seam。 - `SetMax` 与 counters 分离更新；降到当前占用以下时会出现 overcommitted 状态， 需要先确定“拒绝降容”或“允许排空”的正式契约。 - Snapshot Store 永久保留见过的 Proxy ID 对应 Capacity；短 TTL、高换 IP 场景下 diff --git a/internal/domain/proxy/capacity_test.go b/internal/domain/proxy/capacity_test.go new file mode 100644 index 0000000..428aeff --- /dev/null +++ b/internal/domain/proxy/capacity_test.go @@ -0,0 +1,191 @@ +package proxy + +import ( + "errors" + "sync" + "sync/atomic" + "testing" +) + +func TestReservationCancelReleasesReservedCapacity(t *testing.T) { + capacity := NewCapacity(1) + reservation, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() = false, want reservation") + } + if err := reservation.Cancel(); err != nil { + t.Fatalf("Cancel() error = %v", err) + } + assertCapacityCounters(t, capacity, 0, 0) + + if err := reservation.Cancel(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("second Cancel() error = %v, want ErrReservationFinished", err) + } + if err := reservation.Commit(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("Commit() after Cancel error = %v, want ErrReservationFinished", err) + } + if err := reservation.Release(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("Release() after Cancel error = %v, want ErrReservationFinished", err) + } + + reused, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() after Cancel = false, want released slot") + } + if err := reused.Cancel(); err != nil { + t.Fatalf("reused Cancel() error = %v", err) + } + assertCapacityCounters(t, capacity, 0, 0) +} + +func TestReservationCommitAndReleaseAreSingleUse(t *testing.T) { + capacity := NewCapacity(1) + reservation, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() = false, want reservation") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit() error = %v", err) + } + assertCapacityCounters(t, capacity, 1, 0) + + if err := reservation.Commit(); !errors.Is(err, ErrReservationCommitted) { + t.Fatalf("second Commit() error = %v, want ErrReservationCommitted", err) + } + if err := reservation.Cancel(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("Cancel() after Commit error = %v, want ErrReservationFinished", err) + } + assertCapacityCounters(t, capacity, 1, 0) + + if err := reservation.Release(); err != nil { + t.Fatalf("Release() error = %v", err) + } + if err := reservation.Release(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("second Release() error = %v, want ErrReservationFinished", err) + } + if err := reservation.Commit(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("Commit() after Release error = %v, want ErrReservationFinished", err) + } + assertCapacityCounters(t, capacity, 0, 0) +} + +func TestReleaseBeforeCommitDoesNotConsumeReservation(t *testing.T) { + capacity := NewCapacity(1) + reservation, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() = false, want reservation") + } + if err := reservation.Release(); !errors.Is(err, ErrReservationFinished) { + t.Fatalf("Release() before Commit error = %v, want ErrReservationFinished", err) + } + assertCapacityCounters(t, capacity, 0, 1) + + if err := reservation.Cancel(); err != nil { + t.Fatalf("Cancel() after rejected Release error = %v", err) + } + assertCapacityCounters(t, capacity, 0, 0) +} + +func TestConcurrentReservationTerminationPreservesCounters(t *testing.T) { + for iteration := range 1_000 { + capacity := NewCapacity(1) + reservation, ok := capacity.Reserve() + if !ok { + t.Fatalf("iteration %d Reserve() = false", iteration) + } + + start := make(chan struct{}) + results := make(chan terminationResult, 2) + var wait sync.WaitGroup + wait.Add(2) + go func() { + defer wait.Done() + <-start + results <- terminationResult{operation: "commit", err: reservation.Commit()} + }() + go func() { + defer wait.Done() + <-start + results <- terminationResult{operation: "cancel", err: reservation.Cancel()} + }() + close(start) + wait.Wait() + close(results) + + var succeeded string + for result := range results { + if result.err == nil { + if succeeded != "" { + t.Fatalf("iteration %d operations %s and %s both succeeded", iteration, succeeded, result.operation) + } + succeeded = result.operation + continue + } + if !errors.Is(result.err, ErrReservationFinished) { + t.Fatalf("iteration %d %s error = %v, want ErrReservationFinished", iteration, result.operation, result.err) + } + } + + switch succeeded { + case "commit": + assertCapacityCounters(t, capacity, 1, 0) + if err := reservation.Release(); err != nil { + t.Fatalf("iteration %d Release() error = %v", iteration, err) + } + case "cancel": + assertCapacityCounters(t, capacity, 0, 0) + default: + t.Fatalf("iteration %d has no successful termination", iteration) + } + assertCapacityCounters(t, capacity, 0, 0) + } +} + +func TestConcurrentReleaseSucceedsOnce(t *testing.T) { + capacity := NewCapacity(1) + reservation, ok := capacity.Reserve() + if !ok { + t.Fatal("Reserve() = false, want reservation") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit() error = %v", err) + } + + var succeeded atomic.Int64 + var unexpected atomic.Int64 + var wait sync.WaitGroup + for range 100 { + wait.Add(1) + go func() { + defer wait.Done() + err := reservation.Release() + switch { + case err == nil: + succeeded.Add(1) + case !errors.Is(err, ErrReservationFinished): + unexpected.Add(1) + } + }() + } + wait.Wait() + + if succeeded.Load() != 1 || unexpected.Load() != 0 { + t.Fatalf("Release() results = success:%d unexpected:%d, want 1 and 0", succeeded.Load(), unexpected.Load()) + } + assertCapacityCounters(t, capacity, 0, 0) +} + +type terminationResult struct { + operation string + err error +} + +func assertCapacityCounters(t *testing.T, capacity *Capacity, active, reserved int64) { + t.Helper() + if got := capacity.Active(); got != active { + t.Fatalf("Active() = %d, want %d", got, active) + } + if got := capacity.Reserved(); got != reserved { + t.Fatalf("Reserved() = %d, want %d", got, reserved) + } +} diff --git a/progress.md b/progress.md index dadd8ae..495ee9d 100644 --- a/progress.md +++ b/progress.md @@ -25,6 +25,9 @@ - 架构证据审计确认 Provider 分布式 Leader、Health Reducer/Checker、Prometheus 指标模块和四个生产命令仍缺实现；追踪矩阵已把这些条目的配置/领域基础与 完整运行时证据拆开描述。 +- 新增独立 Capacity 生命周期测试，覆盖 Cancel 后复用、重复 Commit/Cancel/ + Release、Release-before-Commit、1,000 轮并发 Commit/Cancel 及 100 并发 + Release；Proxy、Dispatch、Snapshot、Gateway Server 定向测试全部通过。 - 一次 `rg` 同时包含不存在的 `cmd` 路径，以及两次使用 PowerShell 不展开的 通配路径，分别返回退出码 2/123；后续改用实际目录和 `-g` 过滤，不重复原命令。 - 已实现共享 `httpapi`、`httpsecurity` 与 `httpserver`，统一严格 JSON、Problem、