refactor(flowcontrol)!: UsageLimitPolicy fills a framework-owned ceilings buffer

ComputeLimit returned a slice parallel to the priorities argument, and
the dispatch loop indexed it unchecked: a policy returning a short slice
panicked the processor. The policy now writes into a caller-provided
buffer that the framework sizes to len(priorities) and pre-fills with
1.0, so a mis-sized result is unrepresentable and an unwritten entry
fails open instead of carrying stale state. The processor reuses one
buffer across cycles, removing the per-cycle allocation on the 1ms
dispatch path.

A batch call (rather than per-band) is retained deliberately: duty-cycle
policies such as soft-reflective-ceiling advance shared state once per
dispatch cycle and require all gated bands to observe the same open or
closed decision within a cycle.

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

Author: LukeAVanDrie

pkg/epp/config/loader/flowcontrol_test.go

@@ -448,13 +448,12 @@ func TestBuildFlowControlConfig(t *testing.T) {
 	handle := newFlowControlTestHandle(t)
 
 	const funcPolicyName = "func-policy"
-	handle.AddPlugin(funcPolicyName, usagelimits.NewPolicyFunc(funcPolicyName, func(_ context.Context, _ float64, priorities []int) []float64 {
-		result := make([]float64, len(priorities))
-		for i := range result {
-			result[i] = 0.8
-		}
-		return result
-	}))
+	handle.AddPlugin(funcPolicyName, usagelimits.NewPolicyFunc(funcPolicyName,
+		func(_ context.Context, _ float64, _ []int, ceilings []float64) {
+			for i := range ceilings {
+				ceilings[i] = 0.8
+			}
+		}))
 
 	const structPolicyName = "struct-policy"
 	handle.AddPlugin(structPolicyName, &constantPointEightPolicy{})
@@ -490,7 +489,7 @@ func TestBuildFlowControlConfig(t *testing.T) {
 			apiConfig: nil,
 			assertion: func(t *testing.T, cfg *flowcontrol.Config) {
 				require.NotNil(t, cfg.UsageLimitPolicy, "UsageLimitPolicy should be resolved even when not explicitly configured")
-				ceilings := cfg.UsageLimitPolicy.ComputeLimit(context.Background(), 0.5, []int{0})
+				ceilings := computeLimits(t, cfg.UsageLimitPolicy, 0.5, []int{0})
 				assert.Equal(t, []float64{1.0}, ceilings, "Default noop policy should return 1.0 (no gating)")
 			},
 		},
@@ -501,7 +500,7 @@ func TestBuildFlowControlConfig(t *testing.T) {
 			},
 			assertion: func(t *testing.T, cfg *flowcontrol.Config) {
 				require.NotNil(t, cfg.UsageLimitPolicy, "UsageLimitPolicy should be resolved from the handle")
-				ceilings := cfg.UsageLimitPolicy.ComputeLimit(context.Background(), 0.5, []int{0})
+				ceilings := computeLimits(t, cfg.UsageLimitPolicy, 0.5, []int{0})
 				assert.Equal(t, []float64{1.0}, ceilings, "Noop policy should return 1.0 (no gating)")
 			},
 		},
@@ -512,7 +511,6 @@ func TestBuildFlowControlConfig(t *testing.T) {
 			},
 			assertion: func(t *testing.T, cfg *flowcontrol.Config) {
 				require.NotNil(t, cfg.UsageLimitPolicy)
-				ctx := context.Background()
 				for _, tc := range []struct {
 					name       string
 					priority   int
@@ -522,7 +520,7 @@ func TestBuildFlowControlConfig(t *testing.T) {
 					{"half saturation", 1, 0.5},
 					{"full saturation", 5, 1.0},
 				} {
-					assert.Equal(t, []float64{0.8}, cfg.UsageLimitPolicy.ComputeLimit(ctx, tc.saturation, []int{tc.priority}),
+					assert.Equal(t, []float64{0.8}, computeLimits(t, cfg.UsageLimitPolicy, tc.saturation, []int{tc.priority}),
 						"func-based policy should return 0.8 at %s", tc.name)
 				}
 			},
@@ -534,7 +532,6 @@ func TestBuildFlowControlConfig(t *testing.T) {
 			},
 			assertion: func(t *testing.T, cfg *flowcontrol.Config) {
 				require.NotNil(t, cfg.UsageLimitPolicy)
-				ctx := context.Background()
 				for _, tc := range []struct {
 					name       string
 					priority   int
@@ -544,7 +541,7 @@ func TestBuildFlowControlConfig(t *testing.T) {
 					{"half saturation", 1, 0.5},
 					{"full saturation", 5, 1.0},
 				} {
-					assert.Equal(t, []float64{0.8}, cfg.UsageLimitPolicy.ComputeLimit(ctx, tc.saturation, []int{tc.priority}),
+					assert.Equal(t, []float64{0.8}, computeLimits(t, cfg.UsageLimitPolicy, tc.saturation, []int{tc.priority}),
 						"struct-based policy should return 0.8 at %s", tc.name)
 				}
 			},
@@ -608,12 +605,22 @@ func (p *constantPointEightPolicy) TypedName() fwkplugin.TypedName {
 	}
 }
 
-func (p *constantPointEightPolicy) ComputeLimit(_ context.Context, _ float64, priorities []int) []float64 {
-	result := make([]float64, len(priorities))
-	for i := range result {
-		result[i] = 0.8
+func (p *constantPointEightPolicy) ComputeLimit(_ context.Context, _ float64, _ []int, ceilings []float64) {
+	for i := range ceilings {
+		ceilings[i] = 0.8
 	}
-	return result
 }
 
 var _ fwkfc.UsageLimitPolicy = (*constantPointEightPolicy)(nil)
+
+// computeLimits invokes a UsageLimitPolicy with a framework-style output buffer (pre-filled with
+// 1.0, sized to priorities) and returns the filled ceilings.
+func computeLimits(t *testing.T, p fwkfc.UsageLimitPolicy, saturation float64, priorities []int) []float64 {
+	t.Helper()
+	ceilings := make([]float64, len(priorities))
+	for i := range ceilings {
+		ceilings[i] = 1.0
+	}
+	p.ComputeLimit(context.Background(), saturation, priorities, ceilings)
+	return ceilings
+}

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

@@ -87,6 +87,11 @@ type Processor struct {
 	// it needs no synchronization.
 	poolEmpty bool
 
+	// ceilings is the reusable output buffer handed to the UsageLimitPolicy each dispatch cycle,
+	// avoiding a per-cycle allocation. Only accessed from the Run goroutine, so it needs no
+	// synchronization.
+	ceilings []float64
+
 	// wg is used to wait for background tasks (cleanup sweep) to complete on shutdown.
 	wg             sync.WaitGroup
 	isShuttingDown atomic.Bool
@@ -381,7 +386,8 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool {
 	metrics.RecordFlowControlPoolSaturation(p.poolName, saturation)
 
 	priorities := p.registry.AllOrderedPriorityLevels()
-	ceilings := p.usageLimitPolicy.ComputeLimit(ctx, saturation, priorities)
+	ceilings := p.ceilingsBuffer(len(priorities))
+	p.usageLimitPolicy.ComputeLimit(ctx, saturation, priorities, ceilings)
 
 	for i, priority := range priorities {
 		// --- Viability Check (Saturation/HoL Blocking) ---
@@ -423,6 +429,20 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool {
 	return false
 }
 
+// ceilingsBuffer returns the reusable ceilings buffer sized to n, every element reset to 1.0 (no
+// gating). Pre-filling guarantees that an entry the policy does not write fails open rather than
+// carrying a stale value from the previous cycle.
+func (p *Processor) ceilingsBuffer(n int) []float64 {
+	if cap(p.ceilings) < n {
+		p.ceilings = make([]float64, n)
+	}
+	buf := p.ceilings[:n]
+	for i := range buf {
+		buf[i] = 1.0
+	}
+	return buf
+}
+
 // selectItem applies the configured fairness and ordering policies to select a single item.
 func (p *Processor) selectItem(
 	ctx context.Context,

pkg/epp/framework/interface/flowcontrol/plugins.go

@@ -143,32 +143,42 @@ type SaturationDetector interface {
 //
 // Integration:
 // This policy is called during dispatch decision-making, before a request is allowed to proceed. For each
-// priority band, the returned ceiling is compared against current saturation. If saturation exceeds the
+// priority band, the computed ceiling is compared against current saturation. If saturation exceeds the
 // ceiling for a given priority, requests at that priority are gated (not dispatched). The dispatch loop
 // visits bands from highest to lowest priority and stops at the first gated band; lower bands are not
 // considered on that call.
 //
-// Conformance: Implementations MUST ensure all methods are goroutine-safe. Returned ceilings MUST be
+// The framework calls ComputeLimit exactly once per dispatch cycle. This is a contract term, not an
+// implementation detail: dispatch-spreading policies use the call itself as their time base (one tick
+// per cycle), and computing all ceilings in one call is what lets every gated band observe the same
+// open/closed decision within a cycle. The batch shape exists to preserve both properties.
+//
+// Conformance: Implementations MUST ensure all methods are goroutine-safe. Computed ceilings MUST be
 // monotonically non-increasing in the given priority order (highest priority first): because the
 // dispatch loop stops at the first gated band, a lower band whose ceiling exceeds that of a higher band
 // can be marked open on calls where it is unreachable, starving it.
 type UsageLimitPolicy interface {
 	plugin.Plugin
 
 	// ComputeLimit calculates usage ceilings for all currently active priority levels based on current
-	// saturation. The plugin observes the active priority domain (which changes dynamically as workloads
-	// come and go) and computes relative ceilings from scratch on each call.
+	// saturation, writing the ceiling for the n-th priority into the n-th element of the
+	// caller-provided ceilings buffer. The plugin observes the active priority domain (which changes
+	// dynamically as workloads come and go) and computes relative ceilings from scratch on each call.
+	//
+	// The framework guarantees len(ceilings) == len(priorities) and pre-fills every element with 1.0
+	// (no gating), so an entry the plugin does not write fails open. Writing into the framework-owned
+	// buffer means a result of the wrong size cannot exist; there is no return value to validate.
 	//
 	// Parameters:
 	//   - ctx: Request context for logging, tracing, etc.
 	//   - saturation: Current pool-wide resource saturation as a fraction [0.0, 1.0]
-	//   - priorities: Ordered list of currently active priority levels (highest first)
-	//
-	// Returns:
-	//   - ceilings: Computed ceiling for each given priority (n-th ceiling is assigned to the given n-th priority)
+	//   - priorities: Ordered list of currently active priority levels (highest first).
+	//     The slice is a shared snapshot owned by the framework: read-only, and it MUST NOT be
+	//     retained after the call returns.
+	//   - ceilings: Output buffer owned by the framework, valid only for the duration of the call.
+	//     Ceiling semantics per element:
 	//     - 0.0 = fully gated (cannot dispatch regardless of current saturation)
 	//     - 1.0 = no gating (can dispatch until fully saturated)
 	//     - Values between 0.0 and 1.0 reserve capacity headroom
-	//
-	ComputeLimit(ctx context.Context, saturation float64, priorities []int) (ceilings []float64)
+	ComputeLimit(ctx context.Context, saturation float64, priorities []int, ceilings []float64)
 }

pkg/epp/framework/plugins/flowcontrol/usagelimits/priorityholdback/priority_holdback.go

@@ -56,13 +56,13 @@ type priorityHoldbackPolicy struct {
 	name      string
 	cMax      float64
 	cMin      float64
-	computeFn func(cMin, cMax float64, priorities []int) (ceilings []float64)
+	computeFn func(cMin, cMax float64, priorities []int, ceilings []float64)
 }
 
 var _ flowcontrol.UsageLimitPolicy = &priorityHoldbackPolicy{}
 
 func newPriorityHoldbackPolicy(cfg config) *priorityHoldbackPolicy {
-	var fn func(cMin, cMax float64, priorities []int) (ceilings []float64)
+	var fn func(cMin, cMax float64, priorities []int, ceilings []float64)
 	switch cfg.domain {
 	case domainRank:
 		fn = computeLimitStepwiseSpread
@@ -95,37 +95,36 @@ func (p *priorityHoldbackPolicy) TypedName() plugin.TypedName {
 	}
 }
 
-// ComputeLimit returns an admission ceiling for each priority. With a single active priority,
-// holdback is bypassed (ceiling = cMax) to preserve work-conserving behavior.
-func (p *priorityHoldbackPolicy) ComputeLimit(_ context.Context, _ float64, priorities []int) (ceilings []float64) {
+// ComputeLimit writes an admission ceiling for each priority into the caller-provided buffer.
+// With a single active priority, holdback is bypassed (ceiling = cMax) to preserve
+// work-conserving behavior.
+func (p *priorityHoldbackPolicy) ComputeLimit(_ context.Context, _ float64, priorities []int, ceilings []float64) {
 	if len(priorities) == 0 {
-		return []float64{}
+		return
 	}
 	if len(priorities) == 1 {
-		return []float64{p.cMax}
+		ceilings[0] = p.cMax
+		return
 	}
 	// Ceilings are monotonically decreasing as priorities are ordered from highest to lowest per UsageLimitPolicy contract.
 	// New strategies (e.g. sigmoid/static definition) could require explicit monotizing sweep.
-	return p.computeFn(p.cMin, p.cMax, priorities)
+	p.computeFn(p.cMin, p.cMax, priorities, ceilings)
 }
 
 // computeLimitStepwiseSpread divides [cMin, cMax] into equal steps by rank.
 // c_i = cMax - i * (cMax - cMin) / (N - 1)
-func computeLimitStepwiseSpread(cMin, cMax float64, priorities []int) (ceilings []float64) {
-	ceilings = make([]float64, len(priorities))
+func computeLimitStepwiseSpread(cMin, cMax float64, priorities []int, ceilings []float64) {
 	spread := cMax - cMin
 	n := float64(len(priorities) - 1)
 	for i := range priorities {
 		ceilings[i] = cMax - float64(i)*spread/n
 	}
-	return ceilings
 }
 
 // computeLimitLinearProportional scales ceilings proportionally to numerical priority values.
 // r_i = (p_i - pMin) / (pMax - pMin)
 // c_i = cMin + r_i * (cMax - cMin)
-func computeLimitLinearProportional(cMin, cMax float64, priorities []int) (ceilings []float64) {
-	ceilings = make([]float64, len(priorities))
+func computeLimitLinearProportional(cMin, cMax float64, priorities []int, ceilings []float64) {
 	pMin := float64(priorities[len(priorities)-1])
 	pMax := float64(priorities[0])
 	pRange := pMax - pMin
@@ -135,12 +134,11 @@ func computeLimitLinearProportional(cMin, cMax float64, priorities []int) (ceili
 		for i := range priorities {
 			ceilings[i] = cMax
 		}
-		return ceilings
+		return
 	}
 	spread := cMax - cMin
 	for i := range priorities {
 		r := (float64(priorities[i]) - pMin) / pRange
 		ceilings[i] = cMin + r*spread
 	}
-	return ceilings
 }