refactor(flowcontrol): unify on a single priority queue

pkg/epp/config/loader/flowcontrol_test.go

@@ -65,7 +65,6 @@ func newFlowControlTestHandle(t *testing.T) fwkplugin.Handle {
 			Type: edf.EDFOrderingPolicyType,
 			Name: edf.EDFOrderingPolicyType,
 		},
-		RequiredQueueCapabilitiesV: []fwkfc.QueueCapability{fwkfc.CapabilityPriorityConfigurable},
 	})
 	handle.AddPlugin(usagelimits.StaticUsageLimitPolicyType, usagelimits.DefaultPolicy())
 	return handle

pkg/epp/flowcontrol/contracts/errors.go

@@ -28,9 +28,6 @@ var (
 	// ErrPriorityBandNotFound indicates that a requested priority band does not exist in the registry configuration.
 	ErrPriorityBandNotFound = errors.New("priority band not found")
 
-	// ErrPolicyQueueIncompatible indicates that a selected policy is not compatible with the capabilities of the queue.
-	ErrPolicyQueueIncompatible = errors.New("policy is not compatible with queue capabilities")
-
 	// ErrInvalidQueueItemHandle indicates that a QueueItemHandle provided to a SafeQueue operation (e.g.,
 	// SafeQueue.Remove()) is not valid for that queue, has been invalidated, or does not correspond to an actual item in
 	// the queue.

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

@@ -159,21 +159,19 @@ func (m *MockEndpointCandidates) Locate(ctx context.Context, requestMetadata map
 // It is used for tests that need to control the exact return values of a queue's methods without simulating the queue's
 // internal logic or state.
 type MockSafeQueue struct {
-	NameV         string
-	CapabilitiesV []flowcontrol.QueueCapability
-	LenV          int
-	ByteSizeV     uint64
-	PeekV         flowcontrol.QueueItemAccessor
-	AddFunc       func(item flowcontrol.QueueItemAccessor)
-	RemoveFunc    func(handle flowcontrol.QueueItemHandle) (flowcontrol.QueueItemAccessor, error)
-	CleanupFunc   func(predicate contracts.PredicateFunc) []flowcontrol.QueueItemAccessor
-	DrainFunc     func() []flowcontrol.QueueItemAccessor
+	NameV       string
+	LenV        int
+	ByteSizeV   uint64
+	PeekV       flowcontrol.QueueItemAccessor
+	AddFunc     func(item flowcontrol.QueueItemAccessor)
+	RemoveFunc  func(handle flowcontrol.QueueItemHandle) (flowcontrol.QueueItemAccessor, error)
+	CleanupFunc func(predicate contracts.PredicateFunc) []flowcontrol.QueueItemAccessor
+	DrainFunc   func() []flowcontrol.QueueItemAccessor
 }
 
-func (m *MockSafeQueue) Name() string                                { return m.NameV }
-func (m *MockSafeQueue) Capabilities() []flowcontrol.QueueCapability { return m.CapabilitiesV }
-func (m *MockSafeQueue) Len() int                                    { return m.LenV }
-func (m *MockSafeQueue) ByteSize() uint64                            { return m.ByteSizeV }
+func (m *MockSafeQueue) Name() string     { return m.NameV }
+func (m *MockSafeQueue) Len() int         { return m.LenV }
+func (m *MockSafeQueue) ByteSize() uint64 { return m.ByteSizeV }
 
 func (m *MockSafeQueue) Peek() flowcontrol.QueueItemAccessor {
 	return m.PeekV
@@ -331,9 +329,8 @@ func (m *MockManagedQueue) Drain() []flowcontrol.QueueItemAccessor {
 	return drained
 }
 
-func (m *MockManagedQueue) FlowKey() flowcontrol.FlowKey                { return m.FlowKeyV }
-func (m *MockManagedQueue) Name() string                                { return "" }
-func (m *MockManagedQueue) Capabilities() []flowcontrol.QueueCapability { return nil }
+func (m *MockManagedQueue) FlowKey() flowcontrol.FlowKey { return m.FlowKeyV }
+func (m *MockManagedQueue) Name() string                 { return "" }
 func (m *MockManagedQueue) OrderingPolicy() flowcontrol.OrderingPolicy {
 	if m.OrderingPolicyFunc != nil {
 		return m.OrderingPolicyFunc()

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

@@ -50,8 +50,8 @@ func MustRegisterQueue(name RegisteredQueueName, constructor QueueConstructor) {
 	RegisteredQueues[name] = constructor
 }
 
-// NewQueueFromName creates a new SafeQueue given its registered name and the OrderingPolicy that will be optionally
-// used to configure the queue (provided it declares CapabilityPriorityConfigurable).
+// NewQueueFromName creates a new SafeQueue given its registered name and the OrderingPolicy used to
+// configure the queue's ordering.
 // This is called by the FlowRegistry during initialization of a flow's ManagedQueue.
 func NewQueueFromName(name RegisteredQueueName, policy flowcontrol.OrderingPolicy) (contracts.SafeQueue, error) {
 	mu.RLock()

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

@@ -18,7 +18,6 @@ package queue
 
 import (
 	"fmt"
-	"slices"
 	"sort"
 	"sync"
 	"sync/atomic"
@@ -170,8 +169,6 @@ func TestQueueConformance(t *testing.T) {
 				assert.Zero(t, q.Len(), "A new queue should have a length of 0")
 				assert.Zero(t, q.ByteSize(), "A new queue should have a byte size of 0")
 				assert.Equal(t, string(queueName), q.Name(), "Name() should return the registered name of the queue")
-				assert.NotNil(t, q.Capabilities(), "Capabilities() should not return a nil slice")
-				assert.NotEmpty(t, q.Capabilities(), "Capabilities() should return at least one capability")
 			})
 
 			t.Run("LifecycleAndOrdering_DefaultFIFO", func(t *testing.T) {
@@ -192,38 +189,35 @@ func TestQueueConformance(t *testing.T) {
 				testLifecycleAndOrdering(t, q, itemsInFIFOOrder, "DefaultFIFO")
 			})
 
-			qForCapCheck, err := constructor(enqueueTimePolicy)
-			if err == nil && slices.Contains(qForCapCheck.Capabilities(), flowcontrol.CapabilityPriorityConfigurable) {
-				t.Run("LifecycleAndOrdering_PriorityConfigurable_ByteSize", func(t *testing.T) {
-					t.Parallel()
-					q, err := constructor(byteSizePolicy)
-					require.NoError(t, err, "Setup: creating queue with byteSizePolicy should not fail")
-
-					itemLarge := mocks.NewMockQueueItemAccessor(100, "itemLarge_prio", flowKey)
-					itemSmall := mocks.NewMockQueueItemAccessor(20, "itemSmall_prio", flowKey)
-					itemMedium := mocks.NewMockQueueItemAccessor(50, "itemMedium_prio", flowKey)
-
-					itemsInByteSizeOrder := []*mocks.MockQueueItemAccessor{itemSmall, itemMedium, itemLarge}
-					testLifecycleAndOrdering(t, q, itemsInByteSizeOrder, "PriorityByteSize")
-				})
-
-				t.Run("LifecycleAndOrdering_PriorityConfigurable_LIFO", func(t *testing.T) {
-					t.Parallel()
-					q, err := constructor(reverseEnqueueTimePolicy)
-					require.NoError(t, err, "Setup: creating queue with reverseEnqueueTimePolicy should not fail")
-
-					now := time.Now()
-					item1 := mocks.NewMockQueueItemAccessor(100, "item1_lifo", flowKey)
-					item1.EnqueueTimeV = now.Add(-2 * time.Second) // Earliest
-					item2 := mocks.NewMockQueueItemAccessor(50, "item2_lifo", flowKey)
-					item2.EnqueueTimeV = now.Add(-1 * time.Second) // Middle
-					item3 := mocks.NewMockQueueItemAccessor(20, "item3_lifo", flowKey)
-					item3.EnqueueTimeV = now // Latest
-
-					itemsInLIFOOrder := []*mocks.MockQueueItemAccessor{item3, item2, item1}
-					testLifecycleAndOrdering(t, q, itemsInLIFOOrder, "PriorityLIFO")
-				})
-			}
+			t.Run("LifecycleAndOrdering_PriorityConfigurable_ByteSize", func(t *testing.T) {
+				t.Parallel()
+				q, err := constructor(byteSizePolicy)
+				require.NoError(t, err, "Setup: creating queue with byteSizePolicy should not fail")
+
+				itemLarge := mocks.NewMockQueueItemAccessor(100, "itemLarge_prio", flowKey)
+				itemSmall := mocks.NewMockQueueItemAccessor(20, "itemSmall_prio", flowKey)
+				itemMedium := mocks.NewMockQueueItemAccessor(50, "itemMedium_prio", flowKey)
+
+				itemsInByteSizeOrder := []*mocks.MockQueueItemAccessor{itemSmall, itemMedium, itemLarge}
+				testLifecycleAndOrdering(t, q, itemsInByteSizeOrder, "PriorityByteSize")
+			})
+
+			t.Run("LifecycleAndOrdering_PriorityConfigurable_LIFO", func(t *testing.T) {
+				t.Parallel()
+				q, err := constructor(reverseEnqueueTimePolicy)
+				require.NoError(t, err, "Setup: creating queue with reverseEnqueueTimePolicy should not fail")
+
+				now := time.Now()
+				item1 := mocks.NewMockQueueItemAccessor(100, "item1_lifo", flowKey)
+				item1.EnqueueTimeV = now.Add(-2 * time.Second) // Earliest
+				item2 := mocks.NewMockQueueItemAccessor(50, "item2_lifo", flowKey)
+				item2.EnqueueTimeV = now.Add(-1 * time.Second) // Middle
+				item3 := mocks.NewMockQueueItemAccessor(20, "item3_lifo", flowKey)
+				item3.EnqueueTimeV = now // Latest
+
+				itemsInLIFOOrder := []*mocks.MockQueueItemAccessor{item3, item2, item1}
+				testLifecycleAndOrdering(t, q, itemsInLIFOOrder, "PriorityLIFO")
+			})
 
 			t.Run("Remove_InvalidHandle", func(t *testing.T) {
 				t.Parallel()

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

@@ -29,7 +29,7 @@ import (
 //
 // This queue provides a concurrent-safe priority queue whose ordering is maintained by an internal
 // container/heap. Items are ordered by the configured OrderingPolicy, with the highest-priority
-// item (per the policy) at the head. It advertises the CapabilityPriorityConfigurable capability.
+// item (per the policy) at the head.
 //
 // Each item's position in the heap is tracked on its handle, enabling O(log n) targeted removal.
 const PriorityQueueName = "PriorityQueue"
@@ -125,11 +125,6 @@ func (pq *priorityQueue) Name() string {
 	return PriorityQueueName
 }
 
-// Capabilities returns the capabilities of the queue.
-func (pq *priorityQueue) Capabilities() []flowcontrol.QueueCapability {
-	return []flowcontrol.QueueCapability{flowcontrol.CapabilityPriorityConfigurable}
-}
-
 // Len returns the number of items in the queue.
 func (pq *priorityQueue) Len() int {
 	pq.mu.RLock()