diff --git a/rust/fetch/src/builder.rs b/rust/fetch/src/builder.rs index 7f64a80..035df1b 100644 --- a/rust/fetch/src/builder.rs +++ b/rust/fetch/src/builder.rs @@ -64,6 +64,7 @@ impl FetchBuilder { fetch_options: FetchOptions { timeout: Some(defaults::default_timeout_options()), retry: Some(defaults::default_retry_options()), + connect_timeout_ms: None, }, container_options: FetchContainerOptions::default(), default_init: None, @@ -85,6 +86,21 @@ impl FetchBuilder { self } + /// Set the connect timeout in milliseconds. + /// + /// This bounds only the connection-establishment phase. When set, the + /// underlying reqwest client is built with `connect_timeout(...)`, so a + /// connect that never completes (e.g. a black-holed SYN to a dead pod IP + /// still lingering in a ClusterIP's iptables) fails in ~this window and the + /// configured retry can land on a live endpoint — instead of stalling until + /// the whole-request timeout ([`with_timeout`]). Slow-but-alive handlers are + /// unaffected. Leaving it unset preserves the previous no-connect-timeout + /// behavior. + pub fn with_connect_timeout(mut self, ms: u64) -> Self { + self.fetch_options.connect_timeout_ms = Some(ms); + self + } + /// Configure retry behavior. pub fn with_retry(mut self, options: RetryOptions) -> Self { self.fetch_options.retry = Some(options); diff --git a/rust/fetch/src/client.rs b/rust/fetch/src/client.rs index 1562574..21144f1 100644 --- a/rust/fetch/src/client.rs +++ b/rust/fetch/src/client.rs @@ -1,6 +1,7 @@ //! Core fetch client with full pipeline: hooks, timeout, retry, rate limit, circuit breaker. use std::collections::HashMap; +use std::time::Duration; use serde::de::DeserializeOwned; use tracing; @@ -75,8 +76,14 @@ async fn acquire_with_retry( async fn do_single_request( url: &str, init: &RequestInit, + connect_timeout_ms: Option, ) -> Result, FetchError> { - let client = reqwest::Client::new(); + let client = match connect_timeout_ms { + Some(ms) => reqwest::Client::builder() + .connect_timeout(Duration::from_millis(ms)) + .build()?, + None => reqwest::Client::new(), + }; let mut request_builder = client.request(init.method.to_reqwest(), url); @@ -220,11 +227,18 @@ pub async fn fetch( .as_ref() .map(|t| t.timeout_ms) .unwrap_or(defaults::DEFAULT_TIMEOUT_MS); + let connect_timeout_ms = opts.connect_timeout_ms; let operation = |_attempt: u32| { let url = url_clone.clone(); let init = init_clone.clone(); - async move { timeout::with_timeout(timeout_ms, do_single_request::(&url, &init)).await } + async move { + timeout::with_timeout( + timeout_ms, + do_single_request::(&url, &init, connect_timeout_ms), + ) + .await + } }; // 4. Execute with retry (or just once if no retry options) @@ -232,7 +246,11 @@ pub async fn fetch( retry::execute_with_retry(retry_opts, operation).await } else { // No retry, just execute once with timeout - timeout::with_timeout(timeout_ms, do_single_request::(&url, &init)).await + timeout::with_timeout( + timeout_ms, + do_single_request::(&url, &init, connect_timeout_ms), + ) + .await }; // Record success/failure with circuit breaker diff --git a/rust/fetch/src/lib.rs b/rust/fetch/src/lib.rs index 9862ddc..d1733e5 100644 --- a/rust/fetch/src/lib.rs +++ b/rust/fetch/src/lib.rs @@ -76,6 +76,7 @@ pub async fn fetch( init: RequestInit, ) -> Result, FetchError> { let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(defaults::default_timeout_options()), retry: Some(defaults::default_retry_options()), }; diff --git a/rust/fetch/src/types.rs b/rust/fetch/src/types.rs index fc52400..ffd0644 100644 --- a/rust/fetch/src/types.rs +++ b/rust/fetch/src/types.rs @@ -164,6 +164,12 @@ pub struct FetchOptions { pub timeout: Option, /// Retry configuration. pub retry: Option, + /// Connect timeout in milliseconds. When set, the underlying reqwest client + /// is built with `connect_timeout(...)` so a connection that never + /// establishes (e.g. a black-holed SYN to a stale/dead endpoint IP) fails + /// fast instead of blocking until the whole-request `timeout`. `None` + /// (default) preserves the previous behavior with no connect timeout. + pub connect_timeout_ms: Option, } /// HTTP method type. diff --git a/rust/fetch/tests/connect_timeout_tests.rs b/rust/fetch/tests/connect_timeout_tests.rs new file mode 100644 index 0000000..bd71c9f --- /dev/null +++ b/rust/fetch/tests/connect_timeout_tests.rs @@ -0,0 +1,54 @@ +//! Test that a bounded connect timeout fails fast on a black-holed connect +//! instead of stalling until the (much larger) whole-request timeout. +//! +//! Regression coverage for SMOODEV-2498 / SMOODEV-2481: api-prime's ~16s stalls +//! were fresh SYNs to dead pod IPs still lingering in a ClusterIP's iptables. +//! Without a connect timeout, reqwest waits the full whole-request timeout; with +//! one, the connect fails in ~the configured window and retry lands on a live pod. + +use std::time::{Duration, Instant}; + +use serde_json::Value; +use smooai_fetch::client; +use smooai_fetch::error::FetchError; +use smooai_fetch::types::{FetchOptions, Method, RequestInit, TimeoutOptions}; + +#[tokio::test] +async fn connect_timeout_fails_fast_on_black_hole() { + // 10.255.255.1 is a non-routable RFC1918 address with (almost certainly) no + // host answering: the SYN is dropped/black-holed, so the connect never + // establishes and would otherwise hang until the whole-request timeout. + let url = "http://10.255.255.1:80/anything"; + + let connect_timeout_ms = 500; + // Whole-request timeout is 10x the connect timeout — if the connect timeout + // is NOT honored, this test would take ~10s and the elapsed assertion fails. + let options = FetchOptions { + connect_timeout_ms: Some(connect_timeout_ms), + timeout: Some(TimeoutOptions { timeout_ms: 5_000 }), + retry: None, + }; + + let init = RequestInit { + method: Method::GET, + ..Default::default() + }; + + let start = Instant::now(); + let result = client::fetch::(url, init, Some(options), None, None, None, None).await; + let elapsed = start.elapsed(); + + // The connect must fail (not succeed against a black hole). + assert!(result.is_err(), "expected connect to fail, got {result:?}"); + // reqwest surfaces a connect timeout as a request error, not our whole-request + // Timeout variant (which would mean the connect timeout never fired). + match result.unwrap_err() { + FetchError::Request(_) => {} + other => panic!("expected FetchError::Request from connect timeout, got {other:?}"), + } + // Must fail in roughly the connect window, well under the 5s whole timeout. + assert!( + elapsed < Duration::from_secs(3), + "connect timeout did not fire fast: elapsed {elapsed:?} (connect_timeout was {connect_timeout_ms}ms)" + ); +} diff --git a/rust/fetch/tests/fetch_tests.rs b/rust/fetch/tests/fetch_tests.rs index 1ce7ae5..8c9ec9a 100644 --- a/rust/fetch/tests/fetch_tests.rs +++ b/rust/fetch/tests/fetch_tests.rs @@ -34,6 +34,7 @@ async fn test_basic_get_request() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -78,6 +79,7 @@ async fn test_basic_post_request_with_body() { body: Some(r#"{"key":"value"}"#.to_string()), }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -117,6 +119,7 @@ async fn test_failed_request_404() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -163,6 +166,7 @@ async fn test_error_response_with_type_code_message() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -210,6 +214,7 @@ async fn test_non_json_response() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -251,6 +256,7 @@ async fn test_request_with_custom_headers() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -284,6 +290,7 @@ async fn test_schema_validation_error_on_type_mismatch() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -321,6 +328,7 @@ async fn test_default_method_is_get() { let url = format!("{}/default-method", mock_server.uri()); let init = RequestInit::default(); let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; diff --git a/rust/fetch/tests/hooks_tests.rs b/rust/fetch/tests/hooks_tests.rs index d5d359e..647a216 100644 --- a/rust/fetch/tests/hooks_tests.rs +++ b/rust/fetch/tests/hooks_tests.rs @@ -44,6 +44,7 @@ async fn test_pre_request_hook_modifies_url() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -94,6 +95,7 @@ async fn test_pre_request_hook_adds_headers() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -131,6 +133,7 @@ async fn test_pre_request_hook_returns_none_passes_through() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -175,6 +178,7 @@ async fn test_post_response_success_hook_modifies_response() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -219,6 +223,7 @@ async fn test_post_response_success_hook_returns_none_passes_through() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -267,6 +272,7 @@ async fn test_post_response_error_hook_replaces_error() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -317,6 +323,7 @@ async fn test_post_response_error_hook_returns_none_passes_through() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -386,6 +393,7 @@ async fn test_all_hooks_together() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; diff --git a/rust/fetch/tests/retry_tests.rs b/rust/fetch/tests/retry_tests.rs index 1c60184..7a0f04c 100644 --- a/rust/fetch/tests/retry_tests.rs +++ b/rust/fetch/tests/retry_tests.rs @@ -131,6 +131,7 @@ async fn test_retry_succeeds_after_failures() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 2, @@ -185,6 +186,7 @@ async fn test_retry_exhausted() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 2, @@ -241,6 +243,7 @@ async fn test_non_retryable_error_not_retried() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 3, @@ -291,6 +294,7 @@ async fn test_retry_with_retry_after_header() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 10000 }), retry: Some(RetryOptions { attempts: 1, @@ -353,6 +357,7 @@ async fn test_fast_first_skips_initial_delay() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 1, @@ -428,6 +433,7 @@ async fn test_on_rejection_retry_decision_overrides_delay() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 1, @@ -486,6 +492,7 @@ async fn test_on_rejection_abort_stops_retry_loop() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 3, @@ -535,6 +542,7 @@ async fn test_on_rejection_default_falls_through_to_exponential() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 1, @@ -585,6 +593,7 @@ async fn test_on_rejection_skip_consumes_attempt_without_sleep() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: Some(RetryOptions { attempts: 2, diff --git a/rust/fetch/tests/timeout_tests.rs b/rust/fetch/tests/timeout_tests.rs index fb92a93..7663310 100644 --- a/rust/fetch/tests/timeout_tests.rs +++ b/rust/fetch/tests/timeout_tests.rs @@ -29,6 +29,7 @@ async fn test_request_completes_before_timeout() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 5000 }), retry: None, }; @@ -63,6 +64,7 @@ async fn test_request_times_out() { ..Default::default() }; let options = FetchOptions { + connect_timeout_ms: None, timeout: Some(TimeoutOptions { timeout_ms: 200 }), retry: None, };