test: cover capacity reservation lifecycle
This commit is contained in:
youfak
parent
f46d79511f
commit
125740f58d
@ -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
|
||||
|
||||
@ -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 分类。
|
||||
|
||||
@ -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` 在虚拟时钟下准确。
|
||||
|
||||
@ -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 场景下
|
||||
|
||||
191
internal/domain/proxy/capacity_test.go
Normal file
191
internal/domain/proxy/capacity_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
@ -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、
|
||||
|
||||
Loading…
Reference in New Issue
Block a user