e2e tests, fix config and adjust naming

kamilkisiela · kamilkisiela · commit a0aa887abd45 · 2025-11-25T13:07:35.000+01:00
diff --git a/docs/README.md b/docs/README.md
@@ -1871,7 +1871,7 @@ Optional per-subgraph configurations that will override the default configuratio
 
 |Name|Type|Description|Required|
 |----|----|-----------|--------|
-|[**Additional Properties**](#traffic_shapingsubgraphsadditionalproperties)|`object`||yes|
+|[**Additional Properties**](#traffic_shapingsubgraphsadditionalproperties)|`object`|||
 
 <a name="traffic_shapingsubgraphsadditionalproperties"></a>
 #### traffic\_shaping\.subgraphs\.additionalProperties: object
@@ -1880,9 +1880,9 @@ Optional per-subgraph configurations that will override the default configuratio
 
 |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**|`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|
+|**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/>||
+|**pool\_idle\_timeout**|`string`, `null`|Timeout for idle sockets being kept-alive.<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/>||
 
 **Additional Properties:** not allowed  
 
diff --git a/e2e/configs/timeout_per_subgraph_dynamic.router.yaml b/e2e/configs/timeout_per_subgraph_dynamic.router.yaml
@@ -0,0 +1,18 @@
+# yaml-language-server: $schema=../../router-config.schema.json
+supergraph:
+  source: file
+  path: ../supergraph.graphql
+traffic_shaping:
+  all:
+    request_timeout: 2s
+    # Disable deduplication to better hunt for deadlocks in tests
+    dedupe_enabled: false
+  subgraphs:
+    accounts:
+      request_timeout:
+        expression: |
+          if (.request.headers["x-timeout"] == "short") {
+            "10s"
+          } else {
+            "15s"
+          }
diff --git a/e2e/configs/timeout_per_subgraph_static.router.yaml b/e2e/configs/timeout_per_subgraph_static.router.yaml
@@ -0,0 +1,12 @@
+# yaml-language-server: $schema=../../router-config.schema.json
+supergraph:
+  source: file
+  path: ../supergraph.graphql
+traffic_shaping:
+  all:
+    request_timeout: 2s
+    # Disable deduplication to better hunt for deadlocks in tests
+    dedupe_enabled: false
+  subgraphs:
+    accounts:
+      request_timeout: 5s
diff --git a/e2e/src/lib.rs b/e2e/src/lib.rs
@@ -16,3 +16,5 @@ mod probes;
 mod supergraph;
 #[cfg(test)]
 mod testkit;
+#[cfg(test)]
+mod timeout_per_subgraph;
diff --git a/e2e/src/timeout_per_subgraph.rs b/e2e/src/timeout_per_subgraph.rs
@@ -0,0 +1,125 @@
+#[cfg(test)]
+mod override_subgraph_urls_e2e_tests {
+    use ntex::web::test;
+    use sonic_rs::{from_slice, json, Value};
+
+    use crate::testkit::{
+        init_graphql_request, init_router_from_config_file, wait_for_readiness, SubgraphsServer,
+    };
+
+    #[ntex::test]
+    async fn should_not_deadlock_when_overriding_subgraph_timeout_statically() {
+        let _subgraphs_server = SubgraphsServer::start().await;
+        let app = init_router_from_config_file("configs/timeout_per_subgraph_static.router.yaml")
+            .await
+            .unwrap();
+        wait_for_readiness(&app.app).await;
+
+        let req1 = init_graphql_request("{ users { id } }", None);
+        let req2 = init_graphql_request("{ users { id } }", None);
+        let req3 = init_graphql_request("{ users { id } }", None);
+        let req4 = init_graphql_request("{ users { id } }", None);
+
+        let (resp1, resp2, resp3, resp4) = tokio::join!(
+            test::call_service(&app.app, req1.to_request()),
+            test::call_service(&app.app, req2.to_request()),
+            test::call_service(&app.app, req3.to_request()),
+            test::call_service(&app.app, req4.to_request())
+        );
+
+        assert!(resp1.status().is_success(), "Expected 200 OK");
+        assert!(resp2.status().is_success(), "Expected 200 OK");
+        assert!(resp3.status().is_success(), "Expected 200 OK");
+        assert!(resp4.status().is_success(), "Expected 200 OK");
+
+        let expected_json = json!({
+          "data": {
+            "users": [
+              {
+                "id": "1"
+              },
+              {
+                "id": "2"
+              },
+              {
+                "id": "3"
+              },
+              {
+                "id": "4"
+              },
+              {
+                "id": "5"
+              },
+              {
+                "id": "6"
+              }
+            ]
+          }
+        });
+
+        for resp in [resp1, resp2, resp3, resp4] {
+            let body = test::read_body(resp).await;
+            let json_body: Value = from_slice(&body).unwrap();
+            assert_eq!(json_body, expected_json);
+        }
+    }
+
+    #[ntex::test]
+    async fn should_not_deadlock_when_overriding_subgraph_timeout_dynamically() {
+        let _subgraphs_server = SubgraphsServer::start().await;
+        let app = init_router_from_config_file("configs/timeout_per_subgraph_dynamic.router.yaml")
+            .await
+            .unwrap();
+        wait_for_readiness(&app.app).await;
+
+        // We want to ensure that concurrent requests with different timeout settings
+        // do not cause deadlocks. We are not testing the actual timeout duration here.
+        let req1 = init_graphql_request("{ users { id } }", None).header("x-timeout", "short");
+        let req2 = init_graphql_request("{ users { id } }", None).header("x-timeout", "long");
+        let req3 = init_graphql_request("{ users { id } }", None).header("x-timeout", "short");
+        let req4 = init_graphql_request("{ users { id } }", None).header("x-timeout", "long");
+
+        let (resp1, resp2, resp3, resp4) = tokio::join!(
+            test::call_service(&app.app, req1.to_request()),
+            test::call_service(&app.app, req2.to_request()),
+            test::call_service(&app.app, req3.to_request()),
+            test::call_service(&app.app, req4.to_request())
+        );
+
+        assert!(resp1.status().is_success(), "Expected 200 OK");
+        assert!(resp2.status().is_success(), "Expected 200 OK");
+        assert!(resp3.status().is_success(), "Expected 200 OK");
+        assert!(resp4.status().is_success(), "Expected 200 OK");
+
+        let expected_json = json!({
+          "data": {
+            "users": [
+              {
+                "id": "1"
+              },
+              {
+                "id": "2"
+              },
+              {
+                "id": "3"
+              },
+              {
+                "id": "4"
+              },
+              {
+                "id": "5"
+              },
+              {
+                "id": "6"
+              }
+            ]
+          }
+        });
+
+        for resp in [resp1, resp2, resp3, resp4] {
+            let body = test::read_body(resp).await;
+            let json_body: Value = from_slice(&body).unwrap();
+            assert_eq!(json_body, expected_json);
+        }
+    }
+}
diff --git a/lib/executor/src/executors/map.rs b/lib/executor/src/executors/map.rs
@@ -41,7 +41,7 @@ type SubgraphEndpoint = String;
 type ExecutorsBySubgraphMap =
     DashMap<SubgraphName, DashMap<SubgraphEndpoint, SubgraphExecutorBoxedArc>>;
 type EndpointsBySubgraphMap = DashMap<SubgraphName, SubgraphEndpoint>;
-type ExpressionsBySubgraphMap = HashMap<SubgraphName, VrlProgram>;
+type EndpointExpressionsBySubgraphMap = HashMap<SubgraphName, VrlProgram>;
 type TimeoutsBySubgraph = DashMap<SubgraphName, DurationOrProgram>;
 
 struct ResolvedSubgraphConfig<'a> {
@@ -55,8 +55,8 @@ pub struct SubgraphExecutorMap {
     /// Mapping from subgraph name to endpoint for quick lookup
     /// based on supergrah sdl and static overrides from router's config.
     static_endpoints_by_subgraph: EndpointsBySubgraphMap,
-    /// Mapping from subgraph name to VRL expression program
-    expressions_by_subgraph: ExpressionsBySubgraphMap,
+    /// Mapping from subgraph name to VRL expression program for endpoint resolution.
+    endpoint_expressions_by_subgraph: EndpointExpressionsBySubgraphMap,
     timeouts_by_subgraph: TimeoutsBySubgraph,
     config: Arc<HiveRouterConfig>,
     client: Arc<HttpClient>,
@@ -79,7 +79,7 @@ impl SubgraphExecutorMap {
         SubgraphExecutorMap {
             executors_by_subgraph: Default::default(),
             static_endpoints_by_subgraph: Default::default(),
-            expressions_by_subgraph: Default::default(),
+            endpoint_expressions_by_subgraph: Default::default(),
             config,
             client: Arc::new(client),
             semaphores_by_origin: Default::default(),
@@ -103,7 +103,8 @@ impl SubgraphExecutorMap {
             let endpoint_str = match endpoint_str {
                 Some(UrlOrExpression::Url(url)) => url,
                 Some(UrlOrExpression::Expression { expression }) => {
-                    subgraph_executor_map.register_expression(&subgraph_name, expression)?;
+                    subgraph_executor_map
+                        .register_endpoint_expression(&subgraph_name, expression)?;
                     &original_endpoint_str
                 }
                 None => &original_endpoint_str,
@@ -202,7 +203,7 @@ impl SubgraphExecutorMap {
         subgraph_name: &str,
         client_request: &ClientRequestDetails<'_, '_>,
     ) -> Result<Option<SubgraphExecutorBoxedArc>, SubgraphExecutorError> {
-        if let Some(expression) = self.expressions_by_subgraph.get(subgraph_name) {
+        if let Some(expression) = self.endpoint_expressions_by_subgraph.get(subgraph_name) {
             let original_url_value = VrlValue::Bytes(Bytes::from(
                 self.static_endpoints_by_subgraph
                     .get(subgraph_name)
@@ -274,15 +275,15 @@ impl SubgraphExecutorMap {
     /// Registers a VRL expression for the given subgraph name.
     /// The expression can later be used to resolve the endpoint URL cheaply,
     /// without needing to recompile it every time.
-    fn register_expression(
+    fn register_endpoint_expression(
         &mut self,
         subgraph_name: &str,
         expression: &str,
     ) -> Result<(), SubgraphExecutorError> {
         let program = compile_expression(expression, None).map_err(|err| {
             SubgraphExecutorError::EndpointExpressionBuild(subgraph_name.to_string(), err)
         })?;
-        self.expressions_by_subgraph
+        self.endpoint_expressions_by_subgraph
             .insert(subgraph_name.to_string(), program);
 
         Ok(())
diff --git a/lib/router-config/src/traffic_shaping.rs b/lib/router-config/src/traffic_shaping.rs
@@ -45,9 +45,11 @@ pub struct TrafficShapingExecutorSubgraphConfig {
     /// Timeout for idle sockets being kept-alive.
     #[serde(
         deserialize_with = "humantime_serde::deserialize",
-        serialize_with = "humantime_serde::serialize"
+        serialize_with = "humantime_serde::serialize",
+        skip_serializing_if = "Option::is_none",
+        default = "default_subgraph_pool_idle_timeout"
     )]
-    #[schemars(with = "String")]
+    #[schemars(with = "Option<String>")]
     pub pool_idle_timeout: Option<Duration>,
 
     /// Enables/disables request deduplication to subgraphs.
@@ -118,6 +120,10 @@ pub struct TrafficShapingExecutorGlobalConfig {
     pub request_timeout: DurationOrExpression,
 }
 
+fn default_subgraph_pool_idle_timeout() -> Option<Duration> {
+    None
+}
+
 fn default_request_timeout() -> DurationOrExpression {
     DurationOrExpression::Duration(Duration::from_secs(30))
 }
