fix(flowcontrol): call up() after down() in maxMinHeap.Remove

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

@@ -285,9 +285,12 @@ func (h *maxMinHeap) Remove(handle flowcontrol.QueueItemHandle) (flowcontrol.Que
 		delete(h.handles, handle)
 		h.byteSize.Add(^item.OriginalRequest().ByteSize() + 1) // Atomic subtraction
 
-		// The swapped item at index i might violate the heap property, so we must restore it
-		// by bubbling the item down the heap.
+		// The swapped item at index i might violate the heap property in either direction:
+		//   - downward: it is less extreme than its descendants -> down() fixes this.
+		//   - upward:   it is more extreme than its ancestors   -> up() fixes this.
+		// Both calls are needed because the replacement can come from a different subtree.
 		h.down(i)
+		h.up(i)
 	} else {
 		// It's the last item, just remove it.
 		h.items = h.items[:n]

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

@@ -17,6 +17,7 @@ limitations under the License.
 package queue
 
 import (
+	"fmt"
 	"math"
 	"testing"
 	"time"
@@ -61,6 +62,77 @@ func TestMaxMinHeap_InternalProperty(t *testing.T) {
 	}
 }
 
+// TestMaxMinHeap_Remove_CrossSubtreeSwap demonstrates that Remove corrupts the heap when the replacement element
+// (swapped from the end of the array) belongs to a different subtree than the removed element.
+// The root cause is that Remove calls down() but not up(), so ancestor-level violations go unfixed.
+func TestMaxMinHeap_Remove_CrossSubtreeSwap(t *testing.T) {
+	t.Parallel()
+
+	// 6-element max-min heap (FCFS: earlier time = higher priority):
+	//
+	//        0          L0 (max) -- highest priority
+	//       / \
+	//      5   50       L1 (min) -- lowest priority in subtrees
+	//     / \   \
+	//    1   4   40     L2 (max) -- leaves
+	//
+	// Left subtree of root:  {5, 1, 4}   (index 1, bounded by min-ancestor time=5)
+	// Right subtree of root: {50, 40}    (index 2, bounded by min-ancestor time=50)
+	times := []int{0, 5, 50, 1, 4, 40}
+
+	q := newMaxMinHeap(enqueueTimePolicy)
+	flowKey := flowcontrol.FlowKey{ID: "f", Priority: 0}
+	base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
+	for i, sec := range times {
+		item := mocks.NewMockQueueItemAccessor(10, fmt.Sprintf("t%d", sec), flowKey)
+		item.EnqueueTimeV = base.Add(time.Duration(sec) * time.Second)
+		hi := &heapItem{item: item, index: i}
+		item.SetHandle(hi)
+		q.items = append(q.items, item)
+		q.handles[item.Handle()] = hi
+	}
+	q.byteSize.Store(60)
+
+	// Remove index 3 (time=1, left subtree leaf). Last element (index 5, time=40, right subtree) swaps in.
+	// down(3) has no children -> no-op. Heap becomes [0, 5, 50, 40, 4].
+	// Node 1 (min, time=5) must be lowest priority in its subtree {5, 40, 4}.
+	// But time=40 has lower priority than time=5 -> VIOLATION.
+	_, err := q.Remove(q.items[3].Handle())
+	require.NoError(t, err)
+
+	// Verify every node is the max (or min) of its entire subtree -- not just its direct children.
+	for i := range q.items {
+		isMin := isMinLevel(i)
+		for d := range q.items {
+			if !isAncestor(i, d) {
+				continue
+			}
+			if isMin {
+				require.Falsef(t, q.policy.Less(q.items[i], q.items[d]),
+					"min-level node %d (t=%v) has higher priority than descendant %d (t=%v)",
+					i, q.items[i].EnqueueTime().Sub(base)/time.Second,
+					d, q.items[d].EnqueueTime().Sub(base)/time.Second)
+			} else {
+				require.Falsef(t, q.policy.Less(q.items[d], q.items[i]),
+					"max-level node %d (t=%v) has lower priority than descendant %d (t=%v)",
+					i, q.items[i].EnqueueTime().Sub(base)/time.Second,
+					d, q.items[d].EnqueueTime().Sub(base)/time.Second)
+			}
+		}
+	}
+}
+
+// isAncestor reports whether a is a strict ancestor of d in a binary heap.
+func isAncestor(a, d int) bool {
+	if d <= a {
+		return false
+	}
+	for d > a {
+		d = (d - 1) / 2
+	}
+	return d == a
+}
+
 // assertHeapProperty checks if the slice of items satisfies the max-min heap property.
 func assertHeapProperty(t *testing.T, h *maxMinHeap, msgAndArgs ...any) {
 	t.Helper()