feat(executor): support https schema for subgraph calls

feat(router, config): add support for loading supergraph from hive
feat(config): improve file-loading for supergraph

Author: dotansimha

Cargo.lock

@@ -49,3 +49,4 @@ thiserror = "2.0.14"
 xxhash-rust = { version = "0.8.15", features = ["xxh3"] }
 tokio = { version = "1.47.1", features = ["full"] }
 rand = "0.9.2"
+reqwest = "0.12.23"

Cargo.toml

@@ -25,6 +25,7 @@ futures = { workspace = true }
 graphql-parser = { workspace = true }
 graphql-tools = { workspace = true }
 serde = { workspace = true }
+reqwest = { workspace = true }
 sonic-rs = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }

bin/router/Cargo.toml

@@ -2,6 +2,7 @@ mod http_utils;
 mod logger;
 mod pipeline;
 mod shared_state;
+mod supergraph;
 
 use std::sync::Arc;
 
@@ -18,8 +19,6 @@ use ntex::{
     web::{self, HttpRequest},
 };
 
-use hive_router_query_planner::utils::parsing::parse_schema;
-
 async fn graphql_endpoint_handler(
     mut request: HttpRequest,
     body_bytes: Bytes,
@@ -33,10 +32,8 @@ pub async fn router_entrypoint() -> Result<(), Box<dyn std::error::Error>> {
     let router_config = load_config(config_path)?;
     configure_logging(&router_config.log);
 
-    let supergraph_sdl = router_config.supergraph.load().await?;
-    let parsed_schema = parse_schema(&supergraph_sdl);
     let addr = router_config.http.address();
-    let shared_state = RouterSharedState::new(parsed_schema, router_config);
+    let shared_state = RouterSharedState::new(router_config).await?;
 
     web::HttpServer::new(move || {
         web::App::new()

bin/router/src/lib.rs

@@ -1,19 +1,23 @@
 use std::sync::Arc;
 
-use graphql_parser::schema::Document;
 use graphql_tools::validation::{utils::ValidationError, validate::ValidationPlan};
 use hive_router_config::HiveRouterConfig;
 use hive_router_plan_executor::{
+    executors::error::SubgraphExecutorError,
     introspection::schema::{SchemaMetadata, SchemaWithMetadata},
     SubgraphExecutorMap,
 };
 use hive_router_query_planner::{
-    planner::{plan_nodes::QueryPlan, Planner},
+    planner::{plan_nodes::QueryPlan, Planner, PlannerError},
     state::supergraph_state::SupergraphState,
+    utils::parsing::parse_schema,
 };
 use moka::future::Cache;
 
-use crate::pipeline::normalize::GraphQLNormalizationPayload;
+use crate::{
+    pipeline::normalize::GraphQLNormalizationPayload,
+    supergraph::{base::LoadSupergraphError, resolve_from_config},
+};
 
 pub struct RouterSharedState {
     pub schema_metadata: SchemaMetadata,
@@ -27,23 +31,32 @@ pub struct RouterSharedState {
     pub router_config: HiveRouterConfig,
 }
 
+#[derive(Debug, thiserror::Error)]
+pub enum RouterSharedStateError {
+    #[error("Failed to load supergraph: {0}")]
+    SupergraphInitError(#[from] LoadSupergraphError),
+    #[error("Failed to init planner: {0}")]
+    PlannerInitError(#[from] PlannerError),
+    #[error("Failed to init executor: {0}")]
+    ExecutorInitError(#[from] SubgraphExecutorError),
+}
+
 impl RouterSharedState {
-    pub fn new(
-        parsed_supergraph_sdl: Document<'static, String>,
-        router_config: HiveRouterConfig,
-    ) -> Arc<Self> {
+    pub async fn new(router_config: HiveRouterConfig) -> Result<Arc<Self>, RouterSharedStateError> {
+        let mut supergraph_source_loader = resolve_from_config(&router_config.supergraph).await?;
+        supergraph_source_loader.reload().await?;
+        let supergraph_sdl = supergraph_source_loader.current().unwrap();
+        let parsed_supergraph_sdl = parse_schema(supergraph_sdl);
         let supergraph_state = SupergraphState::new(&parsed_supergraph_sdl);
-        let planner =
-            Planner::new_from_supergraph(&parsed_supergraph_sdl).expect("failed to create planner");
+        let planner = Planner::new_from_supergraph(&parsed_supergraph_sdl)?;
         let schema_metadata = planner.consumer_schema.schema_metadata();
 
         let subgraph_executor_map = SubgraphExecutorMap::from_http_endpoint_map(
             supergraph_state.subgraph_endpoint_map,
             router_config.traffic_shaping.clone(),
-        )
-        .expect("Failed to create subgraph executor map");
+        )?;
 
-        Arc::new(Self {
+        Ok(Arc::new(Self {
             schema_metadata,
             planner,
             validation_plan: graphql_tools::validation::rules::default_rules_validation_plan(),
@@ -53,6 +66,6 @@ impl RouterSharedState {
             parse_cache: moka::future::Cache::new(1000),
             normalize_cache: moka::future::Cache::new(1000),
             router_config,
-        })
+        }))
     }
 }

bin/router/src/shared_state.rs

@@ -0,0 +1,15 @@
+use async_trait::async_trait;
+
+#[derive(Debug, thiserror::Error)]
+pub enum LoadSupergraphError {
+    #[error("Failed to read supergraph file: {0}")]
+    ReadFileError(#[from] std::io::Error),
+    #[error("Failed to read supergraph from network: {0}")]
+    NetworkError(#[from] reqwest::Error),
+}
+
+#[async_trait]
+pub trait SupergraphLoader {
+    async fn reload(&mut self) -> Result<(), LoadSupergraphError>;
+    fn current(&self) -> Option<&str>;
+}

bin/router/src/supergraph/base.rs

@@ -0,0 +1,47 @@
+use async_trait::async_trait;
+use hive_router_config::primitives::file_path::FilePath;
+use tracing::{debug, trace};
+
+use crate::supergraph::base::{LoadSupergraphError, SupergraphLoader};
+
+pub struct SupergraphFileLoader {
+    file_path: FilePath,
+    current: Option<String>,
+}
+
+#[async_trait]
+impl SupergraphLoader for SupergraphFileLoader {
+    async fn reload(&mut self) -> Result<(), LoadSupergraphError> {
+        debug!(
+            "Reloading supergraph from file path: '{}'",
+            self.file_path.0
+        );
+        let content = tokio::fs::read_to_string(&self.file_path.0).await?;
+        trace!(
+            "Supergraph loaded from file path: '{}', content: {}",
+            self.file_path.0,
+            content
+        );
+
+        self.current = Some(content);
+        Ok(())
+    }
+
+    fn current(&self) -> Option<&str> {
+        self.current.as_deref()
+    }
+}
+
+impl SupergraphFileLoader {
+    pub async fn new(file_path: &FilePath) -> Result<Box<Self>, LoadSupergraphError> {
+        debug!(
+            "Creating supergraph source from file path: '{}'",
+            file_path.0
+        );
+
+        Ok(Box::new(Self {
+            file_path: file_path.clone(),
+            current: None,
+        }))
+    }
+}

bin/router/src/supergraph/file.rs

@@ -0,0 +1,64 @@
+use async_trait::async_trait;
+use http::header::USER_AGENT;
+use tracing::debug;
+
+use crate::supergraph::base::{LoadSupergraphError, SupergraphLoader};
+
+static USER_AGENT_VALUE: &str = "hive-router";
+static TIMEOUT: u64 = 10;
+static AUTH_HEADER_NAME: &str = "X-Hive-CDN-Key";
+
+pub struct SupergraphHiveConsoleLoader {
+    endpoint: String,
+    key: String,
+    current: Option<String>,
+    http_client: reqwest::Client,
+}
+
+#[async_trait]
+impl SupergraphLoader for SupergraphHiveConsoleLoader {
+    async fn reload(&mut self) -> Result<(), LoadSupergraphError> {
+        debug!(
+            "Fetching supergraph from Hive Console CDN: '{}'",
+            self.endpoint
+        );
+
+        let response = self
+            .http_client
+            .get(&self.endpoint)
+            .header(AUTH_HEADER_NAME, &self.key)
+            .header(USER_AGENT, USER_AGENT_VALUE)
+            .timeout(std::time::Duration::from_secs(TIMEOUT))
+            .send()
+            .await
+            .map_err(LoadSupergraphError::NetworkError)?;
+
+        let content = response
+            .text()
+            .await
+            .map_err(LoadSupergraphError::NetworkError)?;
+
+        self.current = Some(content);
+        Ok(())
+    }
+
+    fn current(&self) -> Option<&str> {
+        self.current.as_deref()
+    }
+}
+
+impl SupergraphHiveConsoleLoader {
+    pub async fn new(endpoint: &str, key: &str) -> Result<Box<Self>, LoadSupergraphError> {
+        debug!(
+            "Creating supergraph source from Hive Console CDN: '{}'",
+            endpoint
+        );
+
+        Ok(Box::new(Self {
+            endpoint: endpoint.to_string(),
+            key: key.to_string(),
+            current: None,
+            http_client: reqwest::Client::new(),
+        }))
+    }
+}

bin/router/src/supergraph/hive.rs

@@ -0,0 +1,25 @@
+use hive_router_config::supergraph::SupergraphSource;
+
+use crate::supergraph::{
+    base::{LoadSupergraphError, SupergraphLoader},
+    file::SupergraphFileLoader,
+    hive::SupergraphHiveConsoleLoader,
+};
+use tracing::debug;
+
+pub mod base;
+pub mod file;
+pub mod hive;
+
+pub async fn resolve_from_config(
+    config: &SupergraphSource,
+) -> Result<Box<dyn SupergraphLoader>, LoadSupergraphError> {
+    debug!("Resolving supergraph from config: {:?}", config);
+
+    match config {
+        SupergraphSource::File { path } => Ok(SupergraphFileLoader::new(path).await?),
+        SupergraphSource::HiveConsole { endpoint, key } => {
+            Ok(SupergraphHiveConsoleLoader::new(endpoint, key).await?)
+        }
+    }
+}

bin/router/src/supergraph/mod.rs

@@ -28,9 +28,10 @@ tracing = { workspace = true }
 thiserror = { workspace = true }
 xxhash-rust = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }
+
 dashmap = "6.0.0"
 ahash = "0.8.12"
-
+hyper-tls = "0.6.0"
 hyper-util = { version = "0.1.16", features = [
   "client",
   "client-legacy",