feat(flowcontrol): Implement the FlowRegistry

This commit introduces the complete, concrete implementation of the
`FlowRegistry`, the stateful control plane for the flow control system.
It provides a scalable, concurrent-safe, and robust foundation for
managing the lifecycle of all flows, queues, and shards.

The architecture is designed to prioritize data path performance and
strict correctness for control plane state transitions, decoupling the
two via a sharded, asynchronously signaled design.

Key architectural features include:

- **Serialized Control Plane (Actor Model):** All administrative
  operations and internal state change events are processed serially by
  a single background event loop. This fundamental design choice
  eliminates race conditions for complex, multi-step operations like
  shard scaling and garbage collection, guaranteeing correctness.

- **Sharded Data Plane:** The registry's state is partitioned across
  multiple `registryShard` instances. This allows the data path
  (enqueue/dispatch operations) to scale linearly with the number of
  workers and CPU cores by using fine-grained, per-priority-band locks
  to minimize global contention.

- **Asynchronous, Lock-Free Signaling:** A sophisticated, lock-free
  atomic state machine is used for signaling between the data path and
  the control plane (e.g., for queue empty/non-empty transitions). This
  ensures the data path is never blocked by control plane backpressure,
  guarantees strictly ordered signals, and prevents lost transitions
  even under high contention.

- **"Trust but Verify" Garbage Collection:** A periodic scanner
  identifies candidate idle flows using an eventually-consistent cache
  ("Trust"). Before deletion, it performs a "stop-the-world" live check
  (locking a single priority band across all shards) to confirm idleness
  against the ground truth ("Verify"). This provides strong consistency
  precisely when needed while minimizing data path disruption.

- **Immutable Flow Identity (`FlowKey`):** The `FlowKey` (ID + Priority)
  is treated as an immutable identifier. To change the priority of
  traffic, a caller simply registers a new flow. The old flow is
  gracefully and automatically garbage collected once it becomes idle.
  This design avoids complex and error-prone state migration logic.

Author: LukeAVanDrie

GEMINI.md

@@ -0,0 +1,29 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package contracts defines the boundaries and service interfaces for the Flow Control system.
+//
+// Adhering to a "Ports and Adapters" (Hexagonal) architectural style, these interfaces decouple the core
+// `controller.FlowController` engine from its dependencies. They establish the required behaviors and system invariants
+// that concrete implementations must uphold.
+//
+// The primary contracts are:
+//
+//   - `FlowRegistry`: The interface for the stateful control plane that manages the lifecycle of flows, queues, and
+//     policies.
+//
+//   - `SaturationDetector`: The interface for a component that provides real-time load signals.
+package contracts

pkg/epp/flowcontrol/contracts/doc.go

@@ -18,18 +18,23 @@ package contracts
 
 import "errors"
 
-// Registry Errors
 var (
-	// ErrFlowInstanceNotFound indicates that a requested flow instance (a `ManagedQueue`) does not exist in the registry
-	// shard, either because the flow is not registered or the specific instance (e.g., a draining queue at a particular
-	// priority) is not present.
+	// ErrFlowInstanceNotFound indicates that a requested flow instance (a `ManagedQueue`) does not exist.
 	ErrFlowInstanceNotFound = errors.New("flow instance not found")
 
-	// ErrPriorityBandNotFound indicates that a requested priority band does not exist in the registry because it was not
-	// part of the initial configuration.
+	// ErrFlowIDEmpty indicates that a flow specification was provided with an empty flow ID.
+	ErrFlowIDEmpty = errors.New("flow ID cannot be empty")
+
+	// 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 it
-	// is intended to operate on. For example, a policy requiring priority-based peeking is used with a simple FIFO queue.
+	// 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")
+
+	// ErrInvalidShardCount indicates that an invalid shard count was provided (e.g., zero or negative).
+	ErrInvalidShardCount = errors.New("invalid shard count")
+
+	// ErrMaxFlowsExceeded indicates that a new flow could not be registered because the configured maximum number of
+	// concurrent flows has been reached.
+	ErrMaxFlowsExceeded = errors.New("maximum number of concurrent flows exceeded")
 )

pkg/epp/flowcontrol/contracts/errors.go

@@ -14,23 +14,109 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package contracts defines the service interfaces that decouple the core `controller.FlowController` engine from its
-// primary dependencies. In alignment with a "Ports and Adapters" (or "Hexagonal") architectural style, these
-// interfaces represent the "ports" through which the engine communicates.
-//
-// This package contains the primary service contracts for the Flow Registry, which acts as the control plane for all
-// flow state and configuration.
 package contracts
 
 import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/framework"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol/types"
 )
 
-// RegistryShard defines the read-oriented interface that a `controller.FlowController` worker uses to access its
-// specific slice (shard) of the `FlowRegistry's` state. It provides a concurrent-safe view of all flow instances, which
-// are uniquely identified by their composite `types.FlowKey`. It is the primary contract for performing dispatch
-// operations.
+// FlowRegistry is the complete interface for the global control plane. An implementation of this interface is the single
+// source of truth for all flow control state and configuration.
+//
+// # Conformance
+//
+// All methods MUST be goroutine-safe. Implementations are expected to perform complex updates (e.g.,
+// `RegisterOrUpdateFlow`) atomically.
+//
+// # System Invariants
+//
+// Concrete implementations MUST uphold the following invariants:
+//  1. Shard Consistency: All configured priority bands and registered flow instances must exist on every Active shard.
+//     Plugin instance types must be consistent for a given flow across all shards.
+//  2. Flow Instance Uniqueness: Each unique `types.FlowKey` (`ID` + `Priority`) corresponds to exactly one managed flow
+//     instance.
+//  3. Capacity Partitioning: Global and per-band capacity limits must be uniformly partitioned across all Active
+//     shards.
+//
+// # Flow Lifecycle
+//
+// A flow instance (identified by its immutable `FlowKey`) has a simple lifecycle:
+//
+//   - Registered: Known to the `FlowRegistry` via `RegisterOrUpdateFlow`.
+//   - Idle: Queues are empty across all Active and Draining shards.
+//   - Garbage Collected (Unregistered): The registry MUST automatically garbage collect flows after they have remained
+//     Idle for a configurable duration.
+//
+// # Shard Lifecycle
+//
+// When a shard is decommissioned, it is marked inactive (`IsActive() == false`) to prevent new enqueues. The shard
+// continues to drain and is deleted only after it is empty.
+type FlowRegistry interface {
+	FlowRegistryAdmin
+	ShardProvider
+}
+
+// FlowRegistryAdmin defines the administrative interface for the global control plane.
+//
+// # Dynamic Update Strategies
+//
+// The contract specifies behaviors for handling dynamic updates, prioritizing stability and correctness:
+//
+//   - Immutable Flow Identity (`types.FlowKey`): The system treats the `FlowKey` (`ID` + `Priority`) as the immutable
+//     identifier. Changing the priority of traffic requires registering a new `FlowKey`. The old flow instance is
+//     automatically garbage collected when Idle. This design eliminates complex priority migration logic.
+//
+//   - Graceful Draining (Shard Scale-Down): Decommissioned shards enter a Draining state. They stop accepting new
+//     requests but continue to be processed for dispatch until empty.
+//
+//   - Self-Balancing (Shard Scale-Up): When new shards are added, the `controller.FlowController`'s distribution logic
+//     naturally utilizes them, funneling new requests to the less-loaded shards. Existing queued items are not
+//     migrated.
+type FlowRegistryAdmin interface {
+	// RegisterOrUpdateFlow handles the registration of a new flow instance or the update of an existing instance's
+	// specification (for the same `types.FlowKey`). The operation is atomic across all shards.
+	//
+	// Since the `FlowKey` (including `Priority`) is immutable, this method cannot change a flow's priority.
+	// To change priority, the caller should simply register the new `FlowKey`; the old flow instance will be
+	// automatically garbage collected when it becomes Idle.
+	//
+	// Returns errors wrapping:
+	// - `ErrFlowIDEmpty`: if the provided specification has an empty `FlowKey.ID`.
+	// - `ErrPriorityBandNotFound`: if the priority level in the `FlowKey` is not configured in the registry.
+	// - `ErrMaxFlowsExceeded`: if registering this new flow would exceed the configured global maximum.
+	// It may also return internal errors if plugin instantiation fails.
+	RegisterOrUpdateFlow(spec types.FlowSpecification) error
+
+	// UpdateShardCount dynamically adjusts the number of internal state shards.
+	//
+	// The implementation MUST atomically re-partition capacity allocations across all active shards.
+	// Returns an error wrapping `ErrInvalidShardCount` if `n` is not positive.
+	UpdateShardCount(n int) error
+
+	// Stats returns globally aggregated statistics for the entire `FlowRegistry`.
+	Stats() AggregateStats
+
+	// ShardStats returns a slice of statistics, one for each internal shard. This provides visibility for debugging and
+	// monitoring per-shard behavior (e.g., identifying hot or stuck shards).
+	ShardStats() []ShardStats
+}
+
+// ShardProvider defines the interface for discovering available shards.
+//
+// A "shard" is an internal, parallel execution unit that allows the `controller.FlowController`'s core dispatch logic
+// to be parallelized, preventing a CPU bottleneck at high request rates. The `FlowRegistry`'s state is sharded to
+// support this parallelism by reducing lock contention.
+//
+// Consumers MUST check `RegistryShard.IsActive()` before routing new work to a shard to avoid sending requests to a
+// Draining shard.
+type ShardProvider interface {
+	// Shards returns a slice of accessors, one for each internal state shard (Active and Draining).
+	Shards() []RegistryShard
+}
+
+// RegistryShard defines the interface for accessing a specific slice (shard) of the `FlowRegistry's` state.
+// It provides a concurrent-safe view for `controller.FlowController` workers.
 //
 // # Conformance
 //
@@ -46,12 +132,12 @@ type RegistryShard interface {
 	// ManagedQueue retrieves the managed queue for the given, unique `types.FlowKey`. This is the primary method for
 	// accessing a specific flow's queue for either enqueueing or dispatching requests.
 	//
-	// Returns an error wrapping `ErrPriorityBandNotFound` if the priority specified in the key is not configured, or
+	// Returns an error wrapping `ErrPriorityBandNotFound` if the priority specified in the `key` is not configured, or
 	// `ErrFlowInstanceNotFound` if no instance exists for the given `key`.
 	ManagedQueue(key types.FlowKey) (ManagedQueue, error)
 
 	// IntraFlowDispatchPolicy retrieves a flow's configured `framework.IntraFlowDispatchPolicy` for this shard,
-	// identified by its unique `FlowKey`.
+	// identified by its unique `types.FlowKey`.
 	// The registry guarantees that a non-nil default policy (as configured at the priority-band level) is returned if
 	// none is specified for the flow.
 	// Returns an error wrapping `ErrFlowInstanceNotFound` if the flow instance does not exist.
@@ -63,8 +149,8 @@ type RegistryShard interface {
 	InterFlowDispatchPolicy(priority uint) (framework.InterFlowDispatchPolicy, error)
 
 	// PriorityBandAccessor retrieves a read-only accessor for a given priority level, providing a view of the band's
-	// state as seen by this specific shard. This is the primary entry point for inter-flow dispatch policies that
-	// need to inspect and compare multiple flow queues within the same priority band.
+	// state as seen by this specific shard. This is the primary entry point for inter-flow dispatch policies that need to
+	// inspect and compare multiple flow queues within the same priority band.
 	// Returns an error wrapping `ErrPriorityBandNotFound` if the priority level is not configured.
 	PriorityBandAccessor(priority uint) (framework.PriorityBandAccessor, error)
 
@@ -80,24 +166,33 @@ type RegistryShard interface {
 }
 
 // ManagedQueue defines the interface for a flow's queue instance on a specific shard.
-// It wraps an underlying `framework.SafeQueue`, augmenting it with lifecycle validation against the `FlowRegistry` and
-// integrating atomic statistics updates.
+// It acts as a stateful decorator around an underlying `framework.SafeQueue`.
 //
 // # Conformance
 //
-//   - All methods (including those embedded from `framework.SafeQueue`) MUST be goroutine-safe.
-//   - All mutating methods (`Add()`, `Remove()`, `Cleanup()`, `Drain()`) MUST atomically update relevant statistics
-//     (e.g., queue length, byte size).
+//   - All methods MUST be goroutine-safe.
+//   - All mutating methods (`Add()`, `Remove()`, etc.) MUST ensure that the underlying queue state and the statistics
+//     (`Len`, `ByteSize`) are updated atomically relative to each other.
 type ManagedQueue interface {
 	framework.SafeQueue
 
-	// FlowQueueAccessor returns a read-only, flow-aware accessor for this queue.
-	// This accessor is primarily used by policy plugins to inspect the queue's state in a structured way.
-	//
+	// FlowQueueAccessor returns a read-only, flow-aware accessor for this queue, used by policy plugins.
 	// Conformance: This method MUST NOT return nil.
 	FlowQueueAccessor() framework.FlowQueueAccessor
 }
 
+// AggregateStats holds globally aggregated statistics for the entire `FlowRegistry`.
+type AggregateStats struct {
+	// TotalCapacityBytes is the globally configured maximum total byte size limit across all priority bands and shards.
+	TotalCapacityBytes uint64
+	// TotalByteSize is the total byte size of all items currently queued across the entire system.
+	TotalByteSize uint64
+	// TotalLen is the total number of items currently queued across the entire system.
+	TotalLen uint64
+	// PerPriorityBandStats maps each configured priority level to its globally aggregated statistics.
+	PerPriorityBandStats map[uint]PriorityBandStats
+}
+
 // ShardStats holds statistics for a single internal shard within the `FlowRegistry`.
 type ShardStats struct {
 	// TotalCapacityBytes is the optional, maximum total byte size limit aggregated across all priority bands within this
@@ -110,35 +205,22 @@ type ShardStats struct {
 	// TotalLen is the total number of items currently queued across all priority bands within this shard.
 	TotalLen uint64
 	// PerPriorityBandStats maps each configured priority level to its statistics within this shard.
+	// The capacity values within represent this shard's partition of the global band capacity.
 	// The key is the numerical priority level.
 	// All configured priority levels are guaranteed to be represented.
 	PerPriorityBandStats map[uint]PriorityBandStats
 }
 
-// DeepCopy returns a deep copy of the `ShardStats`.
-func (s *ShardStats) DeepCopy() ShardStats {
-	if s == nil {
-		return ShardStats{}
-	}
-	newStats := *s
-	if s.PerPriorityBandStats != nil {
-		newStats.PerPriorityBandStats = make(map[uint]PriorityBandStats, len(s.PerPriorityBandStats))
-		for k, v := range s.PerPriorityBandStats {
-			newStats.PerPriorityBandStats[k] = v.DeepCopy()
-		}
-	}
-	return newStats
-}
-
 // PriorityBandStats holds aggregated statistics for a single priority band.
 type PriorityBandStats struct {
 	// Priority is the numerical priority level this struct describes.
 	Priority uint
-	// PriorityName is an optional, human-readable name for the priority level (e.g., "Critical", "Sheddable").
+	// PriorityName is a human-readable name for the priority band (e.g., "Critical", "Sheddable").
+	// The registry configuration requires this field, so it is guaranteed to be non-empty.
 	PriorityName string
-	// CapacityBytes is the configured maximum total byte size for this priority band, aggregated across all items in
-	// all flow queues within this band. If scoped to a shard, its value represents the configured band limit for the
-	// `FlowRegistry` partitioned for this shard.
+	// CapacityBytes is the configured maximum total byte size for this priority band.
+	// When viewed via `AggregateStats`, this is the global limit. When viewed via `ShardStats`, this is the partitioned
+	// value for that specific shard.
 	// The `controller.FlowController` enforces this limit.
 	// A default non-zero value is guaranteed if not configured.
 	CapacityBytes uint64
@@ -147,11 +229,3 @@ type PriorityBandStats struct {
 	// Len is the total number of items currently queued in this priority band.
 	Len uint64
 }
-
-// DeepCopy returns a deep copy of the `PriorityBandStats`.
-func (s *PriorityBandStats) DeepCopy() PriorityBandStats {
-	if s == nil {
-		return PriorityBandStats{}
-	}
-	return *s
-}