test(flowcontrol): harden and reorganize the flow-control benchmark suite (llm-d#1638)

* test(flowcontrol): harden and reorganize the flow-control benchmark suite

Test-only changes to pkg/epp/flowcontrol/benchmark; no production code is touched.

- Re-scope the full-path benchmark (FullPathStress -> FullPath) to what it
  uniquely measures: dispatch throughput and allocation overhead through the real
  producer -> concurrency detector -> controller path. Drop the off-layer producer
  leak-counter assertion (covered by the inflightload producer's own unit tests),
  report d/s like the other benchmarks, and fix a b.Fatalf called from inside
  RunParallel (invalid off the benchmark goroutine).

- Fix the mock benchDetector so the PerformanceMatrix L sweep is meaningful: it
  accumulates in-flight up to the limit and reports saturated once exceeded (real
  W>L backpressure), while self-healing the empty-queue idle grants the processor's
  1ms dispatch ticker would otherwise leak (which previously latched and hung at
  L=1). Distinguish leaked idle grants from genuine load saturation via a release
  counter. Add detector unit tests covering saturation-at-limit and self-heal.

- Replace fixed bootstrap time.Sleep calls with a warmup request that blocks until
  dispatch -- deterministic readiness instead of arbitrary timing.

- Reorganize into doc.go (two-family overview) + helpers_test.go + benchmark_test.go,
  and dedup the shared registry/policy setup.

- Add a bench-smoke make target and CI step (go test -bench=. -benchtime=1x) so
  benchmark bodies execute on every PR; go test only compiles them otherwise,
  letting runtime breakage (panics, deadlocks) go undetected.

Signed-off-by: Luke Van Drie <lukevandrie@google.com>

* test(flowcontrol): correct FullPath backpressure comment

The inline comment claimed the FullPath benchmark sustains W>L queuing, but
worker count rounds down to <= L and the real detector's in-flight load is
released near-instantly, so dispatch is effectively free-flowing. Reword to
match the function godoc: FullPath prices dispatch + the detector's per-cycle
DynamicAttribute reads, not sustained backpressure.

Signed-off-by: Luke Van Drie <lukevandrie@google.com>

* test(flowcontrol): address review feedback on benchmark suite

- Use the llm-d Authors copyright header on the new files (doc.go,
  detector_test.go) per review.
- Fail fast if the PerformanceMatrix warmup request is not dispatched,
  matching the full-path harness, instead of silently timing a dead
  dispatch loop.
- Fail fast if a matrix coordinate dispatches zero requests, so a
  wedged dispatch path produces a red build instead of plausible
  zero rows.
- Drop the matrix harness TTL from 5m to 30s: nothing in a throughput
  benchmark legitimately waits minutes, and a tight TTL bounds a
  wedged coordinate to seconds of CI time instead of minutes per cell.
- Document the liveness bias of the benchDetector's stuck-read reclaim
  heuristic: true queueing still reports saturated while releases flow;
  a stalled release transiently over-admits rather than latching.

Signed-off-by: Luke Van Drie <lukevandrie@google.com>

---------

Signed-off-by: Luke Van Drie <lukevandrie@google.com>

Author: LukeAVanDrie

.github/workflows/ci-pr-checks.yaml

@@ -113,6 +113,16 @@ jobs:
           GO_BUILD_CACHE_VOL: ${{ steps.go-cache.outputs.build }}
         run: make test-unit
 
+      # Anti-rot: execute the benchmark bodies once so runtime breakage (panics,
+      # deadlocks) is caught. `go test` only compiles benchmarks; it never runs
+      # them without -bench.
+      - name: Run benchmark smoke
+        shell: bash
+        env:
+          GO_MOD_CACHE_VOL: ${{ steps.go-cache.outputs.mod }}
+          GO_BUILD_CACHE_VOL: ${{ steps.go-cache.outputs.build }}
+        run: make bench-smoke
+
       - name: Run hermetic integration tests
         shell: bash
         env:

Makefile

@@ -315,6 +315,11 @@ bench-tokenizer: image-build-builder ## Run external tokenizer + scorer benchmar
 	@printf "Run 'EXTERNAL_TOKENIZER_ENABLED=true KV_CACHE_ENABLED=true make env-dev-kind' first.\n\n"
 	$(BUILDER_RUN_CLUSTER) 'go test -bench=. -benchmem -count=5 -timeout=5m ./test/profiling/tokenizerbench/'
 
+.PHONY: bench-smoke
+bench-smoke: image-build-builder ## Smoke-run the flowcontrol benchmarks once (-benchtime=1x) to catch runtime rot
+	@printf "\033[33;1m==== Running Flow Control Benchmark Smoke ====\033[0m\n"
+	$(BUILDER_RUN) 'go test -run=^$$ -bench=. -benchtime=1x -timeout=5m ./pkg/epp/flowcontrol/benchmark/...'
+
 .PHONY: post-deploy-test
 post-deploy-test: ## Run post deployment tests
 	@echo "Success!"

pkg/epp/flowcontrol/benchmark/benchmark_test.go

@@ -27,29 +27,17 @@ import (
 	"testing"
 	"time"
 
-	k8stypes "k8s.io/apimachinery/pkg/types"
-
-	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts/mocks"
 	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/controller"
-	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/registry"
 	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/types"
-	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol"
-	fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol"
 	requesthandling "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requesthandling"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling"
-	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/fairness/globalstrict"
-	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/ordering/fcfs"
-	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency"
-	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits"
-	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload"
-	igwtestutils "github.com/llm-d/llm-d-router/test/utils"
 )
 
 // BenchmarkFlowController_PerformanceMatrix evaluates throughput across a matrix of variables.
-// It systematically evaluates the impact of strict egress limits, data parallelism, priority
-// levels, flow density, and concurrent connections.
+// It systematically evaluates the impact of strict egress limits, priority levels, flow density,
+// and concurrent connections.
 func BenchmarkFlowController_PerformanceMatrix(b *testing.B) {
 	if testing.Short() {
 		b.Skip("skipping PerformanceMatrix in short mode")
@@ -84,11 +72,17 @@ func BenchmarkFlowController_PerformanceMatrix(b *testing.B) {
 // runMatrixCoordinate executes a single coordinate of the performance hypercube.
 func runMatrixCoordinate(b *testing.B, m benchMatrix) {
 	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel() // Ensure SUT goroutines are torn down even when the coordinate fails fatally.
 
 	fc, detector := setupBenchmarkHarness(ctx, b, m.priorities, m.limit, nil, nil)
 
-	// Yield briefly to allow the background supervisor to bootstrap the data plane.
-	time.Sleep(10 * time.Millisecond)
+	// Warm up: block on one request until it's dispatched, proving the dispatch loop is live before
+	// timing. Release the slot per the loop's immediate-drain protocol (a no-op under free-flow).
+	warmup := &benchRequest{key: flowcontrol.FlowKey{ID: "warmup", Priority: 0}, byteSize: 1024}
+	if outcome, err := fc.EnqueueAndWait(ctx, warmup); outcome != types.QueueOutcomeDispatched {
+		b.Fatalf("warmup request was not dispatched: outcome=%v, err=%v", outcome, err)
+	}
+	detector.Release()
 
 	reqs := make([]*benchRequest, m.flows)
 	for i := 0; i < int(m.flows); i++ {
@@ -174,6 +168,13 @@ func runMatrixCoordinate(b *testing.B, m benchMatrix) {
 	elapsed := b.Elapsed().Seconds()
 	telemetry.report(b, elapsed)
 
+	// Every coordinate expects flow (W>L or free-flow), so zero dispatches means the dispatch path
+	// wedged mid-run: fail loudly instead of publishing plausible-looking zero rows. Assert on the
+	// benchmark goroutine (never inside RunParallel, where FailNow is invalid).
+	if telemetry.dispatchCount.Load() == 0 {
+		b.Fatalf("coordinate %s dispatched zero requests; the dispatch path is wedged", m.name())
+	}
+
 	cancel()                          // Graceful teardown to prevent async skewing of subsequent coordinates.
 	time.Sleep(50 * time.Millisecond) // Wait for SUT background goroutines to terminate.
 }
@@ -282,140 +283,48 @@ func BenchmarkFlowController_MassCancellation(b *testing.B) {
 	}
 }
 
-// fullPathBenchHarness holds shared setup state for full-path benchmarks that
-// wire a real InFlightLoadProducer, real concurrency detector, and real
-// persistent endpoints into the FlowController.
-type fullPathBenchHarness struct {
-	fc       *controller.FlowController
-	producer *inflightload.InFlightLoadProducer
-	epMeta   *fwkdl.EndpointMetadata
-	schedEp  scheduling.Endpoint
-}
-
-// setupFullPathBenchmark creates the shared harness for benchmarks that exercise
-// the complete flow control data path: producer -> detector -> controller -> cleanup.
-func setupFullPathBenchmark(ctx context.Context, b *testing.B, name string, numPriorities int) *fullPathBenchHarness {
-	b.Helper()
-	if numPriorities < 1 {
-		numPriorities = 1
-	}
-	handle := igwtestutils.NewTestHandle(ctx)
-
-	producerName := name + "-producer"
-	producerPlugin, err := inflightload.InFlightLoadProducerFactory(
-		producerName, fwkplugin.StrictDecoder([]byte(`{}`)), handle,
-	)
-	if err != nil {
-		b.Fatal(err)
-	}
-	producer := producerPlugin.(*inflightload.InFlightLoadProducer)
-
-	detectorPlugin, err := concurrency.ConcurrencyDetectorFactory(
-		name+"-detector",
-		fwkplugin.StrictDecoder([]byte(fmt.Sprintf(
-			`{"maxConcurrency": 100, "inFlightLoadProducerName": %q}`, producerName,
-		))),
-		handle,
-	)
-	if err != nil {
-		b.Fatal(err)
-	}
-	realDetector := detectorPlugin.(flowcontrol.SaturationDetector)
-
-	epMeta := &fwkdl.EndpointMetadata{
-		NamespacedName: k8stypes.NamespacedName{Name: "pod-1", Namespace: "default"},
-	}
-	ep := fwkdl.NewEndpoint(epMeta, fwkdl.NewMetrics())
-	if err := producer.Extract(ctx, fwkdl.EndpointEvent{
-		Type: fwkdl.EventAddOrUpdate, Endpoint: ep,
-	}); err != nil {
-		b.Fatal(err)
-	}
-
-	fPolicy, err := globalstrict.GlobalStrictFairnessPolicyFactory(registry.DefaultFairnessPolicyRef, nil, handle)
-	if err != nil {
-		b.Fatal(err)
-	}
-	handle.AddPlugin(registry.DefaultFairnessPolicyRef, fPolicy)
-
-	oPolicy, err := fcfs.FCFSOrderingPolicyFactory(registry.DefaultOrderingPolicyRef, nil, handle)
-	if err != nil {
-		b.Fatal(err)
-	}
-	handle.AddPlugin(registry.DefaultOrderingPolicyRef, oPolicy)
-
-	defaults := registry.PriorityBandPolicyDefaults{
-		OrderingPolicy: oPolicy.(flowcontrol.OrderingPolicy),
-		FairnessPolicy: fPolicy.(flowcontrol.FairnessPolicy),
-	}
-	reg := setupRegistry(b, defaults, priorityLevels(numPriorities))
-
-	fc := controller.NewFlowController(ctx, name+"-bench", &controller.Config{
-		DefaultRequestTTL:        5 * time.Minute,
-		ExpiryCleanupInterval:    1 * time.Hour,
-		EnqueueChannelBufferSize: 2000,
-	}, controller.Deps{
-		Registry:           reg,
-		SaturationDetector: realDetector,
-		EndpointCandidates: &mocks.MockEndpointCandidates{Candidates: []fwkdl.Endpoint{ep}},
-		UsageLimitPolicy:   usagelimits.DefaultPolicy(),
-	})
-
-	time.Sleep(10 * time.Millisecond)
-
-	schedEp := scheduling.NewEndpoint(epMeta, fwkdl.NewMetrics(), nil)
-
-	return &fullPathBenchHarness{
-		fc:       fc,
-		producer: producer,
-		epMeta:   epMeta,
-		schedEp:  schedEp,
-	}
-}
-
-// BenchmarkFlowController_FullPathStress exercises the complete flow control
-// data path under realistic production conditions: real InFlightLoadProducer,
-// real concurrency detector reading DynamicAttributes from persistent endpoints,
-// real FlowController with backpressure from a concurrency-gated detector,
-// multiple priority bands, concurrent workers creating unique flows (agentic
-// churn), and the full PreRequest -> dispatch -> StartOfStream -> EndOfStream
-// lifecycle per request.
+// BenchmarkFlowController_FullPath measures dispatch throughput and allocation overhead of the
+// complete flow-control data path using REAL components: a real InFlightLoadProducer feeding a real
+// concurrency SaturationDetector, which gates a real FlowController. Unlike the throughput
+// microbenchmarks (which use the mock benchDetector), this captures the cost of the detector's
+// per-dispatch DynamicAttribute reads against live producer state in the hot path.
 //
-// What it catches:
-//   - PluginState entry leaks (addedTokensEntry not cleaned up by OnEvicted)
-//   - DynamicAttribute closure leaks (producer tracker never decremented)
-//   - Cross-band counter drift (stats routed to wrong priority band)
-//   - Registry flow infrastructure leaks under concurrent churn
-//   - Contention-induced allocation growth (lock convoy, sync.Map thrashing)
+// Each iteration runs the full request lifecycle: EnqueueAndWait (admission) -> PreRequest
+// (producer records in-flight load) -> ResponseBody StartOfStream / EndOfStream (producer releases
+// it). Workers spread across priority bands and mint a unique flow per request for registry churn.
 //
 // Run with:
 //
-//	go test -bench=FullPathStress -benchtime=5000x -count=3 -run=^$ ./pkg/epp/flowcontrol/benchmark/
+//	go test -bench=FullPath -run=^$ ./pkg/epp/flowcontrol/benchmark/
 //
-// Stable B/op across runs = no leak. Counter assertion at the end catches
-// tracker drift that B/op alone would miss.
-func BenchmarkFlowController_FullPathStress(b *testing.B) {
+// Reports d/s (dispatch throughput). Producer-level leak correctness is covered by the inflightload
+// producer's own unit tests, not here.
+func BenchmarkFlowController_FullPath(b *testing.B) {
 	const numPriorities = 4
 
 	ctx := b.Context()
-	h := setupFullPathBenchmark(ctx, b, "stress", numPriorities)
+	h := setupFullPathBenchmark(ctx, b, "fullpath", numPriorities)
 
 	sosResp := &requestcontrol.Response{StartOfStream: true}
 	eosResp := &requestcontrol.Response{EndOfStream: true}
 	profileResults := map[string]*scheduling.ProfileRunResult{
 		"decode": {TargetEndpoints: []scheduling.Endpoint{h.schedEp}},
 	}
 
-	// Concurrency-gated: each dispatched request consumes a slot. Workers
-	// release immediately after ResponseBody, creating sustained backpressure
-	// (W workers > L limit). This forces real queuing in the dispatch cycle.
+	telemetry := newBenchmarkTelemetry()
+
+	// Concurrency-gated by the real detector: each dispatched request records in-flight load (via
+	// PreRequest) and releases it after ResponseBody. Load is held only briefly, so this measures
+	// free-flowing dispatch plus the detector's per-cycle DynamicAttribute read cost rather than
+	// sustained W>L queuing.
 	var globalReqID atomic.Uint64
 
 	b.ResetTimer()
 	b.ReportAllocs()
 	b.SetParallelism(max(100/runtime.GOMAXPROCS(0), 1))
 
 	b.RunParallel(func(pb *testing.PB) {
+		var local threadTelemetry
 		for pb.Next() {
 			id := globalReqID.Add(1)
 			reqID := fmt.Sprintf("req-%d", id)
@@ -427,34 +336,37 @@ func BenchmarkFlowController_FullPathStress(b *testing.B) {
 				byteSize: 512,
 			}
 			outcome, _ := h.fc.EnqueueAndWait(ctx, fcReq)
-
 			if outcome != types.QueueOutcomeDispatched {
-				b.Fatalf("request %s was not dispatched: %v", reqID, outcome)
+				telemetry.recordReject(&local)
+				continue
 			}
+			telemetry.recordDispatch(&local)
 
-			// 2. Post-scheduling: producer tracks the request on the endpoint.
+			// 2. Post-scheduling: producer records in-flight load on the endpoint, which the
+			//    detector reads to compute saturation.
 			infReq := &scheduling.InferenceRequest{
 				RequestID: reqID,
 				Body:      &requesthandling.InferenceRequestBody{TokenizedPrompt: &requesthandling.TokenizedPrompt{PerPromptTokens: [][]uint32{benchTokenIDs}}},
 			}
 			schedResult := &scheduling.SchedulingResult{ProfileResults: profileResults}
 			h.producer.PreRequest(ctx, infReq, schedResult)
 
-			// 3. Response lifecycle: release counters.
+			// 3. Response lifecycle: release the in-flight load.
 			infReq.SchedulingResult = schedResult
 			h.producer.ResponseBody(ctx, infReq, sosResp, h.epMeta)
 			h.producer.ResponseBody(ctx, infReq, eosResp, h.epMeta)
 		}
+		telemetry.commit(&local)
 	})
 
 	b.StopTimer()
-	b.ReportMetric(float64(globalReqID.Load()), "total-requests")
+	telemetry.report(b, b.Elapsed().Seconds())
 
-	finalRequests := h.producer.GetRequests(h.epMeta.NamespacedName.String())
-	finalTokens := h.producer.GetTokens(h.epMeta.NamespacedName.String())
-	if finalRequests != 0 || finalTokens != 0 {
-		b.Errorf("counter leak detected after %d requests across %d priorities: requests=%d, tokens=%d",
-			globalReqID.Load(), numPriorities, finalRequests, finalTokens)
+	// A fully saturated full-path run must dispatch every request; a rejection
+	// means the throughput numbers are skewed. Assert on the benchmark goroutine
+	// (never inside RunParallel, where FailNow is invalid).
+	if rejects := telemetry.rejectCount.Load(); rejects > 0 {
+		b.Fatalf("full-path benchmark saw %d rejected requests; throughput numbers are unreliable", rejects)
 	}
 }