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:
+640
-44
@@ -1,38 +1,126 @@
|
||||
//! 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.
|
||||
//!
|
||||
//! Credential injection happens per-request (from
|
||||
//! `OperationContext.capabilities`), not at client construction — the
|
||||
//! 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::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use arc_swap::ArcSwap;
|
||||
use http::Extensions;
|
||||
use reqwest::ClientBuilder;
|
||||
use reqwest_middleware::ClientWithMiddleware;
|
||||
use reqwest_retry::policies::ExponentialBackoff;
|
||||
use reqwest_retry::RetryTransientMiddleware;
|
||||
use reqwest_retry::{DefaultRetryableStrategy, Jitter, RetryDecision, RetryPolicy};
|
||||
use thiserror::Error;
|
||||
|
||||
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;
|
||||
|
||||
/// 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)]
|
||||
pub struct ClientCertConfig {
|
||||
pub cert_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)]
|
||||
pub struct HttpClientConfig {
|
||||
/// Idle connections kept per host; `None` uses the reqwest default.
|
||||
pub pool_max_idle_per_host: Option<usize>,
|
||||
/// Overall per-request timeout (default 30 s, off with `None`).
|
||||
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,
|
||||
/// 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>,
|
||||
/// Mutual-TLS identity presented to upstreams.
|
||||
pub client_cert: Option<ClientCertConfig>,
|
||||
}
|
||||
|
||||
@@ -40,8 +128,19 @@ impl Default for HttpClientConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
pool_max_idle_per_host: None,
|
||||
request_timeout: None,
|
||||
retry_policy: ExponentialBackoff::builder().build_with_max_retries(3),
|
||||
request_timeout: Some(DEFAULT_REQUEST_TIMEOUT),
|
||||
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,
|
||||
client_cert: None,
|
||||
}
|
||||
@@ -92,8 +191,13 @@ impl std::fmt::Debug for 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> {
|
||||
let client = build_client(&config)?;
|
||||
let client = build_client_sync(&config)?;
|
||||
Ok(Self {
|
||||
inner: ArcSwap::from_pointee(client),
|
||||
config: ArcSwap::from_pointee(config),
|
||||
@@ -108,31 +212,209 @@ impl SharedHttpClient {
|
||||
self.config.load_full()
|
||||
}
|
||||
|
||||
pub fn reload(&self, config: HttpClientConfig) -> Result<(), HttpClientBuildError> {
|
||||
let client = build_client(&config)?;
|
||||
/// Rebuild the underlying client and swap it in for new callers
|
||||
/// (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.inner.store(Arc::new(client));
|
||||
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();
|
||||
builder = builder.redirect(same_host_redirect_policy());
|
||||
if let Some(pool_max_idle) = config.pool_max_idle_per_host {
|
||||
builder = builder.pool_max_idle_per_host(pool_max_idle);
|
||||
}
|
||||
if let Some(timeout) = config.request_timeout {
|
||||
builder = builder.timeout(timeout);
|
||||
}
|
||||
if let Some(ca_bundle_path) = &config.ca_bundle {
|
||||
let pem =
|
||||
std::fs::read(ca_bundle_path).map_err(|source| HttpClientBuildError::CaBundleRead {
|
||||
path: ca_bundle_path.clone(),
|
||||
source,
|
||||
})?;
|
||||
let certs = reqwest::Certificate::from_pem_bundle(&pem).map_err(|source| {
|
||||
if let Some(timeout) = config.connect_timeout {
|
||||
builder = builder.connect_timeout(timeout);
|
||||
}
|
||||
if let Some(timeout) = config.read_timeout {
|
||||
builder = builder.read_timeout(timeout);
|
||||
}
|
||||
if let Some(pem) = &ca_pem {
|
||||
let certs = reqwest::Certificate::from_pem_bundle(pem).map_err(|source| {
|
||||
HttpClientBuildError::CaBundleParse {
|
||||
path: ca_bundle_path.clone(),
|
||||
path: config
|
||||
.ca_bundle
|
||||
.clone()
|
||||
.unwrap_or_else(|| PathBuf::from("<ca-bundle>")),
|
||||
source,
|
||||
}
|
||||
})?;
|
||||
@@ -140,33 +422,27 @@ fn build_client(config: &HttpClientConfig) -> Result<ClientWithMiddleware, HttpC
|
||||
builder = builder.add_root_certificate(cert);
|
||||
}
|
||||
}
|
||||
if let Some(client_cert_cfg) = &config.client_cert {
|
||||
let cert_pem = std::fs::read(&client_cert_cfg.cert_pem).map_err(|source| {
|
||||
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())
|
||||
if let Some((cert_pem, key_pem)) = &client_pems {
|
||||
let identity = reqwest::Identity::from_pem(concat_pem(cert_pem, key_pem).as_slice())
|
||||
.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,
|
||||
})?;
|
||||
builder = builder.identity(identity);
|
||||
}
|
||||
let reqwest_client = builder.build().map_err(HttpClientBuildError::Build)?;
|
||||
let client = reqwest_middleware::ClientBuilder::new(reqwest_client)
|
||||
.with(RetryTransientMiddleware::new_with_policy(
|
||||
.with(RetryGateMiddleware::new(
|
||||
config.retry_policy,
|
||||
config.max_total_retry_duration,
|
||||
))
|
||||
.with(RetryAfterMiddleware::with_capacity(
|
||||
.with(RetryAfterMiddleware::with_capacity_and_ceiling(
|
||||
DEFAULT_RETRY_AFTER_CAPACITY,
|
||||
config.retry_after_ceiling,
|
||||
))
|
||||
.build();
|
||||
Ok(client)
|
||||
@@ -186,14 +462,16 @@ fn concat_pem(cert: &[u8], key: &[u8]) -> Vec<u8> {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::time::SystemTime;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
|
||||
fn minimal_config() -> HttpClientConfig {
|
||||
HttpClientConfig {
|
||||
pool_max_idle_per_host: Some(8),
|
||||
request_timeout: Some(Duration::from_secs(30)),
|
||||
retry_policy: ExponentialBackoff::builder().build_with_max_retries(2),
|
||||
ca_bundle: None,
|
||||
client_cert: None,
|
||||
..HttpClientConfig::default()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -209,18 +487,18 @@ mod tests {
|
||||
assert_eq!(request.url().as_str(), "https://api.example.com/v1/chat");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reload_swaps_the_client_returned_by_client() {
|
||||
#[tokio::test]
|
||||
async fn reload_swaps_the_client_returned_by_client() {
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let before = http.client();
|
||||
let new_config = HttpClientConfig {
|
||||
pool_max_idle_per_host: Some(32),
|
||||
request_timeout: Some(Duration::from_secs(10)),
|
||||
retry_policy: ExponentialBackoff::builder().build_with_max_retries(5),
|
||||
ca_bundle: 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();
|
||||
assert!(
|
||||
!Arc::ptr_eq(&before, &after),
|
||||
@@ -228,7 +506,6 @@ mod tests {
|
||||
);
|
||||
let config = http.config();
|
||||
assert_eq!(config.pool_max_idle_per_host, Some(32));
|
||||
assert_eq!(config.request_timeout, Some(Duration::from_secs(10)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -243,20 +520,25 @@ mod tests {
|
||||
fn default_config_has_sensible_defaults() {
|
||||
let config = HttpClientConfig::default();
|
||||
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.client_cert.is_none());
|
||||
assert_eq!(config.retry_policy.max_n_retries, Some(3));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reload_with_ca_bundle_missing_file_errors() {
|
||||
#[tokio::test]
|
||||
async fn reload_with_ca_bundle_missing_file_errors() {
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let bad_config = HttpClientConfig {
|
||||
ca_bundle: Some(PathBuf::from("/nonexistent/ca-bundle.pem")),
|
||||
..minimal_config()
|
||||
};
|
||||
let err = http.reload(bad_config).unwrap_err();
|
||||
let err = http.reload(bad_config).await.unwrap_err();
|
||||
assert!(matches!(err, HttpClientBuildError::CaBundleRead { .. }));
|
||||
}
|
||||
|
||||
@@ -326,4 +608,318 @@ mod tests {
|
||||
let config = HttpClientConfig::default();
|
||||
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
@@ -3,10 +3,24 @@
|
||||
//!
|
||||
//! Inlined (MIT, from `melotic/reqwest-retry-after`) so the upstream's
|
||||
//! unbounded `HashMap<Url, SystemTime>` storage can be bounded for a
|
||||
//! long-running process. The bound is enforced via LRU eviction: when
|
||||
//! the map is at capacity, the entry with the earliest deadline is
|
||||
//! evicted first (those are the most likely to have already elapsed
|
||||
//! and are the cheapest to drop).
|
||||
//! long-running process.
|
||||
//!
|
||||
//! Policy (review-001 FWD-05 / FWD-11):
|
||||
//!
|
||||
//! - 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::sync::Mutex;
|
||||
@@ -20,35 +34,58 @@ use url::Url;
|
||||
const RETRY_AFTER_HEADER: &str = "retry-after";
|
||||
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 {
|
||||
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();
|
||||
if let Ok(secs) = trimmed.parse::<u64>() {
|
||||
return SystemTime::now()
|
||||
.checked_add(Duration::from_secs(secs))
|
||||
.filter(|deadline| *deadline > SystemTime::now());
|
||||
let parsed = if let Ok(secs) = trimmed.parse::<u64>() {
|
||||
SystemTime::now().checked_add(Duration::from_secs(secs))?
|
||||
} else {
|
||||
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)
|
||||
.ok()
|
||||
.filter(|deadline| *deadline > SystemTime::now())
|
||||
Some(clamped)
|
||||
}
|
||||
|
||||
pub struct RetryAfterMiddleware {
|
||||
deadlines: Mutex<HashMap<Url, SystemTime>>,
|
||||
capacity: usize,
|
||||
ceiling: Duration,
|
||||
}
|
||||
|
||||
impl RetryAfterMiddleware {
|
||||
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 {
|
||||
deadlines: Mutex::new(HashMap::with_capacity(capacity.min(128))),
|
||||
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) {
|
||||
let mut deadlines = self.deadlines.lock().unwrap_or_else(|e| e.into_inner());
|
||||
if !deadlines.contains_key(&url) && deadlines.len() >= self.capacity {
|
||||
@@ -58,12 +95,22 @@ impl RetryAfterMiddleware {
|
||||
}
|
||||
|
||||
fn evict(&self, deadlines: &mut HashMap<Url, SystemTime>) {
|
||||
if let Some(evict_url) = deadlines
|
||||
let now = SystemTime::now();
|
||||
let expired = deadlines
|
||||
.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())
|
||||
{
|
||||
deadlines.remove(&evict_url);
|
||||
deadlines.remove(&farthest);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -73,13 +120,20 @@ impl RetryAfterMiddleware {
|
||||
}
|
||||
|
||||
async fn maybe_sleep_for(&self, url: &Url) {
|
||||
if let Some(deadline) = self.deadline_for(url) {
|
||||
if let Ok(remaining) = deadline.duration_since(SystemTime::now()) {
|
||||
if !remaining.is_zero() {
|
||||
tokio::time::sleep(remaining).await;
|
||||
}
|
||||
}
|
||||
let Some(deadline) = self.deadline_for(url) else {
|
||||
return;
|
||||
};
|
||||
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) {
|
||||
@@ -90,7 +144,7 @@ impl RetryAfterMiddleware {
|
||||
.get(RETRY_AFTER_HEADER)
|
||||
.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);
|
||||
}
|
||||
}
|
||||
@@ -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]
|
||||
impl Middleware for RetryAfterMiddleware {
|
||||
async fn handle(
|
||||
@@ -127,7 +195,7 @@ impl Middleware for RetryAfterMiddleware {
|
||||
let req_url = req.url().clone();
|
||||
self.maybe_sleep_for(&req_url).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)
|
||||
}
|
||||
}
|
||||
@@ -150,33 +218,83 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
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 lower = now.checked_add(Duration::from_secs(4)).unwrap();
|
||||
let upper = now.checked_add(Duration::from_secs(6)).unwrap();
|
||||
assert!(deadline > lower && deadline < upper);
|
||||
assert!(deadline > now);
|
||||
assert!(deadline < now + Duration::from_secs(6));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_retry_after_http_date() {
|
||||
let deadline =
|
||||
parse_retry_after("Wed, 21 Oct 2099 07:28:00 GMT").expect("HTTP-date 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::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());
|
||||
}
|
||||
|
||||
#[test]
|
||||
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!(
|
||||
deadline.is_none(),
|
||||
mw.deadline_for_test(&target).is_none(),
|
||||
"a deadline already in the past must not be recorded"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_retry_after_invalid_yields_none() {
|
||||
assert!(parse_retry_after("not-a-date").is_none());
|
||||
assert!(parse_retry_after("").is_none());
|
||||
let ceiling = Duration::from_secs(300);
|
||||
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]
|
||||
@@ -190,21 +308,38 @@ mod tests {
|
||||
}
|
||||
|
||||
#[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 u1 = url("https://a.example.com");
|
||||
let u2 = url("https://b.example.com");
|
||||
let u3 = url("https://c.example.com");
|
||||
mw.record_test(u1.clone(), SystemTime::now() + Duration::from_secs(100));
|
||||
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));
|
||||
assert_eq!(mw.len(), 2, "capacity must be enforced");
|
||||
assert!(
|
||||
mw.deadline_for_test(&u2).is_none(),
|
||||
"entry with the earliest deadline must be evicted"
|
||||
mw.deadline_for_test(&u1).is_none(),
|
||||
"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());
|
||||
}
|
||||
|
||||
@@ -225,32 +360,17 @@ mod tests {
|
||||
}
|
||||
|
||||
#[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 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"));
|
||||
mw.record_if_throttled(target.clone(), &response);
|
||||
let deadline = mw
|
||||
.deadline_for_test(&target)
|
||||
.expect("429 with Retry-After records a 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(origin.clone(), &response);
|
||||
assert!(
|
||||
mw.deadline_for_test(&origin).is_some(),
|
||||
"the effective (post-redirect) URL carries the deadline"
|
||||
);
|
||||
mw.record_if_throttled(target.clone(), &response);
|
||||
let deadline = mw
|
||||
.deadline_for_test(&target)
|
||||
.expect("503 with Retry-After HTTP-date records a deadline");
|
||||
assert!(deadline > SystemTime::now());
|
||||
assert_eq!(redirector.host_str(), Some("redirector.example.com"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -283,8 +403,39 @@ mod tests {
|
||||
mw.maybe_sleep_for(&target).await;
|
||||
let elapsed = SystemTime::now().duration_since(started).unwrap();
|
||||
assert!(
|
||||
elapsed >= Duration::from_millis(40),
|
||||
"middleware must sleep until the deadline elapses"
|
||||
elapsed >= Duration::from_millis(37),
|
||||
"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"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user