Timeout

Author: ardatan

Cargo.lock

@@ -708,7 +708,6 @@ 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 {
@@ -723,7 +722,7 @@ impl<'exec, 'req> Executor<'exec, 'req> {
             subgraph_name: node.service_name.clone(),
             response: self
                 .executors
-                .execute(&node.service_name, subgraph_request)
+                .execute(&node.service_name, subgraph_request, self.client_request)
                 .await
                 .into(),
         }))

lib/executor/src/execution/plan.rs

@@ -5,13 +5,11 @@ 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<'exec, 'req>(
+    async fn execute<'a>(
         &self,
-        execution_request: HttpExecutionRequest<'exec, 'req>,
+        execution_request: HttpExecutionRequest<'a>,
     ) -> HttpExecutionResponse;
 
     fn to_boxed_arc<'a>(self) -> Arc<Box<dyn SubgraphExecutor + Send + Sync + 'a>>
@@ -28,19 +26,18 @@ pub type SubgraphExecutorBoxedArc = Arc<Box<SubgraphExecutorType>>;
 
 pub type SubgraphRequestExtensions = HashMap<String, Value>;
 
-pub struct HttpExecutionRequest<'exec, 'req> {
-    pub query: &'exec str,
+pub struct HttpExecutionRequest<'a> {
+    pub query: &'a str,
     pub dedupe: bool,
-    pub operation_name: Option<&'exec str>,
+    pub operation_name: Option<&'a str>,
     // TODO: variables could be stringified before even executing the request
-    pub variables: Option<HashMap<&'exec str, &'exec sonic_rs::Value>>,
+    pub variables: Option<HashMap<&'a str, &'a 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

@@ -1,11 +1,7 @@
-use std::collections::BTreeMap;
 use std::sync::Arc;
-use std::time::Duration;
 
-use crate::execution::client_request_details::ClientRequestDetails;
 use crate::executors::common::HttpExecutionResponse;
 use crate::executors::dedupe::{request_fingerprint, ABuildHasher, SharedResponse};
-use crate::utils::expression::execute_expression_with_value;
 use dashmap::DashMap;
 use tokio::sync::OnceCell;
 
@@ -21,8 +17,6 @@ use hyper_tls::HttpsConnector;
 use hyper_util::client::legacy::{connect::HttpConnector, Client};
 use tokio::sync::Semaphore;
 use tracing::debug;
-use vrl::compiler::Program as VrlProgram;
-use vrl::core::Value as VrlValue;
 
 use crate::executors::common::HttpExecutionRequest;
 use crate::executors::error::SubgraphExecutorError;
@@ -33,12 +27,6 @@ use crate::utils::consts::COMMA;
 use crate::utils::consts::QUOTE;
 use crate::{executors::common::SubgraphExecutor, json_writer::write_and_escape_string};
 
-#[derive(Debug)]
-pub enum DurationOrProgram {
-    Duration(Duration),
-    Program(Box<VrlProgram>),
-}
-
 #[derive(Debug)]
 pub struct HTTPSubgraphExecutor {
     pub subgraph_name: String,
@@ -48,7 +36,6 @@ pub struct HTTPSubgraphExecutor {
     pub semaphore: Arc<Semaphore>,
     pub dedupe_enabled: bool,
     pub in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
-    pub timeout: DurationOrProgram,
 }
 
 const FIRST_VARIABLE_STR: &[u8] = b",\"variables\":{";
@@ -64,7 +51,6 @@ impl HTTPSubgraphExecutor {
         semaphore: Arc<Semaphore>,
         dedupe_enabled: bool,
         in_flight_requests: Arc<DashMap<u64, Arc<OnceCell<SharedResponse>>, ABuildHasher>>,
-        timeout: DurationOrProgram,
     ) -> Self {
         let mut header_map = HeaderMap::new();
         header_map.insert(
@@ -84,13 +70,12 @@ impl HTTPSubgraphExecutor {
             semaphore,
             dedupe_enabled,
             in_flight_requests,
-            timeout,
         }
     }
 
     fn build_request_body(
         &self,
-        execution_request: &HttpExecutionRequest<'_, '_>,
+        execution_request: &HttpExecutionRequest<'_>,
     ) -> Result<Vec<u8>, SubgraphExecutorError> {
         let mut body = Vec::with_capacity(4096);
         body.put(FIRST_QUOTE_STR);
@@ -151,7 +136,6 @@ impl HTTPSubgraphExecutor {
         &self,
         body: Vec<u8>,
         headers: HeaderMap,
-        client_request: &ClientRequestDetails<'_, '_>,
     ) -> Result<SharedResponse, SubgraphExecutorError> {
         let mut req = hyper::Request::builder()
             .method(http::Method::POST)
@@ -166,75 +150,9 @@ impl HTTPSubgraphExecutor {
 
         debug!("making http request to {}", self.endpoint.to_string());
 
-        let timeout = match &self.timeout {
-            DurationOrProgram::Duration(dur) => *dur,
-            DurationOrProgram::Program(program) => {
-                let value =
-                    VrlValue::Object(BTreeMap::from([("request".into(), client_request.into())]));
-                let result = execute_expression_with_value(program, value).map_err(|err| {
-                    SubgraphExecutorError::TimeoutExpressionResolution(
-                        self.subgraph_name.to_string(),
-                        err.to_string(),
-                    )
-                })?;
-                match result {
-                    VrlValue::Integer(i) => {
-                        if i < 0 {
-                            return Err(SubgraphExecutorError::TimeoutExpressionResolution(
-                                self.subgraph_name.to_string(),
-                                "Timeout expression resolved to a negative integer".to_string(),
-                            ));
-                        }
-                        std::time::Duration::from_millis(i as u64)
-                    }
-                    VrlValue::Float(f) => {
-                        let f = f.into_inner();
-                        if f < 0.0 {
-                            return Err(SubgraphExecutorError::TimeoutExpressionResolution(
-                                self.subgraph_name.to_string(),
-                                "Timeout expression resolved to a negative float".to_string(),
-                            ));
-                        }
-                        std::time::Duration::from_millis(f as u64)
-                    }
-                    VrlValue::Bytes(b) => {
-                        let s = std::str::from_utf8(&b).map_err(|e| {
-                            SubgraphExecutorError::TimeoutExpressionResolution(
-                                self.subgraph_name.to_string(),
-                                format!("Failed to parse duration string from bytes: {}", e),
-                            )
-                        })?;
-                        humantime::parse_duration(s).map_err(|e| {
-                            SubgraphExecutorError::TimeoutExpressionResolution(
-                                self.subgraph_name.to_string(),
-                                format!("Failed to parse duration string '{}': {}", s, e),
-                            )
-                        })?
-                    }
-                    other => {
-                        return Err(SubgraphExecutorError::TimeoutExpressionResolution(
-                                self.subgraph_name.to_string(),
-                            format!(
-                                "Timeout expression resolved to an unexpected type: {}. Expected a non-negative integer/float (ms) or a duration string.",
-                                other.kind()
-                            ),
-                        ));
-                    }
-                }
-            }
-        };
-
-        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())
-            })?;
+        let res = self.http_client.request(req).await.map_err(|e| {
+            SubgraphExecutorError::RequestFailure(self.endpoint.to_string(), e.to_string())
+        })?;
 
         debug!(
             "http request to {} completed, status: {}",
@@ -291,9 +209,9 @@ impl HTTPSubgraphExecutor {
 #[async_trait]
 impl SubgraphExecutor for HTTPSubgraphExecutor {
     #[tracing::instrument(skip_all, fields(subgraph_name = %self.subgraph_name))]
-    async fn execute<'exec, 'req>(
+    async fn execute<'a>(
         &self,
-        execution_request: HttpExecutionRequest<'exec, 'req>,
+        execution_request: HttpExecutionRequest<'a>,
     ) -> HttpExecutionResponse {
         let body = match self.build_request_body(&execution_request) {
             Ok(body) => body,
@@ -315,10 +233,7 @@ 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();
-            return match self
-                ._send_request(body, headers, execution_request.client_request)
-                .await
-            {
+            return match self._send_request(body, headers).await {
                 Ok(shared_response) => HttpExecutionResponse {
                     body: shared_response.body,
                     headers: shared_response.headers,
@@ -350,8 +265,7 @@ 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, execution_request.client_request)
-                        .await
+                    self._send_request(body, headers).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.