feat: Adapt flow control to per-request saturation (kubernetes-sigs#1622)

This commit refactors the flow control `ShardProcessor` to align with
the new `SaturationDetector` contract (introduced in 7d84fb9), which
evaluates saturation for a specific set of candidate pods rather than
for the entire pool.

This change fundamentally alters the dispatching logic to prioritize
strict fairness and priority over work conservation.

The `BandFilter` abstraction has been removed, and the `ShardProcessor`
now performs a post-selection viability check. After policies select
the fairest request, the `SaturationDetector` is called with the
candidate pods for only that specific request. If the check fails, the
processor stops the entire dispatch cycle for the current tick,
enforcing Head-of-Line blocking to prevent priority inversion.

This new model correctly upholds a strict fairness and priority
contract. However, it introduces a known trade-off where the system may
leave resources idle if the fairest request is blocked, rather than
finding other viable work (the "noisy neighbor" problem).

Author: LukeAVanDrie

pkg/epp/flowcontrol/contracts/mocks/mocks.go

@@ -34,6 +34,7 @@ import (
 	"fmt"
 	"sync"
 
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/contracts"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
@@ -112,12 +113,12 @@ func (m *MockRegistryShard) Stats() contracts.ShardStats {
 
 // MockSaturationDetector is a simple "stub-style" mock for testing.
 type MockSaturationDetector struct {
-	IsSaturatedFunc func(ctx context.Context) bool
+	IsSaturatedFunc func(ctx context.Context, candidatePods []metrics.PodMetrics) bool
 }
 
-func (m *MockSaturationDetector) IsSaturated(ctx context.Context) bool {
+func (m *MockSaturationDetector) IsSaturated(ctx context.Context, candidatePods []metrics.PodMetrics) bool {
 	if m.IsSaturatedFunc != nil {
-		return m.IsSaturatedFunc(ctx)
+		return m.IsSaturatedFunc(ctx, candidatePods)
 	}
 	return false
 }

pkg/epp/flowcontrol/contracts/saturationdetector.go

@@ -16,7 +16,11 @@ limitations under the License.
 
 package contracts
 
-import "context"
+import (
+	"context"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+)
 
 // SaturationDetector defines the contract for a component that provides real-time load signals to the
 // `controller.FlowController`.
@@ -32,8 +36,8 @@ import "context"
 //
 // Implementations MUST be goroutine-safe.
 type SaturationDetector interface {
-	// IsSaturated returns true if the system's backend resources are considered saturated.
+	// IsSaturated returns true if the system's backend resources are considered saturated for a set of candidate pods.
 	// `controller.FlowController`'s dispatch workers call this method to decide whether to pause or throttle dispatch
 	// operations to prevent overwhelming the backends.
-	IsSaturated(ctx context.Context) bool
+	IsSaturated(ctx context.Context, candidatePods []metrics.PodMetrics) bool
 }

pkg/epp/flowcontrol/controller/controller.go

@@ -58,7 +58,7 @@ type shardProcessor interface {
 // This enables dependency injection for testing.
 type shardProcessorFactory func(
 	shard contracts.RegistryShard,
-	dispatchFilter internal.BandFilter,
+	saturationDetector contracts.SaturationDetector,
 	clock clock.Clock,
 	expiryCleanupInterval time.Duration,
 	enqueueChannelBufferSize int,
@@ -130,15 +130,15 @@ func NewFlowController(
 	// Use the real shard processor implementation by default.
 	fc.shardProcessorFactory = func(
 		shard contracts.RegistryShard,
-		dispatchFilter internal.BandFilter,
+		saturationDetector contracts.SaturationDetector,
 		clock clock.Clock,
 		expiryCleanupInterval time.Duration,
 		enqueueChannelBufferSize int,
 		logger logr.Logger,
 	) shardProcessor {
 		return internal.NewShardProcessor(
 			shard,
-			dispatchFilter,
+			saturationDetector,
 			clock,
 			expiryCleanupInterval,
 			enqueueChannelBufferSize,
@@ -310,10 +310,9 @@ func (fc *FlowController) getOrStartWorker(shard contracts.RegistryShard) *manag
 
 	// Construct a new worker, but do not start its processor goroutine yet.
 	processorCtx, cancel := context.WithCancel(fc.parentCtx)
-	dispatchFilter := internal.NewSaturationFilter(fc.saturationDetector)
 	processor := fc.shardProcessorFactory(
 		shard,
-		dispatchFilter,
+		fc.saturationDetector,
 		fc.clock,
 		fc.config.ExpiryCleanupInterval,
 		fc.config.EnqueueChannelBufferSize,

pkg/epp/flowcontrol/controller/controller_test.go

@@ -209,7 +209,7 @@ type mockShardProcessorFactory struct {
 
 func (f *mockShardProcessorFactory) new(
 	shard contracts.RegistryShard,
-	_ internal.BandFilter,
+	_ contracts.SaturationDetector,
 	_ clock.Clock,
 	_ time.Duration,
 	_ int,
@@ -640,7 +640,7 @@ func TestFlowController_Lifecycle(t *testing.T) {
 		h := newUnitHarness(t, t.Context(), Config{}, &mockRegistryClient{})
 		h.fc.shardProcessorFactory = func(
 			shard contracts.RegistryShard,
-			_ internal.BandFilter,
+			_ contracts.SaturationDetector,
 			_ clock.Clock,
 			_ time.Duration,
 			_ int,