feat(flowcontrol): support scoped request queue TTLs

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -564,6 +564,12 @@ type PriorityBandConfig struct {
 	// effectively remove the bound, set an explicit large value.
 	MaxRequests *resource.Quantity `json:"maxRequests,omitempty"`
 
+	// +optional
+	// DefaultRequestTTL bounds how long a request may wait in this priority band before it is evicted.
+	// If omitted, the global DefaultRequestTTL is used. An explicit value replaces the global default;
+	// "0s" makes queue wait unbounded for this band.
+	DefaultRequestTTL *metav1.Duration `json:"defaultRequestTTL,omitempty"`
+
 	// +optional
 	// FairnessPolicyRef specifies the name of the policy that governs flow selection.
 	// If omitted, the system default ("global-strict-fairness-policy") is used.
@@ -587,6 +593,10 @@ func (pbc PriorityBandConfig) String() string {
 		parts = append(parts, fmt.Sprintf("MaxRequests: %d", pbc.MaxRequests.Value()))
 	}
 
+	if pbc.DefaultRequestTTL != nil {
+		parts = append(parts, fmt.Sprintf("DefaultRequestTTL: %s", pbc.DefaultRequestTTL.Duration))
+	}
+
 	if pbc.FairnessPolicyRef != "" {
 		parts = append(parts, "FairnessPolicyRef: "+pbc.FairnessPolicyRef)
 	}

apix/config/v1alpha1/endpointpickerconfig_types_test.go

@@ -63,13 +63,17 @@ func TestStringers(t *testing.T) {
 				MaxBytes:          resource.NewQuantity(1024, resource.DecimalSI),
 				DefaultRequestTTL: &metav1.Duration{Duration: 30 * time.Second},
 				PriorityBands: []PriorityBandConfig{
-					{Priority: 10, MaxBytes: resource.NewQuantity(512, resource.DecimalSI)},
+					{
+						Priority:          10,
+						MaxBytes:          resource.NewQuantity(512, resource.DecimalSI),
+						DefaultRequestTTL: &metav1.Duration{Duration: 5 * time.Second},
+					},
 				},
 				SaturationDetector: &SaturationDetectorConfig{
 					PluginRef: "test-plugin",
 				},
 			},
-			want: "{MaxBytes: 1024, MaxRequests: unlimited, DefaultRequestTTL: 30s, PriorityBands: [{Priority: 10, MaxBytes: 512}], SaturationDetector: {PluginRef: test-plugin}}",
+			want: "{MaxBytes: 1024, MaxRequests: unlimited, DefaultRequestTTL: 30s, PriorityBands: [{Priority: 10, MaxBytes: 512, DefaultRequestTTL: 5s}], SaturationDetector: {PluginRef: test-plugin}}",
 		},
 		{
 			name: "RequestHandlerConfig",

pkg/epp/config/loader/configloader_test.go

@@ -745,6 +745,11 @@ func TestInstantiateAndConfigure(t *testing.T) {
 				require.NotNil(t, cfg.FlowControlConfig, "FlowControl config should be loaded")
 				require.Contains(t, cfg.FlowControlConfig.Registry.PriorityBands, 100, "Should contain priority band 100")
 				band := cfg.FlowControlConfig.Registry.PriorityBands[100]
+				require.NotNil(t, band.DefaultRequestTTL)
+				require.Equal(t, 5*time.Minute, *band.DefaultRequestTTL)
+				unboundedBand := cfg.FlowControlConfig.Registry.PriorityBands[-1]
+				require.NotNil(t, unboundedBand.DefaultRequestTTL)
+				require.Zero(t, *unboundedBand.DefaultRequestTTL)
 
 				// Verify custom policies.
 				require.Equal(t, "customFCFS", band.OrderingPolicy.TypedName().Name,

pkg/epp/config/loader/flowcontrol.go

@@ -145,7 +145,11 @@ func buildPriorityBand(
 	band *configapi.PriorityBandConfig,
 	label string,
 ) (*registry.PriorityBandConfig, error) {
-	bandOpts := make([]registry.PriorityBandConfigOption, 0, 4)
+	bandOpts := make([]registry.PriorityBandConfigOption, 0, 5)
+
+	if band.DefaultRequestTTL != nil {
+		bandOpts = append(bandOpts, registry.WithBandDefaultRequestTTL(band.DefaultRequestTTL.Duration))
+	}
 
 	maxBytes, err := resolveQuantity(band.MaxBytes, label+" MaxBytes")
 	if err != nil {

pkg/epp/config/loader/flowcontrol_test.go

@@ -19,10 +19,12 @@ package loader
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 
 	configapi "github.com/llm-d/llm-d-router/apix/config/v1alpha1"
@@ -110,6 +112,33 @@ func TestBuildRegistryConfig(t *testing.T) {
 					"DefaultPriorityBand template MaxBytes should be translated")
 			},
 		},
+		{
+			name: "ShouldResolveBandRequestTTLs",
+			apiConfig: &configapi.FlowControlConfig{
+				DefaultPriorityBand: &configapi.PriorityBandConfig{
+					DefaultRequestTTL: &metav1.Duration{Duration: 10 * time.Second},
+				},
+				DefaultNegativePriorityBand: &configapi.PriorityBandConfig{
+					DefaultRequestTTL: &metav1.Duration{Duration: 20 * time.Second},
+				},
+				PriorityBands: []configapi.PriorityBandConfig{
+					{Priority: 1},
+					{Priority: 2, DefaultRequestTTL: &metav1.Duration{}},
+					{Priority: 3, DefaultRequestTTL: &metav1.Duration{Duration: 5 * time.Second}},
+				},
+			},
+			assertion: func(t *testing.T, cfg *registry.Config) {
+				assert.Nil(t, cfg.PriorityBands[1].DefaultRequestTTL)
+				require.NotNil(t, cfg.PriorityBands[2].DefaultRequestTTL)
+				assert.Zero(t, *cfg.PriorityBands[2].DefaultRequestTTL)
+				require.NotNil(t, cfg.PriorityBands[3].DefaultRequestTTL)
+				assert.Equal(t, 5*time.Second, *cfg.PriorityBands[3].DefaultRequestTTL)
+				require.NotNil(t, cfg.DefaultPriorityBand.DefaultRequestTTL)
+				assert.Equal(t, 10*time.Second, *cfg.DefaultPriorityBand.DefaultRequestTTL)
+				require.NotNil(t, cfg.DefaultNegativePriorityBand.DefaultRequestTTL)
+				assert.Equal(t, 20*time.Second, *cfg.DefaultNegativePriorityBand.DefaultRequestTTL)
+			},
+		},
 		{
 			name: "ShouldSucceed_WithKubernetesQuantityFormat",
 			apiConfig: &configapi.FlowControlConfig{
@@ -255,6 +284,15 @@ func TestBuildRegistryConfig(t *testing.T) {
 		},
 
 		// --- Validation Errors ---
+		{
+			name: "ShouldError_WithNegativePriorityBandRequestTTL",
+			apiConfig: &configapi.FlowControlConfig{
+				PriorityBands: []configapi.PriorityBandConfig{
+					{Priority: 1, DefaultRequestTTL: &metav1.Duration{Duration: -time.Second}},
+				},
+			},
+			expectedErr: "defaultRequestTTL cannot be negative",
+		},
 		{
 			name: "ShouldError_WithNegativeGlobalMaxBytes",
 			apiConfig: &configapi.FlowControlConfig{

pkg/epp/config/loader/testdata_test.go

@@ -364,10 +364,14 @@ schedulingProfiles:
 featureGates:
 - flowControl
 flowControl:
+  defaultRequestTTL: 1m
   priorityBands:
   - priority: 100
+    defaultRequestTTL: 5m
     orderingPolicyRef: customFCFS
     fairnessPolicyRef: customFairness
+  - priority: -1
+    defaultRequestTTL: 0s
 `
 
 // successParserConfigText tests that configuration with parser plugin is correctly loaded.

pkg/epp/flowcontrol/README.md

@@ -64,7 +64,9 @@ Tuning knobs, all under the `flowControl:` config section:
   behavior for sheddable traffic).
 * `defaultRequestTTL` — the queue-wait budget against a pool that has endpoints, and the other way a
   request is shed. Keep it under the client or gateway deadline, and size it to the time-to-first-token
-  budget you are willing to spend waiting on a saturated pool.
+  budget you are willing to spend waiting on a saturated pool. Priority-band entries and templates
+  may replace the global value, including with `0s` for unbounded queue wait. Clients may shorten the
+  selected operator bound with `x-llm-d-inference-ttl` using Go duration syntax.
 * `noEndpointRequestTTL` — the queue-wait budget that replaces `defaultRequestTTL` while the pool has
   no endpoints, where the queue acts as a scale-from-zero waiting room. Left unset it follows
   `defaultRequestTTL`, so splitting the regimes is opt-in. Size it above pod startup (image pull plus

pkg/epp/flowcontrol/contracts/registry.go

@@ -159,6 +159,8 @@ type ActiveFlowConnection interface {
 	GetDataPlane() FlowRegistryDataPlane
 	// FlowKey returns the immutable identity of the flow this connection is pinned to.
 	FlowKey() flowcontrol.FlowKey
+	// DefaultRequestTTL returns the queue-wait bound configured for the leased priority band and whether it was set.
+	DefaultRequestTTL() (time.Duration, bool)
 }
 
 // ManagedQueue defines the interface for a flow's queue.

pkg/epp/flowcontrol/controller/config.go

@@ -54,9 +54,7 @@ const (
 
 // Config holds the configuration for the `FlowController`.
 type Config struct {
-	// DefaultRequestTTL is the default Time-To-Live applied to requests that do not specify their own
-	// TTL hint. Because the admission adapter does not currently plumb a per-request hint, this value
-	// governs every request entering flow control while the candidate pool has endpoints.
+	// DefaultRequestTTL is the fallback Time-To-Live for priority bands that do not configure one.
 	// Optional: Defaults to `defaultRequestTTL` (60s). An explicit zero disables eviction in that
 	// regime, and, unless `NoEndpointRequestTTL` overrides it, in the empty-pool regime as well; such
 	// requests are then bounded only by request context cancellation (client disconnect or gateway

pkg/epp/flowcontrol/controller/controller.go

@@ -256,13 +256,18 @@ func (fc *FlowController) EnqueueAndWait(
 		req.InferencePoolName(),
 		req.ModelName(), req.TargetModelName(), reqBytes)
 
-	// 1. Create the derived context that governs this request's lifecycle (Parent Cancellation + TTL).
-	reqCtx, cancel, enqueueTime, saturationTTL := fc.createRequestContext(ctx, req)
-	defer cancel()
+	// Capture the logical enqueue time before acquiring the flow. The band's TTL is only available
+	// from the acquired connection, but time spent acquiring it still counts against the queue budget.
+	enqueueTime := fc.clock.Now()
 
 	// 2. Acquire a lease for the Flow.
 	// We hold this lease for the entire duration of the request (Distribution + Queueing).
 	err := fc.withConnectionWithFallback(req, func(conn contracts.ActiveFlowConnection, effectiveReq flowcontrol.FlowControlRequest) error {
+		bandDefaultRequestTTL, bandDefaultRequestTTLSet := conn.DefaultRequestTTL()
+		reqCtx, cancel, saturationTTL := fc.createRequestContext(
+			ctx, effectiveReq, bandDefaultRequestTTL, bandDefaultRequestTTLSet, enqueueTime,
+		)
+		defer cancel()
 
 		select { // Non-blocking check on controller lifecycle.
 		case <-fc.parentCtx.Done():
@@ -361,17 +366,8 @@ func (fc *FlowController) tryDistribution(
 	saturationTTL time.Duration,
 	conn contracts.ActiveFlowConnection,
 ) (*internal.FlowItem, error) {
-	// The item carries the saturation-regime budget: it is the request's own queue-wait budget, and ordering policies
-	// read it as such. A caller deadline that falls inside it clamps it, since the request cannot outlive its caller.
-	effectiveTTL := saturationTTL
-	if deadline, ok := reqCtx.Deadline(); ok {
-		if ttl := deadline.Sub(enqueueTime); ttl > 0 && (effectiveTTL <= 0 || ttl < effectiveTTL) {
-			effectiveTTL = ttl
-		}
-	}
-
 	// We must create a fresh FlowItem on each attempt as finalization is per-lifecycle.
-	item := internal.NewItem(req, effectiveTTL, enqueueTime, fc.logger)
+	item := internal.NewItem(req, saturationTTL, enqueueTime, fc.logger)
 
 	dp := conn.GetDataPlane()
 	_, err := dp.ManagedQueue(conn.FlowKey())
@@ -393,9 +389,9 @@ func (fc *FlowController) tryDistribution(
 	// for handoff has not reached a queue, so it is not waiting on an endpoint to appear and the no-endpoint budget does
 	// not describe it; the regime-aware budget takes over once the processor owns the item.
 	distributeCtx := reqCtx
-	if effectiveTTL > 0 {
+	if saturationTTL > 0 {
 		var cancel context.CancelFunc
-		distributeCtx, cancel = context.WithDeadlineCause(reqCtx, enqueueTime.Add(effectiveTTL), types.ErrTTLExpired)
+		distributeCtx, cancel = context.WithDeadlineCause(reqCtx, enqueueTime.Add(saturationTTL), types.ErrTTLExpired)
 		defer cancel()
 	}
 
@@ -467,21 +463,37 @@ func (fc *FlowController) awaitFinalization(
 func (fc *FlowController) createRequestContext(
 	ctx context.Context,
 	req flowcontrol.FlowControlRequest,
-) (context.Context, context.CancelFunc, time.Time, time.Duration) {
-	enqueueTime := fc.clock.Now()
-	saturationTTL := req.InitialEffectiveTTL()
-	if saturationTTL <= 0 {
-		saturationTTL = fc.config.DefaultRequestTTL
+	bandDefaultRequestTTL time.Duration,
+	bandDefaultRequestTTLSet bool,
+	enqueueTime time.Time,
+) (context.Context, context.CancelFunc, time.Duration) {
+	saturationTTL := fc.config.DefaultRequestTTL
+	if bandDefaultRequestTTLSet {
+		saturationTTL = bandDefaultRequestTTL
+	}
+	// A request may make the selected operator bound stricter, but not extend it.
+	if requestTTL := req.InitialEffectiveTTL(); requestTTL > 0 && (saturationTTL <= 0 || requestTTL < saturationTTL) {
+		saturationTTL = requestTTL
 	}
 
 	// A zero budget in either regime disables eviction there, so no backstop can be derived.
+	var reqCtx context.Context
+	var cancel context.CancelFunc
 	if saturationTTL > 0 && fc.config.NoEndpointRequestTTL > 0 {
 		backstop := max(saturationTTL, fc.config.NoEndpointRequestTTL) + 2*fc.config.ExpiryCleanupInterval
-		reqCtx, cancel := context.WithDeadlineCause(ctx, enqueueTime.Add(backstop), types.ErrTTLExpired)
-		return reqCtx, cancel, enqueueTime, saturationTTL
+		reqCtx, cancel = context.WithDeadlineCause(ctx, enqueueTime.Add(backstop), types.ErrTTLExpired)
+	} else {
+		reqCtx, cancel = context.WithCancel(ctx)
+	}
+
+	// Ordering policies use the saturation budget, clamped to a caller deadline that fires sooner.
+	if deadline, ok := reqCtx.Deadline(); ok {
+		if lifecycleTTL := deadline.Sub(enqueueTime); lifecycleTTL > 0 &&
+			(saturationTTL <= 0 || lifecycleTTL < saturationTTL) {
+			saturationTTL = lifecycleTTL
+		}
 	}
-	reqCtx, cancel := context.WithCancel(ctx)
-	return reqCtx, cancel, enqueueTime, saturationTTL
+	return reqCtx, cancel, saturationTTL
 }
 
 // distributeRequest submits an item to the processor with graceful backpressure.
@@ -502,6 +514,11 @@ func (fc *FlowController) distributeRequest(
 	item *internal.FlowItem,
 ) error {
 	reqID := item.OriginalRequest().ID()
+	select {
+	case <-ctx.Done():
+		return fmt.Errorf("%w: request not accepted: %w", types.ErrRejected, ctx.Err())
+	default:
+	}
 	if err := fc.processor.Submit(item); err == nil {
 		return nil
 	}