test(flowcontrol): add failing test for maxMinHeap.Remove heap corruption

Remove swaps position i with the last element then calls down() but
not up(). When the last element comes from a different subtree, the
ancestor relationship can break:

  Before:              After Remove(index=3, time=1):

        0                    0
       / \                  / \
      5   50               5   50
     / \    \             / \
    1   4   40          40   4

  Last element (index 5, time=40) swaps into index 3.
  down(3): no children, no-op.
  Node 1 (min-level, time=5) must be lowest-priority in {5, 40, 4}.
  time=40 has lower priority than time=5 -- heap property violated.

Signed-off-by: RishabhSaini <rishabhsaini01@gmail.com>

Author: RishabhSaini

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,75 @@ 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=%d) has higher priority than descendant %d (t=%d)",
+					i, times[i], 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 has lower priority than descendant %d", i, d)
+			}
+		}
+	}
+}
+
+// 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()