Configure HTTP/2 timeouts via environment

crates/pulsing-actor/src/connect/mod.rs

@@ -55,13 +55,32 @@ impl PulsingConnect {
 
     /// Connect with multiple gateway addresses for failover.
     pub async fn connect_multi(gateway_addrs: Vec<SocketAddr>) -> Result<Arc<Self>> {
+        let config = Http2Config::from_env()?;
+        Self::connect_multi_with_config(gateway_addrs, config).await
+    }
+
+    /// Connect to a single gateway with an explicit HTTP/2 configuration.
+    pub async fn connect_with_config(
+        gateway_addr: SocketAddr,
+        config: Http2Config,
+    ) -> Result<Arc<Self>> {
+        Self::connect_multi_with_config(vec![gateway_addr], config).await
+    }
+
+    /// Connect to multiple gateways with an explicit HTTP/2 configuration.
+    ///
+    /// The explicit configuration takes precedence over timeout environment variables.
+    pub async fn connect_multi_with_config(
+        gateway_addrs: Vec<SocketAddr>,
+        config: Http2Config,
+    ) -> Result<Arc<Self>> {
         if gateway_addrs.is_empty() {
             return Err(PulsingError::from(RuntimeError::Other(
                 "At least one gateway address is required".into(),
             )));
         }
 
-        let http_client = Arc::new(Http2Client::new(Http2Config::default()));
+        let http_client = Arc::new(Http2Client::new(config));
         http_client.start_background_tasks();
 
         let active = gateway_addrs[0];

crates/pulsing-actor/src/system/config.rs

@@ -84,6 +84,14 @@ impl SystemConfig {
         Self::default()
     }
 
+    /// Create a standalone config using HTTP/2 timeout environment variables.
+    pub fn from_env() -> Result<Self> {
+        Ok(Self {
+            http2_config: Http2Config::from_env()?,
+            ..Self::default()
+        })
+    }
+
     /// Create config with specific address
     pub fn with_addr(addr: SocketAddr) -> Self {
         Self {
@@ -92,6 +100,15 @@ impl SystemConfig {
         }
     }
 
+    /// Create a config with a specific address and HTTP/2 timeout environment variables.
+    pub fn with_addr_from_env(addr: SocketAddr) -> Result<Self> {
+        Ok(Self {
+            addr,
+            http2_config: Http2Config::from_env()?,
+            ..Self::default()
+        })
+    }
+
     /// Add seed nodes for cluster joining
     pub fn with_seeds(mut self, seeds: Vec<SocketAddr>) -> Self {
         self.seed_nodes = seeds;
@@ -273,13 +290,22 @@ impl ActorSystemBuilder {
         self
     }
 
+    /// Set an explicit HTTP/2 configuration.
+    ///
+    /// Explicit configuration takes precedence over timeout environment variables.
+    pub fn http2_config(mut self, config: Http2Config) -> Self {
+        self.http2_config = Some(config);
+        self
+    }
+
     /// Enable TLS with passphrase
     #[cfg(feature = "tls")]
     pub fn tls(mut self, passphrase: &str) -> Result<Self> {
-        let http2_config = self
-            .http2_config
-            .take()
-            .unwrap_or_default()
+        let base_config = match self.http2_config.take() {
+            Some(config) => config,
+            None => Http2Config::from_env()?,
+        };
+        let http2_config = base_config
             .with_tls(passphrase)
             .map_err(|e| PulsingError::from(RuntimeError::Other(e.to_string())))?;
         self.http2_config = Some(http2_config);
@@ -326,11 +352,16 @@ impl ActorSystemBuilder {
 
         let head_addr = Self::parse_optional_addr("head node address", self.head_addr)?;
 
+        let http2_config = match self.http2_config {
+            Some(config) => config,
+            None => Http2Config::from_env()?,
+        };
+
         let config = SystemConfig {
             addr,
             seed_nodes,
             gossip_config: self.gossip_config.unwrap_or_default(),
-            http2_config: self.http2_config.unwrap_or_default(),
+            http2_config,
             default_mailbox_capacity: self.mailbox_capacity.unwrap_or(DEFAULT_MAILBOX_SIZE),
             is_head_node: self.is_head_node,
             head_addr,

crates/pulsing-actor/src/transport/http2/client.rs

@@ -20,12 +20,12 @@ use crate::transport::tensor::{
     TensorTransportRouter,
 };
 use bytes::Bytes;
-use futures::{Stream, StreamExt, TryStreamExt};
+use futures::{Stream, StreamExt};
 use http_body_util::{BodyExt, Full, Limited, StreamBody};
 use hyper::body::{Frame, Incoming};
 use hyper::{Method, Request};
 use opentelemetry::Context;
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::convert::Infallible;
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -58,6 +58,13 @@ pub trait FaultInjector: Send + Sync {
     fn inject(&self, ctx: &FaultInjectContext) -> Option<PulsingError>;
 }
 
+fn stream_error_to_pulsing(error: anyhow::Error) -> PulsingError {
+    match error.downcast::<PulsingError>() {
+        Ok(error) => error,
+        Err(error) => PulsingError::from(RuntimeError::Other(error.to_string())),
+    }
+}
+
 /// HTTP/2 client with connection pooling, retry, and timeout support.
 pub struct Http2Client {
     pool: Arc<ConnectionPool>,
@@ -299,7 +306,7 @@ impl Http2Client {
                     Ok(None) => None,
                     Err(e) => Some(Err(e)),
                 },
-                Err(e) => Some(Err(PulsingError::from(RuntimeError::Other(e.to_string())))),
+                Err(e) => Some(Err(stream_error_to_pulsing(e))),
             }
         });
 
@@ -742,7 +749,7 @@ impl Http2Client {
                         Ok(None) => None,
                         Err(e) => Some(Err(e)),
                     },
-                    Err(e) => Some(Err(PulsingError::from(RuntimeError::Other(e.to_string())))),
+                    Err(e) => Some(Err(stream_error_to_pulsing(e))),
                 }
             });
 
@@ -781,36 +788,80 @@ impl Http2Client {
         cancel: CancellationToken,
         timeout: Duration,
     ) -> impl Stream<Item = anyhow::Result<StreamFrame>> {
-        let parser = Arc::new(Mutex::new(BinaryFrameParser::new()));
-        let start = std::time::Instant::now();
-
-        http_body_util::BodyStream::new(body)
-            .take_while(move |_| {
-                let cancelled = cancel.is_cancelled();
-                let timed_out = start.elapsed() > timeout;
-                async move { !cancelled && !timed_out }
-            })
-            .map(move |result| {
-                let parser = parser.clone();
-                async move {
-                    let frame = result.map_err(|e| anyhow::anyhow!("Body read: {}", e))?;
-                    let data = frame
-                        .into_data()
-                        .map_err(|_| anyhow::anyhow!("Not data frame"))?;
-
-                    let mut parser = parser.lock().await;
-                    parser.push(&data);
-
-                    let frames = parser
-                        .parse_all()
-                        .into_iter()
-                        .map(|r| r.map_err(|e| anyhow::anyhow!("{}", e)))
-                        .collect::<Vec<_>>();
-                    Ok::<_, anyhow::Error>(futures::stream::iter(frames))
+        struct BodyFrameState {
+            body: http_body_util::BodyStream<Incoming>,
+            parser: BinaryFrameParser,
+            pending: VecDeque<anyhow::Result<StreamFrame>>,
+            cancel: CancellationToken,
+            deadline: tokio::time::Instant,
+            timeout_ms: u64,
+            finished: bool,
+        }
+
+        let state = BodyFrameState {
+            body: http_body_util::BodyStream::new(body),
+            parser: BinaryFrameParser::new(),
+            pending: VecDeque::new(),
+            cancel,
+            deadline: tokio::time::Instant::now() + timeout,
+            timeout_ms: timeout.as_millis() as u64,
+            finished: false,
+        };
+
+        futures::stream::unfold(state, |mut state| async move {
+            loop {
+                if state.finished {
+                    return None;
                 }
-            })
-            .buffer_unordered(1)
-            .try_flatten()
+
+                if let Some(frame) = state.pending.pop_front() {
+                    return Some((frame, state));
+                }
+
+                tokio::select! {
+                    _ = state.cancel.cancelled() => {
+                        return None;
+                    }
+                    _ = tokio::time::sleep_until(state.deadline) => {
+                        state.finished = true;
+                        let error = PulsingError::from(RuntimeError::request_timeout(
+                            state.timeout_ms,
+                        ));
+                        return Some((Err(anyhow::Error::new(error)), state));
+                    }
+                    next = state.body.next() => {
+                        let result = next?;
+
+                        let frame = match result {
+                            Ok(frame) => frame,
+                            Err(error) => {
+                                state.finished = true;
+                                return Some((
+                                    Err(anyhow::anyhow!("Body read: {}", error)),
+                                    state,
+                                ));
+                            }
+                        };
+                        let data = match frame.into_data() {
+                            Ok(data) => data,
+                            Err(_) => {
+                                state.finished = true;
+                                return Some((Err(anyhow::anyhow!("Not data frame")), state));
+                            }
+                        };
+
+                        state.parser.push(&data);
+                        state.pending.extend(
+                            state
+                                .parser
+                                .parse_all()
+                                .into_iter()
+                                .map(|result| result.map_err(anyhow::Error::new)),
+                        );
+                    }
+                }
+            }
+        })
     }
 
     /// Send a request to the given address
@@ -904,6 +955,18 @@ impl Http2ClientBuilder {
         }
     }
 
+    /// Create a builder using HTTP/2 timeout environment variables.
+    ///
+    /// Subsequent builder calls take precedence over environment values.
+    pub fn from_env() -> Result<Self> {
+        Ok(Self {
+            http2_config: Http2Config::from_env()?,
+            pool_config: None,
+            retry_config: None,
+            fault_injector: None,
+        })
+    }
+
     /// Set fault injector for testing / chaos engineering.
     pub fn fault_injector(mut self, injector: Option<Arc<dyn FaultInjector>>) -> Self {
         self.fault_injector = injector;