From bfb6bdb25cf321fa0cbcd08dbfbf276555f54cf9 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 2 May 2026 20:13:18 -0400 Subject: [PATCH] fix(cli): bound server readiness probes Use HTTP health checks with short deadlines for managed server readiness and add finite control-plane request timeouts for CLI/server clients. Keep stream bodies uncapped so SSE attach flows can remain long-lived. --- .../2026-05-02-server-readiness-timeouts.md | 308 ++++++++++++++++++ .../fabro-cli/src/commands/server/start.rs | 89 ++++- lib/crates/fabro-cli/src/server_client.rs | 84 ++++- lib/crates/fabro-client/src/client.rs | 108 +++++- lib/crates/fabro-client/src/target.rs | 8 +- lib/crates/fabro-test/src/lib.rs | 5 +- 6 files changed, 569 insertions(+), 33 deletions(-) create mode 100644 docs/superpowers/plans/2026-05-02-server-readiness-timeouts.md diff --git a/docs/superpowers/plans/2026-05-02-server-readiness-timeouts.md b/docs/superpowers/plans/2026-05-02-server-readiness-timeouts.md new file mode 100644 index 000000000..1dc9a2184 --- /dev/null +++ b/docs/superpowers/plans/2026-05-02-server-readiness-timeouts.md @@ -0,0 +1,308 @@ +# Server Readiness Timeout Hardening Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Prevent CLI/server readiness and health-check paths from hanging when a socket accepts connections but never returns an HTTP response. + +**Architecture:** Use bounded HTTP `/health` probes for managed local server readiness instead of raw socket-connect readiness, and give CLI HTTP transports a finite request timeout. Keep long-lived streams separate so SSE/log-follow behavior is not accidentally capped by short control-plane timeouts. + +**Tech Stack:** Rust, Tokio, reqwest via `fabro_http`, `cargo nextest`, pinned nightly rustfmt/clippy. + +--- + +## File Structure + +- Modify `lib/crates/fabro-cli/src/server_client.rs` + - Own CLI transport construction and bounded server health checks. + - Add regression coverage for silent TCP peers and preserve the existing Unix silent-peer test. +- Modify `lib/crates/fabro-cli/src/commands/server/start.rs` + - Replace daemon-start raw socket readiness with bounded `/health` readiness for both Unix and TCP binds. +- Modify `lib/crates/fabro-client/src/client.rs` + - Add a bounded default control-plane HTTP client for explicit `ServerTarget` transports. + - Do not apply this timeout to caller-supplied transports or event stream bodies. +- Modify `lib/crates/fabro-client/src/target.rs` + - Apply the same bounded public HTTP client defaults used by OAuth/public-target helper construction. +- Modify `lib/crates/fabro-test/src/lib.rs` + - Add per-probe timeout to `twin_openai()` readiness polling. + +## Task 1: Add CLI HTTP Target Hang Regression + +**Files:** +- Modify: `lib/crates/fabro-cli/src/server_client.rs` + +- [x] **Step 1: Write a failing TCP silent-peer test** + +Add this test in `server_client.rs` `mod tests` near the Unix silent-peer regression: + +```rust +#[tokio::test] +async fn http_target_transport_times_out_when_peer_accepts_without_http_response() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + if let Ok((_stream, _addr)) = listener.accept().await { + sleep(Duration::from_secs(10)).await; + } + }); + + let target = ServerTarget::http_url(format!("http://{addr}")).unwrap(); + let client = connect_target_api_client_bundle(&target).await.unwrap(); + let result = time::timeout(Duration::from_millis(750), client.get_health()).await; + + server.abort(); + assert!( + result.is_ok(), + "HTTP target health check should return its own timeout error instead of hanging" + ); + assert!(result.unwrap().is_err()); +} +``` + +- [x] **Step 2: Verify the test fails for the right reason** + +Run: + +```bash +cargo nextest run -p fabro-cli http_target_transport_times_out_when_peer_accepts_without_http_response --status-level fail --final-status-level fail --show-progress none +``` + +Expected: FAIL after the outer 750ms timeout because the HTTP request hangs. + +## Task 2: Add Bounded CLI Control-Plane HTTP Clients + +**Files:** +- Modify: `lib/crates/fabro-cli/src/server_client.rs` +- Modify: `lib/crates/fabro-client/src/client.rs` +- Modify: `lib/crates/fabro-client/src/target.rs` + +- [x] **Step 1: Define shared timeout constants** + +In `server_client.rs`, keep the existing health-probe constant and add a control-plane request timeout: + +```rust +const SERVER_HEALTH_PROBE_TIMEOUT: Duration = Duration::from_millis(250); +const CLI_CONTROL_PLANE_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); +``` + +In `fabro-client/src/client.rs`, add: + +```rust +const DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(30); +``` + +In `fabro-client/src/target.rs`, add the same constant with the same value. + +- [x] **Step 2: Apply the timeout only to control-plane client construction** + +In `server_client.rs`, update `connect_cli_target_transport()` HTTP and Unix builders: + +```rust +let mut builder = cli_http_client_builder().timeout(CLI_CONTROL_PLANE_REQUEST_TIMEOUT); +``` + +and: + +```rust +let mut builder = cli_http_client_builder() + .unix_socket(path) + .no_proxy() + .timeout(CLI_CONTROL_PLANE_REQUEST_TIMEOUT); +``` + +In `fabro-client/src/client.rs`, update `connect_target_transport()`: + +```rust +let mut builder = + fabro_http::HttpClientBuilder::new().timeout(DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT); +``` + +and for Unix: + +```rust +let mut builder = fabro_http::HttpClientBuilder::new() + .unix_socket(path) + .no_proxy() + .timeout(DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT); +``` + +In `fabro-client/src/target.rs`, update `build_public_http_client()` builders the same way. + +- [x] **Step 3: Verify TCP silent-peer regression passes** + +Run: + +```bash +cargo nextest run -p fabro-cli http_target_transport_times_out_when_peer_accepts_without_http_response --status-level fail --final-status-level fail --show-progress none +``` + +Expected: PASS. + +## Task 3: Replace Daemon Raw Socket Readiness With `/health` + +**Files:** +- Modify: `lib/crates/fabro-cli/src/commands/server/start.rs` + +- [x] **Step 1: Write daemon readiness helpers** + +Replace `try_connect()` with a helper that builds a short-timeout HTTP client for the resolved `Bind` and calls `/health`: + +```rust +const SERVER_START_HEALTH_PROBE_TIMEOUT: Duration = Duration::from_millis(250); + +async fn try_health(bind: &Bind) -> bool { + let (base_url, client) = match build_health_client(bind) { + Ok(bundle) => bundle, + Err(_) => return false, + }; + + let response = time::timeout( + SERVER_START_HEALTH_PROBE_TIMEOUT, + client.get(format!("{base_url}/health")).send(), + ) + .await; + + matches!(response, Ok(Ok(response)) if response.status().is_success()) +} + +fn build_health_client(bind: &Bind) -> Result<(String, fabro_http::HttpClient)> { + match bind { + Bind::Tcp(addr) => Ok(( + format!("http://{addr}"), + fabro_http::HttpClientBuilder::new() + .no_proxy() + .timeout(SERVER_START_HEALTH_PROBE_TIMEOUT) + .build()?, + )), + Bind::Unix(path) => Ok(( + "http://fabro".to_string(), + fabro_http::HttpClientBuilder::new() + .unix_socket(path) + .no_proxy() + .timeout(SERVER_START_HEALTH_PROBE_TIMEOUT) + .build()?, + )), + } +} +``` + +Update the startup loop condition from: + +```rust +if try_connect(&daemon.bind).await { +``` + +to: + +```rust +if try_health(&daemon.bind).await { +``` + +Remove unused imports for `TcpStream` and `UnixStream`. + +- [x] **Step 2: Add a startup helper unit test for silent TCP readiness** + +Add a test in `start.rs` tests: + +```rust +#[tokio::test] +async fn try_health_returns_false_for_tcp_peer_that_accepts_without_http_response() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + if let Ok((_stream, _addr)) = listener.accept().await { + tokio::time::sleep(Duration::from_secs(10)).await; + } + }); + + let ready = try_health(&Bind::Tcp(addr)).await; + + server.abort(); + assert!(!ready); +} +``` + +If clippy flags absolute paths, import `tokio::net::TcpListener` and use the existing `time` import. + +- [x] **Step 3: Verify daemon startup tests** + +Run: + +```bash +cargo nextest run -p fabro-cli 'commands::server::start::tests' --status-level fail --final-status-level fail --show-progress none +cargo nextest run -p fabro-cli 'cmd::server_start' --status-level fail --final-status-level fail --show-progress none +``` + +Expected: PASS. + +## Task 4: Harden Test-Only Twin Readiness Polling + +**Files:** +- Modify: `lib/crates/fabro-test/src/lib.rs` + +- [x] **Step 1: Bound each `twin_openai()` readiness probe** + +In `twin_openai()`, replace: + +```rust +if let Ok(resp) = client.get(&healthz_url).send().await { +``` + +with: + +```rust +let response = time::timeout( + std::time::Duration::from_millis(250), + client.get(&healthz_url).send(), +) +.await; +if let Ok(Ok(resp)) = response { +``` + +- [x] **Step 2: Verify test-support coverage** + +Run: + +```bash +cargo nextest run -p fabro-test twin_openai --status-level fail --final-status-level fail --show-progress none +``` + +If no test matches that filter, run: + +```bash +cargo nextest run -p fabro-test --status-level fail --final-status-level fail --show-progress none +``` + +Expected: PASS. + +## Task 5: Final Verification + +**Files:** +- Verify all changed files. + +- [x] **Step 1: Run focused regressions** + +```bash +cargo nextest run -p fabro-cli unix_socket_probe_times_out_when_peer_accepts_without_http_response --status-level fail --final-status-level fail --show-progress none +cargo nextest run -p fabro-cli http_target_transport_times_out_when_peer_accepts_without_http_response --status-level fail --final-status-level fail --show-progress none +cargo nextest run -p fabro-cli concurrent_autostart_converges_on_one_shared_daemon_and_cleans_up --status-level fail --final-status-level fail --show-progress none +``` + +Expected: all PASS. + +- [x] **Step 2: Run package tests and style checks** + +```bash +cargo nextest run -p fabro-cli --status-level fail --final-status-level fail --show-progress none +cargo nextest run -p fabro-test --status-level fail --final-status-level fail --show-progress none +cargo +nightly-2026-04-14 fmt --check --all +cargo +nightly-2026-04-14 clippy -p fabro-cli -p fabro-client -p fabro-test --all-targets -- -D warnings +``` + +Expected: all PASS. If nextest reports unrelated leaky tests but exit code is 0, record the leaky test names in the final handoff. + +## Assumptions + +- A 30s control-plane request timeout is acceptable for CLI API calls such as `ps`, `doctor`, `settings`, auth refresh, and model listing. +- Long-lived streams are not created through these default control-plane transports in a way that should be capped at 30s; if a verification failure proves otherwise, move the timeout to only startup/health-specific clients instead of global default target transport. +- The existing Unix silent-peer fix in `server_client.rs` remains part of the baseline change and should not be reverted. diff --git a/lib/crates/fabro-cli/src/commands/server/start.rs b/lib/crates/fabro-cli/src/commands/server/start.rs index cb87cdcf5..e83fabe51 100644 --- a/lib/crates/fabro-cli/src/commands/server/start.rs +++ b/lib/crates/fabro-cli/src/commands/server/start.rs @@ -4,7 +4,7 @@ )] use std::path::{Path, PathBuf}; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::{Context, Result, anyhow, bail}; use fabro_config::RuntimeDirectory; @@ -18,13 +18,14 @@ use fabro_static::EnvVars; use fabro_types::settings::{LogDestination, ServerAuthMethod}; use fabro_util::printer::Printer; use fabro_util::terminal::Styles; -use tokio::net::{TcpStream, UnixStream}; use tokio::process::Command as TokioCommand; use tokio::task::spawn_blocking; use tokio::time; use crate::local_server; +const SERVER_START_HEALTH_PROBE_TIMEOUT: Duration = Duration::from_millis(250); + pub(crate) struct ForegroundServerLogBootstrap { #[expect(dead_code, reason = "held for its Drop to release the server lock")] lock_file: std::fs::File, @@ -349,9 +350,9 @@ async fn execute_daemon( let poll_interval = Duration::from_millis(50); let timeout = Duration::from_secs(5); - let mut elapsed = Duration::ZERO; + let deadline = Instant::now() + timeout; - while elapsed < timeout { + while Instant::now() < deadline { let daemon = match ServerDaemon::read(&runtime_directory) { Ok(daemon) => daemon, Err(err) => { @@ -362,7 +363,7 @@ async fn execute_daemon( } }; if let Some(daemon) = daemon { - if try_connect(&daemon.bind).await { + if try_health(&daemon.bind).await { if announce { let pid = child.id().unwrap_or_default(); maybe_warn_host_port_fallback(bind, &daemon.bind, printer); @@ -394,8 +395,11 @@ async fn execute_daemon( bail!("Server exited during startup with status {status}"); } - time::sleep(poll_interval).await; - elapsed += poll_interval; + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + time::sleep(poll_interval.min(remaining)).await; } ServerDaemon::remove(&runtime_directory); @@ -456,15 +460,47 @@ async fn acquire_lock(runtime_directory: &RuntimeDirectory) -> Result bool { - let connect_timeout = Duration::from_millis(100); +async fn try_health(bind: &Bind) -> bool { + let Ok((base_url, client)) = build_health_client(bind) else { + return false; + }; + + let response = time::timeout( + SERVER_START_HEALTH_PROBE_TIMEOUT, + client.get(format!("{base_url}/health")).send(), + ) + .await; + + matches!(response, Ok(Ok(response)) if response.status().is_success()) +} + +fn build_health_client(bind: &Bind) -> Result<(String, fabro_http::HttpClient)> { match bind { - Bind::Tcp(addr) => time::timeout(connect_timeout, TcpStream::connect(addr)) - .await - .is_ok_and(|r| r.is_ok()), - Bind::Unix(path) => time::timeout(connect_timeout, UnixStream::connect(path)) - .await - .is_ok_and(|r| r.is_ok()), + Bind::Tcp(addr) => Ok(( + format!("http://{addr}"), + fabro_http::HttpClientBuilder::new() + .no_proxy() + .timeout(SERVER_START_HEALTH_PROBE_TIMEOUT) + .build()?, + )), + Bind::Unix(path) => { + #[cfg(unix)] + { + Ok(( + "http://fabro".to_string(), + fabro_http::HttpClientBuilder::new() + .unix_socket(path) + .no_proxy() + .timeout(SERVER_START_HEALTH_PROBE_TIMEOUT) + .build()?, + )) + } + #[cfg(not(unix))] + { + let _ = path; + bail!("Unix-socket HTTP client is not supported on this platform") + } + } } } @@ -495,17 +531,22 @@ fn read_log_tail(log_path: &Path, lines: usize) -> String { #[cfg(test)] mod tests { - use fabro_config::bind::BindRequest; + use std::time::Duration; + + use fabro_config::bind::{Bind, BindRequest}; use fabro_server::serve::ServeArgs; use fabro_static::EnvVars; use fabro_types::settings::LogDestination; use fabro_util::Home; use fabro_util::printer::Printer; use temp_env::with_var; + use tokio::net::TcpListener; use tokio::runtime::Runtime; + use tokio::time; use super::{ ensure_storage_server_autostart_allowed, execute_daemon, prepare_foreground_server_log, + try_health, }; fn runtime() -> Runtime { @@ -545,6 +586,22 @@ destination = "{destination}" } } + #[tokio::test] + async fn try_health_returns_false_for_tcp_peer_that_accepts_without_http_response() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + if let Ok((_stream, _addr)) = listener.accept().await { + time::sleep(Duration::from_secs(10)).await; + } + }); + + let ready = try_health(&Bind::Tcp(addr)).await; + + server.abort(); + assert!(!ready); + } + #[test] fn ensure_server_running_for_storage_errors_when_install_mode_is_required() { let temp_home = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index 86646aec3..952957a68 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -13,12 +13,15 @@ use fabro_config::bind::Bind; pub(crate) use fabro_types::RunProjection; use fabro_types::UserSettings; use fabro_util::dev_token; -use tokio::time::sleep; +use tokio::time::{self, sleep}; use crate::args::ServerTargetArgs; use crate::commands::server::start; use crate::user_config::{self, cli_http_client_builder}; +const SERVER_HEALTH_PROBE_TIMEOUT: Duration = Duration::from_millis(250); +const CLI_CONTROL_PLANE_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + fn refreshable_oauth( target: &ServerTarget, credential: Option<&Credential>, @@ -129,7 +132,11 @@ async fn connect_local_api_client_bundle( .dev_token_path(); let token = wait_for_runtime_dev_token(&runtime_token_path).await?; let http_client = connect_unix_socket_http_client(&path, Some(&token)).await?; - Ok(Client::from_http_client("http://fabro", http_client)) + Ok(Client::builder() + .transport("http://fabro", http_client) + .request_timeout(CLI_CONTROL_PLANE_REQUEST_TIMEOUT) + .connect() + .await?) } Bind::Tcp(addr) => { let target = ServerTarget::http_url(format!("http://{addr}"))?; @@ -154,7 +161,8 @@ async fn build_client( ) -> Result { let mut builder = Client::builder() .target(target.clone()) - .transport_connector(build_cli_transport_connector(target)); + .transport_connector(build_cli_transport_connector(target)) + .request_timeout(CLI_CONTROL_PLANE_REQUEST_TIMEOUT); if let Some((base_url, http_client)) = transport { builder = builder.transport(base_url, http_client); } @@ -305,10 +313,20 @@ fn should_bypass_proxy_for_http_target(api_url: &str) -> bool { } async fn check_server_ready(http_client: &fabro_http::HttpClient) -> Result<()> { - match http_client.get("http://fabro/health").send().await { - Ok(response) if response.status().is_success() => Ok(()), - Ok(response) => bail!("server health check returned status {}", response.status()), - Err(err) => Err(anyhow!(err)), + let response = match time::timeout( + SERVER_HEALTH_PROBE_TIMEOUT, + http_client.get("http://fabro/health").send(), + ) + .await + { + Ok(Ok(response)) => response, + Ok(Err(err)) => return Err(anyhow!(err)), + Err(_) => bail!("server health check timed out"), + }; + + match response { + response if response.status().is_success() => Ok(()), + response => bail!("server health check returned status {}", response.status()), } } @@ -333,6 +351,9 @@ mod tests { use fabro_client::{AuthEntry, DevTokenEntry, OAuthEntry, StoredSubject}; use httpmock::Method::{GET, POST}; use serde_json::json; + use tokio::net::TcpListener; + #[cfg(unix)] + use tokio::net::UnixListener; use super::*; @@ -357,6 +378,55 @@ mod tests { assert!(matches!(credential, Some(Credential::DevToken(found)) if found == token)); } + #[cfg(unix)] + #[tokio::test] + async fn unix_socket_probe_times_out_when_peer_accepts_without_http_response() { + let dir = tempfile::tempdir().unwrap(); + let socket_path = dir.path().join("hung.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + let server = tokio::spawn(async move { + if let Ok((stream, _addr)) = listener.accept().await { + let _stream = stream; + sleep(Duration::from_secs(10)).await; + } + }); + + let result = time::timeout( + Duration::from_millis(500), + try_connect_unix_socket_http_client(&socket_path, None), + ) + .await; + + server.abort(); + assert!( + result.is_ok(), + "Unix socket health probe should return its own timeout error instead of hanging" + ); + assert!(result.unwrap().is_err()); + } + + #[tokio::test] + async fn http_target_transport_times_out_when_peer_accepts_without_http_response() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + if let Ok((_stream, _addr)) = listener.accept().await { + sleep(Duration::from_secs(10)).await; + } + }); + + let target = ServerTarget::http_url(format!("http://{addr}")).unwrap(); + let client = connect_target_api_client_bundle(&target).await.unwrap(); + let result = time::timeout(Duration::from_millis(750), client.get_health()).await; + + server.abort(); + assert!( + result.is_ok(), + "HTTP target health check should return its own timeout error instead of hanging" + ); + assert!(result.unwrap().is_err()); + } + #[test] fn resolve_local_tcp_credential_uses_live_oauth_entry() { let dir = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-client/src/client.rs b/lib/crates/fabro-client/src/client.rs index 4b6775618..312b9c3da 100644 --- a/lib/crates/fabro-client/src/client.rs +++ b/lib/crates/fabro-client/src/client.rs @@ -20,6 +20,7 @@ use futures::future::BoxFuture; use serde::{Deserialize, Serialize}; use tokio::fs::File; use tokio::sync::Mutex; +use tokio::time; use tokio_util::io::ReaderStream; use crate::credential::Credential; @@ -31,6 +32,10 @@ use crate::session::OAuthSession; use crate::target::ServerTarget; use crate::{AuthEntry, OAuthEntry, StoredSubject, sse}; +const DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(30); +const DEFAULT_HEALTH_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(250); + type TransportFuture = BoxFuture<'static, Result<(fabro_http::HttpClient, String)>>; pub struct RunEventStream { @@ -58,6 +63,7 @@ pub struct Client { oauth_session: Option, refresh_lock: Arc>, transport_connector: Option, + request_timeout: Option, } #[derive(Clone)] @@ -72,6 +78,7 @@ pub struct ClientBuilder { oauth_session: Option, transport: Option<(String, fabro_http::HttpClient)>, transport_connector: Option, + request_timeout: Option, } #[derive(Debug, Deserialize)] @@ -206,6 +213,12 @@ impl ClientBuilder { self } + #[must_use] + pub fn request_timeout(mut self, timeout: std::time::Duration) -> Self { + self.request_timeout = Some(timeout); + self + } + pub async fn connect(self) -> Result { let bearer_token = self .credential @@ -217,6 +230,11 @@ impl ClientBuilder { .as_ref() .map(|session| session.target.clone()) }); + let uses_default_target_transport = + self.transport.is_none() && self.transport_connector.is_none() && target.is_some(); + let request_timeout = self.request_timeout.or_else(|| { + uses_default_target_transport.then_some(DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT) + }); let transport_connector = self .transport_connector .or_else(|| target.map(default_transport_connector)); @@ -236,6 +254,7 @@ impl ClientBuilder { oauth_session: self.oauth_session, refresh_lock: Arc::new(Mutex::new(())), transport_connector, + request_timeout, }) } } @@ -260,6 +279,7 @@ impl Client { oauth_session: None, refresh_lock: Arc::new(Mutex::new(())), transport_connector: None, + request_timeout: None, } } @@ -302,6 +322,17 @@ impl Client { .expect("client state lock should not be poisoned") = state; } + async fn with_request_timeout(&self, future: impl Future) -> Result { + let Some(timeout) = self.request_timeout else { + return Ok(future.await); + }; + + match time::timeout(timeout, future).await { + Ok(value) => Ok(value), + Err(_) => bail!("server request timed out after {timeout:?}"), + } + } + async fn send_api( &self, request: F, @@ -317,7 +348,10 @@ impl Client { E: serde::Serialize + std::fmt::Debug + Send + Sync + 'static, { let state = self.current_state(); - match request.clone()(state.client.clone()).await { + match self + .with_request_timeout(Box::pin(request.clone()(state.client.clone()))) + .await? + { Ok(response) => Ok(response), Err(err) => { let mapped = classify_api_error(err).await; @@ -325,7 +359,10 @@ impl Client { if let Some(failed_token) = state.bearer_token.as_deref() { self.refresh_access_token(failed_token).await?; let state = self.current_state(); - return request(state.client.clone()).await.map_err(map_api_error); + return self + .with_request_timeout(Box::pin(request(state.client.clone()))) + .await? + .map_err(map_api_error); } } Err(mapped.error) @@ -473,8 +510,9 @@ impl Client { T: Into, { let state = self.current_state(); - let response = request.clone()(state.http_client.clone()) - .await + let response = self + .with_request_timeout(Box::pin(request.clone()(state.http_client.clone()))) + .await? .map_err(Into::into)?; match classify_http_response(response).await? { Ok(response) => Ok(Ok(response)), @@ -483,8 +521,9 @@ impl Client { if let Some(failed_token) = state.bearer_token.as_deref() { self.refresh_access_token(failed_token).await?; let state = self.current_state(); - let response = request(state.http_client.clone()) - .await + let response = self + .with_request_timeout(Box::pin(request(state.http_client.clone()))) + .await? .map_err(Into::into)?; return classify_http_response(response).await; } @@ -658,8 +697,17 @@ impl Client { } pub async fn get_health(&self) -> Result<()> { - self.send_api(|client| async move { client.get_health().send().await }) - .await?; + match time::timeout( + DEFAULT_HEALTH_REQUEST_TIMEOUT, + self.send_api(|client| async move { client.get_health().send().await }), + ) + .await + { + Ok(result) => { + result?; + } + Err(_) => bail!("server health check timed out"), + } Ok(()) } @@ -1550,6 +1598,7 @@ fn add_pr_upgrade_hint(err: anyhow::Error) -> anyhow::Error { #[cfg(test)] mod tests { use std::sync::Arc; + use std::time::Duration; use chrono::Duration as ChronoDuration; use fabro_util::exit; @@ -1643,6 +1692,49 @@ mod tests { server.abort(); } + #[tokio::test] + async fn request_timeout_does_not_cap_stream_body_after_headers() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = vec![0_u8; 4096]; + let read = stream.read(&mut request).await.unwrap(); + let request = String::from_utf8_lossy(&request[..read]); + assert!( + request.starts_with("GET /api/v1/attach HTTP/1.1"), + "unexpected attach request: {request}" + ); + + let body = b"data: hello\n\n"; + let headers = format!( + "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ); + stream.write_all(headers.as_bytes()).await.unwrap(); + time::sleep(Duration::from_millis(100)).await; + stream.write_all(body).await.unwrap(); + }); + + let target = ServerTarget::http_url(format!("http://{addr}")).unwrap(); + let client = Client::builder() + .target(target) + .request_timeout(Duration::from_millis(50)) + .connect() + .await + .unwrap(); + + let mut stream = client.attach_events(&[]).await.unwrap(); + let chunk = time::timeout(Duration::from_millis(500), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + + assert_eq!(chunk, Bytes::from_static(b"data: hello\n\n")); + server.await.unwrap(); + } + async fn oauth_client( server: &MockServer, ) -> (tempfile::TempDir, Client, AuthStore, ServerTarget) { diff --git a/lib/crates/fabro-client/src/target.rs b/lib/crates/fabro-client/src/target.rs index 442ec952d..c17577181 100644 --- a/lib/crates/fabro-client/src/target.rs +++ b/lib/crates/fabro-client/src/target.rs @@ -4,6 +4,9 @@ use std::str::FromStr; use anyhow::{Result, bail}; +const DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(30); + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub enum ServerTarget { HttpUrl(CanonicalHttpUrl), @@ -50,7 +53,9 @@ impl ServerTarget { pub fn build_public_http_client(&self) -> Result<(fabro_http::HttpClient, String)> { if let Some(api_url) = self.as_http_url() { - let http_client = fabro_http::HttpClientBuilder::new().build()?; + let http_client = fabro_http::HttpClientBuilder::new() + .timeout(DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT) + .build()?; return Ok((http_client, api_url.to_string())); } @@ -63,6 +68,7 @@ impl ServerTarget { let http_client = fabro_http::HttpClientBuilder::new() .unix_socket(path) .no_proxy() + .timeout(DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT) .build()?; Ok((http_client, "http://fabro".to_string())) } diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index 807e441f3..77841bce7 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -2268,7 +2268,10 @@ pub async fn twin_openai() -> &'static TwinOpenAi { let client = test_http_client(); let healthz_url = format!("http://127.0.0.1:{}/healthz", addr.port()); for _ in 0..50 { - if let Ok(resp) = client.get(&healthz_url).send().await { + let response = + time::timeout(Duration::from_millis(250), client.get(&healthz_url).send()) + .await; + if let Ok(Ok(resp)) = response { let status = resp.status(); if status == fabro_http::StatusCode::OK { return TwinOpenAi { base_url };