epp: add turn-priority strategy to program-aware fairness

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

@@ -17,7 +17,9 @@ 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.
+* **Selects a queue to dispatch from** using the configured strategy. Two are supported:
+  * `las` (Least Attained Service): programs with the lowest accumulated service score highest.
+  * `turn-priority`: scoring engages only under contention. With no flow waiting nothing is dispatched; with exactly one waiting flow it is dispatched directly. When two or more flows are waiting, each scores by its head request's `prio = turn_number + turnPriorityTimeWeight * time_elapsed_in_seconds`, highest first. The elapsed term ages waiting flows up so lower-turn flows are not starved. Turn number is the program's dispatched-request count plus the waiting request itself: each program (fairness ID) has its own counter, advanced once per dispatch. The counter resets when idle program state is evicted: a program idle past the eviction TTL has likely lost its KV cache, so on return it restarts from turn one and re-earns priority from a cold state.
 * **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.
 * **Evicts idle program state** on a periodic sweep so per-program memory and Prometheus label series do not accumulate forever.
 
@@ -51,11 +53,12 @@ flowControl:
 
 | Field | Default | Description |
 |---|---|---|
-| `strategy` | `las` | Scoring strategy. Only `las` is supported. |
+| `strategy` | `las` | Scoring strategy: `las` or `turn-priority`. |
 | `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`. |
+| `turnPriorityTimeWeight` | `0.5` | (`turn-priority`) Weight on the elapsed-seconds term of the head-request priority score. |
 | `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`. |
 

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

@@ -32,6 +32,8 @@ type Config struct {
 	LASWeightHeadWait  float64 `json:"lasWeightHeadWait,omitempty"`
 	LASDecayFactor     float64 `json:"lasDecayFactor,omitempty"`
 	LASHalfLifeSeconds float64 `json:"lasHalfLifeSeconds,omitempty"`
+
+	TurnPriorityTimeWeight float64 `json:"turnPriorityTimeWeight,omitempty"`
 }
 
 func DefaultConfig() Config {
@@ -43,6 +45,8 @@ func DefaultConfig() Config {
 		LASWeightHeadWait:    0.2,
 		LASDecayFactor:       0.99997,
 		LASHalfLifeSeconds:   0,
+
+		TurnPriorityTimeWeight: 0.5,
 	}
 }
 
@@ -65,6 +69,9 @@ func (c Config) validate() error {
 	if c.LASHalfLifeSeconds < 0 {
 		return fmt.Errorf("lasHalfLifeSeconds must be >= 0, got %v", c.LASHalfLifeSeconds)
 	}
+	if c.TurnPriorityTimeWeight < 0 {
+		return fmt.Errorf("turnPriorityTimeWeight must be >= 0, got %v", c.TurnPriorityTimeWeight)
+	}
 	return nil
 }
 

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

@@ -32,19 +32,27 @@ func TestFactory_LASConfig(t *testing.T) {
 	require.NotNil(t, p)
 }
 
+func TestFactory_TurnPriorityConfig(t *testing.T) {
+	cfg := `{"strategy":"turn-priority","turnPriorityTimeWeight":0.5}`
+	p, err := ProgramAwarePluginFactory("test", decoder(cfg), nil)
+	require.NoError(t, err)
+	require.NotNil(t, p)
+}
+
 func TestFactory_UnknownStrategy(t *testing.T) {
 	_, err := ProgramAwarePluginFactory("test", decoder(`{"strategy":"wfq"}`), nil)
 	require.Error(t, err)
 }
 
 func TestFactory_InvalidConfig(t *testing.T) {
 	cases := map[string]string{
-		"negative ttl":       `{"evictionTtlSeconds":-1}`,
-		"zero sweep":         `{"evictionSweepSeconds":0}`,
-		"negative weight":    `{"lasWeightService":-0.1}`,
-		"decay factor > 1":   `{"lasDecayFactor":1.5}`,
-		"decay factor 0":     `{"lasDecayFactor":0}`,
-		"negative half life": `{"lasHalfLifeSeconds":-1}`,
+		"negative ttl":         `{"evictionTtlSeconds":-1}`,
+		"zero sweep":           `{"evictionSweepSeconds":0}`,
+		"negative weight":      `{"lasWeightService":-0.1}`,
+		"decay factor > 1":     `{"lasDecayFactor":1.5}`,
+		"decay factor 0":       `{"lasDecayFactor":0}`,
+		"negative half life":   `{"lasHalfLifeSeconds":-1}`,
+		"negative turn weight": `{"turnPriorityTimeWeight":-0.1}`,
 	}
 	for name, cfg := range cases {
 		t.Run(name, func(t *testing.T) {

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

@@ -36,8 +36,10 @@ func newStrategy(cfg Config) (Strategy, error) {
 			decayFactor:     cfg.LASDecayFactor,
 			halfLifeSeconds: cfg.LASHalfLifeSeconds,
 		}, nil
+	case turnPriorityStrategyName:
+		return &turnPriorityStrategy{timeWeight: cfg.TurnPriorityTimeWeight}, nil
 	default:
-		return nil, fmt.Errorf("unknown scoring strategy %q: only \"las\" is supported", cfg.Strategy)
+		return nil, fmt.Errorf("unknown scoring strategy %q: only \"las\" and \"turn-priority\" are supported", cfg.Strategy)
 	}
 }
 

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

@@ -0,0 +1,94 @@
+package programaware
+
+import (
+	"math"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol"
+	fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol"
+	fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling"
+)
+
+const turnPriorityStrategyName = "turn-priority"
+
+var _ Strategy = &turnPriorityStrategy{}
+
+// turnPriorityStrategy scores each flow by its head request's priority
+//
+//	prio = turnNumber + timeWeight * elapsedSeconds
+//
+// and picks the highest. The elapsed term ages waiting flows up so a flow with
+// a lower turn number is not starved indefinitely. The strategy keeps no
+// accumulated per-program state.
+type turnPriorityStrategy struct {
+	timeWeight float64
+}
+
+func (s *turnPriorityStrategy) Name() string { return turnPriorityStrategyName }
+
+func (s *turnPriorityStrategy) Pick(_ int, queues map[string]QueueInfo) flowcontrol.FlowQueueAccessor {
+	// Collect the flows with pending work. Scoring only orders flows against
+	// each other, so it is meaningful only under contention: with no waiting
+	// flow there is nothing to dispatch, and with exactly one the choice is
+	// forced. Score only when two or more flows are waiting.
+	waiting := make([]QueueInfo, 0, len(queues))
+	for _, qi := range queues {
+		if qi.Len == 0 || qi.Queue.Peek() == nil {
+			continue
+		}
+		waiting = append(waiting, qi)
+	}
+
+	switch len(waiting) {
+	case 0:
+		return nil
+	case 1:
+		return waiting[0].Queue
+	}
+
+	var best flowcontrol.FlowQueueAccessor
+	bestScore := math.Inf(-1)
+	now := time.Now()
+
+	for _, qi := range waiting {
+		head := qi.Queue.Peek()
+		elapsed := now.Sub(head.EnqueueTime()).Seconds()
+		if elapsed < 0 {
+			elapsed = 0
+		}
+		score := float64(turnNumberFor(qi.Metrics)) + s.timeWeight*elapsed
+		if score > bestScore {
+			bestScore = score
+			best = qi.Queue
+		}
+	}
+
+	return best
+}
+
+func (s *turnPriorityStrategy) OnPreRequest(_ *ProgramMetrics, _ *fwksched.InferenceRequest) {}
+
+func (s *turnPriorityStrategy) OnCompleted(_ *ProgramMetrics, _ *fwksched.InferenceRequest, _ *fwkrc.Response) {
+}
+
+func (s *turnPriorityStrategy) EvictProgram(_ string) {}
+
+func (s *turnPriorityStrategy) Collectors() []prometheus.Collector { return nil }
+
+// turnNumberFor returns the turn number of a program's head request: the count
+// of requests already dispatched for that program plus the waiting request
+// itself (dispatchedCount+1).
+//
+// The counter is program-scoped: each program (fairness ID) has its own count,
+// and every dispatch advances it by one. The count resets to zero when idle
+// program state is evicted: a program that goes idle past the eviction TTL has
+// likely lost its KV cache, so on return it restarts from turn one and re-earns
+// priority from a cold state.
+func turnNumberFor(metrics *ProgramMetrics) int64 {
+	if metrics == nil {
+		return 1
+	}
+	return metrics.DispatchedCount() + 1
+}

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

@@ -0,0 +1,95 @@
+package programaware
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTurnPriority_Name(t *testing.T) {
+	assert.Equal(t, "turn-priority", (&turnPriorityStrategy{}).Name())
+}
+
+func TestTurnPriority_Pick_PrefersLongerWait(t *testing.T) {
+	s := &turnPriorityStrategy{timeWeight: 0.5}
+	now := time.Now()
+
+	// Both programs have zero dispatches (equal turn), so the longer-waiting
+	// head wins on the elapsed term.
+	idA, qA := makeInfo("alpha", now.Add(-5*time.Second))
+	idB, qB := makeInfo("beta", now.Add(-1*time.Second))
+	queues := map[string]QueueInfo{idA: qA, idB: qB}
+
+	got := s.Pick(0, queues)
+	require.NotNil(t, got)
+	assert.Equal(t, "alpha", got.FlowKey().ID)
+}
+
+func TestTurnPriority_Pick_PrefersHigherTurn(t *testing.T) {
+	s := &turnPriorityStrategy{timeWeight: 0.5}
+	now := time.Now()
+
+	// Equal wait, but beta has more dispatched requests, so its turn number is
+	// higher and it wins.
+	idA, qA := makeInfo("alpha", now.Add(-1*time.Second))
+	idB, qB := makeInfo("beta", now.Add(-1*time.Second))
+	for i := 0; i < 10; i++ {
+		qB.Metrics.RecordDispatched(time.Time{})
+	}
+	queues := map[string]QueueInfo{idA: qA, idB: qB}
+
+	got := s.Pick(0, queues)
+	require.NotNil(t, got)
+	assert.Equal(t, "beta", got.FlowKey().ID)
+}
+
+func TestTurnPriority_TurnNumberFromDispatchedCount(t *testing.T) {
+	// Waiting request is the next turn: dispatchedCount + 1.
+	assert.Equal(t, int64(1), turnNumberFor(&ProgramMetrics{}))
+	assert.Equal(t, int64(1), turnNumberFor(nil))
+
+	m := &ProgramMetrics{}
+	m.RecordDispatched(time.Time{})
+	m.RecordDispatched(time.Time{})
+	assert.Equal(t, int64(3), turnNumberFor(m))
+}
+
+func TestTurnPriority_Pick_SkipsEmptyQueues(t *testing.T) {
+	s := &turnPriorityStrategy{timeWeight: 0.5}
+	queues := map[string]QueueInfo{
+		"empty": {Queue: makeQueue("empty", 0, time.Time{}), Metrics: &ProgramMetrics{}, Len: 0},
+	}
+	assert.Nil(t, s.Pick(0, queues))
+}
+
+func TestTurnPriority_Pick_SingleWaitingFlowDispatchesDirectly(t *testing.T) {
+	s := &turnPriorityStrategy{timeWeight: 0.5}
+
+	// A just-enqueued head scores ~0; with only one waiting flow it must still
+	// be picked, proving the single-flow path bypasses scoring.
+	id, qi := makeInfo("only", time.Now())
+	queues := map[string]QueueInfo{
+		id:      qi,
+		"empty": {Queue: makeQueue("empty", 0, time.Time{}), Metrics: &ProgramMetrics{}, Len: 0},
+	}
+
+	got := s.Pick(0, queues)
+	require.NotNil(t, got)
+	assert.Equal(t, "only", got.FlowKey().ID)
+}
+
+func TestTurnPriority_Pick_ZeroWeightIgnoresElapsed(t *testing.T) {
+	s := &turnPriorityStrategy{timeWeight: 0}
+	now := time.Now()
+
+	// With timeWeight 0 and equal turn numbers, all scores tie; a flow is still
+	// selected rather than nil.
+	idA, qA := makeInfo("alpha", now.Add(-5*time.Second))
+	idB, qB := makeInfo("beta", now.Add(-1*time.Second))
+	queues := map[string]QueueInfo{idA: qA, idB: qB}
+
+	got := s.Pick(0, queues)
+	require.NotNil(t, got)
+}