fix(flowcontrol/bench): restore L backpressure in benchDetector self-heal

The previous detector fix (f527a2b) stopped the L=1 hang but over-corrected:
it reclaimed the prior grant on every Saturation call, pinning inFlight at <=1
so the mock never reported saturated for any limit>=1. That silently removed the
W>L backpressure the PerformanceMatrix sweep exists to measure -- every L
behaved as free-flow.

Restore the original reserve-in-read accounting (inFlight accumulates to L under
load and Saturation returns 1.0 once exceeded), and self-heal only the genuine
leak source: the processor's 1ms dispatch ticker reserves a permit on empty-queue
idle cycles that never dispatch and so are never Released. Distinguish leaked idle
grants from real load saturation via releaseCount -- while releases are flowing
the saturation is genuine; two consecutive saturated reads with zero releases
between them mean the outstanding grants are leaked idle permits, so reclaim one.
This keeps L=1 from latching (no hang) without capping accumulation.

Known approximation: under severe scheduler starvation a saturated band may
over-admit by ~1; this only perturbs the d/s numbers, never fails (the matrix
does not assert on dispatch outcomes).

Tests: add TestBenchDetector_SaturatesAtLimit (the backpressure assertion whose
absence let the regression through); reframe the no-leak test as a no-permanent-
latch/self-heal guard; replace the pendingGrant-specific test with a
release-recovers-capacity test. All deterministic, pass under -race.

Verified: L=1 vs L=100 throughput now separates by ~10x (was identical);
full bench smoke completes with no hang.

Author: LukeAVanDrie

pkg/epp/flowcontrol/benchmark/detector_test.go

@@ -23,19 +23,49 @@ import (
 
 // TestBenchDetector_NoLeakOnNonDispatchingCycles is the regression guard for the leak that made
 // matrix coordinates at L=1 hang for the full request TTL: the processor's dispatchCycle evaluates
-// Saturation once per cycle but may not dispatch (empty queue, or head-of-line gating), so a
-// detector that reserved a permit per call without reclaiming it would latch to saturated forever.
+// Saturation on a 1ms ticker even when the queue is empty, so those grants never dispatch and are
+// never Released. The detector may report saturated briefly but must self-heal -- it must never
+// latch, so grants keep recurring and no saturated run exceeds the stuck threshold.
 func TestBenchDetector_NoLeakOnNonDispatchingCycles(t *testing.T) {
 	d := &benchDetector{}
 	d.concurrencyLimit.Store(1)
 
-	// Simulate many dispatch cycles that evaluate saturation but never dispatch (hence never Release).
-	// A leaking detector latches at >= 1.0.
+	grants, run, maxRun := 0, 0, 0
 	for i := 0; i < 100; i++ {
+		if d.Saturation(context.Background(), nil) < 1.0 {
+			grants++
+			run = 0
+		} else {
+			run++
+			if run > maxRun {
+				maxRun = run
+			}
+		}
+	}
+	if grants == 0 {
+		t.Fatal("detector latched saturated across all idle cycles (permanent leak)")
+	}
+	if maxRun > stuckReadThreshold {
+		t.Fatalf("idle saturated run = %d, want <= %d (self-heal should reclaim)", maxRun, stuckReadThreshold)
+	}
+}
+
+// TestBenchDetector_SaturatesAtLimit is the assertion whose absence let the backpressure regression
+// through: with the limit reached by in-flight (un-Released) grants, the next Saturation must report
+// saturated. This is the W>L backpressure the PerformanceMatrix sweep exists to measure.
+func TestBenchDetector_SaturatesAtLimit(t *testing.T) {
+	d := &benchDetector{}
+	d.concurrencyLimit.Store(2)
+
+	// Two grants whose requests are still in flight (not yet Released) occupy both slots.
+	for i := 0; i < 2; i++ {
 		if got := d.Saturation(context.Background(), nil); got >= 1.0 {
-			t.Fatalf("detector latched saturated on idle cycle %d: got %v, want < 1.0", i, got)
+			t.Fatalf("grant %d within limit must not be saturated: got %v", i, got)
 		}
 	}
+	if got := d.Saturation(context.Background(), nil); got < 1.0 {
+		t.Fatalf("over-limit Saturation must report saturated: got %v, want >= 1.0", got)
+	}
 }
 
 // TestBenchDetector_GrantReleaseBalanced verifies the common path: a cycle grants
@@ -57,21 +87,27 @@ func TestBenchDetector_GrantReleaseBalanced(t *testing.T) {
 	}
 }
 
-// TestBenchDetector_OverlappingGrantsConserveCount exercises the race-prone interleaving
-// deterministically: a second cycle grants before the first dispatch's worker releases, then both
-// releases arrive. The shared CompareAndSwap must free each permit exactly once -- no leak, no
-// underflow.
-func TestBenchDetector_OverlappingGrantsConserveCount(t *testing.T) {
+// TestBenchDetector_ReleaseRecoversCapacity verifies the steady-state load path: grants fill the
+// limit and report saturated, Releases free the slots back to zero, and freed capacity is grantable
+// again -- the conserve-count property under sustained dispatch.
+func TestBenchDetector_ReleaseRecoversCapacity(t *testing.T) {
 	d := &benchDetector{}
-	d.concurrencyLimit.Store(1)
+	d.concurrencyLimit.Store(2)
 
-	d.Saturation(context.Background(), nil) // grant #1 (dispatched, release pending)
-	d.Saturation(context.Background(), nil) // reclaims #1, grants #2 (dispatched)
-	d.Release()                             // frees the outstanding grant
-	d.Release()                             // stale: CAS fails, no-op
+	d.Saturation(context.Background(), nil) // grant 1 (in flight)
+	d.Saturation(context.Background(), nil) // grant 2 (in flight) -> at limit
+	if got := d.Saturation(context.Background(), nil); got < 1.0 {
+		t.Fatalf("at-limit Saturation must be saturated: got %v, want >= 1.0", got)
+	}
 
+	d.Release() // both in-flight requests complete
+	d.Release()
 	if got := d.inFlight.Load(); got != 0 {
-		t.Fatalf("inFlight = %d after balanced grants/releases, want 0", got)
+		t.Fatalf("inFlight = %d after releasing all grants, want 0", got)
+	}
+
+	if got := d.Saturation(context.Background(), nil); got >= 1.0 {
+		t.Fatalf("grant after releases must not be saturated: got %v", got)
 	}
 }
 

pkg/epp/flowcontrol/benchmark/helpers_test.go

@@ -89,56 +89,80 @@ type testDetector interface {
 // parking goroutines. The processor's dispatch loop calls Saturation once per cycle and then
 // dispatches at most one item; the benchmark worker calls Release after that item is dispatched.
 //
-// A permit is reserved optimistically inside Saturation and freed by whichever happens first: the
-// worker's Release (the cycle dispatched) or the next Saturation call (the cycle did not).
-// Reclaiming the previous, unconsumed grant on the next call keeps idle or head-of-line-gated
-// cycles from leaking permits -- a leak that would otherwise permanently saturate the detector at
-// low limits (e.g. L=1, where one leaked permit deadlocks every request until its TTL expires).
+// Saturation optimistically reserves a permit (inFlight++); Release frees it. Under the matrix's
+// W>L load the queue is always full, so every grant leads to a dispatch and a Release: inFlight
+// accumulates to L and Saturation reports 1.0 once exceeded, which is the backpressure the matrix
+// measures.
 //
-// The dispatch loop is single-threaded, so at most one grant is outstanding at a time; pendingGrant
-// tracks it. Release and the reclaim path share a CompareAndSwap on pendingGrant, so each grant is
-// freed exactly once even when the two race.
+// The complication: the processor also runs dispatchCycle on a 1ms ticker when the queue is EMPTY
+// (notably at startup). Those idle grants never dispatch and so are never Released; a naive detector
+// would latch saturated forever and deadlock every request until its TTL (the L=1 hang). We
+// distinguish leaked idle grants from genuine load saturation with releaseCount: while releases are
+// flowing the saturation is real; if two consecutive saturated reads see zero releases between them
+// the outstanding grants are leaked idle permits, so we reclaim one. The 1.0/0.99 split sits below
+// usagelimits.DefaultPolicy's 1.0 ceiling, so a grant never trips head-of-line gating.
+//
+// stuckReads and lastReleaseCount are touched only by the single-threaded dispatch loop (Saturation)
+// and so need no synchronization; inFlight and releaseCount are atomic (Release runs on workers).
 type benchDetector struct {
 	flowcontrol.SaturationDetector
 	concurrencyLimit atomic.Int64
-	// _ prevents false sharing between atomic counters on multicore CPU cache lines.
+	// _ prevents false sharing between concurrencyLimit and the hot atomic counters below.
 	_            [48]byte
 	inFlight     atomic.Int64
-	pendingGrant atomic.Int64
+	releaseCount atomic.Int64
+
+	stuckReads       int
+	lastReleaseCount int64
 }
 
-// Release frees the permit held by the dispatch that just completed. If the next Saturation call
-// already reclaimed it, the CompareAndSwap fails and this is a no-op, preventing a double free.
+// stuckReadThreshold is the number of consecutive saturated reads with zero intervening releases
+// after which the detector treats the outstanding grants as leaked idle permits and reclaims one.
+// >1 so that a single slow Release under load does not cause a spurious reclaim.
+const stuckReadThreshold = 2
+
+// Release frees the permit held by the dispatch that just completed.
 func (d *benchDetector) Release() {
 	if d.concurrencyLimit.Load() <= 0 {
 		return
 	}
-	if d.pendingGrant.CompareAndSwap(1, 0) {
-		d.inFlight.Add(-1)
-	}
+	d.inFlight.Add(-1)
+	d.releaseCount.Add(1)
 }
 
-// Saturation reclaims the previous cycle's grant if no dispatch consumed it, then optimistically
-// reserves a permit for this cycle.
+// Saturation reserves a permit for this cycle, reporting saturated once the limit is exceeded and
+// self-healing leaked idle grants (see the type doc).
 func (d *benchDetector) Saturation(ctx context.Context, candidates []fwkdl.Endpoint) float64 {
 	limit := d.concurrencyLimit.Load()
 	if limit <= 0 {
 		return 0.0 // Free-flow.
 	}
 
-	// Reclaim a grant left unconsumed by a previous, non-dispatching cycle.
-	if d.pendingGrant.CompareAndSwap(1, 0) {
-		d.inFlight.Add(-1)
-	}
-
 	// Optimistically reserve a permit for this cycle.
 	if d.inFlight.Add(1) <= limit {
-		d.pendingGrant.Store(1)
+		d.stuckReads = 0
 		return 0.99 // Return < 1.0 so the dispatcher proceeds.
 	}
 
 	// Capacity exceeded; roll back the speculative reservation.
 	d.inFlight.Add(-1)
+
+	rc := d.releaseCount.Load()
+	if rc != d.lastReleaseCount {
+		// Releases are flowing: this is genuine load saturation, not a leak.
+		d.lastReleaseCount = rc
+		d.stuckReads = 0
+		return 1.0
+	}
+
+	// No releases since the last saturated read. After a couple of these the outstanding grants must
+	// be leaked idle permits (an empty-queue dispatch tick reserved but never dispatched), so reclaim
+	// one to keep the detector from latching.
+	d.stuckReads++
+	if d.stuckReads >= stuckReadThreshold {
+		d.inFlight.Add(-1)
+		d.stuckReads = 0
+	}
 	return 1.0 // Saturated - forces the Flow Control layer to hold the item.
 }