perf(flowcontrol): index active queues so dispatch scans O(active) flows

pkg/epp/flowcontrol/registry/managedqueue.go

@@ -63,6 +63,13 @@ type managedQueue struct {
 	// onStatsDelta is the callback used to propagate statistics changes up to the registry.
 	onStatsDelta propagateStatsDeltaFunc
 
+	// onActiveTransition is invoked when the queue transitions between empty and non-empty, so the
+	// owning priority band can maintain its index of active (non-empty) queues. Transitions are
+	// detected under `mu`, which serializes them per queue. The callback must be lock-free or take
+	// only leaf locks (never the registry mutex): it runs inside this queue's critical section.
+	// May be nil (e.g., in isolated unit tests).
+	onActiveTransition func(mq *managedQueue, active bool)
+
 	// --- State Protected by `mu` ---
 
 	// mu protects all mutating operations. It ensures that any changes to the underlying `queue` and the updates to the
@@ -92,14 +99,16 @@ func newManagedQueue(
 	key flowcontrol.FlowKey,
 	logger logr.Logger,
 	onStatsDelta propagateStatsDeltaFunc,
+	onActiveTransition func(mq *managedQueue, active bool),
 ) *managedQueue {
 	mqLogger := logger.WithName("managed-queue").WithValues("flowKey", key)
 	mq := &managedQueue{
-		queue:        queue,
-		policy:       policy,
-		key:          key,
-		onStatsDelta: onStatsDelta,
-		logger:       mqLogger,
+		queue:              queue,
+		policy:             policy,
+		key:                key,
+		onStatsDelta:       onStatsDelta,
+		onActiveTransition: onActiveTransition,
+		logger:             mqLogger,
 	}
 	mq.flowQueueAccessor = &flowQueueAccessor{mq: mq}
 	return mq
@@ -187,6 +196,17 @@ func (mq *managedQueue) propagateStatsDeltaLocked(lenDelta, byteSizeDelta int64)
 	}
 	mq.byteSize.Add(byteSizeDelta)
 
+	// Maintain the band's active-queue index on empty<->non-empty transitions. `mu` is held, so
+	// transitions are strictly alternating per queue; `newLen == lenDelta` implies the previous
+	// length was zero.
+	if mq.onActiveTransition != nil && lenDelta != 0 {
+		if lenDelta > 0 && newLen == lenDelta {
+			mq.onActiveTransition(mq, true)
+		} else if newLen == 0 {
+			mq.onActiveTransition(mq, false)
+		}
+	}
+
 	// Propagate the delta up to the registry. This propagation is lock-free and eventually consistent.
 	mq.onStatsDelta(mq.key.Priority, lenDelta, byteSizeDelta)
 }

pkg/epp/flowcontrol/registry/managedqueue_test.go

@@ -70,7 +70,7 @@ func newMqHarness(t *testing.T, queue contracts.SafeQueue, key flowcontrol.FlowK
 	propagator := &mockStatsPropagator{}
 	mockPolicy := &fwkfcmocks.MockOrderingPolicy{}
 
-	mq := newManagedQueue(queue, mockPolicy, key, logr.Discard(), propagator.propagate)
+	mq := newManagedQueue(queue, mockPolicy, key, logr.Discard(), propagator.propagate, nil)
 	require.NotNil(t, mq, "Test setup: newManagedQueue must return a valid instance")
 
 	return &mqTestHarness{

pkg/epp/flowcontrol/registry/registry_helpers.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"sync"
 
 	"github.com/llm-d/llm-d-router/pkg/common/observability/logging"
 	"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts"
@@ -50,6 +51,29 @@ type priorityBand struct {
 
 	// priorityBandAccessor is a preallocated flowcontrol.PriorityBandAccessor for this priorityBand
 	priorityBandAccessor *priorityBandAccessor
+
+	// activeQueues indexes the subset of `queues` that currently hold items, keyed by logical ID
+	// (values are *managedQueue). It is maintained by each queue's empty<->non-empty transitions
+	// (serialized per queue under the queue's own mutex) and read lock-free by IterateQueues, which
+	// keeps the dispatch hot path O(active flows) with zero allocation instead of O(registered
+	// flows) with a snapshot. The view is eventually consistent: a queue is always present here by
+	// the time an Add returns, but may linger briefly after draining; readers must tolerate
+	// observing an empty queue.
+	activeQueues sync.Map
+}
+
+// setQueueActivity is the onActiveTransition callback for this band's queues. It runs inside the
+// queue's critical section, so it must remain lock-free (sync.Map only, never the registry mutex).
+func (b *priorityBand) setQueueActivity(mq *managedQueue, active bool) {
+	if active {
+		b.activeQueues.Store(mq.key.ID, mq)
+	} else {
+		// Deactivation must be conditional on the entry still belonging to this queue. A cleanup-sweep
+		// worker can drain a queue through a handle resolved before deleteFlow removed it, and a
+		// successor queue may have been registered under the same ID in the interim; an unconditional
+		// delete would hide that live, non-empty successor from IterateQueues.
+		b.activeQueues.CompareAndDelete(mq.key.ID, mq)
+	}
 }
 
 // initPriorityBand constructs the runtime state for a single priority level and registers it within the registry.
@@ -161,7 +185,7 @@ func (fr *FlowRegistry) synchronizeFlow(
 
 	fr.logger.V(logging.TRACE).Info("Creating new queue for flow instance.", "flowKey", key)
 
-	mq := newManagedQueue(q, policy, key, fr.logger, fr.propagateStatsDelta)
+	mq := newManagedQueue(q, policy, key, fr.logger, fr.propagateStatsDelta, band.setQueueActivity)
 	band.queues[key.ID] = mq
 }
 
@@ -185,6 +209,7 @@ func (fr *FlowRegistry) deleteFlow(key flowcontrol.FlowKey) {
 			}
 		}
 		delete(band.queues, key.ID)
+		band.activeQueues.Delete(key.ID)
 	}
 }
 
@@ -244,22 +269,16 @@ func (a *priorityBandAccessor) Queue(id string) flowcontrol.FlowQueueAccessor {
 	return mq.FlowQueueAccessor()
 }
 
-// IterateQueues executes the given `callback` for each FlowQueueAccessor in this priority band.
+// IterateQueues executes the given `callback` for each active (non-empty) FlowQueueAccessor in
+// this priority band.
 //
-// To minimize lock contention, this implementation snapshots the queue accessors under a read lock and then executes
-// the callback on the snapshot, outside of the lock. This ensures that a potentially slow policy (the callback) does
-// not block other operations on the registry.
+// It ranges over the band's lock-free active-queue index, so it takes no registry lock and
+// performs no allocation, and its cost scales with the number of flows that currently hold items
+// rather than the number of registered flows. The view is eventually consistent: a queue drained
+// concurrently with iteration may still be visited, so callbacks must tolerate Len() == 0; a
+// queue is guaranteed to be visible once the Add that made it non-empty has returned.
 func (a *priorityBandAccessor) IterateQueues(callback func(queue flowcontrol.FlowQueueAccessor) bool) {
-	a.registry.mu.RLock()
-	accessors := make([]flowcontrol.FlowQueueAccessor, 0, len(a.band.queues))
-	for _, mq := range a.band.queues {
-		accessors = append(accessors, mq.FlowQueueAccessor())
-	}
-	a.registry.mu.RUnlock()
-
-	for _, accessor := range accessors {
-		if !callback(accessor) {
-			return
-		}
-	}
+	a.band.activeQueues.Range(func(_, v any) bool {
+		return callback(v.(*managedQueue).FlowQueueAccessor())
+	})
 }

pkg/epp/flowcontrol/registry/registry_helpers_test.go

@@ -264,20 +264,71 @@ func TestRegistry_PriorityBandAccessor(t *testing.T) {
 		t.Run("IterateQueues", func(t *testing.T) {
 			t.Parallel()
 
-			t.Run("ShouldVisitAllQueuesInBand", func(t *testing.T) {
+			t.Run("ShouldVisitAllActiveQueuesInBand", func(t *testing.T) {
 				t.Parallel()
+				h := newTestHarness(t) // Isolated harness: this subtest mutates queue contents.
+				accessor, err := h.registry.PriorityBandAccessor(highPriority)
+				require.NoError(t, err)
+				h.addItem(h.highPriorityKey1, 100)
+				h.addItem(h.highPriorityKey2, 100)
+
 				var iteratedKeys []flowcontrol.FlowKey
 				accessor.IterateQueues(func(queue flowcontrol.FlowQueueAccessor) bool {
 					iteratedKeys = append(iteratedKeys, queue.FlowKey())
 					return true
 				})
 				expectedKeys := []flowcontrol.FlowKey{h.highPriorityKey1, h.highPriorityKey2}
 				assert.ElementsMatch(t, expectedKeys, iteratedKeys,
-					"IterateQueues must visit every registered flow in the band exactly once")
+					"IterateQueues must visit every active (non-empty) flow in the band exactly once")
+			})
+
+			t.Run("ShouldSkipEmptyQueues", func(t *testing.T) {
+				t.Parallel()
+				h := newTestHarness(t) // Isolated harness: this subtest mutates queue contents.
+				accessor, err := h.registry.PriorityBandAccessor(highPriority)
+				require.NoError(t, err)
+				h.addItem(h.highPriorityKey1, 100)
+
+				var iteratedKeys []flowcontrol.FlowKey
+				accessor.IterateQueues(func(queue flowcontrol.FlowQueueAccessor) bool {
+					iteratedKeys = append(iteratedKeys, queue.FlowKey())
+					return true
+				})
+				assert.Equal(t, []flowcontrol.FlowKey{h.highPriorityKey1}, iteratedKeys,
+					"IterateQueues must visit only flows whose queues hold items; registered-but-empty flows are skipped")
+			})
+
+			t.Run("ShouldTrackEmptinessTransitions", func(t *testing.T) {
+				t.Parallel()
+				h := newTestHarness(t) // Isolated harness: this subtest mutates queue contents.
+				accessor, err := h.registry.PriorityBandAccessor(highPriority)
+				require.NoError(t, err)
+
+				countVisited := func() int {
+					var n int
+					accessor.IterateQueues(func(queue flowcontrol.FlowQueueAccessor) bool {
+						n++
+						return true
+					})
+					return n
+				}
+
+				item := h.addItem(h.highPriorityKey1, 100)
+				assert.Equal(t, 1, countVisited(), "A flow must become visible once its queue holds an item")
+				h.removeItem(h.highPriorityKey1, item)
+				assert.Equal(t, 0, countVisited(), "A flow must stop being visited once its queue drains")
+				h.addItem(h.highPriorityKey1, 100)
+				assert.Equal(t, 1, countVisited(), "A drained flow must become visible again on re-add")
 			})
 
 			t.Run("ShouldExitEarly_WhenCallbackReturnsFalse", func(t *testing.T) {
 				t.Parallel()
+				h := newTestHarness(t) // Isolated harness: this subtest mutates queue contents.
+				accessor, err := h.registry.PriorityBandAccessor(highPriority)
+				require.NoError(t, err)
+				h.addItem(h.highPriorityKey1, 100)
+				h.addItem(h.highPriorityKey2, 100)
+
 				var iterationCount int
 				accessor.IterateQueues(func(queue flowcontrol.FlowQueueAccessor) bool {
 					iterationCount++
@@ -307,12 +358,16 @@ func TestRegistry_PriorityBandAccessor(t *testing.T) {
 					}
 				}()
 
-				// Goroutine B: The Modifier (constantly writing)
+				// Goroutine B: The Modifier (constantly writing). Item add/remove cycles exercise the
+				// active-queue index transitions racing against iteration; flow create/delete cycles
+				// exercise registry topology changes.
 				go func() {
 					defer wg.Done()
 					for i := range 100 {
 						key := flowcontrol.FlowKey{ID: fmt.Sprintf("new-flow-%d", i), Priority: highPriority}
 						h.synchronizeFlow(key)
+						item := h.addItem(key, 100)
+						h.removeItem(key, item)
 						h.registry.mu.Lock()
 						h.registry.deleteFlow(key)
 						h.registry.mu.Unlock()
@@ -348,6 +403,45 @@ func TestRegistry_PriorityBandAccessor(t *testing.T) {
 	})
 }
 
+// TestRegistry_IterateQueues_StaleDrainDoesNotHideReincarnatedQueue exercises the interleaving
+// where a cleanup-sweep worker drains a queue through a handle resolved before deleteFlow removed
+// that queue, after a successor queue was registered under the same flow ID and became active. The
+// stale drain's empty transition must not remove the successor's active-queue index entry: a
+// non-empty registered queue that IterateQueues skips is invisible to both dispatch and future
+// sweeps, so its requests would hang until flow GC.
+func TestRegistry_IterateQueues_StaleDrainDoesNotHideReincarnatedQueue(t *testing.T) {
+	t.Parallel()
+	h := newTestHarness(t)
+	key := h.highPriorityKey1
+
+	// A sweep worker resolves a handle to a queue holding finalized-but-unswept items.
+	h.addItem(key, 100)
+	staleMQ, err := h.registry.ManagedQueue(key)
+	require.NoError(t, err, "Setup: resolving the pre-deletion queue handle must succeed")
+
+	// GC collects the idle flow (deleteFlow tolerates non-empty queues by design).
+	h.registry.mu.Lock()
+	h.registry.deleteFlow(key)
+	h.registry.mu.Unlock()
+
+	// The flow is re-registered under the same ID and receives a new request.
+	h.synchronizeFlow(key)
+	h.addItem(key, 100)
+
+	// The stale sweep drain must not delete the reincarnated queue's index entry.
+	staleMQ.Cleanup(func(flowcontrol.QueueItemAccessor) bool { return true })
+
+	accessor, err := h.registry.PriorityBandAccessor(highPriority)
+	require.NoError(t, err, "Setup: getting the band accessor must succeed")
+	var visited []string
+	accessor.IterateQueues(func(q flowcontrol.FlowQueueAccessor) bool {
+		visited = append(visited, q.FlowKey().ID)
+		return true
+	})
+	assert.Contains(t, visited, key.ID,
+		"a non-empty registered queue must remain visible to IterateQueues after a stale drain of its predecessor")
+}
+
 // --- Lifecycle and State Management Tests ---
 
 func TestRegistry_SynchronizeFlow(t *testing.T) {

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

@@ -63,7 +63,9 @@ type PriorityBandAccessor interface {
 	// Returns nil if the ID is not found in this group.
 	Queue(id string) FlowQueueAccessor
 
-	// IterateQueues executes the given callback for each active Flow in this group.
+	// IterateQueues executes the given callback for each active Flow in this group. A Flow is active while its queue
+	// holds at least one item; implementations may skip empty queues entirely. The view may be eventually consistent
+	// under concurrent mutation, so callbacks must tolerate a visited queue reporting Len() == 0.
 	// Iteration stops if the callback returns false. The order of iteration is not guaranteed unless specified by
 	// the implementation.
 	IterateQueues(callback func(flow FlowQueueAccessor) (keepIterating bool))

pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/README.md

@@ -18,7 +18,7 @@ Choose this policy when:
 * **Identifies programs** via the fairness ID header carried on each request.
 * **Tracks per-program metrics** through the request lifecycle: queue wait time, dispatched count, in-flight count, last completion time, and (LAS) attained service in weighted tokens.
 * **Selects a queue to dispatch from** using the configured strategy. Currently `las` (Least Attained Service) is supported; programs with the lowest accumulated service score highest.
-* **Decays attained service** for inactive programs so a long-idle program is not penalized indefinitely. Both wall-clock half-life and per-Pick factor decay are supported.
+* **Decays attained service** in wall-clock time so a long-idle program is not penalized indefinitely. Decay is applied lazily whenever a program's service is read or accumulated, so it does not depend on the program being visited by the dispatch loop. Two parameterizations are supported: an explicit half-life, or a per-second decay factor.
 * **Evicts idle program state** on a periodic sweep so per-program memory and Prometheus label series do not accumulate forever.
 
 ## Unit of Fairness
@@ -54,8 +54,8 @@ flowControl:
 | `strategy` | `las` | Scoring strategy. Only `las` is supported. |
 | `lasWeightService` | `0.8` | Weight on the inverted attained-service signal. Higher values prioritize underserved programs more aggressively. |
 | `lasWeightHeadWait` | `0.2` | Weight on the head-of-queue age. Acts as a tiebreaker on cold start when programs have equal attained service. |
-| `lasDecayFactor` | `0.99997` | Per-Pick decay factor applied to inactive programs when `lasHalfLifeSeconds` is `0`. Must be in `(0, 1]`. Coupled to Pick rate. |
-| `lasHalfLifeSeconds` | `0` | Wall-clock half-life of attained service for inactive programs. When `> 0` it overrides `lasDecayFactor`. |
+| `lasDecayFactor` | `0.99997` | Per-second decay factor applied to attained service when `lasHalfLifeSeconds` is `0`. Must be in `(0, 1]`. |
+| `lasHalfLifeSeconds` | `60` | Wall-clock half-life of attained service. When `> 0` it overrides `lasDecayFactor`. |
 | `evictionTtlSeconds` | `3600` | A program with no completion in this window is evicted from the metrics map. |
 | `evictionSweepSeconds` | `300` | How often the eviction sweep runs. Must be `> 0`. |
 
@@ -75,7 +75,7 @@ The plugin exports two shared collectors and one strategy-owned collector under
 
 * **Abandoned requests block eviction**: Requests abandoned after dispatch leave `inFlight` non-zero, and the eviction sweep skips any program with non-zero `inFlight`, so its `ProgramMetrics` entry and Prometheus series persist indefinitely.
 * **Memory and label-series growth**: A new program ID adds a `ProgramMetrics` entry plus per-program Prometheus label series. The eviction sweep bounds growth, but a workload with rapidly churning program IDs (e.g. a fresh ID per request) will see TTL-bounded accumulation. Choose a TTL that matches your churn rate.
-* **Decay tuning depends on workload**: `lasDecayFactor` is per-Pick, so its effective half-life depends on the cluster's pick rate. Use `lasHalfLifeSeconds` for predictable wall-clock decay.
+* **Decay accrues during generation**: Decay is continuous wall-clock time, including while a program's requests are in flight; the service consumed by an in-flight request is added back, post-decay, when it completes. With half-lives well above typical generation times this effect is negligible.
 
 ## Related Documentation
 

pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin.go

@@ -42,7 +42,7 @@ func DefaultConfig() Config {
 		LASWeightService:     0.8,
 		LASWeightHeadWait:    0.2,
 		LASDecayFactor:       0.99997,
-		LASHalfLifeSeconds:   0,
+		LASHalfLifeSeconds:   60,
 	}
 }
 
@@ -201,6 +201,9 @@ func (p *ProgramAwarePlugin) Pick(_ context.Context, band flowcontrol.PriorityBa
 		return nil, nil //nolint:nilnil
 	}
 
+	// IterateQueues visits only active (non-empty) queues. That is sufficient: attained-service
+	// decay is time-anchored inside the strategy, so an idle program's service ages out without its
+	// queue being visited.
 	infos := make(map[string]QueueInfo)
 	band.IterateQueues(func(queue flowcontrol.FlowQueueAccessor) bool {
 		if queue == nil {