From a3ae2ebbc87957d89cb7a689fb8edfe30e4a5a3d Mon Sep 17 00:00:00 2001 From: rita-aga <36133358+rita-aga@users.noreply.github.com> Date: Sat, 24 Oct 2026 01:07:09 +0000 Subject: [PATCH] fix(server): reuse the S3 client and retry a stalled blob GET Every BlobStore::s3 built a new reqwest client, so each get paid a fresh TCP and TLS handshake. On 2026-10-23 those handshakes were the transport: successful s3 gets were idle for nearly their whole duration, p95 about 1.5s with a 2.5s tail, while semaphore wait stayed around 5ms. Gets still idle at about 8s were cancelled before they recorded an outcome, and the File response failed. Share one process-wide client (pool idle 45s, connect timeout 2s) and retry a buffered GET once inside a 7s budget. HTTP statuses are not retried. Puts and streams share the client without the retry. --- crates/temper-server/src/blob_store.rs | 68 ++--- .../src/blob_store/s3_transport.rs | 277 ++++++++++++++++++ .../temper-server/src/blob_store/streaming.rs | 12 +- .../adrs/0182-s3-blob-get-connection-reuse.md | 57 ++++ 4 files changed, 373 insertions(+), 41 deletions(-) create mode 100644 crates/temper-server/src/blob_store/s3_transport.rs create mode 100644 docs/adrs/0182-s3-blob-get-connection-reuse.md diff --git a/crates/temper-server/src/blob_store.rs b/crates/temper-server/src/blob_store.rs index df39b6860..ebd387091 100644 --- a/crates/temper-server/src/blob_store.rs +++ b/crates/temper-server/src/blob_store.rs @@ -19,6 +19,7 @@ mod keys; mod limits; mod local; mod raw_ingest; +mod s3_transport; mod state; mod streaming; pub(crate) use endpoint::{LocalInternalBlobEndpoint, is_local_internal_blob_endpoint}; @@ -31,6 +32,7 @@ pub(crate) use raw_ingest::BlobIngestProgressPolicy; pub(crate) use raw_ingest::{ BlobIngestAdmissionError, BlobIngestBudget, BlobStageError, MAX_RAW_BLOB_BYTES, }; +use s3_transport::S3Http; pub(crate) use streaming::BlobReadBounded; pub use streaming::{BlobObjectStream, BlobStreamRead, decode_json_base64_stream}; @@ -53,7 +55,7 @@ struct S3BlobStore { key_prefix: Option, access_key: Option, secret_key: Option, - client: reqwest::Client, + http: S3Http, } impl BlobStore { @@ -81,10 +83,27 @@ impl BlobStore { key_prefix, access_key, secret_key, - client: reqwest::Client::builder() - .connect_timeout(std::time::Duration::from_secs(10)) - .build() - .expect("static blob HTTP client configuration must be valid"), // ci-ok: static reqwest builder options + http: S3Http::production(), + }), + } + } + + /// S3 store whose GET retry budget is short enough for a unit test. + #[cfg(test)] + pub(crate) fn s3_with_get_retry( + endpoint: impl Into, + attempt_timeout: Duration, + total_budget: Duration, + ) -> Self { + Self { + staging_root: PathBuf::from("target/blob-get-retry-test"), + backend: BlobStoreBackend::S3(S3BlobStore { + endpoint: endpoint.into().trim_end_matches('/').to_string(), + bucket: "bucket".to_string(), + key_prefix: None, + access_key: None, + secret_key: None, + http: S3Http::for_test(attempt_timeout, 2, total_budget), }), } } @@ -214,7 +233,7 @@ impl S3BlobStore { let span = blob_transport_span(operation, "s3", request_bytes); let result = async { let url = self.object_url(key); - let mut request = self.client.put(&url).body(body.to_vec()); + let mut request = self.http.client().put(&url).body(body.to_vec()); let headers = self .signed_headers(Method::PUT, &url) .map_err(BlobTransportError::message)?; @@ -279,7 +298,8 @@ impl S3BlobStore { let result = async { let url = self.object_url(key); let mut request = self - .client + .http + .client() .put(&url) .header(reqwest::header::CONTENT_LENGTH, content_len) .timeout(raw_ingest::BLOB_BACKEND_OPERATION_TIMEOUT) @@ -341,37 +361,7 @@ impl S3BlobStore { async fn get(&self, key: &str) -> Result>, String> { let started_at = Instant::now(); // determinism-ok: production blob transport metric only let span = blob_transport_span("get", "s3", 0); - let result = async { - let url = self.object_url(key); - let mut request = self.client.get(&url); - let headers = self - .signed_headers(Method::GET, &url) - .map_err(BlobTransportError::message)?; - for (header_name, header_value) in &headers { - request = request.header(header_name, header_value); - } - - let response = request.send().await.map_err(|e| { - BlobTransportError::message(format!("blob GET request failed for '{key}': {e}")) - })?; - let status = response.status(); - if status == StatusCode::NOT_FOUND { - return Ok((status, None)); - } - if !status.is_success() { - return Err(BlobTransportError::status( - format!("blob GET failed for '{key}' with HTTP {status}"), - status, - )); - } - - let bytes = response.bytes().await.map_err(|e| { - BlobTransportError::message(format!("blob GET body read failed for '{key}': {e}")) - })?; - Ok((status, Some(bytes.to_vec()))) - } - .instrument(span.clone()) - .await; + let result = self.read_object(key).instrument(span.clone()).await; match result { Ok((status, bytes)) => { @@ -410,7 +400,7 @@ impl S3BlobStore { let span = blob_transport_span("head", "s3", 0); let result = async { let url = self.object_url(key); - let mut request = self.client.head(&url); + let mut request = self.http.client().head(&url); let headers = self .signed_headers(Method::HEAD, &url) .map_err(BlobTransportError::message)?; diff --git a/crates/temper-server/src/blob_store/s3_transport.rs b/crates/temper-server/src/blob_store/s3_transport.rs new file mode 100644 index 000000000..22b1dd8aa --- /dev/null +++ b/crates/temper-server/src/blob_store/s3_transport.rs @@ -0,0 +1,277 @@ +//! Shared S3 HTTP client and the bounded GET retry. +//! +//! Every `BlobStore::s3` used to build its own `reqwest::Client`. Callers +//! construct a store per read, so the pool died with the call and every GET +//! paid a fresh TCP and TLS handshake. On 2026-10-23 those handshakes were +//! the blob transport: successful `backend:s3, operation:get` spans were +//! idle for essentially their whole duration (busy time under 10ms), p95 +//! about 1.5s, alerted tail about 2.5s, on objects as small as a few tens +//! of kilobytes. Semaphore wait stayed around 5ms, and cache hits stayed +//! near 0ms. Forty-three gets that were still idle at 7.94–7.99s never +//! recorded an outcome: the caller cancelled them first, and the File +//! response failed instead of returning the bytes. +//! +//! The shared client keeps live connections. A stalled attempt is bounded +//! below that cancellation and retried once, so a dead handshake does not +//! spend the whole response budget. + +use std::sync::OnceLock; +use std::time::{Duration, Instant}; + +use reqwest::StatusCode; + +use super::S3BlobStore; +use crate::blob_transport_observability::BlobTransportError; + +/// One attempt, above the 2.5s tail and below the ~8s cancellation. +const S3_GET_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(3); +/// A hung connect should fail inside one attempt, leaving room to retry. +const S3_CONNECT_TIMEOUT: Duration = Duration::from_secs(2); +const S3_GET_ATTEMPTS: u32 = 2; +/// Two attempts still finish before the observed ~8s cancellation. +const S3_GET_TOTAL_BUDGET: Duration = Duration::from_secs(7); +/// Drop idle pooled sockets before a typical NAT or load-balancer idle +/// timeout blackholes them. Reuse is what removes the handshake; keeping +/// a dead socket would put the stall back. +const S3_POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(45); +const S3_TCP_KEEPALIVE: Duration = Duration::from_secs(15); +const S3_HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15); +const S3_HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(4); + +#[derive(Clone, Debug)] +pub(super) struct S3Http { + client: reqwest::Client, + get_attempt_timeout: Duration, + get_attempts: u32, + get_total_budget: Duration, +} + +impl S3Http { + pub(super) fn production() -> Self { + Self { + client: shared_client(), + get_attempt_timeout: S3_GET_ATTEMPT_TIMEOUT, + get_attempts: S3_GET_ATTEMPTS, + get_total_budget: S3_GET_TOTAL_BUDGET, + } + } + + #[cfg(test)] + pub(super) fn for_test( + attempt_timeout: Duration, + attempts: u32, + total_budget: Duration, + ) -> Self { + let client = reqwest::Client::builder() + .connect_timeout(attempt_timeout) + .pool_max_idle_per_host(0) + .build() + .expect("test blob HTTP client configuration must be valid"); + Self { + client, + get_attempt_timeout: attempt_timeout, + get_attempts: attempts.max(1), + get_total_budget: total_budget, + } + } + + pub(super) fn client(&self) -> &reqwest::Client { + &self.client + } +} + +fn shared_client() -> reqwest::Client { + static CLIENT: OnceLock = OnceLock::new(); + CLIENT + .get_or_init(|| { + reqwest::Client::builder() + .connect_timeout(S3_CONNECT_TIMEOUT) + .tcp_keepalive(S3_TCP_KEEPALIVE) + .pool_idle_timeout(S3_POOL_IDLE_TIMEOUT) + .pool_max_idle_per_host(32) + .http2_keep_alive_interval(S3_HTTP2_KEEPALIVE_INTERVAL) + .http2_keep_alive_timeout(S3_HTTP2_KEEPALIVE_TIMEOUT) + .http2_keep_alive_while_idle(true) + .build() + .expect("static blob HTTP client configuration must be valid") + }) + .clone() +} + +struct AttemptError { + retryable: bool, + transport: BlobTransportError, +} + +impl S3BlobStore { + /// Read one object, retrying a stalled or disconnected attempt once. + /// + /// HTTP status answers, including 404 and 5xx, are not retried: S3 + /// already responded. Only a timeout or a connection failure is. + pub(super) async fn read_object( + &self, + key: &str, + ) -> Result<(StatusCode, Option>), BlobTransportError> { + let started = Instant::now(); // determinism-ok: production S3 GET retry budget + let deadline = started + self.http.get_total_budget; + let attempts = self.http.get_attempts.max(1); + let mut last_error = None; + + for attempt in 0..attempts { + let now = Instant::now(); // determinism-ok: production S3 GET retry budget + let remaining = deadline.saturating_duration_since(now); + if attempt > 0 && remaining.is_zero() { + break; + } + let timeout = self + .http + .get_attempt_timeout + .min(remaining.max(Duration::from_millis(1))); + match self.read_object_once(key, timeout).await { + Ok(body) => return Ok(body), + Err(error) if error.retryable && attempt + 1 < attempts => { + last_error = Some(error.transport); + } + Err(error) => return Err(error.transport), + } + } + + Err(last_error.unwrap_or_else(|| { + BlobTransportError::message(format!("blob GET exhausted its retry budget for '{key}'")) + })) + } + + async fn read_object_once( + &self, + key: &str, + timeout: Duration, + ) -> Result<(StatusCode, Option>), AttemptError> { + let url = self.object_url(key); + let headers = self + .signed_headers(reqwest::Method::GET, &url) + .map_err(|message| AttemptError { + retryable: false, + transport: BlobTransportError::message(message), + })?; + let mut request = self.http.client().get(&url).timeout(timeout); + for (header_name, header_value) in &headers { + request = request.header(header_name, header_value); + } + + let response = match request.send().await { + Ok(response) => response, + Err(error) => { + return Err(attempt_transport_error( + error, + format!("blob GET request failed for '{key}'"), + )); + } + }; + let status = response.status(); + if status == StatusCode::NOT_FOUND { + return Ok((status, None)); + } + if !status.is_success() { + return Err(AttemptError { + retryable: false, + transport: BlobTransportError::status( + format!("blob GET failed for '{key}' with HTTP {status}"), + status, + ), + }); + } + + match response.bytes().await { + Ok(bytes) => Ok((status, Some(bytes.to_vec()))), + Err(error) => Err(attempt_transport_error( + error, + format!("blob GET body read failed for '{key}'"), + )), + } + } +} + +fn attempt_transport_error(error: reqwest::Error, context: String) -> AttemptError { + let retryable = + error.is_timeout() || error.is_connect() || error.is_request() || error.is_body(); + AttemptError { + retryable, + transport: BlobTransportError::message(format!("{context}: {error}")), + } +} + +#[cfg(test)] +mod tests { + use std::time::{Duration, Instant}; + + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use crate::blob_store::BlobStore; + + #[tokio::test] + async fn slow_s3_get_retries_within_budget_and_returns_bytes() { + let server = MockServer::start().await; + // Last mounted mock wins. The stalled response is highest priority + // for one hit; the retry then matches the immediate body. + Mock::given(method("GET")) + .and(path("/bucket/obj")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"object-bytes".to_vec())) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/bucket/obj")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(Duration::from_secs(5)) + .set_body_bytes(b"stalled".to_vec()), + ) + .up_to_n_times(1) + .mount(&server) + .await; + + let store = BlobStore::s3_with_get_retry( + server.uri(), + Duration::from_millis(200), + Duration::from_secs(2), + ); + let started = Instant::now(); + let bytes = tokio::time::timeout(Duration::from_secs(2), store.get("obj")) + .await + .expect("slow S3 get must finish inside the response budget") + .expect("retry must turn the stalled attempt into a successful read"); + assert_eq!(bytes.as_deref(), Some(b"object-bytes".as_slice())); + assert!( + started.elapsed() < Duration::from_secs(2), + "retried get took {:?}", + started.elapsed() + ); + } + + #[tokio::test] + async fn s3_get_that_responds_within_the_attempt_succeeds_without_retry() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/bucket/obj")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(Duration::from_millis(100)) + .set_body_bytes(b"once".to_vec()), + ) + .expect(1) + .mount(&server) + .await; + + let store = BlobStore::s3_with_get_retry( + server.uri(), + Duration::from_secs(1), + Duration::from_secs(2), + ); + let bytes = tokio::time::timeout(Duration::from_secs(2), store.get("obj")) + .await + .expect("in-budget S3 get must not be abandoned") + .expect("in-budget S3 get must succeed"); + assert_eq!(bytes.as_deref(), Some(b"once".as_slice())); + } +} diff --git a/crates/temper-server/src/blob_store/streaming.rs b/crates/temper-server/src/blob_store/streaming.rs index 0518c46ea..1f53569a6 100644 --- a/crates/temper-server/src/blob_store/streaming.rs +++ b/crates/temper-server/src/blob_store/streaming.rs @@ -201,7 +201,11 @@ impl S3BlobStore { let span = blob_transport_span("get_bounded", "s3", 0); let result = async { let url = self.object_url(key); - let mut request = self.client.get(&url).timeout(BLOB_BOUNDED_READ_TIMEOUT); + let mut request = self + .http + .client() + .get(&url) + .timeout(BLOB_BOUNDED_READ_TIMEOUT); let headers = self .signed_headers(Method::GET, &url) .map_err(BlobTransportError::message)?; @@ -290,7 +294,11 @@ impl S3BlobStore { async fn get_stream(&self, key: &str, max_bytes: u64) -> Result { let url = self.object_url(key); - let mut request = self.client.get(&url).timeout(BLOB_STREAM_TOTAL_TIMEOUT); + let mut request = self + .http + .client() + .get(&url) + .timeout(BLOB_STREAM_TOTAL_TIMEOUT); let headers = self.signed_headers(Method::GET, &url)?; for (header_name, header_value) in &headers { request = request.header(header_name, header_value); diff --git a/docs/adrs/0182-s3-blob-get-connection-reuse.md b/docs/adrs/0182-s3-blob-get-connection-reuse.md new file mode 100644 index 000000000..f42b32d3b --- /dev/null +++ b/docs/adrs/0182-s3-blob-get-connection-reuse.md @@ -0,0 +1,57 @@ +# ADR-0182: Reuse the S3 blob client and retry a stalled GET + +- Status: Accepted +- Date: 2026-10-24 +- Deciders: Temper core maintainers +- Related: + - ADR-0071: Storage Retry Classification + - ADR-0093: Native Blob Transport Observability + - `crates/temper-server/src/blob_store.rs` + - `crates/temper-server/src/blob_store/s3_transport.rs` + +## Context + +On 2026-10-23 the native blob transport monitors for `service:temperpaw`, +`backend:s3`, `operation:get` flapped all day. The monitors named "wait" +query `temper_blob_native_transport_duration_ms`, not the semaphore +histogram. In the sampled spans, successful gets had a p95 around 1.5s and +a tail around 2.5s. Busy time on those spans was under 10ms; the rest was +socket idle, including reads of a few tens of kilobytes. Semaphore wait +stayed around 5ms. Cache hits (`outcome:cache`) stayed near 0ms. + +Forty-three gets never recorded an outcome. Their durations sat in a 50ms +band from 7.94s to 7.99s, which is the caller giving up while the GET was +still idle. `BlobStore::s3` built a new `reqwest::Client` on every store, +and every read builds a new store, so there was no connection pool. +`connect_timeout` was 10s, longer than that cancellation, and a timed-out +handshake was not retried. The File `$value` response failed instead of +returning bytes that a second attempt would have read. + +ADR-0071 keeps plain timeout text from being retried at the dispatch layer. +A retry of an idempotent GET belongs on the blob client, where a timeout is +a stalled socket rather than a saturated write gate. + +## Decision + +Share one process-wide S3 `reqwest::Client` across blob stores. Keep idle +pooled connections for 45s, send TCP and HTTP/2 keepalives, and bound +connect to 2s so a blackholed handshake fails inside one attempt. + +A buffered GET may try twice inside a 7s budget. Each attempt is capped at +3s. A timeout, connect failure, or dropped body is retried. An HTTP status, +including 404 and 5xx, is not: S3 already answered. Puts and streaming +reads use the same client and do not gain this retry. + +**Why this approach**: The idle-versus-busy split says the time is the +handshake, not the semaphore and not the cache. A shared pool removes that +handshake from the steady state. The 7s budget is inside the observed ~8s +cancellation, so a stalled first attempt can still succeed on the second +before the response is failed. + +## Consequences + +Steady-state gets reuse a live connection. A single dead socket no longer +consumes the whole response budget. A transport that stays stalled for both +attempts still returns an error, now before the caller cancels the read +with no outcome. PUT semantics are unchanged. The client is production I/O; +simulation state does not depend on it.