diff --git a/crates/pulsing-actor/src/connect/mod.rs b/crates/pulsing-actor/src/connect/mod.rs index 9ee104b15..7210a99cf 100644 --- a/crates/pulsing-actor/src/connect/mod.rs +++ b/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) -> Result> { + 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> { + 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, + config: Http2Config, + ) -> Result> { 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]; diff --git a/crates/pulsing-actor/src/system/config.rs b/crates/pulsing-actor/src/system/config.rs index 1aa545c00..5e4a1799a 100644 --- a/crates/pulsing-actor/src/system/config.rs +++ b/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 { + 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 { + Ok(Self { + addr, + http2_config: Http2Config::from_env()?, + ..Self::default() + }) + } + /// Add seed nodes for cluster joining pub fn with_seeds(mut self, seeds: Vec) -> 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 { - 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, diff --git a/crates/pulsing-actor/src/transport/http2/client.rs b/crates/pulsing-actor/src/transport/http2/client.rs index 464f13f73..7bddc2ad7 100644 --- a/crates/pulsing-actor/src/transport/http2/client.rs +++ b/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; } +fn stream_error_to_pulsing(error: anyhow::Error) -> PulsingError { + match error.downcast::() { + 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, @@ -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> { - 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::>(); - Ok::<_, anyhow::Error>(futures::stream::iter(frames)) + struct BodyFrameState { + body: http_body_util::BodyStream, + parser: BinaryFrameParser, + pending: VecDeque>, + 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 { + 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>) -> Self { self.fault_injector = injector; diff --git a/crates/pulsing-actor/src/transport/http2/config.rs b/crates/pulsing-actor/src/transport/http2/config.rs index efb8eda1d..e9548cf12 100644 --- a/crates/pulsing-actor/src/transport/http2/config.rs +++ b/crates/pulsing-actor/src/transport/http2/config.rs @@ -2,10 +2,14 @@ use std::time::Duration; +use crate::error::{PulsingError, Result, RuntimeError}; + #[cfg(feature = "tls")] use super::tls::TlsConfig; -#[cfg(feature = "tls")] -use crate::error::Result; + +pub const HTTP2_CONNECT_TIMEOUT_ENV: &str = "PULSING_HTTP2_CONNECT_TIMEOUT_MS"; +pub const HTTP2_REQUEST_TIMEOUT_ENV: &str = "PULSING_HTTP2_REQUEST_TIMEOUT_MS"; +pub const HTTP2_STREAM_TIMEOUT_ENV: &str = "PULSING_HTTP2_STREAM_TIMEOUT_MS"; /// HTTP/2 transport configuration. #[derive(Debug, Clone)] @@ -63,6 +67,60 @@ impl Http2Config { Self::default() } + /// Build the default HTTP/2 configuration with process environment overrides. + /// + /// Timeout values are positive integer milliseconds: + /// - `PULSING_HTTP2_CONNECT_TIMEOUT_MS` + /// - `PULSING_HTTP2_REQUEST_TIMEOUT_MS` + /// - `PULSING_HTTP2_STREAM_TIMEOUT_MS` + /// + /// Environment variables are read once when this function is called. + pub fn from_env() -> Result { + Self::from_env_reader(|name| match std::env::var(name) { + Ok(value) => Ok(Some(value)), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(std::env::VarError::NotUnicode(_)) => Err("value is not valid Unicode".to_string()), + }) + } + + fn from_env_reader(mut read: F) -> Result + where + F: FnMut(&str) -> std::result::Result, String>, + { + let defaults = Self::default(); + let (connect_timeout, connect_from_env) = read_timeout_env( + &mut read, + HTTP2_CONNECT_TIMEOUT_ENV, + defaults.connect_timeout, + )?; + let (request_timeout, request_from_env) = read_timeout_env( + &mut read, + HTTP2_REQUEST_TIMEOUT_ENV, + defaults.request_timeout, + )?; + let (stream_timeout, stream_from_env) = + read_timeout_env(&mut read, HTTP2_STREAM_TIMEOUT_ENV, defaults.stream_timeout)?; + + let config = Self { + connect_timeout, + request_timeout, + stream_timeout, + ..defaults + }; + + tracing::debug!( + connect_timeout_ms = config.connect_timeout.as_millis() as u64, + connect_timeout_source = if connect_from_env { "env" } else { "default" }, + request_timeout_ms = config.request_timeout.as_millis() as u64, + request_timeout_source = if request_from_env { "env" } else { "default" }, + stream_timeout_ms = config.stream_timeout.as_millis() as u64, + stream_timeout_source = if stream_from_env { "env" } else { "default" }, + "Resolved HTTP/2 timeout configuration" + ); + + Ok(config) + } + pub fn low_latency() -> Self { Self { connect_timeout: Duration::from_secs(2), @@ -221,9 +279,49 @@ impl Http2Config { } } +fn read_timeout_env(read: &mut F, name: &str, default: Duration) -> Result<(Duration, bool)> +where + F: FnMut(&str) -> std::result::Result, String>, +{ + let value = read(name).map_err(|reason| { + PulsingError::from(RuntimeError::invalid_config_value( + name, + "", + reason, + )) + })?; + + match value { + Some(value) => parse_timeout_ms(name, &value).map(|duration| (duration, true)), + None => Ok((default, false)), + } +} + +fn parse_timeout_ms(name: &str, value: &str) -> Result { + let trimmed = value.trim(); + let timeout_ms = trimmed.parse::().map_err(|_| { + PulsingError::from(RuntimeError::invalid_config_value( + name, + value, + "expected a positive integer number of milliseconds", + )) + })?; + + if timeout_ms == 0 { + return Err(PulsingError::from(RuntimeError::invalid_config_value( + name, + value, + "timeout must be greater than zero", + ))); + } + + Ok(Duration::from_millis(timeout_ms)) +} + #[cfg(test)] mod tests { use super::*; + use std::collections::HashMap; #[test] fn test_default_config() { @@ -278,4 +376,65 @@ mod tests { assert_eq!(retry_config.max_retries, 5); assert_eq!(retry_config.initial_delay, Duration::from_millis(200)); } + + #[test] + fn test_from_env_reader_uses_defaults_when_unset() { + let config = Http2Config::from_env_reader(|_| Ok(None)).unwrap(); + + assert_eq!(config.connect_timeout, Duration::from_secs(5)); + assert_eq!(config.request_timeout, Duration::from_secs(30)); + assert_eq!(config.stream_timeout, Duration::from_secs(300)); + } + + #[test] + fn test_from_env_reader_applies_timeout_overrides() { + let values = HashMap::from([ + (HTTP2_CONNECT_TIMEOUT_ENV, "1250".to_string()), + (HTTP2_REQUEST_TIMEOUT_ENV, "45000".to_string()), + (HTTP2_STREAM_TIMEOUT_ENV, "900000".to_string()), + ]); + + let config = Http2Config::from_env_reader(|name| Ok(values.get(name).cloned())).unwrap(); + + assert_eq!(config.connect_timeout, Duration::from_millis(1250)); + assert_eq!(config.request_timeout, Duration::from_secs(45)); + assert_eq!(config.stream_timeout, Duration::from_secs(900)); + } + + #[test] + fn test_from_env_reader_rejects_invalid_timeout() { + let error = Http2Config::from_env_reader(|name| { + Ok((name == HTTP2_REQUEST_TIMEOUT_ENV).then(|| "thirty".to_string())) + }) + .unwrap_err(); + + assert!(error.to_string().contains(HTTP2_REQUEST_TIMEOUT_ENV)); + assert!(error.to_string().contains("thirty")); + } + + #[test] + fn test_from_env_reader_rejects_zero_timeout() { + let error = Http2Config::from_env_reader(|name| { + Ok((name == HTTP2_STREAM_TIMEOUT_ENV).then(|| "0".to_string())) + }) + .unwrap_err(); + + assert!(error.to_string().contains(HTTP2_STREAM_TIMEOUT_ENV)); + assert!(error.to_string().contains("greater than zero")); + } + + #[test] + fn test_from_env_reader_rejects_non_unicode_timeout() { + let error = Http2Config::from_env_reader(|name| { + if name == HTTP2_CONNECT_TIMEOUT_ENV { + Err("value is not valid Unicode".to_string()) + } else { + Ok(None) + } + }) + .unwrap_err(); + + assert!(error.to_string().contains(HTTP2_CONNECT_TIMEOUT_ENV)); + assert!(error.to_string().contains("not valid Unicode")); + } } diff --git a/crates/pulsing-actor/tests/unit/transport/client_tests.rs b/crates/pulsing-actor/tests/unit/transport/client_tests.rs index 9d1719e2c..405ce93c9 100644 --- a/crates/pulsing-actor/tests/unit/transport/client_tests.rs +++ b/crates/pulsing-actor/tests/unit/transport/client_tests.rs @@ -1,14 +1,56 @@ //! HTTP/2 Client tests use crate::common::fixtures::{StreamingHandler, TestCounters, TestHandler}; +use futures::StreamExt; use pulsing_actor::actor::{ActorId, Message}; +use pulsing_actor::error::{PulsingError, RuntimeError}; use pulsing_actor::transport::{ - Http2Client, Http2Config, Http2RemoteTransport, Http2Server, TransportTarget, + Http2Client, Http2Config, Http2RemoteTransport, Http2Server, Http2ServerHandler, + TransportTarget, }; use std::sync::atomic::Ordering; use std::sync::Arc; use tokio_util::sync::CancellationToken; +struct SilentStreamHandler; + +#[async_trait::async_trait] +impl Http2ServerHandler for SilentStreamHandler { + async fn handle_message_full( + &self, + _path: &str, + _msg: Message, + ) -> pulsing_actor::error::Result { + let (tx, rx) = tokio::sync::mpsc::channel(1); + tokio::spawn(async move { + futures::future::pending::<()>().await; + drop(tx); + }); + Ok(Message::from_channel("chunk", rx)) + } + + async fn handle_tell( + &self, + _path: &str, + _msg_type: &str, + _payload: Vec, + ) -> pulsing_actor::error::Result<()> { + Ok(()) + } + + async fn handle_gossip( + &self, + _payload: Vec, + _peer_addr: std::net::SocketAddr, + ) -> pulsing_actor::error::Result>> { + Ok(None) + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } +} + #[tokio::test] async fn test_http2_client_creation() { let client = Http2Client::new(Http2Config::default()); @@ -490,6 +532,50 @@ async fn test_http2_streaming_request() { cancel.cancel(); } +#[tokio::test] +async fn test_http2_silent_stream_returns_timeout_error() { + let cancel = CancellationToken::new(); + let server = Http2Server::new( + "127.0.0.1:0".parse().unwrap(), + Arc::new(SilentStreamHandler), + Http2Config::default(), + cancel.clone(), + ) + .await + .unwrap(); + + let client = Http2Client::new( + Http2Config::default() + .request_timeout(tokio::time::Duration::from_secs(1)) + .stream_timeout(tokio::time::Duration::from_millis(50)), + ); + let response = client + .send_message( + server.local_addr(), + "/actors/silent_stream", + "Request", + Vec::new(), + ) + .await + .unwrap(); + + let Message::Stream { mut stream, .. } = response else { + panic!("Expected streaming response"); + }; + let result = tokio::time::timeout(tokio::time::Duration::from_secs(1), stream.next()) + .await + .expect("stream timeout did not wake the consumer") + .expect("stream ended without a timeout error"); + let error = result.expect_err("stream unexpectedly produced a frame"); + + assert!(matches!( + error, + PulsingError::Runtime(RuntimeError::RequestTimeout { timeout_ms: 50 }) + )); + + cancel.cancel(); +} + #[tokio::test] async fn test_http2_single_request_with_full_api() { let counters = Arc::new(TestCounters::default()); diff --git a/crates/pulsing-py/src/actor.rs b/crates/pulsing-py/src/actor.rs index a662e085e..a87318c11 100644 --- a/crates/pulsing-py/src/actor.rs +++ b/crates/pulsing-py/src/actor.rs @@ -1513,17 +1513,17 @@ pub struct PySystemConfig { #[pymethods] impl PySystemConfig { #[staticmethod] - fn standalone() -> Self { - Self { - inner: SystemConfig::standalone(), - } + fn standalone() -> PyResult { + Ok(Self { + inner: SystemConfig::from_env().map_err(to_pyerr)?, + }) } #[staticmethod] fn with_addr(addr: String) -> PyResult { let socket_addr: SocketAddr = addr.parse().map_err(to_py_value_err)?; Ok(Self { - inner: SystemConfig::with_addr(socket_addr), + inner: SystemConfig::with_addr_from_env(socket_addr).map_err(to_pyerr)?, }) } diff --git a/crates/pulsing-rpymod/src/bindings/message.rs b/crates/pulsing-rpymod/src/bindings/message.rs index 6db4af5dd..748f41d35 100644 --- a/crates/pulsing-rpymod/src/bindings/message.rs +++ b/crates/pulsing-rpymod/src/bindings/message.rs @@ -424,10 +424,11 @@ pub struct PySystemConfig { #[pyclass] impl PySystemConfig { #[pystaticmethod] - fn standalone() -> Self { - Self { - inner: SystemConfig::standalone(), - } + fn standalone(vm: &VirtualMachine) -> PyResult { + Ok(Self { + inner: SystemConfig::from_env() + .map_err(|error| vm.new_value_error(error.to_string()))?, + }) } #[pystaticmethod] @@ -437,7 +438,8 @@ impl PySystemConfig { .parse() .map_err(|e: std::net::AddrParseError| vm.new_value_error(e.to_string()))?; Ok(Self { - inner: SystemConfig::with_addr(socket_addr), + inner: SystemConfig::with_addr_from_env(socket_addr) + .map_err(|error| vm.new_value_error(error.to_string()))?, }) } diff --git a/crates/pulsing-rpymod/src/runtime.rs b/crates/pulsing-rpymod/src/runtime.rs index ed9caf83e..f87fa5319 100644 --- a/crates/pulsing-rpymod/src/runtime.rs +++ b/crates/pulsing-rpymod/src/runtime.rs @@ -60,7 +60,10 @@ fn runtime() -> Result<&'static CliRuntime> { .expect("failed to build tokio runtime"); let handle = runtime.handle().clone(); let system = runtime - .block_on(async { ActorSystem::new(SystemConfig::standalone()).await }) + .block_on(async { + let config = SystemConfig::from_env()?; + ActorSystem::new(config).await + }) .expect("ActorSystem::new failed"); let _ = tx.send((Arc::clone(&system), handle.clone())); runtime.block_on(async { diff --git a/docs/src/guide/reliability.md b/docs/src/guide/reliability.md index c65d4d15b..f4b3dc9cc 100644 --- a/docs/src/guide/reliability.md +++ b/docs/src/guide/reliability.md @@ -18,6 +18,30 @@ result = await asyncio.wait_for(ref.ask({"op": "compute"}), timeout=10.0) For proxy method calls: `await asyncio.wait_for(proxy.compute(), timeout=10.0)`. +### HTTP/2 transport timeouts + +Remote actor traffic and `PulsingConnect` use these process environment variables: + +| Variable | Default | Scope | +| --- | ---: | --- | +| `PULSING_HTTP2_CONNECT_TIMEOUT_MS` | `5000` | TCP connection establishment | +| `PULSING_HTTP2_REQUEST_TIMEOUT_MS` | `30000` | Non-streaming request/response stages | +| `PULSING_HTTP2_STREAM_TIMEOUT_MS` | `300000` | Total streaming response lifetime after response headers | + +Values are positive integer milliseconds. Invalid or zero values fail configuration instead of +silently falling back to defaults. The environment is read when an actor system or connector is +created; restart or recreate it after changing a value. + +```bash +export PULSING_HTTP2_CONNECT_TIMEOUT_MS=10000 +export PULSING_HTTP2_REQUEST_TIMEOUT_MS=60000 +export PULSING_HTTP2_STREAM_TIMEOUT_MS=900000 +``` + +An explicit Rust `Http2Config` takes precedence over these variables. The transport settings do +not impose a deadline on local `ActorRef.ask()` calls or on server-side actor handler execution, +so application-level `asyncio.wait_for()` remains useful for end-to-end deadlines. + ## Retries (application-level) Pulsing does not hide retries for you. If you retry, assume duplicates are possible. @@ -134,5 +158,8 @@ async def process_with_retry(actor, data, max_retries=3): For streaming responses, assume partial streams are possible. Make chunks independently meaningful: +The HTTP/2 stream timeout is a hard total deadline: expiration wakes a silent stream and returns a +request-timeout error. It is not a per-chunk idle timeout. + - include `seq` / offsets / ids per chunk - allow resume or dedup on the client side diff --git a/docs/src/guide/reliability.zh.md b/docs/src/guide/reliability.zh.md index 6f41fa847..53a4bd327 100644 --- a/docs/src/guide/reliability.zh.md +++ b/docs/src/guide/reliability.zh.md @@ -18,6 +18,29 @@ result = await asyncio.wait_for(ref.ask({"op": "compute"}), timeout=10.0) 对 proxy 方法调用：`await asyncio.wait_for(proxy.compute(), timeout=10.0)`。 +### HTTP/2 传输超时 + +远程 Actor 通信和 `PulsingConnect` 使用以下进程环境变量： + +| 环境变量 | 默认值 | 作用范围 | +| --- | ---: | --- | +| `PULSING_HTTP2_CONNECT_TIMEOUT_MS` | `5000` | 建立 TCP 连接 | +| `PULSING_HTTP2_REQUEST_TIMEOUT_MS` | `30000` | 非流式请求/响应阶段 | +| `PULSING_HTTP2_STREAM_TIMEOUT_MS` | `300000` | 收到响应头后的流式响应总生命周期 | + +变量值必须是正整数毫秒。非法值或零值会导致配置失败，不会静默回退到默认值。环境变量在 +ActorSystem 或 Connect 创建时读取一次；修改后需要重新创建客户端或重启进程。 + +```bash +export PULSING_HTTP2_CONNECT_TIMEOUT_MS=10000 +export PULSING_HTTP2_REQUEST_TIMEOUT_MS=60000 +export PULSING_HTTP2_STREAM_TIMEOUT_MS=900000 +``` + +显式传入的 Rust `Http2Config` 优先于环境变量。这些配置不限制本地 `ActorRef.ask()`，也不 +构成服务端 Actor handler 的执行截止时间，因此端到端超时仍建议在应用层使用 +`asyncio.wait_for()`。 + ## 重试（放在业务层） Pulsing 不会替你“隐式重试”。一旦你做重试，就要默认可能出现重复处理。 @@ -134,5 +157,8 @@ async def process_with_retry(actor, data, max_retries=3): 对流式响应要默认可能“部分输出后中断”。建议每个 chunk 自包含： +HTTP/2 流式超时是总生命周期硬截止时间：即使连接完全静默，到期也会唤醒并返回请求超时 +错误；它不是逐 chunk 的空闲超时。 + - 每个 chunk 带 `seq` / offset / id - 客户端可恢复或去重