refactor(flowcontrol): single-ended SafeQueue backed by container/heap (#1690)

* refactor: single-ended queue backed by heap

The dispatch SafeQueue was double-ended, but nothing consumed PeekTail
other than the flowQueueAccessor passthrough itself: dispatch needs only
the head, and eviction uses an independent ordering. With PeekTail gone
the queue is single-ended, so the bespoke max-min heap that gave O(1)
access to both ends is no longer justified.

- Remove the unused PeekTail from the SafeQueue contract and all
  implementers, mocks, and tests; rename PeekHead -> Peek now that there
  is a single end.
- Replace the hand-rolled max-min heap (maxminheap.go) with a standard
  library container/heap priority queue (priorityqueue.go), eliminating
  the custom bubble-up/down logic and the latent bug class it carried.
- Drop the side handle map: membership is validated by index identity
  (a *heapItem lives in exactly one queue), preserving the contract's
  Remove error semantics with less state. O(log n) targeted removal is
  retained via per-item index tracking.
- Rename the registered queue MaxMinHeap -> PriorityQueue.

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

* refactor(flowcontrol): apply review feedback on the priority queue

Addresses review comments from @shmuelk on #1690:

- Correct the new-file license headers to "2026 The llm-d Authors".
- Clarify the PriorityQueue doc: ordering is maintained by an internal
  container/heap (the queue does not itself implement heap.Interface).
- Narrow the Add and Remove critical sections to the shared heap
  mutation. Precondition checks (nil, type assertion) and the atomic
  byteSize update move outside the lock; the isInvalidated and index
  reads stay inside it, since those fields are written under the lock by
  Remove/Cleanup/Drain and reading them lock-free would race.
- Compact Cleanup survivors in place instead of allocating a second
  slice, clearing the vacated tail so removed items are not retained.

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

---------

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

Author: LukeAVanDrie

pkg/epp/flowcontrol/contracts/mocks/mocks.go

@@ -163,8 +163,7 @@ type MockSafeQueue struct {
 	CapabilitiesV []flowcontrol.QueueCapability
 	LenV          int
 	ByteSizeV     uint64
-	PeekHeadV     flowcontrol.QueueItemAccessor
-	PeekTailV     flowcontrol.QueueItemAccessor
+	PeekV         flowcontrol.QueueItemAccessor
 	AddFunc       func(item flowcontrol.QueueItemAccessor)
 	RemoveFunc    func(handle flowcontrol.QueueItemHandle) (flowcontrol.QueueItemAccessor, error)
 	CleanupFunc   func(predicate contracts.PredicateFunc) []flowcontrol.QueueItemAccessor
@@ -176,12 +175,8 @@ func (m *MockSafeQueue) Capabilities() []flowcontrol.QueueCapability { return m.
 func (m *MockSafeQueue) Len() int                                    { return m.LenV }
 func (m *MockSafeQueue) ByteSize() uint64                            { return m.ByteSizeV }
 
-func (m *MockSafeQueue) PeekHead() flowcontrol.QueueItemAccessor {
-	return m.PeekHeadV
-}
-
-func (m *MockSafeQueue) PeekTail() flowcontrol.QueueItemAccessor {
-	return m.PeekTailV
+func (m *MockSafeQueue) Peek() flowcontrol.QueueItemAccessor {
+	return m.PeekV
 }
 
 func (m *MockSafeQueue) Add(item flowcontrol.QueueItemAccessor) {
@@ -366,8 +361,8 @@ func (m *MockManagedQueue) ByteSize() uint64 {
 	return size
 }
 
-// PeekHead returns the first item found in the mock queue. Note: map iteration order is not guaranteed.
-func (m *MockManagedQueue) PeekHead() flowcontrol.QueueItemAccessor {
+// Peek returns the first item found in the mock queue. Note: map iteration order is not guaranteed.
+func (m *MockManagedQueue) Peek() flowcontrol.QueueItemAccessor {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.init()
@@ -376,8 +371,3 @@ func (m *MockManagedQueue) PeekHead() flowcontrol.QueueItemAccessor {
 	}
 	return nil // Queue is empty
 }
-
-// PeekTail is not implemented for this mock.
-func (m *MockManagedQueue) PeekTail() flowcontrol.QueueItemAccessor {
-	return nil
-}

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

@@ -413,7 +413,7 @@ func (sp *Processor) selectItem(
 	}
 	// The queue itself is responsible for explicit ordering via its configured OrderingPolicy.
 	// We simply peek at the head.
-	return queue.PeekHead(), nil
+	return queue.Peek(), nil
 }
 
 // dispatchItem handles the final steps of dispatching an item: removing it from the queue and finalizing its outcome.

pkg/epp/flowcontrol/framework/plugins/queue/benchmark_test.go

@@ -46,10 +46,6 @@ func BenchmarkQueues(b *testing.B) {
 				benchmarkAddPeekRemove(b, q)
 			})
 
-			b.Run("AddPeekTailRemove", func(b *testing.B) {
-				benchmarkAddPeekTailRemove(b, q)
-			})
-
 			b.Run("BulkAddThenBulkRemove", func(b *testing.B) {
 				benchmarkBulkAddThenBulkRemove(b, q)
 			})
@@ -79,10 +75,10 @@ func benchmarkAddRemove(b *testing.B, q contracts.SafeQueue) {
 	})
 }
 
-// benchmarkAddPeekRemove measures the throughput of a serial Add, PeekHead, and Remove sequence. This simulates a
+// benchmarkAddPeekRemove measures the throughput of a serial Add, Peek, and Remove sequence. This simulates a
 // common consumer pattern where a single worker peeks at an item before deciding to process and remove it.
 func benchmarkAddPeekRemove(b *testing.B, q contracts.SafeQueue) {
-	// Pre-add one item so PeekHead doesn't fail on the first iteration.
+	// Pre-add one item so Peek doesn't fail on the first iteration.
 	initialItem := mocks.NewMockQueueItemAccessor(1, "initial", benchmarkFlowKey)
 	q.Add(initialItem)
 
@@ -91,11 +87,11 @@ func benchmarkAddPeekRemove(b *testing.B, q contracts.SafeQueue) {
 	for b.Loop() {
 		item := mocks.NewMockQueueItemAccessor(1, "item", benchmarkFlowKey)
 		q.Add(item)
-		peeked := q.PeekHead()
+		peeked := q.Peek()
 		if peeked == nil {
 			// In a concurrent benchmark, this could happen if the queue becomes empty.
 			// In a serial one, it's a fatal error.
-			b.Fatal("PeekHead failed")
+			b.Fatal("Peek failed")
 		}
 
 		_, err := q.Remove(peeked.Handle())
@@ -121,9 +117,9 @@ func benchmarkBulkAddThenBulkRemove(b *testing.B, q contracts.SafeQueue) {
 
 		// Remove the same number of items
 		for range items {
-			peeked := q.PeekHead()
+			peeked := q.Peek()
 			if peeked == nil {
-				b.Fatal("PeekHead failed")
+				b.Fatal("Peek failed")
 			}
 			if _, err := q.Remove(peeked.Handle()); err != nil {
 				b.Fatalf("Remove failed: %v", err)
@@ -132,31 +128,6 @@ func benchmarkBulkAddThenBulkRemove(b *testing.B, q contracts.SafeQueue) {
 	}
 }
 
-// benchmarkAddPeekTailRemove measures the throughput of a serial Add, PeekTail, and Remove sequence. This is useful for
-// understanding the performance of accessing the lowest-priority item.
-func benchmarkAddPeekTailRemove(b *testing.B, q contracts.SafeQueue) {
-	// Pre-add one item so PeekTail doesn't fail on the first iteration.
-	initialItem := mocks.NewMockQueueItemAccessor(1, "initial", benchmarkFlowKey)
-	q.Add(initialItem)
-
-	b.ReportAllocs()
-
-	for b.Loop() {
-		item := mocks.NewMockQueueItemAccessor(1, "item", benchmarkFlowKey)
-		q.Add(item)
-
-		peeked := q.PeekTail()
-		if peeked == nil {
-			b.Fatal("PeekTail failed")
-		}
-
-		_, err := q.Remove(peeked.Handle())
-		if err != nil {
-			b.Fatalf("Remove failed: %v", err)
-		}
-	}
-}
-
 // benchmarkHighContention simulates a more realistic workload with multiple producers and consumers operating on the
 // queue concurrently.
 func benchmarkHighContention(b *testing.B, q contracts.SafeQueue) {
@@ -190,7 +161,7 @@ func benchmarkHighContention(b *testing.B, q contracts.SafeQueue) {
 	// Consumers drive the benchmark.
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			peeked := q.PeekHead()
+			peeked := q.Peek()
 			if peeked != nil {
 				_, _ = q.Remove(peeked.Handle())
 			}

pkg/epp/flowcontrol/framework/plugins/queue/functional_test.go

@@ -58,7 +58,7 @@ var reverseEnqueueTimePolicy = &mocks.MockOrderingPolicy{
 	},
 }
 
-// testLifecycleAndOrdering is a helper function to execute a standard sequence of Add, PeekHead, and Remove operations
+// testLifecycleAndOrdering is a helper function to execute a standard sequence of Add, Peek, and Remove operations
 // on a queue. It verifies the queue's state (length, byte size) and item ordering based on the provided `itemsInOrder`
 // slice, which should be pre-sorted according to the `comparatorName` (which is just a string for
 // logging/identification).
@@ -71,11 +71,9 @@ func testLifecycleAndOrdering(
 ) {
 	t.Helper()
 
-	// PeekHead/PeekTail on empty queue
-	peeked := q.PeekHead()
-	assert.Nil(t, peeked, "[%s] PeekHead on empty queue should return a nil item", comparatorName)
-	peeked = q.PeekTail()
-	assert.Nil(t, peeked, "[%s] PeekTail on empty queue should return a nil item", comparatorName)
+	// Peek on empty queue
+	peeked := q.Peek()
+	assert.Nil(t, peeked, "[%s] Peek on empty queue should return a nil item", comparatorName)
 
 	// Add items
 	currentExpectedLen := 0
@@ -112,28 +110,20 @@ func testLifecycleAndOrdering(
 	expectedLen := initialLen
 	expectedByteSize := expectedTotalByteSize
 	for i, expectedItem := range itemsInOrder {
-		// Verify PeekHead
-		peeked = q.PeekHead()
-		require.NotNil(t, peeked, "[%s] PeekHead should return a non-nil item (iteration %d)", comparatorName, i)
+		// Verify Peek
+		peeked = q.Peek()
+		require.NotNil(t, peeked, "[%s] Peek should return a non-nil item (iteration %d)", comparatorName, i)
 		assert.Equal(t, expectedItem.OriginalRequest().ID(), peeked.OriginalRequest().ID(),
-			"[%s] PeekHead must return the item (ID: %s) at the head of the queue (iteration %d)",
+			"[%s] Peek must return the item (ID: %s) at the head of the queue (iteration %d)",
 			comparatorName, expectedItem.OriginalRequest().ID(), i)
 		peekedHandle := peeked.Handle()
 		require.NotNil(t, peekedHandle, "[%s] Handle from a peeked item must not be nil (iteration %d)", comparatorName, i)
 		require.False(t, peekedHandle.IsInvalidated(),
 			"[%s] Handle from a peeked item must not be invalidated (iteration %d)", comparatorName, i)
-		assert.Equal(t, expectedLen, q.Len(), "[%s] Len() must be unchanged after PeekHead (iteration %d)",
+		assert.Equal(t, expectedLen, q.Len(), "[%s] Len() must be unchanged after Peek (iteration %d)",
 			comparatorName, i)
 		assert.Equal(t, expectedByteSize, q.ByteSize(),
-			"[%s] ByteSize() must be unchanged after PeekHead (iteration %d)", comparatorName, i)
-
-		// Verify PeekTail
-		peekedTail := q.PeekTail()
-		require.NotNil(t, peekedTail, "[%s] PeekTail should return a non-nil item (iteration %d)", comparatorName, i)
-		// The tail is the last item in the *remaining* ordered slice.
-		expectedTailItem := itemsInOrder[len(itemsInOrder)-1]
-		assert.Equal(t, expectedTailItem.OriginalRequest().ID(), peekedTail.OriginalRequest().ID(),
-			"[%s] PeekTail must return the item with the lowest priority (iteration %d)", comparatorName, i)
+			"[%s] ByteSize() must be unchanged after Peek (iteration %d)", comparatorName, i)
 
 		// Remove the head item
 		removed, removeErr := q.Remove(peekedHandle)
@@ -156,8 +146,8 @@ func testLifecycleAndOrdering(
 	assert.Zero(t, q.Len(), "[%s] Queue length should be 0 after all items are removed", comparatorName)
 	assert.Zero(t, q.ByteSize(), "[%s] Queue byte size should be 0 after all items are removed", comparatorName)
 
-	peeked = q.PeekHead()
-	assert.Nil(t, peeked, "[%s] PeekHead on an empty queue should return a nil item again", comparatorName)
+	peeked = q.Peek()
+	assert.Nil(t, peeked, "[%s] Peek on an empty queue should return a nil item again", comparatorName)
 }
 
 // TestQueueConformance is the main conformance test suite for SafeQueue implementations.
@@ -404,7 +394,7 @@ func TestQueueConformance(t *testing.T) {
 				// Verify remaining items are correct
 				var remainingIDs []string
 				for q.Len() > 0 {
-					peeked := q.PeekHead()
+					peeked := q.Peek()
 					item, _ := q.Remove(peeked.Handle())
 					remainingIDs = append(remainingIDs, item.OriginalRequest().ID())
 				}
@@ -517,13 +507,9 @@ func TestQueueConformance(t *testing.T) {
 							case 2: // Inspect
 								_ = q.Len()
 								_ = q.ByteSize()
-								peeked := q.PeekHead()
-								if q.Len() == 0 {
-									assert.Nil(t, peeked, "PeekHead on empty queue expected nil")
-								}
-								peeked = q.PeekTail()
+								peeked := q.Peek()
 								if q.Len() == 0 {
-									assert.Nil(t, peeked, "PeekTail on empty queue expected nil")
+									assert.Nil(t, peeked, "Peek on empty queue expected nil")
 								}
 							case 3: // Cleanup
 								q.Cleanup(func(item flowcontrol.QueueItemAccessor) bool { return false })

pkg/epp/flowcontrol/framework/plugins/queue/listqueue.go

@@ -202,8 +202,8 @@ func (lq *listQueue) ByteSize() uint64 {
 	return lq.byteSize.Load()
 }
 
-// PeekHead returns the item at the front of the queue without removing it.
-func (lq *listQueue) PeekHead() flowcontrol.QueueItemAccessor {
+// Peek returns the item at the front of the queue without removing it.
+func (lq *listQueue) Peek() flowcontrol.QueueItemAccessor {
 	lq.mu.RLock()
 	defer lq.mu.RUnlock()
 
@@ -213,15 +213,3 @@ func (lq *listQueue) PeekHead() flowcontrol.QueueItemAccessor {
 	element := lq.requests.Front()
 	return element.Value.(flowcontrol.QueueItemAccessor)
 }
-
-// PeekTail returns the item at the back of the queue without removing it.
-func (lq *listQueue) PeekTail() flowcontrol.QueueItemAccessor {
-	lq.mu.RLock()
-	defer lq.mu.RUnlock()
-
-	if lq.requests.Len() == 0 {
-		return nil
-	}
-	element := lq.requests.Back()
-	return element.Value.(flowcontrol.QueueItemAccessor)
-}