feat(router): request selected supergraphs

.changeset/drop_with_schema_from_on_graphql_validation_plugin_hook.md

@@ -0,0 +1,24 @@
+---
+hive-router-plan-executor: major
+hive-router: patch
+---
+
+# Drop `with_schema` from the `on_graphql_validation` plugin hook
+
+Remove `OnGraphQLValidationStartHookPayload::with_schema`. Replacing only the validation schema was unsafe because parsing, introspection, normalization, planning, demand control, execution, coprocessors, and schema-aware caches continued using the request's original supergraph.
+
+Plugins that need a request-specific schema should construct and retain an `Arc<Supergraph>`, then select it in `on_http_request`:
+
+```rust
+fn on_http_request<'req>(
+    &'req self,
+    payload: OnHttpRequestHookPayload<'req>,
+) -> OnHttpRequestHookResult<'req> {
+    payload.set_supergraph(self.supergraph_for_request(&payload));
+    payload.proceed()
+}
+```
+
+Build each variant with `Supergraph::from_sdl` or `Supergraph::from_document` outside the request hot path and reuse the same `Arc<Supergraph>`. The router snapshots the selected supergraph and applies it to the complete request pipeline.
+
+See `plugin_examples/replace_schema` for overriding a configured default and `plugin_examples/feature_flags` for plugin-only supergraph selection.

.changeset/health_readiness_plugin_hooks.md

@@ -0,0 +1,9 @@
+---
+hive-router: minor
+---
+
+# Health and readiness now pass through the plugin `on_http_request` chain and its `on_end` callbacks
+
+This is required because readiness in plugin-only mode must allow the plugin to select a supergraph for that specific readiness request.
+
+Coprocessors still do not run for health or readiness.

.changeset/rename_new_supergraph_data.md

@@ -0,0 +1,21 @@
+---
+hive-router-plan-executor: major
+---
+
+# Rename `on_supergraph_load` end hook payload `new_supergraph_data` field to `new_supergraph`
+
+Bringing consistency across the new supergraph snapshotting practice.
+
+```diff
+fn on_supergraph_reload<'a>(
+    &'a self,
+    payload: OnSupergraphLoadStartHookPayload,
+) -> OnSupergraphLoadStartHookResult<'a> {
+    payload.on_end(|payload| {
+-       let supergraph = payload.new_supergraph_data;
++       let supergraph = payload.new_supergraph;
+        println!("{}", supergraph.public_schema.sdl);
+        payload.proceed()
+    })
+}
+```

.changeset/replace_the_schema_state_in_the_on_http_request_plugin_hook.md

@@ -0,0 +1,14 @@
+---
+hive-router-plan-executor: minor
+hive-router: patch
+---
+
+# Select a supergraph in the `on_http_request` plugin hook
+
+Add `OnHttpRequestHookPayload::set_supergraph`, allowing a plugin to select a stable `Arc<Supergraph>` for an HTTP request or WebSocket upgrade. The selected supergraph is used consistently for validation, introspection, normalization, planning, demand control, execution, coprocessors, usage reporting, and request deduplication.
+
+`Supergraph` contains schema-derived state only and can be built with `Supergraph::from_sdl` or `Supergraph::from_document`. Router-specific state, including subgraph executors and schema-aware caches, remains owned by the router. The router builds configured runtimes eagerly and plugin-selected runtimes lazily, reusing them through a bounded FIFO cache.
+
+Plugins own the lifetime of their supergraphs. Dropping the last `Arc<Supergraph>` retires that supergraph: ordinary in-flight requests finish from their snapshots, active subscriptions close with the schema-reload error, and the router removes any cached plugin runtime in the background. Runtime eviction (when the internal bounded FIFO cache of supergraph runtimes evicts) does not retire a supergraph - if reused later, the router will rebuild the internal supergraph runtime.
+
+See `plugin_examples/replace_schema` for overriding a configured default and `plugin_examples/feature_flags` for plugin-only supergraph selection.

.changeset/supergraph_source_plugin.md

@@ -0,0 +1,11 @@
+---
+hive-router-config: minor
+---
+
+# Add `supergraph.source: plugin` for deployments where plugins are the only source of supergraphs
+
+This source creates no loader and has no configured fallback. Readiness, GraphQL requests, and WebSocket upgrades return service unavailable until the request's plugin selects a usable supergraph.
+
+HTTP readiness checks invoke plugin's `on_http_request`, in order for the readiness check to pass - a supergraph must be selected even during that readiness request.
+
+See `plugin_examples/feature_flags` for `supergraph.source: plugin` with all variants selected by the plugin.

bin/router/src/cache_state.rs

@@ -1,20 +1,10 @@
 use std::sync::Arc;
 
-use graphql_tools::validation::utils::ValidationError;
 use hive_router_internal::telemetry::TelemetryContext;
-use hive_router_query_planner::planner::plan_nodes::QueryPlan;
-use moka::future::Cache;
 use moka::Entry;
 
-use crate::pipeline::normalize::GraphQLNormalizationPayload;
-use crate::pipeline::parser::ParseCacheEntry;
-
-pub struct CacheState {
-    pub parse_cache: Cache<u64, ParseCacheEntry>,
-    pub validate_cache: Cache<u64, Arc<Vec<ValidationError>>>,
-    pub normalize_cache: Cache<u64, Arc<GraphQLNormalizationPayload>>,
-    pub plan_cache: Cache<u64, Arc<QueryPlan>>,
-}
+use crate::schema_state::SchemaState;
+use crate::shared_state::RouterSharedState;
 
 #[derive(Clone, Copy, Debug)]
 pub enum CacheHitMiss {
@@ -65,50 +55,40 @@ impl<K, V> EntryValueHitMissExt<V> for Entry<K, V> {
     }
 }
 
-impl CacheState {
-    pub fn new() -> Self {
-        Self {
-            parse_cache: Cache::new(1000),
-            validate_cache: Cache::new(1000),
-            normalize_cache: Cache::new(1000),
-            plan_cache: Cache::new(1000),
-        }
-    }
-
-    pub fn on_schema_change(&self) {
-        self.plan_cache.invalidate_all();
-        self.validate_cache.invalidate_all();
-        self.normalize_cache.invalidate_all();
-    }
-}
-
 pub fn register_cache_size_observers(
     telemetry_context: Arc<TelemetryContext>,
-    cache_state: Arc<CacheState>,
+    shared_state: Arc<RouterSharedState>,
+    schema_state: Arc<SchemaState>,
 ) {
     let metrics = &telemetry_context.metrics.cache;
 
-    let parse_cache = Arc::clone(&cache_state);
     metrics
         .parse
-        .observe_size_with(move || parse_cache.parse_cache.entry_count());
-
-    let normalize_cache = Arc::clone(&cache_state);
-    metrics
-        .normalize
-        .observe_size_with(move || normalize_cache.normalize_cache.entry_count());
-
-    let validate_cache = Arc::clone(&cache_state);
-    metrics
-        .validate
-        .observe_size_with(move || validate_cache.validate_cache.entry_count());
-
-    let plan_cache = Arc::clone(&cache_state);
-    metrics
-        .plan
-        .observe_size_with(move || plan_cache.plan_cache.entry_count());
-
-    // The demand-control formula cache is owned by `DemandControlRuntime` (it is
-    // schema-scoped), so its size observer is registered in
-    // `SchemaState::new_from_config` instead.
+        .observe_size_with(move || shared_state.parse_cache.entry_count());
+
+    // validate/normalize/plan caches live on `RouterSupergraphRuntime` (one per supergraph variant,
+    // dropped with it on retirement) rather than on the shared state, so sum entry counts across
+    // every runtime currently alive (the configured default plus any plugin-selected ones still cached)
+
+    let validate_schema_state = Arc::clone(&schema_state);
+    metrics.validate.observe_size_with(move || {
+        let mut total = 0;
+        validate_schema_state
+            .for_each_runtime(|runtime| total += runtime.validate_cache.entry_count());
+        total
+    });
+
+    let normalize_schema_state = Arc::clone(&schema_state);
+    metrics.normalize.observe_size_with(move || {
+        let mut total = 0;
+        normalize_schema_state
+            .for_each_runtime(|runtime| total += runtime.normalize_cache.entry_count());
+        total
+    });
+
+    metrics.plan.observe_size_with(move || {
+        let mut total = 0;
+        schema_state.for_each_runtime(|runtime| total += runtime.plan_cache.entry_count());
+        total
+    });
 }

bin/router/src/http_utils/probes.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use ntex::web::{self, Responder};
+use ntex::web::{self, HttpRequest, Responder};
 
 use crate::schema_state::SchemaState;
 
@@ -9,9 +9,10 @@ pub async fn health_check_handler() -> impl Responder {
 }
 
 pub async fn readiness_check_handler(
+    req: HttpRequest,
     schema_state: web::types::State<Arc<SchemaState>>,
 ) -> impl Responder {
-    if schema_state.is_ready() {
+    if schema_state.is_ready(&req) {
         web::HttpResponse::Ok()
     } else {
         web::HttpResponse::ServiceUnavailable()

bin/router/src/lib.rs

@@ -49,7 +49,7 @@ use crate::{
     telemetry::{HeaderExtractor, PrometheusAttached},
 };
 
-use crate::cache_state::{register_cache_size_observers, CacheState};
+use crate::cache_state::register_cache_size_observers;
 pub use crate::plugins::registry::PluginRegistry;
 pub use crate::{schema_state::SchemaState, shared_state::RouterSharedState};
 pub use arc_swap::ArcSwap;
@@ -220,11 +220,13 @@ async fn graphql_endpoint_dispatch(
         if let Some(coprocessor_runtime) = app_state.coprocessor.as_ref() {
             response = match coprocessor_runtime
                 .on_graphql_response(response, request, || {
-                    schema_state
-                        .current_supergraph()
-                        .as_ref()
-                        .as_ref()
-                        .map(|supergraph| supergraph.public_schema.sdl.clone())
+                    // reuse the exact snapshot execution already resolved and stored on the
+                    // request - never re-resolve here, which could observe a different
+                    // generation than the one that actually executed the operation
+                    request
+                        .extensions()
+                        .get::<crate::schema_state::SelectedSupergraph>()
+                        .map(|selected| selected.snapshot.public_schema.sdl.clone())
                 })
                 .await
             {
@@ -440,23 +442,18 @@ pub async fn configure_app_from_config(
     let storage_manager = Arc::new(StorageManager::new(&router_config.storages)?);
     let router_config_arc = Arc::new(router_config);
     let telemetry_context_arc = Arc::new(telemetry_context);
-    let cache_state = Arc::new(CacheState::new());
-
-    if router_config_arc.telemetry.metrics.is_enabled() {
-        register_cache_size_observers(telemetry_context_arc.clone(), cache_state.clone());
-    }
 
     let schema_state = SchemaState::new_from_config(
         bg_tasks_manager,
         telemetry_context_arc.clone(),
         router_config_arc.clone(),
         plugins_arc.clone(),
-        cache_state.clone(),
         active_subscriptions.clone(),
         storage_manager.clone(),
     )
     .await?;
     let schema_state_arc = Arc::new(schema_state);
+
     let mut validation_plan = default_rules_validation_plan();
     if let Some(max_depth_config) = &router_config_arc.limits.max_depth {
         validation_plan.add_rule(Box::new(MaxDepthRule {
@@ -493,19 +490,27 @@ pub async fn configure_app_from_config(
         ));
     }
 
+    let metrics_enabled = router_config_arc.telemetry.metrics.is_enabled();
     let shared_state = Arc::new(RouterSharedState::new(
         router_config_arc,
         persisted_documents_runtime,
         jwt_runtime,
         hive_usage_agent,
         validation_plan,
-        telemetry_context_arc,
+        telemetry_context_arc.clone(),
         plugins_arc,
-        cache_state,
         active_subscriptions.clone(),
         storage_manager,
     )?);
 
+    if metrics_enabled {
+        register_cache_size_observers(
+            telemetry_context_arc,
+            shared_state.clone(),
+            schema_state_arc.clone(),
+        );
+    }
+
     Ok((shared_state, schema_state_arc))
 }
 

bin/router/src/pipeline/authorization/mod.rs

@@ -12,7 +12,6 @@ pub mod metadata;
 
 use std::sync::Arc;
 
-use crate::pipeline::authorization::metadata::AuthorizationMetadataExt;
 use crate::pipeline::error::PipelineError;
 use crate::pipeline::normalize::GraphQLNormalizationPayload;
 use crate::pipeline::nullify::rebuilder::{
@@ -34,7 +33,7 @@ use hive_router_plan_executor::response::graphql_error::GraphQLError;
 use hive_router_query_planner::ast::operation::OperationDefinition;
 
 use hive_router_internal::telemetry::traces::spans::graphql::GraphQLAuthorizeSpan;
-pub use metadata::{AuthorizationMetadataError, UserAuthContext};
+pub use metadata::{AuthorizationMetadataError, AuthorizationMetadataExt, UserAuthContext};
 
 /// Error representing an unauthorized field access.
 ///

bin/router/src/pipeline/coerce_variables.rs

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 
 use hive_router_internal::telemetry::traces::spans::graphql::GraphQLVariableCoercionSpan;
 use hive_router_plan_executor::execution::plan::CoerceVariablesPayload;
-use hive_router_plan_executor::hooks::on_supergraph_load::SupergraphData;
+use hive_router_plan_executor::hooks::on_supergraph_load::SupergraphSnapshot;
 use hive_router_plan_executor::variables::collect_variables;
 use sonic_rs::Value;
 use tracing::{trace, warn};
@@ -12,7 +12,7 @@ use crate::pipeline::normalize::GraphQLNormalizationPayload;
 
 #[inline]
 pub fn coerce_request_variables(
-    supergraph: &SupergraphData,
+    supergraph: &SupergraphSnapshot,
     variables: &mut HashMap<String, Value>,
     normalized_operation: &GraphQLNormalizationPayload,
 ) -> Result<CoerceVariablesPayload, PipelineError> {

bin/router/src/pipeline/demand_control/runtime.rs

@@ -16,7 +16,7 @@ use hive_router_plan_executor::execution::demand_control::{
     DemandControlExecutionOperationContext, DemandControlExecutionSubgraphsContext,
 };
 use hive_router_plan_executor::execution::plan::CoerceVariablesPayload;
-use hive_router_plan_executor::hooks::on_supergraph_load::SupergraphData;
+use hive_router_plan_executor::hooks::on_supergraph_load::SupergraphSnapshot;
 use hive_router_query_planner::ast::operation::{OperationDefinition, SubgraphFetchOperation};
 use hive_router_query_planner::planner::plan_nodes::{PlanNode, QueryPlan};
 use hive_router_query_planner::state::supergraph_state::{OperationKind, SupergraphState};
@@ -85,17 +85,13 @@ impl DemandControlRuntime {
     pub fn formula_cache(&self) -> &Cache<u64, Arc<DemandControlFormulaPlan>> {
         &self.formula_cache
     }
-
-    pub fn invalidate_formula_cache(&self) {
-        self.formula_cache.invalidate_all();
-    }
 }
 
 impl DemandControlRuntime {
     #[allow(clippy::too_many_arguments)]
     pub async fn evaluate<'exec>(
         &self,
-        supergraph: &'exec SupergraphData,
+        supergraph: &'exec SupergraphSnapshot,
         variable_payload: &'exec CoerceVariablesPayload,
         query_plan: &'exec QueryPlan,
         operation_for_plan: &'exec OperationDefinition,

bin/router/src/pipeline/error.rs

@@ -40,6 +40,7 @@ use crate::{
         progressive_override::LabelEvaluationError,
         sse,
     },
+    schema_state::RouterSupergraphRuntimeError,
     RouterSharedState,
 };
 
@@ -197,6 +198,10 @@ pub enum PipelineError {
 
     #[error(transparent)]
     OperationFilterFailed(#[from] OperationFilterError),
+
+    #[error("Supergraph runtime error")]
+    #[strum(serialize = "SUPERGRAPH_RUNTIME_ERROR")]
+    RouterSupergraphRuntimeError(#[from] RouterSupergraphRuntimeError),
 }
 
 #[derive(Clone, Debug, thiserror::Error)]
@@ -305,7 +310,10 @@ impl PipelineError {
             (Self::NoSupergraphAvailable { .. }, _) => StatusCode::SERVICE_UNAVAILABLE,
             (Self::CoprocessorError(err), _) => err.status_code(),
             (Self::RequestContextError(_), _) => StatusCode::INTERNAL_SERVER_ERROR,
+
             (Self::OperationFilterFailed(_), _) => StatusCode::INTERNAL_SERVER_ERROR,
+
+            (Self::RouterSupergraphRuntimeError(_), _) => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 }