scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-06 08:18:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Increase connect timeout to 30s, disable request/stream_read timeouts

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-02-24 09:46:50 -05:00
parent 9665fad133
commit fa828dd8b1
5 changed files with 85 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
crates/llm/src/providers/anthropic.rs
View file

@ -17,8 +17,8 @@ pub struct Adapter {
base_url: String,
default_headers: std::collections::HashMap<String, String>,
client: reqwest::Client,
request_timeout: std::time::Duration,
stream_read_timeout: std::time::Duration,
request_timeout: Option<std::time::Duration>,
stream_read_timeout: Option<std::time::Duration>,
}
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<std::time::Duration>,
}
impl SseReaderState {
@ -935,7 +935,7 @@ impl SseReaderState {
+ 'static,
rate_limit: Option<crate::types::RateLimitInfo>,
json_schema_mode: bool,
stream_read_timeout: std::time::Duration,
stream_read_timeout: Option<std::time::Duration>,
) -> 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 {

33
crates/llm/src/providers/gemini.rs
View file

@ -19,8 +19,8 @@ pub struct Adapter {
base_url: String,
default_headers: std::collections::HashMap<String, String>,
client: reqwest::Client,
request_timeout: std::time::Duration,
stream_read_timeout: std::time::Duration,
request_timeout: Option<std::time::Duration>,
stream_read_timeout: Option<std::time::Duration>,
}
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<crate::types::RateLimitInfo>, stream_read_timeout: std::time::Duration) -> StreamEventStream {
fn process_sse_stream(http_resp: reqwest::Response, model: String, rate_limit: Option<crate::types::RateLimitInfo>, stream_read_timeout: Option<std::time::Duration>) -> 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<crate::types::RateLimitInfo>,
stream_read_timeout: std::time::Duration,
stream_read_timeout: Option<std::time::Duration>,
}
impl SseStreamState {
fn new(http_resp: reqwest::Response, model: String, rate_limit: Option<crate::types::RateLimitInfo>, stream_read_timeout: std::time::Duration) -> Self {
fn new(http_resp: reqwest::Response, model: String, rate_limit: Option<crate::types::RateLimitInfo>, stream_read_timeout: Option<std::time::Duration>) -> 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 {

33
crates/llm/src/providers/openai.rs
View file

@ -22,8 +22,8 @@ pub struct Adapter {
project_id: Option<String>,
default_headers: std::collections::HashMap<String, String>,
client: reqwest::Client,
request_timeout: std::time::Duration,
stream_read_timeout: std::time::Duration,
request_timeout: Option<std::time::Duration>,
stream_read_timeout: Option<std::time::Duration>,
}
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<serde_json::Value>,
rate_limit: Option<crate::types::RateLimitInfo>,
stream_read_timeout: std::time::Duration,
stream_read_timeout: Option<std::time::Duration>,
}
/// 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 {

45
crates/llm/src/providers/openai_compatible.rs
View file

@ -23,8 +23,8 @@ pub struct Adapter {
provider_name: String,
default_headers: std::collections::HashMap<String, String>,
client: reqwest::Client,
request_timeout: std::time::Duration,
stream_read_timeout: std::time::Duration,
request_timeout: Option<std::time::Duration>,
stream_read_timeout: Option<std::time::Duration>,
}
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<crate::types::RateLimitInfo>,
stream_read_timeout: std::time::Duration,
stream_read_timeout: Option<std::time::Duration>,
}
impl StreamState {
@ -626,7 +623,7 @@ impl StreamState {
provider_name: String,
model: String,
rate_limit: Option<crate::types::RateLimitInfo>,
stream_read_timeout: std::time::Duration,
stream_read_timeout: Option<std::time::Duration>,
) -> 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] {

16
crates/llm/src/types.rs
View file

@ -634,16 +634,16 @@ impl From<f64> 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<f64>,
pub stream_read: Option<f64>,
}
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]