feat(flowcontrol): reclamation controller for demand-driven in-flight eviction

Implements the controller from docs/flow-control-eviction.md (PR
llm-d#2061): on HoL blocking, scan blocked bands for queued demand strictly
above the victim head, size revocations as a saturation deficit
against a mean-footprint credit with pending-reclaim debits, and pace
by confirmation (stream termination) plus a grace covering the
sensor's visibility lag, with a timeout so a hung stream cannot wedge
the gate. Bands with unattainable ceilings are excluded from demand.
Adds the flowControl.enableEviction API field and its translation.
New metrics: revocations issued and terminal outcomes, reclaim target,
pending reclaim, confirmation latency.

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

Author: LukeAVanDrie

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -401,6 +401,14 @@ type FlowControlConfig struct {
 	// SaturationDetector specifies which saturation detector plugin to use for both Admission and
 	// Flow Control. If omitted, "utilization-detector" is used by default.
 	SaturationDetector *SaturationDetectorConfig `json:"saturationDetector,omitempty"`
+
+	// +optional
+	// EnableEviction enables demand-driven in-flight eviction. When higher-priority requests are
+	// blocked by pool saturation, lower-priority in-flight requests (priority < 0) may be
+	// terminated to reclaim capacity. Pacing and sizing self-configure from the selected
+	// saturation detector. See docs/flow-control-eviction.md.
+	// Defaults to false.
+	EnableEviction bool `json:"enableEviction,omitempty"`
 }
 
 func (fcc *FlowControlConfig) String() string {
@@ -445,6 +453,10 @@ func (fcc *FlowControlConfig) String() string {
 		parts = append(parts, fmt.Sprintf("SaturationDetector: %v", fcc.SaturationDetector))
 	}
 
+	if fcc.EnableEviction {
+		parts = append(parts, "EnableEviction: true")
+	}
+
 	return "{" + strings.Join(parts, ", ") + "}"
 }
 

pkg/epp/flowcontrol/controller/config.go

@@ -28,6 +28,16 @@ const (
 	defaultExpiryCleanupInterval = 1 * time.Second
 	// defaultEnqueueChannelBufferSize is the default size of a worker's incoming request buffer.
 	defaultEnqueueChannelBufferSize = 100
+	// defaultMaxRevocationsPerDecision caps revocations per reclamation decision, bounding the
+	// damage from mean-footprint misestimation.
+	defaultMaxRevocationsPerDecision = 2
+	// defaultEvictionConfirmationGrace covers the reclaiming stage (engine abort, KV GC, scrape,
+	// staleness window) for the default utilization detector. Deployments pairing eviction with the
+	// concurrency detector can lower it substantially.
+	defaultEvictionConfirmationGrace = 500 * time.Millisecond
+	// defaultEvictionConfirmationTimeout bounds how long an unconfirmed revocation can hold the
+	// reclamation pacing gate closed.
+	defaultEvictionConfirmationTimeout = 10 * time.Second
 )
 
 // Config holds the configuration for the `FlowController`.
@@ -46,6 +56,33 @@ type Config struct {
 	// serial execution loop and allowing the system to handle short bursts of traffic without blocking.
 	// Optional: Defaults to `defaultEnqueueChannelBufferSize` (100).
 	EnqueueChannelBufferSize int
+
+	// EnableEviction enables demand-driven in-flight eviction: when higher-priority requests are
+	// blocked by pool saturation, lower-priority in-flight requests may be terminated to reclaim
+	// capacity. Requires the eviction plumbing to be wired (see Deps.InFlightEvictor).
+	// See docs/flow-control-eviction.md.
+	// Optional: Defaults to false.
+	EnableEviction bool
+
+	// MaxRevocationsPerDecision caps how many revocations a single reclamation decision may issue.
+	// Not exposed through the API configuration: benchmark data shows sizing is deficit-bound, so
+	// this cap rarely binds and is not worth a user-facing knob.
+	// Optional: Defaults to `defaultMaxRevocationsPerDecision` (2).
+	MaxRevocationsPerDecision int
+
+	// EvictionConfirmationGrace is how long a confirmed revocation's pending-reclaim debit keeps
+	// suppressing further reclamation, covering the saturation signal's confirmation-to-visibility
+	// lag. Not exposed through the API configuration: the EPP wiring derives it from the selected
+	// saturation detector, since the correct value is a property of the sensor, not a preference.
+	// Optional: Defaults to `defaultEvictionConfirmationGrace` (500ms), the conservative value for
+	// scraped sensors.
+	EvictionConfirmationGrace time.Duration
+
+	// EvictionConfirmationTimeout bounds how long an unconfirmed revocation can hold the
+	// reclamation pacing gate closed before being treated as confirmed. Not exposed through the
+	// API configuration.
+	// Optional: Defaults to `defaultEvictionConfirmationTimeout` (10s).
+	EvictionConfirmationTimeout time.Duration
 }
 
 func (c *Config) String() string {
@@ -64,20 +101,26 @@ type ConfigOption func(*Config)
 
 // NewConfigFromAPI creates a new Config from the API configuration.
 func NewConfigFromAPI(apiConfig *configapi.FlowControlConfig) (*Config, error) {
-	opts := make([]ConfigOption, 0, 1)
+	opts := make([]ConfigOption, 0, 4)
 	if apiConfig != nil {
 		if apiConfig.DefaultRequestTTL != nil {
 			opts = append(opts, WithDefaultRequestTTL(apiConfig.DefaultRequestTTL.Duration))
 		}
+		if apiConfig.EnableEviction {
+			opts = append(opts, WithEnableEviction(true))
+		}
 	}
 	return NewConfig(opts...)
 }
 
 // NewConfig creates a new Config with the given options, applying defaults and validation.
 func NewConfig(opts ...ConfigOption) (*Config, error) {
 	c := &Config{
-		ExpiryCleanupInterval:    defaultExpiryCleanupInterval,
-		EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize,
+		ExpiryCleanupInterval:       defaultExpiryCleanupInterval,
+		EnqueueChannelBufferSize:    defaultEnqueueChannelBufferSize,
+		MaxRevocationsPerDecision:   defaultMaxRevocationsPerDecision,
+		EvictionConfirmationGrace:   defaultEvictionConfirmationGrace,
+		EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout,
 	}
 
 	for _, opt := range opts {
@@ -111,6 +154,34 @@ func WithEnqueueChannelBufferSize(size int) ConfigOption {
 	}
 }
 
+// WithEnableEviction enables demand-driven in-flight eviction.
+func WithEnableEviction(enabled bool) ConfigOption {
+	return func(c *Config) {
+		c.EnableEviction = enabled
+	}
+}
+
+// WithMaxRevocationsPerDecision sets the per-decision revocation cap.
+func WithMaxRevocationsPerDecision(n int) ConfigOption {
+	return func(c *Config) {
+		c.MaxRevocationsPerDecision = n
+	}
+}
+
+// WithEvictionConfirmationGrace sets the post-confirmation grace period.
+func WithEvictionConfirmationGrace(d time.Duration) ConfigOption {
+	return func(c *Config) {
+		c.EvictionConfirmationGrace = d
+	}
+}
+
+// WithEvictionConfirmationTimeout sets the confirmation timeout.
+func WithEvictionConfirmationTimeout(d time.Duration) ConfigOption {
+	return func(c *Config) {
+		c.EvictionConfirmationTimeout = d
+	}
+}
+
 // validate checks the configuration for validity.
 func (c *Config) validate() error {
 	if c.DefaultRequestTTL < 0 {
@@ -122,5 +193,14 @@ func (c *Config) validate() error {
 	if c.EnqueueChannelBufferSize < 0 {
 		return fmt.Errorf("EnqueueChannelBufferSize cannot be negative, but got %d", c.EnqueueChannelBufferSize)
 	}
+	if c.MaxRevocationsPerDecision < 1 {
+		return fmt.Errorf("MaxRevocationsPerDecision must be at least 1, but got %d", c.MaxRevocationsPerDecision)
+	}
+	if c.EvictionConfirmationGrace < 0 {
+		return fmt.Errorf("EvictionConfirmationGrace cannot be negative, but got %v", c.EvictionConfirmationGrace)
+	}
+	if c.EvictionConfirmationTimeout <= 0 {
+		return fmt.Errorf("EvictionConfirmationTimeout must be positive, but got %v", c.EvictionConfirmationTimeout)
+	}
 	return nil
 }

pkg/epp/flowcontrol/controller/config_test.go

@@ -41,9 +41,12 @@ func TestNewConfig(t *testing.T) {
 			opts:      nil,
 			expectErr: false,
 			expectedCfg: Config{
-				DefaultRequestTTL:        0,
-				ExpiryCleanupInterval:    defaultExpiryCleanupInterval,
-				EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize,
+				DefaultRequestTTL:           0,
+				ExpiryCleanupInterval:       defaultExpiryCleanupInterval,
+				EnqueueChannelBufferSize:    defaultEnqueueChannelBufferSize,
+				MaxRevocationsPerDecision:   defaultMaxRevocationsPerDecision,
+				EvictionConfirmationGrace:   defaultEvictionConfirmationGrace,
+				EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout,
 			},
 		},
 		{
@@ -53,9 +56,12 @@ func TestNewConfig(t *testing.T) {
 			},
 			expectErr: false,
 			expectedCfg: Config{
-				DefaultRequestTTL:        10 * time.Second,
-				ExpiryCleanupInterval:    defaultExpiryCleanupInterval,
-				EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize,
+				DefaultRequestTTL:           10 * time.Second,
+				ExpiryCleanupInterval:       defaultExpiryCleanupInterval,
+				EnqueueChannelBufferSize:    defaultEnqueueChannelBufferSize,
+				MaxRevocationsPerDecision:   defaultMaxRevocationsPerDecision,
+				EvictionConfirmationGrace:   defaultEvictionConfirmationGrace,
+				EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout,
 			},
 		},
 		{
@@ -67,9 +73,12 @@ func TestNewConfig(t *testing.T) {
 			},
 			expectErr: false,
 			expectedCfg: Config{
-				DefaultRequestTTL:        10 * time.Second,
-				ExpiryCleanupInterval:    2 * time.Second,
-				EnqueueChannelBufferSize: 50,
+				DefaultRequestTTL:           10 * time.Second,
+				ExpiryCleanupInterval:       2 * time.Second,
+				EnqueueChannelBufferSize:    50,
+				MaxRevocationsPerDecision:   defaultMaxRevocationsPerDecision,
+				EvictionConfirmationGrace:   defaultEvictionConfirmationGrace,
+				EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout,
 			},
 		},
 		{
@@ -100,6 +109,45 @@ func TestNewConfig(t *testing.T) {
 			},
 			expectErr: true,
 		},
+		{
+			name: "WithEvictionOptions_ShouldUpdateConfig",
+			opts: []ConfigOption{
+				WithEnableEviction(true),
+				WithMaxRevocationsPerDecision(5),
+				WithEvictionConfirmationGrace(50 * time.Millisecond),
+				WithEvictionConfirmationTimeout(30 * time.Second),
+			},
+			expectErr: false,
+			expectedCfg: Config{
+				ExpiryCleanupInterval:       defaultExpiryCleanupInterval,
+				EnqueueChannelBufferSize:    defaultEnqueueChannelBufferSize,
+				EnableEviction:              true,
+				MaxRevocationsPerDecision:   5,
+				EvictionConfirmationGrace:   50 * time.Millisecond,
+				EvictionConfirmationTimeout: 30 * time.Second,
+			},
+		},
+		{
+			name: "ZeroMaxRevocationsPerDecision_ShouldError",
+			opts: []ConfigOption{
+				WithMaxRevocationsPerDecision(0),
+			},
+			expectErr: true,
+		},
+		{
+			name: "NegativeEvictionConfirmationGrace_ShouldError",
+			opts: []ConfigOption{
+				WithEvictionConfirmationGrace(-1 * time.Second),
+			},
+			expectErr: true,
+		},
+		{
+			name: "ZeroEvictionConfirmationTimeout_ShouldError",
+			opts: []ConfigOption{
+				WithEvictionConfirmationTimeout(0),
+			},
+			expectErr: true,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -175,6 +223,19 @@ func TestNewConfigFromAPI(t *testing.T) {
 			},
 			expectedErr: "DefaultRequestTTL cannot be negative",
 		},
+		{
+			name: "EnableEviction_ShouldTranslate_WithInternalDefaults",
+			apiConfig: &configapi.FlowControlConfig{
+				EnableEviction: true,
+			},
+			assertion: func(t *testing.T, cfg *Config) {
+				assert.True(t, cfg.EnableEviction, "EnableEviction should be translated")
+				// Pacing and sizing parameters are internal: defaulted here, derived at wiring time.
+				assert.Equal(t, defaultMaxRevocationsPerDecision, cfg.MaxRevocationsPerDecision)
+				assert.Equal(t, defaultEvictionConfirmationGrace, cfg.EvictionConfirmationGrace)
+				assert.Equal(t, defaultEvictionConfirmationTimeout, cfg.EvictionConfirmationTimeout)
+			},
+		},
 	}
 
 	for _, tc := range testCases {

pkg/epp/flowcontrol/controller/controller.go

@@ -60,6 +60,7 @@ type processorFactory func(
 	cleanupSweepInterval time.Duration,
 	enqueueChannelBufferSize int,
 	logger logr.Logger,
+	reclamation *internal.ReclamationController,
 ) processor
 
 var _ processor = &internal.Processor{}
@@ -105,6 +106,11 @@ type Deps struct {
 	UsageLimitPolicy   flowcontrol.UsageLimitPolicy
 	Clock              clock.WithTicker
 	ProcessorFactory   processorFactory
+
+	// InFlightEvictor enables demand-driven in-flight eviction when Config.EnableEviction is set.
+	// Satisfied by *eviction.RequestEvictor. The FlowController registers the reclamation
+	// controller's Confirm as the evictor's eviction-terminated listener.
+	InFlightEvictor internal.InFlightEvictor
 }
 
 // NewFlowController creates and starts a new FlowController instance.
@@ -147,6 +153,7 @@ func NewFlowController(
 			cleanupSweepInterval time.Duration,
 			enqueueChannelBufferSize int,
 			logger logr.Logger,
+			reclamation *internal.ReclamationController,
 		) processor {
 			return internal.NewProcessor(
 				ctx,
@@ -160,12 +167,36 @@ func NewFlowController(
 				cleanupSweepInterval,
 				enqueueChannelBufferSize,
 				logger,
+				reclamation,
 			)
 		}
 	} else {
 		fc.processorFactory = deps.ProcessorFactory
 	}
 
+	// Demand-driven in-flight eviction is active only when both the config enables it and the
+	// eviction plumbing was wired in. The reclamation controller's Confirm becomes the evictor's
+	// eviction-terminated listener, closing the confirmation loop.
+	var reclamation *internal.ReclamationController
+	if config.EnableEviction && deps.InFlightEvictor != nil {
+		reclamation = internal.NewReclamationController(
+			internal.ReclamationConfig{
+				MaxRevocationsPerDecision: config.MaxRevocationsPerDecision,
+				ConfirmationGrace:         config.EvictionConfirmationGrace,
+				ConfirmationTimeout:       config.EvictionConfirmationTimeout,
+			},
+			deps.InFlightEvictor,
+			deps.Clock,
+			fc.logger,
+			poolName,
+		)
+		deps.InFlightEvictor.SetEvictionTerminatedListener(reclamation.Confirm)
+		fc.logger.V(logutil.DEFAULT).Info("Demand-driven in-flight eviction enabled.",
+			"maxRevocationsPerDecision", config.MaxRevocationsPerDecision,
+			"confirmationGrace", config.EvictionConfirmationGrace,
+			"confirmationTimeout", config.EvictionConfirmationTimeout)
+	}
+
 	// Construct a new worker, but do not start its goroutine yet.
 	fc.processor = fc.processorFactory(
 		fc.parentCtx,
@@ -178,6 +209,7 @@ func NewFlowController(
 		fc.config.ExpiryCleanupInterval,
 		fc.config.EnqueueChannelBufferSize,
 		fc.logger,
+		reclamation,
 	)
 
 	fc.logger.V(logutil.DEFAULT).Info("Starting the Processor.")

pkg/epp/flowcontrol/controller/controller_test.go

@@ -287,6 +287,7 @@ func (f *mockProcessorFactory) new(
 	_ time.Duration,
 	_ int,
 	_ logr.Logger,
+	_ *internal.ReclamationController,
 ) processor {
 	if f.processor != nil {
 		return f.processor

pkg/epp/flowcontrol/controller/internal/processor.go

@@ -75,6 +75,10 @@ type Processor struct {
 	cleanupSweepInterval time.Duration
 	logger               logr.Logger
 
+	// reclamation, when non-nil, enables demand-driven in-flight eviction on HoL blocking.
+	// See docs/flow-control-eviction.md.
+	reclamation *ReclamationController
+
 	// lifecycleCtx controls the processor's lifetime. Monitored by Submit* methods for safe shutdown.
 	lifecycleCtx context.Context
 
@@ -106,6 +110,7 @@ func NewProcessor(
 	cleanupSweepInterval time.Duration,
 	enqueueChannelBufferSize int,
 	logger logr.Logger,
+	reclamation *ReclamationController,
 ) *Processor {
 	return &Processor{
 		registry:             registry,
@@ -119,6 +124,7 @@ func NewProcessor(
 		logger:               logger,
 		lifecycleCtx:         ctx,
 		enqueueChan:          make(chan *FlowItem, enqueueChannelBufferSize),
+		reclamation:          reclamation,
 	}
 }
 
@@ -377,6 +383,9 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool {
 		if saturation >= usageLimit {
 			p.logger.V(logutil.DEBUG).Info("Priority band is saturated; enforcing HoL blocking.",
 				"priority", priority, "saturation", saturation, "usageLimit", usageLimit)
+			if p.reclamation != nil {
+				p.maybeReclaim(ctx, saturation, priorities, ceilings, i)
+			}
 			// Stop the dispatch cycle entirely to respect strict policy decision and prevent priority inversion where
 			// lower-priority work might exacerbate the saturation affecting high-priority work.
 			return false

pkg/epp/flowcontrol/controller/internal/processor_test.go

@@ -142,7 +142,8 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn
 		h.clock,
 		expiryCleanupInterval,
 		100,
-		h.logger)
+		h.logger,
+		nil)
 	require.NotNil(t, h.processor, "NewProcessor should not return nil")
 
 	t.Cleanup(func() { h.Stop() })