113 lines
3.3 KiB
Go
113 lines
3.3 KiB
Go
package pipeline
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// TestRateGuardLimitsConcurrency proves the semaphore REALLY bounds concurrency (WS6 precondition,
|
|
// ревью-2 F4): with max_concurrency=2, no more than 2 goroutines are ever inside the guarded region
|
|
// at once, even under 20 racing callers. A broken guard (missing/over-sized semaphore) trips the
|
|
// observed-max assertion.
|
|
func TestRateGuardLimitsConcurrency(t *testing.T) {
|
|
const cap, workers = 2, 20
|
|
g := newRateGuard(cap, 0)
|
|
var inFlight, maxSeen int64
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < workers; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
rel, err := g.acquire(context.Background())
|
|
if err != nil {
|
|
t.Errorf("acquire: %v", err)
|
|
return
|
|
}
|
|
cur := atomic.AddInt64(&inFlight, 1)
|
|
for {
|
|
m := atomic.LoadInt64(&maxSeen)
|
|
if cur <= m || atomic.CompareAndSwapInt64(&maxSeen, m, cur) {
|
|
break
|
|
}
|
|
}
|
|
time.Sleep(time.Millisecond) // widen the window a concurrency bug would exploit
|
|
atomic.AddInt64(&inFlight, -1)
|
|
rel()
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
if maxSeen > cap {
|
|
t.Fatalf("rate-guard admitted %d concurrent callers, cap is %d", maxSeen, cap)
|
|
}
|
|
if maxSeen == 0 {
|
|
t.Fatalf("no caller was observed in-flight — the guard never admitted anyone")
|
|
}
|
|
}
|
|
|
|
// TestRateGuardNilAndUnlimitedAreNoOps: a nil guard and a zero-config guard both admit instantly and
|
|
// never block (the sequential path / an unconfigured model must cost nothing).
|
|
func TestRateGuardNilAndUnlimitedAreNoOps(t *testing.T) {
|
|
var nilGuard *rateGuard
|
|
rel, err := nilGuard.acquire(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("nil guard acquire: %v", err)
|
|
}
|
|
rel() // must not panic
|
|
unlimited := newRateGuard(0, 0)
|
|
for i := 0; i < 100; i++ {
|
|
rel, err := unlimited.acquire(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("unlimited acquire %d: %v", i, err)
|
|
}
|
|
rel()
|
|
}
|
|
}
|
|
|
|
// TestRateGuardHonoursContextCancel: a caller blocked on a full semaphore unblocks on ctx cancel and
|
|
// does NOT leak a slot (the wave dispatcher must be able to abort a run cleanly).
|
|
func TestRateGuardHonoursContextCancel(t *testing.T) {
|
|
g := newRateGuard(1, 0)
|
|
rel, err := g.acquire(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("first acquire: %v", err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
r, err := g.acquire(ctx) // blocks: the one slot is held
|
|
r()
|
|
done <- err
|
|
}()
|
|
cancel()
|
|
select {
|
|
case err := <-done:
|
|
if err != context.Canceled {
|
|
t.Fatalf("blocked acquire on cancel = %v, want context.Canceled", err)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("acquire did not unblock on ctx cancel")
|
|
}
|
|
rel() // release the first; a fresh acquire must now succeed (no leaked slot)
|
|
r2, err := g.acquire(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("acquire after release: %v", err)
|
|
}
|
|
r2()
|
|
}
|
|
|
|
// TestRateGuardPacingSpacesStarts: with min_interval set, two serial acquires are spaced by at least
|
|
// the interval (best-effort pacing of call STARTS).
|
|
func TestRateGuardPacingSpacesStarts(t *testing.T) {
|
|
g := newRateGuard(0, 30*time.Millisecond)
|
|
start := time.Now()
|
|
r1, _ := g.acquire(context.Background())
|
|
r1()
|
|
r2, _ := g.acquire(context.Background())
|
|
r2()
|
|
if elapsed := time.Since(start); elapsed < 30*time.Millisecond {
|
|
t.Fatalf("paced acquires took %v, want ≥30ms between starts", elapsed)
|
|
}
|
|
}
|