fix(flowcontrol): prevent aggregate stats underflow on flow GC

deleteFlow deducted a non-empty queue's stats from the registry
aggregates without emptying the underlying SafeQueue. The cleanup sweep
resolves ManagedQueue handles before processing, without registry locks,
so a handle resolved before the GC could still drain the queue afterward
and propagate the same deduction a second time. The int64 aggregates
went negative and the uint64 casts in Stats() wrapped to near-MaxUint64,
corrupting capacity checks and the flow_control_* gauges for the process
lifetime.

Drain the queue through the managedQueue wrapper instead, so the
deduction is measured across the actual mutation and stale handles
observe an empty queue.

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

Author: LukeAVanDrie

pkg/epp/flowcontrol/registry/registry_helpers.go

@@ -170,17 +170,18 @@ func (fr *FlowRegistry) deleteFlow(key flowcontrol.FlowKey) {
 	fr.logger.V(logging.DEBUG).Info("Deleting queue instance.", "flowKey", key)
 	if val, ok := fr.priorityBands.Load(key.Priority); ok {
 		band := val.(*priorityBand)
-		// Requests in a queue that are asynchronously finalized (e.g., due to client
-		// stream cancellation or context timeout), they are left in the queue for the
-		// GC process to clean them up, including updating the capacity. Here we remove
-		// a flow queue, potentially with such requests waiting for GC, therefor the
-		// capacity stats are updated here before removing the queue.
+		// Requests that are asynchronously finalized (e.g., due to client stream
+		// cancellation or context timeout) are left in the queue for the cleanup sweep.
+		// A queue deleted here may still hold such items, and the sweep may still hold a
+		// ManagedQueue handle to it (handles are resolved before processing, without
+		// registry locks). Draining through the wrapper both empties the queue and
+		// deducts the stats in one critical section, so a later mutation through a stale
+		// handle observes an empty queue and propagates nothing.
 		if mq, ok := band.queues[key.ID]; ok && mq != nil {
-			// Safe-guard: Deduct any unswept capacity before destroying the queue
-			if mqLen := int64(mq.Len()); mqLen > 0 {
-				fr.logger.V(logging.DEBUG).Info("Deregistering non-empty queue during GC, flushing stats",
-					"flowKey", key, "unsweptCount", mqLen)
-				fr.propagateStatsDelta(key.Priority, -mqLen, -int64(mq.ByteSize()))
+			if mq.Len() > 0 {
+				fr.logger.V(logging.DEBUG).Info("Deregistering non-empty queue during GC, draining unswept items",
+					"flowKey", key, "unsweptCount", mq.Len())
+				mq.Drain()
 			}
 		}
 		delete(band.queues, key.ID)

pkg/epp/flowcontrol/registry/registry_helpers_test.go

@@ -381,6 +381,72 @@ func TestRegistry_DeleteFlow(t *testing.T) {
 		"Accessing a deleted flow must return ErrFlowInstanceNotFound")
 }
 
+// TestRegistry_DeleteFlow_StaleHandleStats covers the interleaving where the cleanup sweep
+// resolves a ManagedQueue handle (processAllQueuesConcurrently phase 1), GC deletes the still
+// non-empty flow, and the sweep then mutates through the stale handle (phase 3). Each item must be
+// deducted from the aggregates exactly once: the uint64 casts in Stats() require the aggregates to
+// never go negative.
+func TestRegistry_DeleteFlow_StaleHandleStats(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name    string
+		staleOp func(t *testing.T, mq contracts.ManagedQueue, item flowcontrol.QueueItemAccessor)
+	}{
+		{
+			name: "Drain",
+			staleOp: func(t *testing.T, mq contracts.ManagedQueue, _ flowcontrol.QueueItemAccessor) {
+				assert.Empty(t, mq.Drain(), "Stale handle must observe an empty queue after deleteFlow")
+			},
+		},
+		{
+			name: "Cleanup",
+			staleOp: func(t *testing.T, mq contracts.ManagedQueue, _ flowcontrol.QueueItemAccessor) {
+				items := mq.Cleanup(func(_ flowcontrol.QueueItemAccessor) bool { return true })
+				assert.Empty(t, items, "Stale handle must observe an empty queue after deleteFlow")
+			},
+		},
+		{
+			name: "Remove",
+			staleOp: func(t *testing.T, mq contracts.ManagedQueue, item flowcontrol.QueueItemAccessor) {
+				_, err := mq.Remove(item.Handle())
+				assert.Error(t, err, "Removing an item already accounted for by deleteFlow must fail")
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			h := newTestHarness(t)
+			item := h.addItem(h.highPriorityKey1, 100)
+
+			// The sweep resolves handles before processing, without holding registry locks in between.
+			mq, err := h.registry.ManagedQueue(h.highPriorityKey1)
+			require.NoError(t, err, "Test setup: resolving the ManagedQueue handle must succeed")
+
+			h.registry.mu.Lock()
+			h.registry.deleteFlow(h.highPriorityKey1)
+			h.registry.mu.Unlock()
+
+			stats := h.registry.Stats()
+			assert.Zero(t, stats.TotalLen, "deleteFlow must deduct the unswept items from the total length")
+			assert.Zero(t, stats.TotalByteSize, "deleteFlow must deduct the unswept items from the total byte size")
+
+			tc.staleOp(t, mq, item)
+
+			stats = h.registry.Stats()
+			assert.Zero(t, stats.TotalLen,
+				"A stale-handle mutation after deleteFlow must not deduct the same items again (uint64 underflow)")
+			assert.Zero(t, stats.TotalByteSize,
+				"A stale-handle mutation after deleteFlow must not deduct the same items again (uint64 underflow)")
+			bandStats := stats.PerPriorityBandStats[highPriority]
+			assert.Zero(t, bandStats.Len, "Per-band length must not underflow after a stale-handle mutation")
+			assert.Zero(t, bandStats.ByteSize, "Per-band byte size must not underflow after a stale-handle mutation")
+		})
+	}
+}
+
 func TestRegistry_DynamicProvisioning(t *testing.T) {
 	t.Parallel()