From fa828dd8b1591fa7611552f2979575c6af04af05 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 24 Feb 2026 09:46:50 -0500 Subject: [PATCH] Increase connect timeout to 30s, disable request/stream_read timeouts LLM API calls should not have HTTP-level request or stream-read timeouts. These are better controlled at the application level via TimeoutConfig (total/per_step). The connect timeout is increased from 10s to 30s since network conditions vary. Co-Authored-By: Claude Opus 4.6 --- crates/llm/src/providers/anthropic.rs | 28 +++++++----- crates/llm/src/providers/gemini.rs | 33 ++++++++------ crates/llm/src/providers/openai.rs | 33 +++++++------- crates/llm/src/providers/openai_compatible.rs | 45 ++++++++++--------- crates/llm/src/types.rs | 16 +++---- 5 files changed, 85 insertions(+), 70 deletions(-) diff --git a/crates/llm/src/providers/anthropic.rs b/crates/llm/src/providers/anthropic.rs index 215864997..fcc0c9c44 100644 --- a/crates/llm/src/providers/anthropic.rs +++ b/crates/llm/src/providers/anthropic.rs @@ -17,8 +17,8 @@ pub struct Adapter { base_url: String, default_headers: std::collections::HashMap, client: reqwest::Client, - request_timeout: std::time::Duration, - stream_read_timeout: std::time::Duration, + request_timeout: Option, + stream_read_timeout: Option, } impl Adapter { @@ -34,8 +34,8 @@ impl Adapter { base_url: DEFAULT_BASE_URL.to_string(), default_headers: std::collections::HashMap::new(), client, - request_timeout: std::time::Duration::from_secs_f64(timeout.request), - stream_read_timeout: std::time::Duration::from_secs_f64(timeout.stream_read), + request_timeout: timeout.request.map(std::time::Duration::from_secs_f64), + stream_read_timeout: timeout.stream_read.map(std::time::Duration::from_secs_f64), } } @@ -57,8 +57,8 @@ impl Adapter { .connect_timeout(std::time::Duration::from_secs_f64(timeout.connect)) .build() .unwrap_or_default(); - self.request_timeout = std::time::Duration::from_secs_f64(timeout.request); - self.stream_read_timeout = std::time::Duration::from_secs_f64(timeout.stream_read); + self.request_timeout = timeout.request.map(std::time::Duration::from_secs_f64); + self.stream_read_timeout = timeout.stream_read.map(std::time::Duration::from_secs_f64); self } @@ -925,7 +925,7 @@ struct SseReaderState { done: bool, /// When true, `tool_use` events for the synthetic tool are converted to text events. json_schema_mode: bool, - stream_read_timeout: std::time::Duration, + stream_read_timeout: Option, } impl SseReaderState { @@ -935,7 +935,7 @@ impl SseReaderState { + 'static, rate_limit: Option, json_schema_mode: bool, - stream_read_timeout: std::time::Duration, + stream_read_timeout: Option, ) -> Self { use futures::StreamExt; Self { @@ -967,7 +967,11 @@ impl SseReaderState { } // Read more bytes from the stream. - match tokio::time::timeout(self.stream_read_timeout, self.byte_stream.next()).await { + let chunk_result = match self.stream_read_timeout { + Some(timeout) => tokio::time::timeout(timeout, self.byte_stream.next()).await, + None => Ok(self.byte_stream.next().await), + }; + match chunk_result { Ok(Some(Ok(chunk))) => { let text = String::from_utf8_lossy(&chunk); self.buffer.push_str(&text); @@ -1142,8 +1146,12 @@ impl ProviderAdapter for Adapter { } let (_api_request, req_builder) = build_api_request(self, request, false); + let mut req = req_builder; + if let Some(t) = self.request_timeout { + req = req.timeout(t); + } let (body, headers) = - send_and_read_response(req_builder.timeout(self.request_timeout), "anthropic", "type").await?; + send_and_read_response(req, "anthropic", "type").await?; let api_resp: ApiResponse = serde_json::from_str(&body).map_err(|e| SdkError::Network { diff --git a/crates/llm/src/providers/gemini.rs b/crates/llm/src/providers/gemini.rs index ebe2fad5f..760ede6f2 100644 --- a/crates/llm/src/providers/gemini.rs +++ b/crates/llm/src/providers/gemini.rs @@ -19,8 +19,8 @@ pub struct Adapter { base_url: String, default_headers: std::collections::HashMap, client: reqwest::Client, - request_timeout: std::time::Duration, - stream_read_timeout: std::time::Duration, + request_timeout: Option, + stream_read_timeout: Option, } impl Adapter { @@ -36,8 +36,8 @@ impl Adapter { base_url: DEFAULT_BASE_URL.to_string(), default_headers: std::collections::HashMap::new(), client, - request_timeout: std::time::Duration::from_secs_f64(timeout.request), - stream_read_timeout: std::time::Duration::from_secs_f64(timeout.stream_read), + request_timeout: timeout.request.map(std::time::Duration::from_secs_f64), + stream_read_timeout: timeout.stream_read.map(std::time::Duration::from_secs_f64), } } @@ -59,8 +59,8 @@ impl Adapter { .connect_timeout(std::time::Duration::from_secs_f64(timeout.connect)) .build() .unwrap_or_default(); - self.request_timeout = std::time::Duration::from_secs_f64(timeout.request); - self.stream_read_timeout = std::time::Duration::from_secs_f64(timeout.stream_read); + self.request_timeout = timeout.request.map(std::time::Duration::from_secs_f64); + self.stream_read_timeout = timeout.stream_read.map(std::time::Duration::from_secs_f64); self } } @@ -571,7 +571,7 @@ async fn send_streaming_request( /// Process a stream of SSE chunks from the Gemini `streamGenerateContent` endpoint /// and yield `StreamEvent` values. -fn process_sse_stream(http_resp: reqwest::Response, model: String, rate_limit: Option, stream_read_timeout: std::time::Duration) -> StreamEventStream { +fn process_sse_stream(http_resp: reqwest::Response, model: String, rate_limit: Option, stream_read_timeout: Option) -> StreamEventStream { Box::pin(stream::unfold( SseStreamState::new(http_resp, model, rate_limit, stream_read_timeout), |mut state| async move { @@ -680,11 +680,11 @@ struct SseStreamState { finished: bool, /// Rate limit info parsed from HTTP response headers. rate_limit: Option, - stream_read_timeout: std::time::Duration, + stream_read_timeout: Option, } impl SseStreamState { - fn new(http_resp: reqwest::Response, model: String, rate_limit: Option, stream_read_timeout: std::time::Duration) -> Self { + fn new(http_resp: reqwest::Response, model: String, rate_limit: Option, stream_read_timeout: Option) -> Self { Self { http_resp, model, @@ -720,7 +720,11 @@ impl SseStreamState { } // Read more bytes from the HTTP response. - match tokio::time::timeout(self.stream_read_timeout, self.http_resp.chunk()).await { + let chunk_result = match self.stream_read_timeout { + Some(timeout) => tokio::time::timeout(timeout, self.http_resp.chunk()).await, + None => Ok(self.http_resp.chunk().await), + }; + match chunk_result { Ok(Ok(Some(bytes))) => { let text = String::from_utf8_lossy(&bytes); self.line_buffer.push_str(&text); @@ -918,10 +922,11 @@ impl ProviderAdapter for Adapter { for (key, value) in &self.default_headers { req = req.header(key, value); } - let (body, headers) = send_gemini_response( - req.json(&api_body).timeout(self.request_timeout), - ) - .await?; + let mut gemini_req = req.json(&api_body); + if let Some(t) = self.request_timeout { + gemini_req = gemini_req.timeout(t); + } + let (body, headers) = send_gemini_response(gemini_req).await?; let api_resp: ApiResponse = serde_json::from_str(&body).map_err(|e| SdkError::Network { diff --git a/crates/llm/src/providers/openai.rs b/crates/llm/src/providers/openai.rs index bcbabca70..64d649d5d 100644 --- a/crates/llm/src/providers/openai.rs +++ b/crates/llm/src/providers/openai.rs @@ -22,8 +22,8 @@ pub struct Adapter { project_id: Option, default_headers: std::collections::HashMap, client: reqwest::Client, - request_timeout: std::time::Duration, - stream_read_timeout: std::time::Duration, + request_timeout: Option, + stream_read_timeout: Option, } impl Adapter { @@ -41,8 +41,8 @@ impl Adapter { project_id: None, default_headers: std::collections::HashMap::new(), client, - request_timeout: std::time::Duration::from_secs_f64(timeout.request), - stream_read_timeout: std::time::Duration::from_secs_f64(timeout.stream_read), + request_timeout: timeout.request.map(std::time::Duration::from_secs_f64), + stream_read_timeout: timeout.stream_read.map(std::time::Duration::from_secs_f64), } } @@ -76,8 +76,8 @@ impl Adapter { .connect_timeout(std::time::Duration::from_secs_f64(timeout.connect)) .build() .unwrap_or_default(); - self.request_timeout = std::time::Duration::from_secs_f64(timeout.request); - self.stream_read_timeout = std::time::Duration::from_secs_f64(timeout.stream_read); + self.request_timeout = timeout.request.map(std::time::Duration::from_secs_f64); + self.stream_read_timeout = timeout.stream_read.map(std::time::Duration::from_secs_f64); self } @@ -496,7 +496,7 @@ struct SseStreamState { emitted_text_start: bool, raw_response: Option, rate_limit: Option, - stream_read_timeout: std::time::Duration, + stream_read_timeout: Option, } /// Extract complete SSE messages from the buffer. @@ -566,7 +566,11 @@ async fn process_next_sse_events( continue; } - match tokio::time::timeout(state.stream_read_timeout, state.byte_stream.next()).await { + let chunk_result = match state.stream_read_timeout { + Some(timeout) => tokio::time::timeout(timeout, state.byte_stream.next()).await, + None => Ok(state.byte_stream.next().await), + }; + match chunk_result { Ok(Some(Ok(bytes))) => { let text = String::from_utf8_lossy(&bytes); state.buffer.push_str(&text); @@ -913,14 +917,11 @@ impl ProviderAdapter for Adapter { let request_body = build_request_body(request, false); let url = format!("{}/responses", self.base_url); - let (body, headers) = send_and_read_response( - self.build_request(&url) - .json(&request_body) - .timeout(self.request_timeout), - "openai", - "type", - ) - .await?; + let mut req = self.build_request(&url).json(&request_body); + if let Some(t) = self.request_timeout { + req = req.timeout(t); + } + let (body, headers) = send_and_read_response(req, "openai", "type").await?; let api_resp: ApiResponse = serde_json::from_str(&body).map_err(|e| SdkError::Network { diff --git a/crates/llm/src/providers/openai_compatible.rs b/crates/llm/src/providers/openai_compatible.rs index 1b4590a9d..21d0415a8 100644 --- a/crates/llm/src/providers/openai_compatible.rs +++ b/crates/llm/src/providers/openai_compatible.rs @@ -23,8 +23,8 @@ pub struct Adapter { provider_name: String, default_headers: std::collections::HashMap, client: reqwest::Client, - request_timeout: std::time::Duration, - stream_read_timeout: std::time::Duration, + request_timeout: Option, + stream_read_timeout: Option, } impl Adapter { @@ -41,8 +41,8 @@ impl Adapter { provider_name: "openai-compatible".to_string(), default_headers: std::collections::HashMap::new(), client, - request_timeout: std::time::Duration::from_secs_f64(timeout.request), - stream_read_timeout: std::time::Duration::from_secs_f64(timeout.stream_read), + request_timeout: timeout.request.map(std::time::Duration::from_secs_f64), + stream_read_timeout: timeout.stream_read.map(std::time::Duration::from_secs_f64), } } @@ -64,8 +64,8 @@ impl Adapter { .connect_timeout(std::time::Duration::from_secs_f64(timeout.connect)) .build() .unwrap_or_default(); - self.request_timeout = std::time::Duration::from_secs_f64(timeout.request); - self.stream_read_timeout = std::time::Duration::from_secs_f64(timeout.stream_read); + self.request_timeout = timeout.request.map(std::time::Duration::from_secs_f64); + self.stream_read_timeout = timeout.stream_read.map(std::time::Duration::from_secs_f64); self } @@ -412,14 +412,11 @@ impl ProviderAdapter for Adapter { let api_body = build_api_request(request, None, &self.provider_name); let url = format!("{}/chat/completions", self.base_url); - let (body, headers) = send_and_read_response( - self.build_request(&url) - .json(&api_body) - .timeout(self.request_timeout), - &self.provider_name, - "type", - ) - .await?; + let mut req = self.build_request(&url).json(&api_body); + if let Some(t) = self.request_timeout { + req = req.timeout(t); + } + let (body, headers) = send_and_read_response(req, &self.provider_name, "type").await?; let api_resp: ApiResponse = serde_json::from_str(&body).map_err(|e| SdkError::Network { @@ -617,7 +614,7 @@ struct StreamState { text_started: bool, done: bool, rate_limit: Option, - stream_read_timeout: std::time::Duration, + stream_read_timeout: Option, } impl StreamState { @@ -626,7 +623,7 @@ impl StreamState { provider_name: String, model: String, rate_limit: Option, - stream_read_timeout: std::time::Duration, + stream_read_timeout: Option, ) -> Self { Self { response, @@ -659,7 +656,11 @@ impl StreamState { return Ok(Some(line)); } - match tokio::time::timeout(self.stream_read_timeout, self.response.chunk()).await { + let chunk_result = match self.stream_read_timeout { + Some(timeout) => tokio::time::timeout(timeout, self.response.chunk()).await, + None => Ok(self.response.chunk().await), + }; + match chunk_result { Ok(Ok(Some(bytes))) => { let text = String::from_utf8_lossy(&bytes); self.buffer.push_str(&text); @@ -914,7 +915,7 @@ mod tests { .body("") .unwrap(), ); - let mut state = StreamState::new(http_resp, "test".into(), "model".into(), None, std::time::Duration::from_secs(30)); + let mut state = StreamState::new(http_resp, "test".into(), "model".into(), None, Some(std::time::Duration::from_secs(30))); // First text chunk should emit TextStart + TextDelta. let chunk1: StreamChunk = serde_json::from_str( @@ -944,7 +945,7 @@ mod tests { .body("") .unwrap(), ); - let mut state = StreamState::new(http_resp, "test".into(), "model".into(), None, std::time::Duration::from_secs(30)); + let mut state = StreamState::new(http_resp, "test".into(), "model".into(), None, Some(std::time::Duration::from_secs(30))); // First tool call chunk (has id and name) -> ToolCallStart. let chunk1: StreamChunk = serde_json::from_str( @@ -973,7 +974,7 @@ mod tests { .body("") .unwrap(), ); - let mut state = StreamState::new(http_resp, "test-provider".into(), "test-model".into(), None, std::time::Duration::from_secs(30)); + let mut state = StreamState::new(http_resp, "test-provider".into(), "test-model".into(), None, Some(std::time::Duration::from_secs(30))); state.response_id = "resp-1".into(); state.response_model = "gpt-4".into(); state.accumulated_text = "Hello world".into(); @@ -1015,7 +1016,7 @@ mod tests { .body("") .unwrap(), ); - let mut state = StreamState::new(http_resp, "test".into(), "model".into(), None, std::time::Duration::from_secs(30)); + let mut state = StreamState::new(http_resp, "test".into(), "model".into(), None, Some(std::time::Duration::from_secs(30))); state.response_id = "resp-1".into(); state.tool_calls.push(AccumulatedToolCall { id: "call_1".into(), @@ -1058,7 +1059,7 @@ mod tests { .body("") .unwrap(), ); - let mut state = StreamState::new(http_resp, "test".into(), "fallback-model".into(), None, std::time::Duration::from_secs(30)); + let mut state = StreamState::new(http_resp, "test".into(), "fallback-model".into(), None, Some(std::time::Duration::from_secs(30))); // response_model is empty, so finish_events should use the request model. let events = state.finish_events(); match &events[0] { diff --git a/crates/llm/src/types.rs b/crates/llm/src/types.rs index b2bbf4734..23f337ab4 100644 --- a/crates/llm/src/types.rs +++ b/crates/llm/src/types.rs @@ -634,16 +634,16 @@ impl From for TimeoutConfig { #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] pub struct AdapterTimeout { pub connect: f64, - pub request: f64, - pub stream_read: f64, + pub request: Option, + pub stream_read: Option, } impl Default for AdapterTimeout { fn default() -> Self { Self { - connect: 10.0, - request: 120.0, - stream_read: 120.0, + connect: 30.0, + request: None, + stream_read: None, } } } @@ -1128,9 +1128,9 @@ mod tests { #[test] fn adapter_timeout_defaults() { let timeout = AdapterTimeout::default(); - assert!((timeout.connect - 10.0).abs() < f64::EPSILON); - assert!((timeout.request - 120.0).abs() < f64::EPSILON); - assert!((timeout.stream_read - 120.0).abs() < f64::EPSILON); + assert!((timeout.connect - 30.0).abs() < f64::EPSILON); + assert!(timeout.request.is_none()); + assert!(timeout.stream_read.is_none()); } #[test]