feat(router): Subgraph Timeout Configuration

ardatan · ardatan · commit f7a56fad9f3b · 2025-10-31T14:38:24.000+03:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/lib/executor/Cargo.toml b/lib/executor/Cargo.toml
@@ -44,6 +44,7 @@ hyper-util = { version = "0.1.16", features = [
   "http2",
   "tokio",
 ] }
+humantime = "2.3.0"
 bytes = "1.10.1"
 itoa = "1.0.15"
 ryu = "1.0.20"
diff --git a/lib/executor/src/execution/plan.rs b/lib/executor/src/execution/plan.rs
@@ -708,6 +708,7 @@ impl<'exec, 'req> Executor<'exec, 'req> {
             representations,
             headers: headers_map,
             extensions: None,
+            client_request: self.client_request,
         };
 
         if let Some(jwt_forwarding_plan) = &self.jwt_forwarding_plan {
@@ -722,7 +723,7 @@ impl<'exec, 'req> Executor<'exec, 'req> {
             subgraph_name: node.service_name.clone(),
             response: self
                 .executors
-                .execute(&node.service_name, subgraph_request, self.client_request)
+                .execute(&node.service_name, subgraph_request)
                 .await
                 .into(),
         }))
diff --git a/lib/executor/src/executors/common.rs b/lib/executor/src/executors/common.rs
@@ -5,11 +5,13 @@ use bytes::Bytes;
 use http::HeaderMap;
 use sonic_rs::Value;
 
+use crate::execution::client_request_details::ClientRequestDetails;
+
 #[async_trait]
 pub trait SubgraphExecutor {
-    async fn execute<'a>(
+    async fn execute<'exec, 'req>(
         &self,
-        execution_request: HttpExecutionRequest<'a>,
+        execution_request: HttpExecutionRequest<'exec, 'req>,
     ) -> HttpExecutionResponse;
 
     fn to_boxed_arc<'a>(self) -> Arc<Box<dyn SubgraphExecutor + Send + Sync + 'a>>
@@ -26,18 +28,19 @@ pub type SubgraphExecutorBoxedArc = Arc<Box<SubgraphExecutorType>>;
 
 pub type SubgraphRequestExtensions = HashMap<String, Value>;
 
-pub struct HttpExecutionRequest<'a> {
-    pub query: &'a str,
+pub struct HttpExecutionRequest<'exec, 'req> {
+    pub query: &'exec str,
     pub dedupe: bool,
-    pub operation_name: Option<&'a str>,
+    pub operation_name: Option<&'exec str>,
     // TODO: variables could be stringified before even executing the request
-    pub variables: Option<HashMap<&'a str, &'a sonic_rs::Value>>,
+    pub variables: Option<HashMap<&'exec str, &'exec sonic_rs::Value>>,
     pub headers: HeaderMap,
     pub representations: Option<Vec<u8>>,
     pub extensions: Option<SubgraphRequestExtensions>,
+    pub client_request: &'exec ClientRequestDetails<'exec, 'req>,
 }
 
-impl HttpExecutionRequest<'_> {
+impl HttpExecutionRequest<'_, '_> {
     pub fn add_request_extensions_field(&mut self, key: String, value: Value) {
         self.extensions
             .get_or_insert_with(HashMap::new)
diff --git a/lib/executor/src/executors/error.rs b/lib/executor/src/executors/error.rs
@@ -20,6 +20,10 @@ pub enum SubgraphExecutorError {
     RequestFailure(String, String),
     #[error("Failed to serialize variable \"{0}\": {1}")]
     VariablesSerializationFailure(String, String),
+    #[error("Failed to resolve VRL expression for timeout. Runtime error: {0}")]
+    TimeoutExpressionResolutionFailure(String),
+    #[error("Request to subgraph \"{0}\" timed out after {1} milliseconds")]
+    RequestTimeout(String, u64),
 }
 
 impl From<SubgraphExecutorError> for GraphQLError {
@@ -61,6 +65,10 @@ impl SubgraphExecutorError {
             SubgraphExecutorError::VariablesSerializationFailure(_, _) => {
                 "SUBGRAPH_VARIABLES_SERIALIZATION_FAILURE"
             }
+            SubgraphExecutorError::TimeoutExpressionResolutionFailure(_) => {
+                "SUBGRAPH_TIMEOUT_EXPRESSION_RESOLUTION_FAILURE"
+            }
+            SubgraphExecutorError::RequestTimeout(_, _) => "SUBGRAPH_REQUEST_TIMEOUT",
         }
     }
 }
diff --git a/lib/executor/src/executors/http.rs b/lib/executor/src/executors/http.rs
@@ -1,9 +1,11 @@
+use std::collections::BTreeMap;
 use std::sync::Arc;
 
+use crate::execution::client_request_details::ClientRequestDetails;
 use crate::executors::common::HttpExecutionResponse;
 use crate::executors::dedupe::{request_fingerprint, ABuildHasher, SharedResponse};
 use dashmap::DashMap;
-use hive_router_config::HiveRouterConfig;
+use hive_router_config::traffic_shaping::DurationOrExpression;
 use tokio::sync::OnceCell;
 
 use async_trait::async_trait;
@@ -18,6 +20,7 @@ use hyper_tls::HttpsConnector;
 use hyper_util::client::legacy::{connect::HttpConnector, Client};
 use tokio::sync::Semaphore;
 use tracing::debug;
+use vrl::core::Value as VrlValue;
 
 use crate::executors::common::HttpExecutionRequest;
 use crate::executors::error::SubgraphExecutorError;
@@ -35,8 +38,9 @@ pub struct HTTPSubgraphExecutor {
     pub http_client: Arc<Client<HttpsConnector<HttpConnector>, Full<Bytes>>>,
     pub header_map: HeaderMap,
     pub semaphore: Arc<Semaphore>,
-    pub config: Arc<HiveRouterConfig>,
+    pub dedupe_enabled: bool,
     pub in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
+    pub timeout: DurationOrExpression,
 }
 
 const FIRST_VARIABLE_STR: &[u8] = b",\"variables\":{";
@@ -50,8 +54,9 @@ impl HTTPSubgraphExecutor {
         endpoint: http::Uri,
         http_client: Arc<HttpClient>,
         semaphore: Arc<Semaphore>,
-        config: Arc<HiveRouterConfig>,
+        dedupe_enabled: bool,
         in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
+        timeout: DurationOrExpression,
     ) -> Self {
         let mut header_map = HeaderMap::new();
         header_map.insert(
@@ -69,14 +74,15 @@ impl HTTPSubgraphExecutor {
             http_client,
             header_map,
             semaphore,
-            config,
+            dedupe_enabled,
             in_flight_requests,
+            timeout,
         }
     }
 
-    fn build_request_body<'a>(
+    fn build_request_body(
         &self,
-        execution_request: &HttpExecutionRequest<'a>,
+        execution_request: &HttpExecutionRequest<'_, '_>,
     ) -> Result<Vec<u8>, SubgraphExecutorError> {
         let mut body = Vec::with_capacity(4096);
         body.put(FIRST_QUOTE_STR);
@@ -137,6 +143,7 @@ impl HTTPSubgraphExecutor {
         &self,
         body: Vec<u8>,
         headers: HeaderMap,
+        client_request: &ClientRequestDetails<'_, '_>,
     ) -> Result<SharedResponse, SubgraphExecutorError> {
         let mut req = hyper::Request::builder()
             .method(http::Method::POST)
@@ -151,9 +158,62 @@ impl HTTPSubgraphExecutor {
 
         debug!("making http request to {}", self.endpoint.to_string());
 
-        let res = self.http_client.request(req).await.map_err(|e| {
-            SubgraphExecutorError::RequestFailure(self.endpoint.to_string(), e.to_string())
-        })?;
+        let timeout = match &self.timeout {
+            DurationOrExpression::Duration(dur) => *dur,
+            DurationOrExpression::Expression(expr) => {
+                let value =
+                    VrlValue::Object(BTreeMap::from([("request".into(), client_request.into())]));
+                let result = expr.execute(value).map_err(|err| {
+                    SubgraphExecutorError::TimeoutExpressionResolutionFailure(err.to_string())
+                })?;
+                match result {
+                    VrlValue::Integer(i) if i >= 0 => std::time::Duration::from_millis(i as u64),
+                    VrlValue::Float(f) => {
+                        let f = f.into_inner();
+                        if f >= 0.0 {
+                            std::time::Duration::from_millis(f as u64)
+                        } else {
+                            return Err(SubgraphExecutorError::TimeoutExpressionResolutionFailure(
+                                "Timeout expression resolved to a negative float".to_string(),
+                            ));
+                        }
+                    }
+                    VrlValue::Bytes(b) => {
+                        let str: String = String::from_utf8(b.to_vec()).map_err(|e| {
+                            SubgraphExecutorError::TimeoutExpressionResolutionFailure(format!(
+                                "Failed to parse duration string from bytes: {}",
+                                e
+                            ))
+                        })?;
+                        let parsed = humantime::parse_duration(&str).map_err(|e| {
+                            SubgraphExecutorError::TimeoutExpressionResolutionFailure(format!(
+                                "Failed to parse duration string '{}': {}",
+                                str, e
+                            ))
+                        })?;
+                        parsed
+                    }
+                    _ => {
+                        return Err(SubgraphExecutorError::TimeoutExpressionResolutionFailure(
+                            "Timeout expression did not resolve to a non-negative integer or float"
+                                .to_string(),
+                        ));
+                    }
+                }
+            }
+        };
+
+        let res = tokio::time::timeout(timeout, self.http_client.request(req))
+            .await
+            .map_err(|_| {
+                SubgraphExecutorError::RequestTimeout(
+                    self.endpoint.to_string(),
+                    timeout.as_millis() as u64,
+                )
+            })?
+            .map_err(|e| {
+                SubgraphExecutorError::RequestFailure(self.endpoint.to_string(), e.to_string())
+            })?;
 
         debug!(
             "http request to {} completed, status: {}",
@@ -210,9 +270,9 @@ impl HTTPSubgraphExecutor {
 #[async_trait]
 impl SubgraphExecutor for HTTPSubgraphExecutor {
     #[tracing::instrument(skip_all, fields(subgraph_name = self.subgraph_name))]
-    async fn execute<'a>(
+    async fn execute<'exec, 'req>(
         &self,
-        execution_request: HttpExecutionRequest<'a>,
+        execution_request: HttpExecutionRequest<'exec, 'req>,
     ) -> HttpExecutionResponse {
         let body = match self.build_request_body(&execution_request) {
             Ok(body) => body,
@@ -230,11 +290,14 @@ impl SubgraphExecutor for HTTPSubgraphExecutor {
             headers.insert(key, value.clone());
         });
 
-        if !self.config.traffic_shaping.dedupe_enabled || !execution_request.dedupe {
+        if !self.dedupe_enabled || !execution_request.dedupe {
             // This unwrap is safe because the semaphore is never closed during the application's lifecycle.
             // `acquire()` only fails if the semaphore is closed, so this will always return `Ok`.
             let _permit = self.semaphore.acquire().await.unwrap();
-            return match self._send_request(body, headers).await {
+            return match self
+                ._send_request(body, headers, execution_request.client_request)
+                .await
+            {
                 Ok(shared_response) => HttpExecutionResponse {
                     body: shared_response.body,
                     headers: shared_response.headers,
@@ -266,7 +329,8 @@ impl SubgraphExecutor for HTTPSubgraphExecutor {
                     // This unwrap is safe because the semaphore is never closed during the application's lifecycle.
                     // `acquire()` only fails if the semaphore is closed, so this will always return `Ok`.
                     let _permit = self.semaphore.acquire().await.unwrap();
-                    self._send_request(body, headers).await
+                    self._send_request(body, headers, execution_request.client_request)
+                        .await
                 };
                 // It's important to remove the entry from the map before returning the result.
                 // This ensures that once the OnceCell is set, no future requests can join it.
diff --git a/lib/executor/src/executors/map.rs b/lib/executor/src/executors/map.rs
@@ -59,7 +59,7 @@ impl SubgraphExecutorMap {
         let client: HttpClient = Client::builder(TokioExecutor::new())
             .pool_timer(TokioTimer::new())
             .pool_idle_timeout(Duration::from_secs(
-                config.traffic_shaping.pool_idle_timeout_seconds,
+                config.traffic_shaping.all.pool_idle_timeout_seconds,
             ))
             .pool_max_idle_per_host(config.traffic_shaping.max_connections_per_host)
             .build(https);
@@ -105,13 +105,12 @@ impl SubgraphExecutorMap {
         Ok(subgraph_executor_map)
     }
 
-    pub async fn execute<'a, 'req>(
+    pub async fn execute<'exec, 'req>(
         &self,
         subgraph_name: &str,
-        execution_request: HttpExecutionRequest<'a>,
-        client_request: &ClientRequestDetails<'a, 'req>,
+        execution_request: HttpExecutionRequest<'exec, 'req>,
     ) -> HttpExecutionResponse {
-        match self.get_or_create_executor(subgraph_name, client_request) {
+        match self.get_or_create_executor(subgraph_name, execution_request.client_request) {
             Ok(Some(executor)) => executor.execute(execution_request).await,
             Err(err) => {
                 error!(
@@ -295,13 +294,33 @@ impl SubgraphExecutorMap {
             .or_insert_with(|| Arc::new(Semaphore::new(self.max_connections_per_host)))
             .clone();
 
+        let mut client = self.client.clone();
+        let mut dedupe_enabled = self.config.traffic_shaping.all.dedupe_enabled;
+        let mut timeout = self.config.traffic_shaping.all.request_timeout.clone();
+        if let Some(subgraph_traffic_shaping_config) =
+            self.config.traffic_shaping.subgraphs.get(subgraph_name)
+        {
+            client = Arc::new(
+                Client::builder(TokioExecutor::new())
+                    .pool_timer(TokioTimer::new())
+                    .pool_idle_timeout(Duration::from_secs(
+                        subgraph_traffic_shaping_config.pool_idle_timeout_seconds,
+                    ))
+                    .pool_max_idle_per_host(self.max_connections_per_host)
+                    .build(HttpsConnector::new()),
+            );
+            dedupe_enabled = subgraph_traffic_shaping_config.dedupe_enabled;
+            timeout = subgraph_traffic_shaping_config.request_timeout.clone();
+        }
+
         let executor = HTTPSubgraphExecutor::new(
             subgraph_name.to_string(),
             endpoint_uri,
-            self.client.clone(),
+            client,
             semaphore,
-            self.config.clone(),
+            dedupe_enabled,
             self.in_flight_requests.clone(),
+            timeout,
         );
 
         self.executors_by_subgraph
diff --git a/lib/router-config/src/traffic_shaping.rs b/lib/router-config/src/traffic_shaping.rs

Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,10 @@ pub enum SubgraphExecutorError {`
`20`	`20`	`RequestFailure(String, String),`
`21`	`21`	`#[error("Failed to serialize variable \"{0}\": {1}")]`
`22`	`22`	`VariablesSerializationFailure(String, String),`
	`23`	`+ #[error("Failed to resolve VRL expression for timeout. Runtime error: {0}")]`
	`24`	`+ TimeoutExpressionResolutionFailure(String),`
	`25`	`+ #[error("Request to subgraph \"{0}\" timed out after {1} milliseconds")]`
	`26`	`+ RequestTimeout(String, u64),`
`23`	`27`	`}`
`24`	`28`
`25`	`29`	`impl From<SubgraphExecutorError> for GraphQLError {`
`@@ -61,6 +65,10 @@ impl SubgraphExecutorError {`
`61`	`65`	`SubgraphExecutorError::VariablesSerializationFailure(_, _) => {`
`62`	`66`	`"SUBGRAPH_VARIABLES_SERIALIZATION_FAILURE"`
`63`	`67`	`}`
	`68`	`+ SubgraphExecutorError::TimeoutExpressionResolutionFailure(_) => {`
	`69`	`+ "SUBGRAPH_TIMEOUT_EXPRESSION_RESOLUTION_FAILURE"`
	`70`	`+ }`
	`71`	`+ SubgraphExecutorError::RequestTimeout(_, _) => "SUBGRAPH_REQUEST_TIMEOUT",`
`64`	`72`	`}`
`65`	`73`	`}`
`66`	`74`	`}`