refactor(flowcontrol): address non-blocking review feedback from llm-d#1354

No functional change. Picks up review nits left on llm-d#1354:

- ensureFlowInfrastructure: drop the redundant priority-band existence
  check. buildFlowComponents already re-checks the same map under its own
  read lock and returns the identical ErrPriorityBandNotFound, so fold the
  not-found case into that single critical section (one lock acquisition
  instead of two on a flow's first request).
- Rename the priority-0 fallback from "demotion" to "fallback"
  (withConnectionWithFallback / fallbackRequest / fallbackKey). Relative to
  the requested priority, falling back to 0 is a demotion for positive
  priorities but a promotion for negative ones, so the neutral name is more
  accurate; note the semantics in the doc comment and log at priority-0.
- SubmitDesiredPriorities: rename the local map to avoid shadowing the
  builtin copy.
- ApplyDesiredPriorities: reword the post-idle re-check comment. Holding
  fr.mu does not exclude a concurrent lock-free pin; the check is a cheap
  best-effort guard and the real safety net is pinLeasedResource's
  stale-object protection plus the priority-0 fallback and self-healing GC.

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

Author: LukeAVanDrie

pkg/epp/flowcontrol/controller/controller.go

@@ -239,7 +239,7 @@ func (fc *FlowController) EnqueueAndWait(
 
 	// 2. Acquire a lease for the Flow.
 	// We hold this lease for the entire duration of the request (Distribution + Queueing).
-	err := fc.withConnectionWithDemotion(req, func(conn contracts.ActiveFlowConnection, effectiveReq flowcontrol.FlowControlRequest) error {
+	err := fc.withConnectionWithFallback(req, func(conn contracts.ActiveFlowConnection, effectiveReq flowcontrol.FlowControlRequest) error {
 
 		select { // Non-blocking check on controller lifecycle.
 		case <-fc.parentCtx.Done():
@@ -249,7 +249,7 @@ func (fc *FlowController) EnqueueAndWait(
 		}
 
 		// Attempt to distribute the request once, passing the active connection.
-		// effectiveReq carries the demoted flow key when the requested band was not provisioned, so the
+		// effectiveReq carries the fallback flow key when the requested band was not provisioned, so the
 		// item is enqueued under the band that was actually leased.
 		item, err := fc.tryDistribution(reqCtx, effectiveReq, enqueueTime, conn)
 		if err != nil {
@@ -280,19 +280,22 @@ func (fc *FlowController) EnqueueAndWait(
 	return finalOutcome, err
 }
 
-// demotedRequest wraps a FlowControlRequest to override its flow key, so a request demoted to a fallback
+// fallbackRequest wraps a FlowControlRequest to override its flow key, so a request that falls back to a different
 // priority is enqueued under the band that was actually leased rather than its original (unprovisioned) band.
-type demotedRequest struct {
+type fallbackRequest struct {
 	flowcontrol.FlowControlRequest
 	key flowcontrol.FlowKey
 }
 
-func (d demotedRequest) FlowKey() flowcontrol.FlowKey { return d.key }
+func (r fallbackRequest) FlowKey() flowcontrol.FlowKey { return r.key }
 
-// withConnectionWithDemotion acquires a flow connection, demoting to priority 0 when the requested band is not yet
-// provisioned. On demotion, the callback receives a request whose FlowKey reports the demoted priority, ensuring the
-// item is leased, distributed, and enqueued consistently under priority 0; otherwise it receives the original request.
-func (fc *FlowController) withConnectionWithDemotion(
+// withConnectionWithFallback acquires a flow connection, falling back to priority 0 when the requested band is not yet
+// provisioned. On fallback, the callback receives a request whose FlowKey reports priority 0, ensuring the item is
+// leased, distributed, and enqueued consistently under priority 0; otherwise it receives the original request.
+//
+// Note: relative to the requested priority this is a demotion for positive priorities but a promotion for negative
+// ones. It is an availability-first fallback for the brief window before the control plane provisions the band.
+func (fc *FlowController) withConnectionWithFallback(
 	req flowcontrol.FlowControlRequest,
 	fn func(conn contracts.ActiveFlowConnection, effectiveReq flowcontrol.FlowControlRequest) error,
 ) error {
@@ -305,15 +308,15 @@ func (fc *FlowController) withConnectionWithDemotion(
 	}
 
 	fc.logger.V(logutil.DEFAULT).Info(
-		"Priority band not provisioned, demoting request to priority 0",
+		"Priority band not provisioned, falling back to priority 0",
 		"originalPriority", key.Priority,
 		"flowID", key.ID,
 	)
-	demotedKey := key
-	demotedKey.Priority = 0
-	demoted := demotedRequest{FlowControlRequest: req, key: demotedKey}
-	return fc.registry.WithConnection(demotedKey, func(conn contracts.ActiveFlowConnection) error {
-		return fn(conn, demoted)
+	fallbackKey := key
+	fallbackKey.Priority = 0
+	fallback := fallbackRequest{FlowControlRequest: req, key: fallbackKey}
+	return fc.registry.WithConnection(fallbackKey, func(conn contracts.ActiveFlowConnection) error {
+		return fn(conn, fallback)
 	})
 }
 

pkg/epp/flowcontrol/controller/controller_test.go

@@ -1071,11 +1071,11 @@ func TestFlowController_Concurrency_Backpressure(t *testing.T) {
 		"all concurrent requests should be dispatched successfully even under high contention and zero buffer capacity")
 }
 
-// TestFlowController_EnqueueAndWait_DemotionRewritesItemPriority verifies that when the requested priority band is not
+// TestFlowController_EnqueueAndWait_FallbackRewritesItemPriority verifies that when the requested priority band is not
 // provisioned, the request is not only leased at the fallback priority (0) but also enqueued there: the FlowItem handed
-// to the processor must report the demoted flow key, not the original (unprovisioned) priority. Without this, the
-// processor looks up a managed queue at the missing band and rejects the demoted request.
-func TestFlowController_EnqueueAndWait_DemotionRewritesItemPriority(t *testing.T) {
+// to the processor must report the fallback flow key, not the original (unprovisioned) priority. Without this, the
+// processor looks up a managed queue at the missing band and rejects the request.
+func TestFlowController_EnqueueAndWait_FallbackRewritesItemPriority(t *testing.T) {
 	t.Parallel()
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -1094,7 +1094,7 @@ func TestFlowController_EnqueueAndWait_DemotionRewritesItemPriority(t *testing.T
 
 	registry := &mockRegistryClient{FlowRegistryDataPlane: &mocks.MockRegistryDataPlane{}}
 	registry.WithConnectionFunc = func(key flowcontrol.FlowKey, fn func(conn contracts.ActiveFlowConnection) error) error {
-		// Only priority 0 is provisioned; any other band is rejected, forcing demotion.
+		// Only priority 0 is provisioned; any other band is rejected, forcing the fallback.
 		if key.Priority != 0 {
 			return fmt.Errorf("band %d: %w", key.Priority, contracts.ErrPriorityBandNotFound)
 		}
@@ -1105,9 +1105,9 @@ func TestFlowController_EnqueueAndWait_DemotionRewritesItemPriority(t *testing.T
 
 	outcome, err := h.fc.EnqueueAndWait(ctx, newTestRequest(flowcontrol.FlowKey{ID: "batch", Priority: originalPriority}))
 
-	require.NoError(t, err, "demoted request should be dispatched, not rejected")
+	require.NoError(t, err, "fallback request should be dispatched, not rejected")
 	assert.Equal(t, types.QueueOutcomeDispatched, outcome)
 	assert.Equal(t, 0, capturedKey.Priority,
-		"demoted request must be enqueued at priority 0, not its original unprovisioned priority")
-	assert.Equal(t, "batch", capturedKey.ID, "demotion must preserve the flow ID")
+		"fallback request must be enqueued at priority 0, not its original unprovisioned priority")
+	assert.Equal(t, "batch", capturedKey.ID, "fallback must preserve the flow ID")
 }

pkg/epp/flowcontrol/registry/registry.go

@@ -202,9 +202,9 @@ func (fr *FlowRegistry) SubmitDesiredPriorities(desired map[int]struct{}) {
 	if desired == nil {
 		desired = map[int]struct{}{}
 	}
-	copy := make(map[int]struct{}, len(desired))
+	desiredCopy := make(map[int]struct{}, len(desired))
 	for priority := range desired {
-		copy[priority] = struct{}{}
+		desiredCopy[priority] = struct{}{}
 	}
 
 	// Drain any stale pending update so only the latest state is queued.
@@ -214,7 +214,7 @@ func (fr *FlowRegistry) SubmitDesiredPriorities(desired map[int]struct{}) {
 	}
 
 	// After draining, the channel always has capacity; this send never blocks.
-	fr.priorityBandUpdateCh <- copy
+	fr.priorityBandUpdateCh <- desiredCopy
 }
 
 // PriorityBandUpdateChannel returns the channel carrying desired priority topology updates.
@@ -256,8 +256,10 @@ func (fr *FlowRegistry) ApplyDesiredPriorities(desired map[int]struct{}) {
 		if !fr.isPriorityBandIdle(priority) {
 			continue
 		}
-		// Re-verify under the lock: a concurrent request may have pinned the band
-		// in the window between isPriorityBandIdle and this check.
+		// Cheap best-effort guard against tearing down a band that just became active. Holding fr.mu does
+		// not actually exclude a concurrent pin: pinLeasedResource is lock-free and never takes fr.mu. The
+		// removal is safe regardless, since pinLeasedResource's stale-object protection backs off a racing
+		// pin and the priority-0 fallback plus the next reconcile/GC cycle self-heal.
 		if val, ok := fr.priorityBandStates.Load(priority); ok {
 			if val.(*priorityBandState).isActive() {
 				continue
@@ -342,14 +344,8 @@ func (fr *FlowRegistry) WithConnection(key flowcontrol.FlowKey, fn func(conn con
 //
 // NOTE: The caller (WithConnection) must already hold a lease on the priority band to prevent GC during this operation.
 func (fr *FlowRegistry) ensureFlowInfrastructure(key flowcontrol.FlowKey) error {
-	fr.mu.RLock()
-	_, exists := fr.config.PriorityBands[key.Priority]
-	fr.mu.RUnlock()
-
-	if !exists {
-		return fmt.Errorf("priority band %d not found: %w", key.Priority, contracts.ErrPriorityBandNotFound)
-	}
-
+	// buildFlowComponents validates that the priority band exists (returning ErrPriorityBandNotFound if not)
+	// under the same read lock it uses to read the topology, so a single acquisition covers both.
 	fr.mu.RLock()
 	components, err := fr.buildFlowComponents(key)
 	fr.mu.RUnlock()