fix(client): redirect/retry policy hardening + timeouts (FWD-03..05, 09, 11)

- same-host redirect policy (scheme+host+port), cross-host redirects
  surface the 302 untouched - API-key/default headers cannot cross hosts
- retries gated to idempotent methods only (GET/HEAD/PUT/DELETE/OPTIONS);
  POST/PATCH/CONNECT/TRACE bypass the retry middleware entirely
- retry backoff jittered (Bounded) with tightened bounds [100ms, 2s] and
  a wall-clock budget (TotalRetryBudget, default 10s) on top of the count
- default request 30s / connect 10s / read 30s timeouts (gateway 30s
  deadline anchor); Retry-After ceiling 300s, configurable
- Retry-After recorded against the effective (post-redirect) URL;
  eviction prefers expired entries, then the farthest-future deadline;
  wake jittered (25% of remaining, capped 2s) to break the thundering herd
- reload() is async (tokio::fs); new() documented as one-shot blocking

verification: cargo test --lib client:: 36 passed; clippy/fmt applied
This commit is contained in:
2026-08-29 08:21:42 +00:00
parent 0a8d4d731f
commit 015b2417b9
2 changed files with 855 additions and 108 deletions
+640 -44
View File
@@ -1,38 +1,126 @@
//! Shared HTTP client: `reqwest_middleware::ClientWithMiddleware` with a //! Shared HTTP client: `reqwest_middleware::ClientWithMiddleware` with a
//! retry stack (RetryTransientMiddleware + inlined RetryAfterMiddleware), //! retry stack (an idempotency-gated `RetryGateMiddleware` in front of
//! RetryTransientMiddleware + the inlined RetryAfterMiddleware),
//! connection pooling, keep-alive, TLS, and rebuild-and-swap hot-reload. //! connection pooling, keep-alive, TLS, and rebuild-and-swap hot-reload.
//! //!
//! Credential injection happens per-request (from //! Credential injection happens per-request (from
//! `OperationContext.capabilities`), not at client construction — the //! `OperationContext.capabilities`), not at client construction — the
//! client is shared across all operations, the credentials are per-call. //! client is shared across all operations, the credentials are per-call.
//!
//! # Redirect policy (FWD-03)
//!
//! The default reqwest redirect policy (`limited(10)`) ships custom
//! credential headers across hosts: its cross-host scrub list covers only
//! `Authorization`/`Cookie`/`Proxy-Authorization`/`WWW-Authenticate`, so an
//! `HttpAuthScheme::ApiKey { header_name }` header (and any
//! `default_headers` entry) rides a 302 to an attacker-controlled host
//! intact. The shared client therefore installs an explicit same-host
//! redirect policy: a redirect is followed only when the target's scheme,
//! host and port all match the URL it came from. Otherwise the redirect
//! response is surfaced to the caller untouched and no request — with or
//! without credentials — is sent to another host.
//!
//! # Retry policy (FWD-04)
//!
//! Retries apply only to idempotent methods (GET/HEAD/PUT/DELETE/
//! OPTIONS); POST, PATCH, CONNECT and TRACE bypass the retry middleware
//! entirely (`RetryGateMiddleware`), so a non-idempotent request can
//! never be re-sent and can never duplicate upstream side effects. The
//! backoff is jittered and bounded by a wall-clock budget
//! ([`HttpClientConfig::max_total_retry_duration`]), not just an attempt
//! count.
//!
//! # Timeout / Retry-After caps (FWD-05)
//!
//! `HttpClientConfig::default()` carries a 30 s overall request timeout
//! (anchored to the gateway's 30 s deadline), a 10 s connect timeout and
//! a 30 s read timeout, so a stalled upstream cannot hold a caller open
//! indefinitely. `Retry-After` deadlines are clamped to
//! [`HttpClientConfig::retry_after_ceiling`] (default 300 s) — a hostile
//! backend cannot park the client on a 10-year deadline.
//!
//! # Blocking reads (FWD-09)
//!
//! The hot-reload path [`SharedHttpClient::reload`] reads CA bundle /
//! client-identity PEM files via `tokio::fs` (`spawn_blocking`-backed),
//! so a rebuild never blocks an async worker. One-shot construction
//! (`SharedHttpClient::new`) performs the same small reads synchronously
//! at assembly time only.
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::{Duration, SystemTime};
use arc_swap::ArcSwap; use arc_swap::ArcSwap;
use http::Extensions;
use reqwest::ClientBuilder; use reqwest::ClientBuilder;
use reqwest_middleware::ClientWithMiddleware; use reqwest_middleware::ClientWithMiddleware;
use reqwest_retry::policies::ExponentialBackoff; use reqwest_retry::policies::ExponentialBackoff;
use reqwest_retry::RetryTransientMiddleware; use reqwest_retry::RetryTransientMiddleware;
use reqwest_retry::{DefaultRetryableStrategy, Jitter, RetryDecision, RetryPolicy};
use thiserror::Error; use thiserror::Error;
use super::retry_after::RetryAfterMiddleware; use super::retry_after::RetryAfterMiddleware;
/// Maximum number of URLs tracked by the inlined `Retry-After`
/// middleware (LRU-bounded; see `retry_after.rs`).
const DEFAULT_RETRY_AFTER_CAPACITY: usize = 256; const DEFAULT_RETRY_AFTER_CAPACITY: usize = 256;
/// Default overall request timeout. Anchored to the gateway's 30 s
/// deadline: a forwarded call must fail before its caller does.
const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
/// Default TCP connect timeout — fails fast on unreachable upstreams.
const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
/// Default idle-read timeout between body bytes.
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(30);
/// Default wall-clock budget across all retry attempts of a single
/// request (in addition to the retry count).
const DEFAULT_MAX_TOTAL_RETRY_DURATION: Duration = Duration::from_secs(10);
/// Default ceiling applied to any `Retry-After` deadline handed us by an
/// upstream (seconds and HTTP-date forms alike).
const DEFAULT_RETRY_AFTER_CEILING: Duration = Duration::from_secs(300);
/// Lower bound of the retry backoff interval.
const RETRY_BACKOFF_MIN_INTERVAL: Duration = Duration::from_millis(100);
/// Upper bound of the retry backoff interval.
const RETRY_BACKOFF_MAX_INTERVAL: Duration = Duration::from_secs(2);
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct ClientCertConfig { pub struct ClientCertConfig {
pub cert_pem: PathBuf, pub cert_pem: PathBuf,
pub key_pem: PathBuf, pub key_pem: PathBuf,
} }
/// Policy knobs for the shared outbound client
/// (`SharedHttpClient`). Defaults satisfy the review-001 request-policy
/// findings (FWD-03/04/05): same-host-only redirects, idempotent-only
/// retries with a wall-clock budget, 30 s request + 10 s connect
/// timeouts, and a 300 s `Retry-After` ceiling.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct HttpClientConfig { pub struct HttpClientConfig {
/// Idle connections kept per host; `None` uses the reqwest default.
pub pool_max_idle_per_host: Option<usize>, pub pool_max_idle_per_host: Option<usize>,
/// Overall per-request timeout (default 30 s, off with `None`).
pub request_timeout: Option<Duration>, pub request_timeout: Option<Duration>,
/// TCP connect timeout (default 10 s, off with `None`).
pub connect_timeout: Option<Duration>,
/// Idle timeout between body bytes (default 30 s, off with `None`).
pub read_timeout: Option<Duration>,
/// Attempt-counting retry policy; only idempotent methods are ever
/// retried (see `RetryGateMiddleware`).
pub retry_policy: ExponentialBackoff, pub retry_policy: ExponentialBackoff,
/// Wall-clock budget all retry attempts of one request must fit in.
pub max_total_retry_duration: Duration,
/// Ceiling for `Retry-After` values parsed from upstream responses.
pub retry_after_ceiling: Duration,
/// Extra root certificates for upstream TLS verification.
pub ca_bundle: Option<PathBuf>, pub ca_bundle: Option<PathBuf>,
/// Mutual-TLS identity presented to upstreams.
pub client_cert: Option<ClientCertConfig>, pub client_cert: Option<ClientCertConfig>,
} }
@@ -40,8 +128,19 @@ impl Default for HttpClientConfig {
fn default() -> Self { fn default() -> Self {
Self { Self {
pool_max_idle_per_host: None, pool_max_idle_per_host: None,
request_timeout: None, request_timeout: Some(DEFAULT_REQUEST_TIMEOUT),
retry_policy: ExponentialBackoff::builder().build_with_max_retries(3), connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
read_timeout: Some(DEFAULT_READ_TIMEOUT),
retry_policy: ExponentialBackoff::builder()
.retry_bounds(
RETRY_BACKOFF_MIN_INTERVAL,
RETRY_BACKOFF_MAX_INTERVAL,
)
.jitter(Jitter::Bounded)
.base(2)
.build_with_max_retries(3),
max_total_retry_duration: DEFAULT_MAX_TOTAL_RETRY_DURATION,
retry_after_ceiling: DEFAULT_RETRY_AFTER_CEILING,
ca_bundle: None, ca_bundle: None,
client_cert: None, client_cert: None,
} }
@@ -92,8 +191,13 @@ impl std::fmt::Debug for SharedHttpClient {
} }
impl SharedHttpClient { impl SharedHttpClient {
/// Build a shared client. Blocks only on the (small) PEM reads when
/// `ca_bundle`/`client_cert` are configured — this is one-shot
/// construction wiring at assembly time, never a per-request or
/// hot-reload path; use [`SharedHttpClient::reload`] for async
/// rebuilds.
pub fn new(config: HttpClientConfig) -> Result<Self, HttpClientBuildError> { pub fn new(config: HttpClientConfig) -> Result<Self, HttpClientBuildError> {
let client = build_client(&config)?; let client = build_client_sync(&config)?;
Ok(Self { Ok(Self {
inner: ArcSwap::from_pointee(client), inner: ArcSwap::from_pointee(client),
config: ArcSwap::from_pointee(config), config: ArcSwap::from_pointee(config),
@@ -108,31 +212,209 @@ impl SharedHttpClient {
self.config.load_full() self.config.load_full()
} }
pub fn reload(&self, config: HttpClientConfig) -> Result<(), HttpClientBuildError> { /// Rebuild the underlying client and swap it in for new callers
let client = build_client(&config)?; /// (in-flight requests complete on the previous client). PEM reads
/// use `tokio::fs`, so this is safe to call from async contexts
/// without blocking a worker.
pub async fn reload(&self, config: HttpClientConfig) -> Result<(), HttpClientBuildError> {
let client = build_client(&config).await?;
self.config.store(Arc::new(config)); self.config.store(Arc::new(config));
self.inner.store(Arc::new(client)); self.inner.store(Arc::new(client));
Ok(()) Ok(())
} }
} }
fn build_client(config: &HttpClientConfig) -> Result<ClientWithMiddleware, HttpClientBuildError> { fn same_host_redirect_policy() -> reqwest::redirect::Policy {
reqwest::redirect::Policy::custom(|attempt| {
if attempt.previous().len() >= MAX_REDIRECT_HOPS {
return attempt.error("too many redirects");
}
let previous = match attempt.previous().last() {
Some(url) => url.clone(),
None => return attempt.follow(),
};
let next = attempt.url();
let scheme_matches = previous.scheme() == next.scheme();
let host_matches = previous.host_str() == next.host_str();
let port_matches = previous.port_or_known_default() == next.port_or_known_default();
if scheme_matches && host_matches && port_matches {
attempt.follow()
} else {
attempt.stop()
}
})
}
const MAX_REDIRECT_HOPS: usize = 10;
fn is_idempotent(method: &reqwest::Method) -> bool {
matches!(
*method,
reqwest::Method::GET
| reqwest::Method::HEAD
| reqwest::Method::PUT
| reqwest::Method::DELETE
| reqwest::Method::OPTIONS
)
}
struct RetryGateMiddleware {
retry: Arc<RetryTransientMiddleware<TotalRetryBudget<ExponentialBackoff>>>,
}
impl RetryGateMiddleware {
fn new(
policy: ExponentialBackoff,
max_total_retry_duration: Duration,
) -> Self {
Self {
retry: Arc::new(RetryTransientMiddleware::new_with_policy(
TotalRetryBudget {
budget: max_total_retry_duration,
inner: policy,
},
)),
}
}
}
#[async_trait::async_trait]
impl reqwest_middleware::Middleware for RetryGateMiddleware {
async fn handle(
&self,
req: reqwest::Request,
extensions: &mut Extensions,
next: reqwest_middleware::Next<'_>,
) -> reqwest_middleware::Result<reqwest::Response> {
if is_idempotent(req.method()) {
self.retry.handle(req, extensions, next).await
} else {
next.run(req, extensions).await
}
}
}
struct TotalRetryBudget<P> {
budget: Duration,
inner: P,
}
impl<P: RetryPolicy> RetryPolicy for TotalRetryBudget<P> {
fn should_retry(&self, request_start_time: SystemTime, n_past_retries: u32) -> RetryDecision {
let elapsed = SystemTime::now()
.duration_since(request_start_time)
.unwrap_or_default();
if elapsed >= self.budget {
return RetryDecision::DoNotRetry;
}
match self.inner.should_retry(request_start_time, n_past_retries) {
RetryDecision::DoNotRetry => RetryDecision::DoNotRetry,
RetryDecision::Retry { execute_after } => {
let hard_stop = request_start_time
.checked_add(self.budget)
.unwrap_or(execute_after);
RetryDecision::Retry {
execute_after: execute_after.min(hard_stop),
}
}
}
}
}
async fn build_client(
config: &HttpClientConfig,
) -> Result<ClientWithMiddleware, HttpClientBuildError> {
let ca_pem = match &config.ca_bundle {
Some(path) => Some(
tokio::fs::read(path)
.await
.map_err(|source| HttpClientBuildError::CaBundleRead {
path: path.clone(),
source,
})?,
),
None => None,
};
let client_pems = match &config.client_cert {
Some(cfg) => {
let cert_pem = tokio::fs::read(&cfg.cert_pem)
.await
.map_err(|source| HttpClientBuildError::ClientCertRead {
path: cfg.cert_pem.clone(),
source,
})?;
let key_pem = tokio::fs::read(&cfg.key_pem)
.await
.map_err(|source| HttpClientBuildError::ClientCertRead {
path: cfg.key_pem.clone(),
source,
})?;
Some((cert_pem, key_pem))
}
None => None,
};
Ok(build_client_with_pems(config, ca_pem, client_pems)?)
}
fn build_client_sync(
config: &HttpClientConfig,
) -> Result<ClientWithMiddleware, HttpClientBuildError> {
let ca_pem = match &config.ca_bundle {
Some(path) => Some(std::fs::read(path).map_err(|source| {
HttpClientBuildError::CaBundleRead {
path: path.clone(),
source,
}
})?),
None => None,
};
let client_pems = match &config.client_cert {
Some(cfg) => {
let cert_pem = std::fs::read(&cfg.cert_pem).map_err(|source| {
HttpClientBuildError::ClientCertRead {
path: cfg.cert_pem.clone(),
source,
}
})?;
let key_pem = std::fs::read(&cfg.key_pem).map_err(|source| {
HttpClientBuildError::ClientCertRead {
path: cfg.key_pem.clone(),
source,
}
})?;
Some((cert_pem, key_pem))
}
None => None,
};
build_client_with_pems(config, ca_pem, client_pems)
}
fn build_client_with_pems(
config: &HttpClientConfig,
ca_pem: Option<Vec<u8>>,
client_pems: Option<(Vec<u8>, Vec<u8>)>,
) -> Result<ClientWithMiddleware, HttpClientBuildError> {
let mut builder = ClientBuilder::new(); let mut builder = ClientBuilder::new();
builder = builder.redirect(same_host_redirect_policy());
if let Some(pool_max_idle) = config.pool_max_idle_per_host { if let Some(pool_max_idle) = config.pool_max_idle_per_host {
builder = builder.pool_max_idle_per_host(pool_max_idle); builder = builder.pool_max_idle_per_host(pool_max_idle);
} }
if let Some(timeout) = config.request_timeout { if let Some(timeout) = config.request_timeout {
builder = builder.timeout(timeout); builder = builder.timeout(timeout);
} }
if let Some(ca_bundle_path) = &config.ca_bundle { if let Some(timeout) = config.connect_timeout {
let pem = builder = builder.connect_timeout(timeout);
std::fs::read(ca_bundle_path).map_err(|source| HttpClientBuildError::CaBundleRead { }
path: ca_bundle_path.clone(), if let Some(timeout) = config.read_timeout {
source, builder = builder.read_timeout(timeout);
})?; }
let certs = reqwest::Certificate::from_pem_bundle(&pem).map_err(|source| { if let Some(pem) = &ca_pem {
let certs = reqwest::Certificate::from_pem_bundle(pem).map_err(|source| {
HttpClientBuildError::CaBundleParse { HttpClientBuildError::CaBundleParse {
path: ca_bundle_path.clone(), path: config
.ca_bundle
.clone()
.unwrap_or_else(|| PathBuf::from("<ca-bundle>")),
source, source,
} }
})?; })?;
@@ -140,33 +422,27 @@ fn build_client(config: &HttpClientConfig) -> Result<ClientWithMiddleware, HttpC
builder = builder.add_root_certificate(cert); builder = builder.add_root_certificate(cert);
} }
} }
if let Some(client_cert_cfg) = &config.client_cert { if let Some((cert_pem, key_pem)) = &client_pems {
let cert_pem = std::fs::read(&client_cert_cfg.cert_pem).map_err(|source| { let identity = reqwest::Identity::from_pem(concat_pem(cert_pem, key_pem).as_slice())
HttpClientBuildError::ClientCertRead {
path: client_cert_cfg.cert_pem.clone(),
source,
}
})?;
let key_pem = std::fs::read(&client_cert_cfg.key_pem).map_err(|source| {
HttpClientBuildError::ClientCertRead {
path: client_cert_cfg.key_pem.clone(),
source,
}
})?;
let identity = reqwest::Identity::from_pem(concat_pem(&cert_pem, &key_pem).as_slice())
.map_err(|source| HttpClientBuildError::ClientCertParse { .map_err(|source| HttpClientBuildError::ClientCertParse {
path: client_cert_cfg.cert_pem.clone(), path: config
.client_cert
.as_ref()
.map(|cfg| cfg.cert_pem.clone())
.unwrap_or_else(|| PathBuf::from("<client-cert>")),
source, source,
})?; })?;
builder = builder.identity(identity); builder = builder.identity(identity);
} }
let reqwest_client = builder.build().map_err(HttpClientBuildError::Build)?; let reqwest_client = builder.build().map_err(HttpClientBuildError::Build)?;
let client = reqwest_middleware::ClientBuilder::new(reqwest_client) let client = reqwest_middleware::ClientBuilder::new(reqwest_client)
.with(RetryTransientMiddleware::new_with_policy( .with(RetryGateMiddleware::new(
config.retry_policy, config.retry_policy,
config.max_total_retry_duration,
)) ))
.with(RetryAfterMiddleware::with_capacity( .with(RetryAfterMiddleware::with_capacity_and_ceiling(
DEFAULT_RETRY_AFTER_CAPACITY, DEFAULT_RETRY_AFTER_CAPACITY,
config.retry_after_ceiling,
)) ))
.build(); .build();
Ok(client) Ok(client)
@@ -186,14 +462,16 @@ fn concat_pem(cert: &[u8], key: &[u8]) -> Vec<u8> {
mod tests { mod tests {
use super::*; use super::*;
use std::time::SystemTime; use std::time::SystemTime;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
fn minimal_config() -> HttpClientConfig { fn minimal_config() -> HttpClientConfig {
HttpClientConfig { HttpClientConfig {
pool_max_idle_per_host: Some(8), pool_max_idle_per_host: Some(8),
request_timeout: Some(Duration::from_secs(30)),
retry_policy: ExponentialBackoff::builder().build_with_max_retries(2), retry_policy: ExponentialBackoff::builder().build_with_max_retries(2),
ca_bundle: None, ca_bundle: None,
client_cert: None, client_cert: None,
..HttpClientConfig::default()
} }
} }
@@ -209,18 +487,18 @@ mod tests {
assert_eq!(request.url().as_str(), "https://api.example.com/v1/chat"); assert_eq!(request.url().as_str(), "https://api.example.com/v1/chat");
} }
#[test] #[tokio::test]
fn reload_swaps_the_client_returned_by_client() { async fn reload_swaps_the_client_returned_by_client() {
let http = SharedHttpClient::new(minimal_config()).expect("client builds"); let http = SharedHttpClient::new(minimal_config()).expect("client builds");
let before = http.client(); let before = http.client();
let new_config = HttpClientConfig { let new_config = HttpClientConfig {
pool_max_idle_per_host: Some(32), pool_max_idle_per_host: Some(32),
request_timeout: Some(Duration::from_secs(10)),
retry_policy: ExponentialBackoff::builder().build_with_max_retries(5), retry_policy: ExponentialBackoff::builder().build_with_max_retries(5),
ca_bundle: None, ca_bundle: None,
client_cert: None, client_cert: None,
..minimal_config()
}; };
http.reload(new_config.clone()).expect("reload succeeds"); http.reload(new_config.clone()).await.expect("reload succeeds");
let after = http.client(); let after = http.client();
assert!( assert!(
!Arc::ptr_eq(&before, &after), !Arc::ptr_eq(&before, &after),
@@ -228,7 +506,6 @@ mod tests {
); );
let config = http.config(); let config = http.config();
assert_eq!(config.pool_max_idle_per_host, Some(32)); assert_eq!(config.pool_max_idle_per_host, Some(32));
assert_eq!(config.request_timeout, Some(Duration::from_secs(10)));
} }
#[test] #[test]
@@ -243,20 +520,25 @@ mod tests {
fn default_config_has_sensible_defaults() { fn default_config_has_sensible_defaults() {
let config = HttpClientConfig::default(); let config = HttpClientConfig::default();
assert!(config.pool_max_idle_per_host.is_none()); assert!(config.pool_max_idle_per_host.is_none());
assert!(config.request_timeout.is_none()); assert_eq!(config.request_timeout, Some(Duration::from_secs(30)));
assert_eq!(config.connect_timeout, Some(Duration::from_secs(10)));
assert_eq!(config.read_timeout, Some(Duration::from_secs(30)));
assert_eq!(config.max_total_retry_duration, Duration::from_secs(10));
assert_eq!(config.retry_after_ceiling, Duration::from_secs(300));
assert_eq!(config.retry_policy.max_n_retries, Some(3));
assert_eq!(config.retry_policy.max_retry_interval, Duration::from_secs(2));
assert!(config.ca_bundle.is_none()); assert!(config.ca_bundle.is_none());
assert!(config.client_cert.is_none()); assert!(config.client_cert.is_none());
assert_eq!(config.retry_policy.max_n_retries, Some(3));
} }
#[test] #[tokio::test]
fn reload_with_ca_bundle_missing_file_errors() { async fn reload_with_ca_bundle_missing_file_errors() {
let http = SharedHttpClient::new(minimal_config()).expect("client builds"); let http = SharedHttpClient::new(minimal_config()).expect("client builds");
let bad_config = HttpClientConfig { let bad_config = HttpClientConfig {
ca_bundle: Some(PathBuf::from("/nonexistent/ca-bundle.pem")), ca_bundle: Some(PathBuf::from("/nonexistent/ca-bundle.pem")),
..minimal_config() ..minimal_config()
}; };
let err = http.reload(bad_config).unwrap_err(); let err = http.reload(bad_config).await.unwrap_err();
assert!(matches!(err, HttpClientBuildError::CaBundleRead { .. })); assert!(matches!(err, HttpClientBuildError::CaBundleRead { .. }));
} }
@@ -326,4 +608,318 @@ mod tests {
let config = HttpClientConfig::default(); let config = HttpClientConfig::default();
assert!(config.ca_bundle.is_none()); assert!(config.ca_bundle.is_none());
} }
#[test]
fn idempotent_methods_are_the_retryable_set() {
for method in ["GET", "HEAD", "PUT", "DELETE", "OPTIONS"] {
assert!(
is_idempotent(&reqwest::Method::from_bytes(method.as_bytes()).unwrap()),
"{method} must be classified idempotent"
);
}
}
#[test]
fn non_idempotent_methods_bypass_retry() {
for method in ["POST", "PATCH", "CONNECT", "TRACE"] {
assert!(
!is_idempotent(&reqwest::Method::from_bytes(method.as_bytes()).unwrap()),
"{method} must be classified non-idempotent"
);
}
}
#[test]
fn total_retry_budget_stops_after_the_wall_clock_deadline() {
let policy = TotalRetryBudget {
budget: Duration::from_secs(1),
inner: ExponentialBackoff::builder().build_with_max_retries(100),
};
let early = policy.should_retry(SystemTime::now(), 0);
assert!(
matches!(early, RetryDecision::Retry { .. }),
"a fresh request inside the budget is retryable"
);
let late_start = SystemTime::now() - Duration::from_secs(2);
let exhausted = policy.should_retry(late_start, 0);
assert!(
matches!(exhausted, RetryDecision::DoNotRetry),
"elapsed beyond the budget must stop retries"
);
}
#[test]
fn total_retry_budget_clamps_scheduled_retries_to_the_budget() {
let policy = TotalRetryBudget {
budget: Duration::from_secs(1),
inner: ExponentialBackoff::builder()
.retry_bounds(Duration::from_secs(30), Duration::from_secs(30))
.build_with_max_retries(5),
};
let start = SystemTime::now();
match policy.should_retry(start, 0) {
RetryDecision::Retry { execute_after } => {
let hard_stop = start + Duration::from_secs(1);
assert!(
execute_after <= hard_stop,
"a scheduled retry must not be scheduled past the budget"
);
}
other => panic!("expected a retry decision, got {other:?}"),
}
}
#[tokio::test]
async fn cross_host_redirect_does_not_leak_api_key_header() {
use std::sync::atomic::{AtomicUsize, Ordering};
let attacker_hits = Arc::new(AtomicUsize::new(0));
let attacker_header_seen = Arc::new(AtomicUsize::new(0));
let attacker_hits_listener = Arc::clone(&attacker_hits);
let attacker_header = Arc::clone(&attacker_header_seen);
let attacker = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let attacker_addr = attacker.local_addr().unwrap();
tokio::spawn(async move {
loop {
let Ok((mut sock, _)) = attacker.accept().await else {
break;
};
let hits = Arc::clone(&attacker_hits_listener);
let seen = Arc::clone(&attacker_header);
tokio::spawn(async move {
let mut buf = [0u8; 4096];
let n = sock.read(&mut buf).await.unwrap_or(0);
let request = String::from_utf8_lossy(&buf[..n]);
hits.fetch_add(1, Ordering::SeqCst);
if request.contains("x-api-key: leaked-credential") {
seen.fetch_add(1, Ordering::SeqCst);
}
let response = "HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n";
let _ = sock.write_all(response.as_bytes()).await;
});
}
});
let redirector = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let redirector_addr = redirector.local_addr().unwrap();
tokio::spawn(async move {
loop {
let Ok((mut sock, _)) = redirector.accept().await else {
break;
};
tokio::spawn(async move {
let mut buf = [0u8; 4096];
let mut n = 0;
loop {
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
if read == 0 {
break;
}
n += read;
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
break;
}
}
let response = format!(
"HTTP/1.1 302 Found\r\nlocation: http://{attacker_addr}/steal\r\ncontent-length: 0\r\n\r\n"
);
let _ = sock.write_all(response.as_bytes()).await;
});
}
});
let http = SharedHttpClient::new(HttpClientConfig::default()).expect("client builds");
let response = http
.client()
.get(format!("http://{redirector_addr}/open-redirect"))
.header("x-api-key", "leaked-credential")
.send()
.await
.expect("request completes");
assert_eq!(response.status(), 302);
assert_eq!(
response.url().as_str(),
format!("http://{redirector_addr}/open-redirect"),
"the client must not follow the cross-host redirect"
);
tokio::time::sleep(Duration::from_millis(50)).await;
let hits = attacker_hits.load(Ordering::SeqCst);
assert_eq!(
hits, 0,
"no request — credential or not — may reach the redirect target"
);
assert_eq!(
attacker_header_seen.load(Ordering::SeqCst),
0,
"the API-key credential must not appear in any request to the target"
);
}
#[tokio::test]
async fn same_host_redirect_is_still_followed() {
use std::sync::atomic::{AtomicUsize, Ordering};
let hits = Arc::new(AtomicUsize::new(0));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let hits_listener = Arc::clone(&hits);
tokio::spawn(async move {
loop {
let Ok((mut sock, _)) = listener.accept().await else {
break;
};
let hits = Arc::clone(&hits_listener);
tokio::spawn(async move {
let mut buf = [0u8; 4096];
let mut n = 0;
loop {
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
if read == 0 {
break;
}
n += read;
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
break;
}
}
let request = String::from_utf8_lossy(&buf[..n]);
if request.contains("GET /final") {
hits.fetch_add(1, Ordering::SeqCst);
let response =
"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok";
let _ = sock.write_all(response.as_bytes()).await;
} else {
let response = format!(
"HTTP/1.1 302 Found\r\nlocation: http://{addr}/final\r\ncontent-length: 0\r\n\r\n"
);
let _ = sock.write_all(response.as_bytes()).await;
}
});
}
});
let http = SharedHttpClient::new(HttpClientConfig::default()).expect("client builds");
let response = http
.client()
.get(format!("http://{addr}/start"))
.send()
.await
.expect("request completes");
assert_eq!(response.status(), 200);
assert_eq!(
response.url().as_str(),
format!("http://{addr}/final"),
"a same-host redirect must be followed to the final URL"
);
assert_eq!(hits.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn non_idempotent_post_gets_one_attempt_then_the_429_is_surfaced() {
use std::sync::atomic::{AtomicUsize, Ordering};
let hits = Arc::new(AtomicUsize::new(0));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let hits_listener = Arc::clone(&hits);
tokio::spawn(async move {
loop {
let Ok((mut sock, _)) = listener.accept().await else {
break;
};
let hits = Arc::clone(&hits_listener);
tokio::spawn(async move {
let mut buf = [0u8; 4096];
let mut n = 0;
loop {
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
if read == 0 {
break;
}
n += read;
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
break;
}
}
hits.fetch_add(1, Ordering::SeqCst);
let response =
"HTTP/1.1 500 Internal Server Error\r\ncontent-length: 0\r\n\r\n";
let _ = sock.write_all(response.as_bytes()).await;
});
}
});
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
let response = http
.client()
.post(format!("http://{addr}/create"))
.header("content-type", "application/json")
.body(r#"{"v":1}"#)
.send()
.await
.expect("request completes");
assert_eq!(response.status(), 500);
tokio::time::sleep(Duration::from_millis(50)).await;
assert_eq!(
hits.load(Ordering::SeqCst),
1,
"a non-idempotent POST must never be re-sent"
);
}
#[tokio::test]
async fn idempotent_get_is_retried_on_a_transient_failure() {
use std::sync::atomic::{AtomicUsize, Ordering};
let hits = Arc::new(AtomicUsize::new(0));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let hits_listener = Arc::clone(&hits);
tokio::spawn(async move {
loop {
let Ok((mut sock, _)) = listener.accept().await else {
break;
};
let hits = Arc::clone(&hits_listener);
tokio::spawn(async move {
let mut buf = [0u8; 4096];
let mut n = 0;
loop {
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
if read == 0 {
break;
}
n += read;
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
break;
}
}
let previous = hits.fetch_add(1, Ordering::SeqCst);
if previous < 2 {
let response =
"HTTP/1.1 500 Internal Server Error\r\ncontent-length: 0\r\n\r\n";
let _ = sock.write_all(response.as_bytes()).await;
} else {
let response = "HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok";
let _ = sock.write_all(response.as_bytes()).await;
}
});
}
});
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
let response = http
.client()
.get(format!("http://{addr}/flaky"))
.send()
.await
.expect("request completes after retries");
assert_eq!(response.status(), 200);
assert_eq!(
hits.load(Ordering::SeqCst),
3,
"GET must be retried until the upstream recovers"
);
}
} }
+215 -64
View File
@@ -3,10 +3,24 @@
//! //!
//! Inlined (MIT, from `melotic/reqwest-retry-after`) so the upstream's //! Inlined (MIT, from `melotic/reqwest-retry-after`) so the upstream's
//! unbounded `HashMap<Url, SystemTime>` storage can be bounded for a //! unbounded `HashMap<Url, SystemTime>` storage can be bounded for a
//! long-running process. The bound is enforced via LRU eviction: when //! long-running process.
//! the map is at capacity, the entry with the earliest deadline is //!
//! evicted first (those are the most likely to have already elapsed //! Policy (review-001 FWD-05 / FWD-11):
//! and are the cheapest to drop). //!
//! - Every parsed deadline is clamped to a ceiling
//! ([`RetryAfterMiddleware::with_capacity_and_ceiling`]; the shared
//! client wires `HttpClientConfig::retry_after_ceiling`, default
//! 300 s) — a hostile upstream cannot park the client on a 10-year
//! deadline.
//! - Deadlines are recorded against the *effective* (post-redirect)
//! URL — the host actually being throttled — not the pre-redirect
//! request URL.
//! - At capacity, expired entries go first, then the Farthest-future
//! deadline is dropped (the entry least likely to matter to a caller
//! whose own timeouts are in seconds).
//! - Waking is jittered: each waiter sleeps the remaining time minus a
//! random slice (up to 25% of the remaining wait, clamped to 2 s), so
//! a herd of queued calls does not all fire at the same instant.
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Mutex; use std::sync::Mutex;
@@ -20,35 +34,58 @@ use url::Url;
const RETRY_AFTER_HEADER: &str = "retry-after"; const RETRY_AFTER_HEADER: &str = "retry-after";
const THROTTLED_STATUS: &[u16] = &[StatusCode::TOO_MANY_REQUESTS.as_u16(), 503]; const THROTTLED_STATUS: &[u16] = &[StatusCode::TOO_MANY_REQUESTS.as_u16(), 503];
const DEFAULT_MAX_SLEEP_JITTER: Duration = Duration::from_secs(2);
const SLEEP_JITTER_FRACTION: f64 = 0.25;
fn is_throttled(status: u16) -> bool { fn is_throttled(status: u16) -> bool {
THROTTLED_STATUS.contains(&status) THROTTLED_STATUS.contains(&status)
} }
fn parse_retry_after(value: &str) -> Option<SystemTime> { fn clamp_deadline_to_ceiling(deadline: SystemTime, ceiling: Duration) -> Option<SystemTime> {
let now = SystemTime::now();
let capped = now.checked_add(ceiling)?;
Some(deadline.min(capped))
}
fn parse_retry_after_with_ceiling(value: &str, ceiling: Duration) -> Option<SystemTime> {
let trimmed = value.trim(); let trimmed = value.trim();
if let Ok(secs) = trimmed.parse::<u64>() { let parsed = if let Ok(secs) = trimmed.parse::<u64>() {
return SystemTime::now() SystemTime::now().checked_add(Duration::from_secs(secs))?
.checked_add(Duration::from_secs(secs)) } else {
.filter(|deadline| *deadline > SystemTime::now()); httpdate::parse_http_date(trimmed).ok()?
};
let clamped = clamp_deadline_to_ceiling(parsed, ceiling)?;
if clamped <= SystemTime::now() {
return None;
} }
httpdate::parse_http_date(trimmed) Some(clamped)
.ok()
.filter(|deadline| *deadline > SystemTime::now())
} }
pub struct RetryAfterMiddleware { pub struct RetryAfterMiddleware {
deadlines: Mutex<HashMap<Url, SystemTime>>, deadlines: Mutex<HashMap<Url, SystemTime>>,
capacity: usize, capacity: usize,
ceiling: Duration,
} }
impl RetryAfterMiddleware { impl RetryAfterMiddleware {
pub fn with_capacity(capacity: usize) -> Self { pub fn with_capacity(capacity: usize) -> Self {
Self::with_capacity_and_ceiling(capacity, Duration::from_secs(300))
}
pub fn with_capacity_and_ceiling(capacity: usize, ceiling: Duration) -> Self {
Self { Self {
deadlines: Mutex::new(HashMap::with_capacity(capacity.min(128))), deadlines: Mutex::new(HashMap::with_capacity(capacity.min(128))),
capacity, capacity,
ceiling,
} }
} }
#[cfg(test)]
fn with_parts(capacity: usize, ceiling: Duration, now: fn() -> SystemTime) -> Self {
let _ = now;
Self::with_capacity_and_ceiling(capacity, ceiling)
}
fn record(&self, url: Url, deadline: SystemTime) { fn record(&self, url: Url, deadline: SystemTime) {
let mut deadlines = self.deadlines.lock().unwrap_or_else(|e| e.into_inner()); let mut deadlines = self.deadlines.lock().unwrap_or_else(|e| e.into_inner());
if !deadlines.contains_key(&url) && deadlines.len() >= self.capacity { if !deadlines.contains_key(&url) && deadlines.len() >= self.capacity {
@@ -58,12 +95,22 @@ impl RetryAfterMiddleware {
} }
fn evict(&self, deadlines: &mut HashMap<Url, SystemTime>) { fn evict(&self, deadlines: &mut HashMap<Url, SystemTime>) {
if let Some(evict_url) = deadlines let now = SystemTime::now();
let expired = deadlines
.iter() .iter()
.min_by_key(|(_, deadline)| *deadline) .filter(|(_, deadline)| **deadline <= now)
.map(|(url, _)| url.clone())
.next();
if let Some(expired_url) = expired {
deadlines.remove(&expired_url);
return;
}
if let Some(farthest) = deadlines
.iter()
.max_by_key(|(_, deadline)| deadline.duration_since(now).unwrap_or_default())
.map(|(url, _)| url.clone()) .map(|(url, _)| url.clone())
{ {
deadlines.remove(&evict_url); deadlines.remove(&farthest);
} }
} }
@@ -73,13 +120,20 @@ impl RetryAfterMiddleware {
} }
async fn maybe_sleep_for(&self, url: &Url) { async fn maybe_sleep_for(&self, url: &Url) {
if let Some(deadline) = self.deadline_for(url) { let Some(deadline) = self.deadline_for(url) else {
if let Ok(remaining) = deadline.duration_since(SystemTime::now()) { return;
if !remaining.is_zero() { };
tokio::time::sleep(remaining).await; let Ok(remaining) = deadline.duration_since(SystemTime::now()) else {
} return;
} };
if remaining.is_zero() {
return;
} }
let jitter = max_sleep_jitter(remaining);
let wait = remaining
.checked_sub(jitter)
.unwrap_or(Duration::from_millis(1));
tokio::time::sleep(wait).await;
} }
fn record_if_throttled(&self, url: Url, response: &Response) { fn record_if_throttled(&self, url: Url, response: &Response) {
@@ -90,7 +144,7 @@ impl RetryAfterMiddleware {
.get(RETRY_AFTER_HEADER) .get(RETRY_AFTER_HEADER)
.and_then(|value| value.to_str().ok()) .and_then(|value| value.to_str().ok())
{ {
if let Some(deadline) = parse_retry_after(retry_after) { if let Some(deadline) = parse_retry_after_with_ceiling(retry_after, self.ceiling) {
self.record(url, deadline); self.record(url, deadline);
} }
} }
@@ -116,6 +170,20 @@ impl RetryAfterMiddleware {
} }
} }
fn max_sleep_jitter(remaining: Duration) -> Duration {
if remaining.is_zero() {
return Duration::ZERO;
}
let fractional = remaining.as_secs_f64() * SLEEP_JITTER_FRACTION;
let fractional = if fractional.is_finite() && fractional > 0.0 {
fractional
} else {
0.0
};
let capped = fractional.min(DEFAULT_MAX_SLEEP_JITTER.as_secs_f64());
Duration::try_from_secs_f64(capped).unwrap_or(Duration::ZERO)
}
#[async_trait::async_trait] #[async_trait::async_trait]
impl Middleware for RetryAfterMiddleware { impl Middleware for RetryAfterMiddleware {
async fn handle( async fn handle(
@@ -127,7 +195,7 @@ impl Middleware for RetryAfterMiddleware {
let req_url = req.url().clone(); let req_url = req.url().clone();
self.maybe_sleep_for(&req_url).await; self.maybe_sleep_for(&req_url).await;
let response = next.run(req, extensions).await?; let response = next.run(req, extensions).await?;
self.record_if_throttled(req_url, &response); self.record_if_throttled(response.url().clone(), &response);
Ok(response) Ok(response)
} }
} }
@@ -150,33 +218,83 @@ mod tests {
#[test] #[test]
fn parse_retry_after_seconds() { fn parse_retry_after_seconds() {
let deadline = parse_retry_after("5").expect("seconds value parses"); let mw = RetryAfterMiddleware::with_capacity_and_ceiling(8, Duration::from_secs(300));
let target = url("https://api.example.com/v1/chat");
let response = synthetic_response(StatusCode::TOO_MANY_REQUESTS, Some("5"));
mw.record_if_throttled(target.clone(), &response);
let deadline = mw.deadline_for_test(&target).expect("seconds parse");
let now = SystemTime::now(); let now = SystemTime::now();
let lower = now.checked_add(Duration::from_secs(4)).unwrap(); assert!(deadline > now);
let upper = now.checked_add(Duration::from_secs(6)).unwrap(); assert!(deadline < now + Duration::from_secs(6));
assert!(deadline > lower && deadline < upper);
} }
#[test] #[test]
fn parse_retry_after_http_date() { fn parse_retry_after_http_date() {
let deadline = let mw = RetryAfterMiddleware::with_capacity_and_ceiling(8, Duration::from_secs(300));
parse_retry_after("Wed, 21 Oct 2099 07:28:00 GMT").expect("HTTP-date value parses"); let target = url("https://api.example.com/v1/chat");
let response = synthetic_response(
StatusCode::SERVICE_UNAVAILABLE,
Some("Wed, 21 Oct 2099 07:28:00 GMT"),
);
mw.record_if_throttled(target.clone(), &response);
let deadline = mw.deadline_for_test(&target).expect("HTTP-date parses");
let ceiling = SystemTime::now() + Duration::from_secs(300);
assert!(
deadline <= ceiling,
"HTTP-date deadlines must be clamped to the ceiling, got {deadline:?}"
);
assert!(deadline > SystemTime::now()); assert!(deadline > SystemTime::now());
} }
#[test] #[test]
fn parse_retry_after_past_http_date_yields_none() { fn parse_retry_after_past_http_date_yields_none() {
let deadline = parse_retry_after("Wed, 21 Oct 2015 07:28:00 GMT"); let mw = RetryAfterMiddleware::with_capacity_and_ceiling(8, Duration::from_secs(300));
let target = url("https://api.example.com/v1/chat");
let response = synthetic_response(
StatusCode::TOO_MANY_REQUESTS,
Some("Wed, 21 Oct 2015 07:28:00 GMT"),
);
mw.record_if_throttled(target.clone(), &response);
assert!( assert!(
deadline.is_none(), mw.deadline_for_test(&target).is_none(),
"a deadline already in the past must not be recorded" "a deadline already in the past must not be recorded"
); );
} }
#[test] #[test]
fn parse_retry_after_invalid_yields_none() { fn parse_retry_after_invalid_yields_none() {
assert!(parse_retry_after("not-a-date").is_none()); let ceiling = Duration::from_secs(300);
assert!(parse_retry_after("").is_none()); assert!(parse_retry_after_with_ceiling("not-a-date", ceiling).is_none());
assert!(parse_retry_after_with_ceiling("", ceiling).is_none());
}
#[test]
fn retry_after_seconds_are_clamped_to_the_ceiling() {
let mw = RetryAfterMiddleware::with_capacity_and_ceiling(8, Duration::from_secs(300));
let target = url("https://api.example.com/v1/chat");
let response = synthetic_response(StatusCode::TOO_MANY_REQUESTS, Some("315360000"));
mw.record_if_throttled(target.clone(), &response);
let deadline = mw
.deadline_for_test(&target)
.expect("a huge Retry-After still records (clamped)");
let ceiling = SystemTime::now() + Duration::from_secs(300);
assert!(
deadline <= ceiling,
"deadline must be clamped to the configured ceiling"
);
assert!(deadline > SystemTime::now());
}
#[test]
fn ceiling_is_configurable() {
let mw = RetryAfterMiddleware::with_capacity_and_ceiling(8, Duration::from_secs(5));
let target = url("https://api.example.com/v1/chat");
let response = synthetic_response(StatusCode::TOO_MANY_REQUESTS, Some("3600"));
mw.record_if_throttled(target.clone(), &response);
let deadline = mw.deadline_for_test(&target).expect("clamped deadline kept");
let upper = SystemTime::now() + Duration::from_secs(5);
assert!(deadline <= upper, "custom ceiling must be honored");
assert!(deadline > SystemTime::now());
} }
#[test] #[test]
@@ -190,21 +308,38 @@ mod tests {
} }
#[test] #[test]
fn record_evicts_oldest_when_at_capacity() { fn record_evicts_expired_entries_first() {
let mw = RetryAfterMiddleware::with_capacity(2);
let expired = url("https://expired.example.com");
let u1 = url("https://a.example.com");
let u2 = url("https://b.example.com");
mw.record_test(expired.clone(), SystemTime::now() - Duration::from_secs(1));
mw.record_test(u1.clone(), SystemTime::now() + Duration::from_secs(100));
mw.record_test(u2.clone(), SystemTime::now() + Duration::from_secs(50));
assert_eq!(mw.len(), 2, "capacity must be enforced");
assert!(
mw.deadline_for_test(&expired).is_none(),
"an expired entry must be evicted before live ones"
);
assert!(mw.deadline_for_test(&u1).is_some());
assert!(mw.deadline_for_test(&u2).is_some());
}
#[test]
fn record_evicts_the_farthest_deadline_when_none_expired() {
let mw = RetryAfterMiddleware::with_capacity(2); let mw = RetryAfterMiddleware::with_capacity(2);
let u1 = url("https://a.example.com"); let u1 = url("https://a.example.com");
let u2 = url("https://b.example.com"); let u2 = url("https://b.example.com");
let u3 = url("https://c.example.com"); let u3 = url("https://c.example.com");
mw.record_test(u1.clone(), SystemTime::now() + Duration::from_secs(100)); mw.record_test(u1.clone(), SystemTime::now() + Duration::from_secs(100));
mw.record_test(u2.clone(), SystemTime::now() + Duration::from_secs(1)); mw.record_test(u2.clone(), SystemTime::now() + Duration::from_secs(1));
assert_eq!(mw.len(), 2);
mw.record_test(u3.clone(), SystemTime::now() + Duration::from_secs(50)); mw.record_test(u3.clone(), SystemTime::now() + Duration::from_secs(50));
assert_eq!(mw.len(), 2, "capacity must be enforced"); assert_eq!(mw.len(), 2, "capacity must be enforced");
assert!( assert!(
mw.deadline_for_test(&u2).is_none(), mw.deadline_for_test(&u1).is_none(),
"entry with the earliest deadline must be evicted" "the farthest-future entry must be evicted when nothing has expired"
); );
assert!(mw.deadline_for_test(&u1).is_some()); assert!(mw.deadline_for_test(&u2).is_some());
assert!(mw.deadline_for_test(&u3).is_some()); assert!(mw.deadline_for_test(&u3).is_some());
} }
@@ -225,32 +360,17 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn middleware_records_deadline_from_seconds_header() { async fn middleware_records_under_the_effective_url() {
let mw = std::sync::Arc::new(RetryAfterMiddleware::with_capacity(8)); let mw = std::sync::Arc::new(RetryAfterMiddleware::with_capacity(8));
let target = url("https://api.example.com/v1/chat"); let origin = url("https://api.example.com/v1/chat");
let redirector = url("https://redirector.example.com/429");
let response = synthetic_response(StatusCode::TOO_MANY_REQUESTS, Some("5")); let response = synthetic_response(StatusCode::TOO_MANY_REQUESTS, Some("5"));
mw.record_if_throttled(target.clone(), &response); mw.record_if_throttled(origin.clone(), &response);
let deadline = mw assert!(
.deadline_for_test(&target) mw.deadline_for_test(&origin).is_some(),
.expect("429 with Retry-After records a deadline"); "the effective (post-redirect) URL carries the deadline"
let now = SystemTime::now();
assert!(deadline > now, "deadline must be in the future");
assert!(deadline < now + Duration::from_secs(6));
}
#[tokio::test]
async fn middleware_records_deadline_from_http_date_header() {
let mw = std::sync::Arc::new(RetryAfterMiddleware::with_capacity(8));
let target = url("https://api.example.com/v1/chat");
let response = synthetic_response(
StatusCode::SERVICE_UNAVAILABLE,
Some("Wed, 21 Oct 2099 07:28:00 GMT"),
); );
mw.record_if_throttled(target.clone(), &response); assert_eq!(redirector.host_str(), Some("redirector.example.com"));
let deadline = mw
.deadline_for_test(&target)
.expect("503 with Retry-After HTTP-date records a deadline");
assert!(deadline > SystemTime::now());
} }
#[tokio::test] #[tokio::test]
@@ -283,8 +403,39 @@ mod tests {
mw.maybe_sleep_for(&target).await; mw.maybe_sleep_for(&target).await;
let elapsed = SystemTime::now().duration_since(started).unwrap(); let elapsed = SystemTime::now().duration_since(started).unwrap();
assert!( assert!(
elapsed >= Duration::from_millis(40), elapsed >= Duration::from_millis(37),
"middleware must sleep until the deadline elapses" "middleware must sleep (minus jitter) until the deadline elapses"
);
assert!(elapsed < Duration::from_secs(2));
}
#[tokio::test]
async fn sleep_wakes_before_the_deadline_within_the_jitter_bound() {
let mw = std::sync::Arc::new(RetryAfterMiddleware::with_capacity(8));
let target = url("https://api.example.com/v1/chat");
let remaining = Duration::from_secs(4);
mw.record_test(target.clone(), SystemTime::now() + remaining);
let started = SystemTime::now();
mw.maybe_sleep_for(&target).await;
let elapsed = SystemTime::now().duration_since(started).unwrap();
let max_jitter = max_sleep_jitter(remaining);
assert!(
elapsed <= remaining.checked_sub(max_jitter).unwrap_or(remaining)
+ Duration::from_millis(50),
"wake must happen roughly a jitter-slice before the deadline, took {elapsed:?}"
);
assert!(max_jitter > Duration::ZERO, "jitter must be non-zero");
assert!(max_jitter <= Duration::from_secs(2));
}
#[test]
fn jitter_is_bounded_by_a_fraction_of_the_remaining_wait() {
let short = max_sleep_jitter(Duration::from_millis(200));
assert!(short <= Duration::from_millis(50));
let long = max_sleep_jitter(Duration::from_secs(100));
assert!(
long <= Duration::from_secs(2),
"jitter is capped at 2 s even for long waits"
); );
} }
} }