feat(router): Subgraph Timeout Configuration

Author: ardatan

Cargo.lock

@@ -15,7 +15,7 @@
 |[**override\_subgraph\_urls**](#override_subgraph_urls)|`object`|Configuration for overriding subgraph URLs.<br/>Default: `{}`<br/>||
 |[**query\_planner**](#query_planner)|`object`|Query planning configuration.<br/>Default: `{"allow_expose":false,"timeout":"10s"}`<br/>||
 |[**supergraph**](#supergraph)|`object`|Configuration for the Federation supergraph source. By default, the router will use a local file-based supergraph source (`./supergraph.graphql`).<br/>||
-|[**traffic\_shaping**](#traffic_shaping)|`object`|Configuration for the traffic-shaping of the executor. Use these configurations to control how requests are being executed to subgraphs.<br/>Default: `{"dedupe_enabled":true,"max_connections_per_host":100,"pool_idle_timeout":"50s"}`<br/>||
+|[**traffic\_shaping**](#traffic_shaping)|`object`|Configuration for the traffic-shaping of the executor. Use these configurations to control how requests are being executed to subgraphs.<br/>Default: `{"all":{"dedupe_enabled":true,"pool_idle_timeout":"50s","request_timeout":{"Duration":"15s"}},"max_connections_per_host":100}`<br/>||
 
 **Additional Properties:** not allowed  
 **Example**
@@ -107,9 +107,12 @@ query_planner:
   timeout: 10s
 supergraph: {}
 traffic_shaping:
-  dedupe_enabled: true
+  all:
+    dedupe_enabled: true
+    pool_idle_timeout: 50s
+    request_timeout:
+      Duration: 15s
   max_connections_per_host: 100
-  pool_idle_timeout: 50s
 
 ```
 
@@ -1815,18 +1818,70 @@ Configuration for the traffic-shaping of the executor. Use these configurations
 
 |Name|Type|Description|Required|
 |----|----|-----------|--------|
-|**dedupe\_enabled**|`boolean`|Enables/disables request deduplication to subgraphs.<br/><br/>When requests exactly matches the hashing mechanism (e.g., subgraph name, URL, headers, query, variables), and are executed at the same time, they will<br/>be deduplicated by sharing the response of other in-flight requests.<br/>Default: `true`<br/>||
+|[**all**](#traffic_shapingall)|`object`|The default configuration that will be applied to all subgraphs, unless overridden by a specific subgraph configuration.<br/>Default: `{"dedupe_enabled":true,"pool_idle_timeout":"50s","request_timeout":{"Duration":"15s"}}`<br/>||
 |**max\_connections\_per\_host**|`integer`|Limits the concurrent amount of requests/connections per host/subgraph.<br/>Default: `100`<br/>Format: `"uint"`<br/>Minimum: `0`<br/>||
+|[**subgraphs**](#traffic_shapingsubgraphs)|`object`|Optional per-subgraph configurations that will override the default configuration for specific subgraphs.<br/>||
+
+**Additional Properties:** not allowed  
+**Example**
+
+```yaml
+all:
+  dedupe_enabled: true
+  pool_idle_timeout: 50s
+  request_timeout:
+    Duration: 15s
+max_connections_per_host: 100
+
+```
+
+<a name="traffic_shapingall"></a>
+### traffic\_shaping\.all: object
+
+The default configuration that will be applied to all subgraphs, unless overridden by a specific subgraph configuration.
+
+
+**Properties**
+
+|Name|Type|Description|Required|
+|----|----|-----------|--------|
+|**dedupe\_enabled**|`boolean`|Enables/disables request deduplication to subgraphs.<br/><br/>When requests exactly matches the hashing mechanism (e.g., subgraph name, URL, headers, query, variables), and are executed at the same time, they will<br/>be deduplicated by sharing the response of other in-flight requests.<br/>Default: `true`<br/>||
 |**pool\_idle\_timeout**|`string`|Timeout for idle sockets being kept-alive.<br/>Default: `"50s"`<br/>||
+|**request\_timeout**||Optional timeout configuration for requests to subgraphs.<br/><br/>Example with a fixed duration:<br/>```yaml<br/>  timeout:<br/>    duration: 5s<br/>```<br/><br/>Or with a VRL expression that can return a duration based on the operation kind:<br/>```yaml<br/>  timeout:<br/>    expression: \|<br/>     if (.request.operation.type == "mutation") {<br/>       "10s"<br/>     } else {<br/>       "15s"<br/>     }<br/>```<br/>Default: `{"Duration":"15s"}`<br/>||
 
 **Additional Properties:** not allowed  
 **Example**
 
 ```yaml
 dedupe_enabled: true
-max_connections_per_host: 100
 pool_idle_timeout: 50s
+request_timeout:
+  Duration: 15s
 
 ```
 
+<a name="traffic_shapingsubgraphs"></a>
+### traffic\_shaping\.subgraphs: object
+
+Optional per-subgraph configurations that will override the default configuration for specific subgraphs.
+
+
+**Additional Properties**
+
+|Name|Type|Description|Required|
+|----|----|-----------|--------|
+|[**Additional Properties**](#traffic_shapingsubgraphsadditionalproperties)|`object`||yes|
+
+<a name="traffic_shapingsubgraphsadditionalproperties"></a>
+#### traffic\_shaping\.subgraphs\.additionalProperties: object
+
+**Properties**
+
+|Name|Type|Description|Required|
+|----|----|-----------|--------|
+|**dedupe\_enabled**|`boolean`, `null`|Enables/disables request deduplication to subgraphs.<br/><br/>When requests exactly matches the hashing mechanism (e.g., subgraph name, URL, headers, query, variables), and are executed at the same time, they will<br/>be deduplicated by sharing the response of other in-flight requests.<br/>|no|
+|**pool\_idle\_timeout\_seconds**|`string`|Timeout for idle sockets being kept-alive.<br/>|yes|
+|**request\_timeout**||Optional timeout configuration for requests to subgraphs.<br/><br/>Example with a fixed duration:<br/>```yaml<br/>  timeout:<br/>    duration: 5s<br/>```<br/><br/>Or with a VRL expression that can return a duration based on the operation kind:<br/>```yaml<br/>  timeout:<br/>    expression: \|<br/>     if (.request.operation.type == "mutation") {<br/>       "10s"<br/>     } else {<br/>       "15s"<br/>     }<br/>```<br/>|no|
+
+**Additional Properties:** not allowed  
 

docs/README.md

@@ -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"

lib/executor/Cargo.toml

@@ -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(),
         }))

lib/executor/src/execution/plan.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)

lib/executor/src/executors/common.rs

@@ -20,6 +20,12 @@ pub enum SubgraphExecutorError {
     RequestFailure(String, String),
     #[error("Failed to serialize variable \"{0}\": {1}")]
     VariablesSerializationFailure(String, String),
+    #[error("Failed to compile VRL expression for timeout for subgraph '{0}'. Please check your VRL expression for syntax errors. Diagnostic: {1}")]
+    RequestTimeoutExpressionBuild(String, String),
+    #[error("Failed to resolve VRL expression for timeout for subgraph '{0}'. Runtime error: {1}")]
+    TimeoutExpressionResolution(String, String),
+    #[error("Request to subgraph \"{0}\" timed out after {1} milliseconds")]
+    RequestTimeout(String, u64),
 }
 
 impl From<SubgraphExecutorError> for GraphQLError {
@@ -61,6 +67,13 @@ impl SubgraphExecutorError {
             SubgraphExecutorError::VariablesSerializationFailure(_, _) => {
                 "SUBGRAPH_VARIABLES_SERIALIZATION_FAILURE"
             }
+            SubgraphExecutorError::TimeoutExpressionResolution(_, _) => {
+                "SUBGRAPH_TIMEOUT_EXPRESSION_RESOLUTION_FAILURE"
+            }
+            SubgraphExecutorError::RequestTimeout(_, _) => "SUBGRAPH_REQUEST_TIMEOUT",
+            SubgraphExecutorError::RequestTimeoutExpressionBuild(_, _) => {
+                "SUBGRAPH_TIMEOUT_EXPRESSION_BUILD_FAILURE"
+            }
         }
     }
 }