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 is responsible for managing the lifecycle of all flows, queues, and
shards.

Key features of this implementation include:

- An actor-based, serialized event loop that processes all state changes
  to ensure correctness and eliminate race conditions in the control
  plane.
- A robust garbage collection system for idle flows and drained
  components, using a "Trust but Verify" pattern to safely handle races
  between the data path and control plane.
- A well-defined component lifecycle (Active, Draining, Drained) with
  atomic state transitions and exactly-once edge signaling.
- A sharded architecture where the `FlowRegistry` orchestrates the
  `registryShard` data plane slices.

Author: LukeAVanDrie

pkg/epp/flowcontrol/contracts/doc.go

@@ -0,0 +1,28 @@
+/*
+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 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 with external systems and pluggable logic.
+//
+// The two primary contracts defined here are:
+//
+//   - `FlowRegistry`: The interface for the stateful control plane that manages the lifecycle of all flows, queues, and
+//     policies.
+//
+//   - `SaturationDetector`: The interface for a component that provides real-time load signals, allowing the dispatch
+//     engine to react to backend saturation.
+package contracts

pkg/epp/flowcontrol/contracts/errors.go

@@ -20,16 +20,20 @@ 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")
 
+	// 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 because it was not
 	// part of the initial 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 = errors.New("policy is not compatible with queue capabilities")
+
+	// ErrInvalidShardCount indicates that an invalid shard count was provided (e.g., zero).
+	ErrInvalidShardCount = errors.New("invalid shard count")
 )

pkg/epp/flowcontrol/contracts/registry.go

@@ -14,18 +14,144 @@ 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"
 )
 
+// FlowRegistry is the complete interface for the global control plane, composed of administrative functions and the
+// ability to provide shard accessors. A concrete implementation of this interface is the single source of truth for all
+// flow control state and configuration.
+//
+// # Conformance
+//
+// All methods defined in this interface (including those embedded) MUST be goroutine-safe.
+// Implementations are expected to perform complex updates (e.g., RegisterOrUpdateFlow, UpdateShardCount) atomically to
+// preserve system invariants.
+//
+// # Invariants
+//
+// Concrete implementations of FlowRegistry MUST uphold the following invariants across all operations:
+//  1. Shard Consistency: All configured priority bands and logical flows must be represented on every active internal
+//     shard. Plugin instance types (e.g., the specific SafeQueue implementation or policy plugins) must be consistent
+//     for a given flow or band across all shards.
+//  2. Flow Instance Uniqueness per Band: For any given logical flow, there can be a maximum of one ManagedQueue
+//     instance per priority band. An instance can be either 'active' or 'draining'.
+//  3. Single Active Instance per Flow: For any given logical flow, there can be a maximum of one *active* ManagedQueue
+//     instance across all priority bands. All other instances for that flow must be in a 'draining' state.
+//  4. Capacity Partitioning Consistency: Global and per-band capacity limits are uniformly partitioned across all
+//     active shards. The sum of the capacity limits allocated to each shard must not exceed the globally configured
+//     limits.
+//
+// # Flow Lifecycle States
+//
+//   - Registered: A logical flow is 'registered' when it is known to the FlowRegistry. It has exactly one 'active'
+//     instance across all priority bands and zero or more 'draining' instances.
+//   - Active: A specific instance of a flow within a priority band is 'active' if it is the designated target for all
+//     new enqueues for that logical flow.
+//   - Draining: A flow instance is 'draining' if it no longer accepts new enqueues but still contains items that are
+//     eligible for dispatch. This occurs after a priority change or when a flow is unregistered.
+//   - Unregistered: A logical flow is 'unregistered' after a call to UnregisterFlow. It has no 'active' instances,
+//     though 'draining' instances may still exist until their queues are empty.
+//
+// # Shard Garbage Collection
+//
+// When a shard is decommissioned via UpdateShardCount, the FlowRegistry must ensure a graceful shutdown. It must mark
+// the shard as inactive to prevent new enqueues, allow the FlowController to continue draining its queues, and only
+// delete the shard's state after the associated worker has fully terminated and all queues are empty.
+type FlowRegistry interface {
+	FlowRegistryAdmin
+	ShardProvider
+}
+
+// FlowRegistryAdmin defines the administrative interface for the global control plane. This interface is intended for
+// external systems (like a Kubernetes operator) to configure flows, manage system parallelism, and query aggregated
+// statistics for observability.
+//
+// # Design Rationale for Dynamic Update Strategies
+//
+// The `FlowRegistryAdmin` contract specifies precise behaviors for handling dynamic updates. These strategies were chosen
+// to prioritize system stability, correctness, and minimal disruption:
+//
+//   - Graceful Draining (for Priority/Shard Lifecycle Changes): For operations that change a flow's priority or
+//     decommission a shard, the affected queue instances are marked as inactive but are not immediately deleted. They enter
+//     a "draining" state where they no longer accept new requests but are still processed for dispatch. This ensures that
+//     requests already accepted by the system are processed to completion. Crucially, requests in a draining queue continue
+//     to be dispatched according to the priority level and policies they were enqueued with, ensuring consistency.
+//
+//   - Atomic Queue Migration (Future Design for Incompatible Intra-Flow Policy Changes): When an intra-flow policy is
+//     updated to one that is incompatible with the existing queue data structure, the designed future behavior is a
+//     full "drain and re-enqueue" migration. This more disruptive operation is necessary to guarantee correctness. A
+//     simpler "graceful drain"—by creating a second instance of the same flow in the same priority band—is not used
+//     because it would violate the system's "one flow instance per band" invariant. This invariant is critical because
+//     it ensures that inter-flow policies operate on a clean set of distinct flows, stateful intra-flow policies have a
+//     single authoritative view of their flow's state, and lookups are unambiguous. Note: This atomic migration is a
+//     future design consideration and is not implemented in the current version.
+//
+//   - Self-Balancing on Shard Scale-Up: When new shards are added via `UpdateShardCount`, the framework relies on the
+//     `FlowController`'s request distribution logic (e.g., a "Join the Shortest Queue by Bytes (JSQ-Bytes)" strategy) to
+//     naturally funnel *new* requests to the less-loaded shards. This design choice strategically avoids the complexity of
+//     actively migrating or rebalancing existing items that are already queued on other shards, promoting system stability
+//     during scaling events.
+type FlowRegistryAdmin interface {
+	// RegisterOrUpdateFlow handles the registration of a new flow or the update of an existing flow's specification.
+	// This method orchestrates complex state transitions atomically across all managed shards.
+	//
+	// # Dynamic Update Behaviors
+	//
+	//   - Priority Changes: If a flow's priority level changes, its current active `ManagedQueue` instance is marked
+	//     as inactive to drain existing requests. A new instance is activated at the new priority level. If a flow is
+	//     updated to a priority level where an instance is already draining (e.g., during a rapid rollback), that
+	//     draining instance is re-activated.
+	//
+	// # Returns
+	//
+	//   - nil on success.
+	//   - An error wrapping ErrFlowIDEmpty if spec.ID is empty.
+	//   - An error wrapping ErrPriorityBandNotFound if spec.Priority refers to an unconfigured priority level.
+	//   - Other errors if internal creation/activation of policy or queue instances fail.
+	RegisterOrUpdateFlow(spec types.FlowSpecification) error
+
+	// UpdateShardCount dynamically adjusts the number of internal state shards, triggering a state rebalance.
+	//
+	// # Dynamic Update Behaviors
+	//
+	//   - On Increase: New, empty state shards are initialized with all registered flows. The FlowController's request
+	//     distribution logic will naturally balance load to these new shards over time.
+	//   - On Decrease: A specified number of existing shards are marked as inactive. They stop accepting new requests
+	//     but continue to drain existing items. They are fully removed only after their queues are empty.
+	//
+	// The implementation MUST atomically re-partition capacity allocations across all active shards when the count
+	// changes.
+	UpdateShardCount(n uint) 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 a minimal interface for consumers that need to discover and iterate over available shards.
+//
+// A "shard" is an internal, parallel execution unit that allows the FlowController's core dispatch logic to be
+// parallelized. Consumers of this interface, such as a request distributor, MUST check `RegistryShard.IsActive()` before
+// routing new work to a shard to ensure they do not send requests to a shard that is gracefully draining.
+type ShardProvider interface {
+	// Shards returns a slice of accessors, one for each internal state shard.
+	//
+	// A "shard" is an internal, parallel execution unit that allows the 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.
+	//
+	// The returned slice includes accessors for both active and draining shards. Consumers MUST use IsActive() to
+	// determine if new work should be routed to a shard. Callers should not modify the returned slice.
+	Shards() []RegistryShard
+}
+
 // 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 the necessary methods for a worker to perform its
 // dispatch operations by accessing queues and policies in a concurrent-safe manner.
@@ -100,6 +226,18 @@ type ManagedQueue interface {
 	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
@@ -117,6 +255,7 @@ type ShardStats struct {
 	PerPriorityBandStats map[uint]PriorityBandStats
 }
 
+// DeepCopy returns a deep copy of the `ShardStats`.
 // DeepCopy returns a deep copy of the `ShardStats`.
 func (s *ShardStats) DeepCopy() ShardStats {
 	if s == nil {
@@ -150,6 +289,7 @@ type PriorityBandStats struct {
 	Len uint64
 }
 
+// DeepCopy returns a deep copy of the `PriorityBandStats`.
 // DeepCopy returns a deep copy of the `PriorityBandStats`.
 func (s *PriorityBandStats) DeepCopy() PriorityBandStats {
 	if s == nil {