#![cfg_attr(not(test), deny(clippy::disallowed_methods))]
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use bytes::Bytes;
use reqwest::Method;
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use serde::Serialize;
use serde::de::DeserializeOwned;
use crate::config::RetryBudgetConfig;
use crate::deadline::{DEADLINE_HEADER, Deadline};
use crate::retry_budget::{RetryBudget, RetryBudgets, RetryKind};
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum ClientError {
#[error("outbound HTTP request failed: {0}")]
Request(#[from] reqwest::Error),
#[error("JSON error: {0}")]
Json(#[from] serde_json::Error),
#[error("no mock registered for {0} {1}")]
NoMock(String, String),
#[error("outbound circuit breaker is open")]
CircuitBreakerOpen,
#[error("{0}")]
FaultInjected(String),
#[error("outbound request to {host} throttled locally: the host rejects too many requests")]
ThrottledLocally {
host: String,
},
#[error("{0}")]
ReplayedRequestFailure(String),
#[error(
"outbound HTTP is not recorded in a failure capsule and is blocked during replay: {0} {1}"
)]
BlockedDuringReplay(String, String),
#[error("SSRF policy blocked address: {0}")]
SsrfBlocked(String),
#[error("too many redirects (max {0})")]
TooManyRedirects(usize),
#[error("redirect rejected: {0}")]
RedirectRejected(String),
#[error("invalid or unresolvable URL: {0}")]
InvalidUrl(String),
#[error("{0}")]
IncompatiblePinRedirect(&'static str),
#[error("{0}")]
PinRequiresDomainHost(&'static str),
#[error("{0}")]
PinNotAllowedWithSsrfSafe(&'static str),
#[error("simulated network: {0}")]
SimNetwork(String),
#[error("request deadline exceeded before the outbound request could start")]
DeadlineExceeded,
}
#[derive(Debug)]
pub struct Response {
status: reqwest::StatusCode,
headers: HeaderMap,
body: Bytes,
url: Option<reqwest::Url>,
}
impl Response {
pub const fn status(&self) -> reqwest::StatusCode {
self.status
}
pub const fn headers(&self) -> &HeaderMap {
&self.headers
}
pub fn is_success(&self) -> bool {
self.status.is_success()
}
pub const fn url(&self) -> Option<&reqwest::Url> {
self.url.as_ref()
}
pub fn json<T: DeserializeOwned>(self) -> Result<T, ClientError> {
serde_json::from_slice(&self.body).map_err(ClientError::Json)
}
pub fn text(self) -> String {
String::from_utf8_lossy(&self.body).into_owned()
}
pub fn bytes(self) -> Bytes {
self.body
}
}
#[must_use]
pub fn is_blocked_ip(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => is_blocked_ipv4(v4),
IpAddr::V6(v6) => is_blocked_ipv6(v6),
}
}
#[must_use]
pub fn is_public_ip(ip: IpAddr) -> bool {
!is_blocked_ip(ip)
}
fn is_blocked_ipv4(ip: Ipv4Addr) -> bool {
let [a, b, c, _d] = ip.octets();
if a == 0 {
return true;
}
if a == 10 {
return true;
}
if a == 100 && (64..=127).contains(&b) {
return true;
}
if a == 127 {
return true;
}
if a == 169 && b == 254 {
return true;
}
if a == 172 && (16..=31).contains(&b) {
return true;
}
if a == 192 && b == 0 && c == 0 {
return true;
}
if a == 192 && b == 0 && c == 2 {
return true;
}
if a == 192 && b == 88 && c == 99 {
return true;
}
if a == 192 && b == 168 {
return true;
}
if a == 198 && (18..=19).contains(&b) {
return true;
}
if a == 198 && b == 51 && c == 100 {
return true;
}
if a == 203 && b == 0 && c == 113 {
return true;
}
if a >= 224 {
return true;
}
false
}
fn embedded_ipv4(ip: Ipv6Addr) -> Option<Ipv4Addr> {
if let Some(v4) = ip.to_ipv4_mapped() {
return Some(v4);
}
let segs = ip.segments();
if segs[0] == 0
&& segs[1] == 0
&& segs[2] == 0
&& segs[3] == 0
&& segs[4] == 0xffff
&& segs[5] == 0
{
let o = ip.octets();
return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
}
let o = ip.octets();
if o[0] == 0x00
&& o[1] == 0x64
&& o[2] == 0xff
&& o[3] == 0x9b
&& o[4..12].iter().all(|&b| b == 0)
{
return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
}
if o[0] == 0x20 && o[1] == 0x02 {
return Some(Ipv4Addr::new(o[2], o[3], o[4], o[5]));
}
ip.to_ipv4()
}
fn is_blocked_ipv6(ip: Ipv6Addr) -> bool {
if let Some(v4) = embedded_ipv4(ip)
&& is_blocked_ipv4(v4)
{
return true;
}
let segs = ip.segments();
if segs[0] == 0x0064 && segs[1] == 0xff9b && segs[2] == 0x0001 {
return true;
}
if ip == Ipv6Addr::UNSPECIFIED {
return true;
}
if ip == Ipv6Addr::LOCALHOST {
return true;
}
if segs[0] & 0xfe00 == 0xfc00 {
return true;
}
if segs[0] & 0xffc0 == 0xfe80 {
return true;
}
if segs[0] & 0xffc0 == 0xfec0 {
return true;
}
if segs[0] & 0xff00 == 0xff00 {
return true;
}
if segs[0] == 0x2001 && segs[1] == 0x0db8 {
return true;
}
let s = segs;
if s[0] == 0x0100 && s[1] == 0 && s[2] == 0 && s[3] == 0 {
return true;
}
if s[0] == 0x2001 && s[1] == 0x0002 && s[2] == 0x0000 {
return true;
}
if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0010 {
return true;
}
if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0020 {
return true;
}
if s[0] == 0x3fff && (s[1] & 0xF000) == 0x0000 {
return true;
}
if s[0] == 0x5f00 {
return true;
}
if s[0] == 0x2620 && s[1] == 0x004f && s[2] == 0x8000 {
return true;
}
false
}
#[derive(Clone, Debug)]
pub struct RetryPolicy {
pub max_retries: u32,
pub retry_idempotent_only: bool,
pub max_retry_after: Duration,
pub request_timeout: Option<Duration>,
pub max_backoff: Duration,
}
impl Default for RetryPolicy {
fn default() -> Self {
Self {
max_retries: 3,
retry_idempotent_only: true,
max_retry_after: Duration::from_secs(10),
request_timeout: Some(Duration::from_secs(30)),
max_backoff: Duration::from_millis(crate::backoff::DEFAULT_HTTP_MAX_BACKOFF_MS),
}
}
}
const BASE_BACKOFF: Duration = Duration::from_millis(100);
const IDEMPOTENCY_KEY: &str = "idempotency-key";
impl RetryPolicy {
fn retry_delay(
&self,
entropy: &dyn crate::entropy::Entropy,
attempt: u32,
hint: Option<Duration>,
) -> Duration {
let backoff = crate::backoff::full_jitter(entropy, BASE_BACKOFF, self.max_backoff, attempt);
let Some(mut hint) = hint else {
return backoff;
};
hint = hint.min(self.max_retry_after);
if let Some(timeout) = self.request_timeout {
hint = hint.min(timeout);
}
crate::backoff::retry_after_wait(hint, backoff)
}
}
const fn is_retryable_response(status: u16) -> bool {
status == 429 || is_retryable_status(status)
}
fn retry_hint(status: u16, headers: &HeaderMap) -> Option<Duration> {
match status {
429 => Some(parse_retry_after(headers).unwrap_or(Duration::from_secs(1))),
503 => parse_retry_after(headers),
_ => None,
}
}
pub(crate) struct MockEntry {
pub(crate) method: Option<Method>,
pub(crate) path: String,
pub(crate) alias: Option<String>,
pub(crate) status: u16,
pub(crate) body: Option<serde_json::Value>,
pub(crate) call_count: Arc<AtomicUsize>,
}
pub(crate) struct MockResponse {
pub(crate) status: u16,
pub(crate) body: Option<serde_json::Value>,
}
pub struct MockRegistry {
entries: Mutex<Vec<MockEntry>>,
}
impl MockRegistry {
#[must_use]
pub const fn new() -> Self {
Self {
entries: Mutex::new(Vec::new()),
}
}
pub(crate) fn register(&self, entry: MockEntry) {
self.entries
.lock()
.expect("mock registry lock poisoned")
.push(entry);
}
pub(crate) fn find_match(
&self,
method: &Method,
url: &str,
alias: Option<&str>,
) -> Option<MockResponse> {
let url_path_owned: String = reqwest::Url::parse(url).map_or_else(
|_| {
let s = url.split_once('?').map_or(url, |(p, _)| p);
s.split_once('#').map_or(s, |(p, _)| p).to_owned()
},
|parsed| parsed.path().to_owned(),
);
let url_path = url_path_owned.as_str();
let found = {
let entries = self.entries.lock().expect("mock registry lock poisoned");
entries.iter().find_map(|entry| {
let method_ok = entry.method.as_ref().is_none_or(|m| m == method);
let path_ok = url_path == entry.path.as_str()
|| url_path
.strip_suffix(entry.path.as_str())
.is_some_and(|prefix| {
prefix.is_empty()
|| prefix.ends_with('/')
|| entry.path.starts_with('/')
});
let alias_ok = entry
.alias
.as_deref()
.is_none_or(|a| alias.is_some_and(|b| a == b));
if method_ok && path_ok && alias_ok {
Some((entry.call_count.clone(), entry.status, entry.body.clone()))
} else {
None
}
})
};
found.map(|(call_count, status, body)| {
call_count.fetch_add(1, Ordering::SeqCst);
MockResponse { status, body }
})
}
}
impl Default for MockRegistry {
fn default() -> Self {
Self::new()
}
}
static OUTBOUND_BLOCKED: AtomicBool = AtomicBool::new(false);
#[cfg_attr(not(feature = "reporting"), allow(dead_code))]
pub(crate) fn block_outbound_for_replay() {
OUTBOUND_BLOCKED.store(true, Ordering::SeqCst);
}
#[must_use]
pub(crate) fn outbound_blocked_for_replay() -> bool {
OUTBOUND_BLOCKED.load(Ordering::SeqCst)
}
#[cfg(feature = "reporting")]
const OPAQUE_HEADER_PREFIX: &str = "[bytes] ";
#[cfg(feature = "reporting")]
fn caller_headers(builder: &RequestBuilder) -> Vec<(String, String)> {
builder
.extra_headers
.iter()
.filter(|(name, _)| {
builder.web_bot_auth.is_none()
|| !matches!(name.as_str(), "signature" | "signature-input")
})
.map(|(name, value)| {
(
name.as_str().to_owned(),
match value.to_str() {
Ok(text) if !text.starts_with(OPAQUE_HEADER_PREFIX) => text.to_owned(),
_ => format!("{OPAQUE_HEADER_PREFIX}{}", value.as_bytes().escape_ascii()),
},
)
})
.collect()
}
#[cfg(feature = "reporting")]
const fn http_error_kind(error: &ClientError) -> crate::capsule::schema::HttpErrorKind {
use crate::capsule::schema::HttpErrorKind as Kind;
match error {
ClientError::Json(_) => Kind::Json,
ClientError::NoMock(..) => Kind::NoMock,
ClientError::CircuitBreakerOpen => Kind::CircuitBreakerOpen,
ClientError::ThrottledLocally { .. } => Kind::ThrottledLocally,
ClientError::SsrfBlocked(_) => Kind::SsrfBlocked,
ClientError::TooManyRedirects(_) => Kind::TooManyRedirects,
ClientError::RedirectRejected(_) => Kind::RedirectRejected,
ClientError::InvalidUrl(_) => Kind::InvalidUrl,
ClientError::FaultInjected(_) => Kind::FaultInjected,
_ => Kind::Transport,
}
}
#[cfg(feature = "reporting")]
fn rebuild_client_error(
kind: Option<crate::capsule::schema::HttpErrorKind>,
text: String,
) -> ClientError {
use crate::capsule::schema::HttpErrorKind as Kind;
match kind {
Some(Kind::CircuitBreakerOpen) => ClientError::CircuitBreakerOpen,
Some(Kind::ThrottledLocally) => text
.strip_prefix("outbound request to ")
.and_then(|rest| rest.split_once(" throttled locally"))
.map_or_else(
|| ClientError::ReplayedRequestFailure(text.clone()),
|(host, _)| ClientError::ThrottledLocally {
host: host.to_owned(),
},
),
Some(Kind::SsrfBlocked) => {
ClientError::SsrfBlocked(strip_prefix_payload(&text, "SSRF policy blocked address: "))
}
Some(Kind::RedirectRejected) => {
ClientError::RedirectRejected(strip_prefix_payload(&text, "redirect rejected: "))
}
Some(Kind::InvalidUrl) => {
ClientError::InvalidUrl(strip_prefix_payload(&text, "invalid or unresolvable URL: "))
}
Some(Kind::FaultInjected) => ClientError::FaultInjected(text),
Some(Kind::TooManyRedirects) => text
.trim_end_matches(')')
.rsplit_once("(max ")
.and_then(|(_, hops)| hops.parse().ok())
.map_or_else(
|| ClientError::ReplayedRequestFailure(text.clone()),
ClientError::TooManyRedirects,
),
Some(Kind::Json | Kind::NoMock | Kind::Transport) | None => {
ClientError::ReplayedRequestFailure(text)
}
}
}
#[cfg(feature = "reporting")]
fn strip_prefix_payload(text: &str, prefix: &str) -> String {
text.strip_prefix(prefix).unwrap_or(text).to_owned()
}
#[cfg(feature = "reporting")]
fn replayed_response(
tape: &crate::capsule::effects::ReplayEffects,
request: &crate::capsule::effects::OutboundRequest<'_>,
) -> Result<Response, ClientError> {
let Some(recorded) = tape.next_http(request) else {
return Err(ClientError::BlockedDuringReplay(
request.method.to_owned(),
crate::capsule::redact::hide_url_userinfo(request.url),
));
};
if let Some(error) = recorded.error {
return Err(rebuild_client_error(recorded.error_kind, error));
}
let mut headers = HeaderMap::new();
for (name, value) in &recorded.response_headers {
if let (Ok(name), Ok(value)) = (
HeaderName::from_bytes(name.as_bytes()),
HeaderValue::from_str(value),
) {
headers.append(name, value);
}
}
Ok(Response {
status: reqwest::StatusCode::from_u16(recorded.status)
.unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR),
headers,
body: Bytes::from(crate::capsule::effects::body_bytes(&recorded.response_body)),
url: recorded
.final_url
.as_deref()
.or(Some(request.url))
.and_then(|url| url::Url::parse(url).ok()),
})
}
#[cfg(feature = "reporting")]
struct OutboundRecorder {
scope: Option<Arc<crate::capsule::CaptureScope>>,
slot: Option<usize>,
max_body_bytes: usize,
method: String,
url: String,
request_headers: Vec<(String, String)>,
request_body: crate::capsule::CapsuleBody,
}
#[cfg(feature = "reporting")]
impl OutboundRecorder {
fn arm(builder: &RequestBuilder) -> Self {
let scope = crate::capsule::current_scope();
let Some(scope) = scope else {
return Self {
scope: None,
slot: None,
max_body_bytes: 0,
method: String::new(),
url: String::new(),
request_headers: Vec::new(),
request_body: crate::capsule::CapsuleBody::Absent,
};
};
let max_body_bytes = scope.settings().max_body_bytes;
let slot = scope.reserve_http();
let request_headers = caller_headers(builder);
Self {
scope: Some(scope),
slot,
max_body_bytes,
method: builder.method.to_string(),
url: builder.url.clone(),
request_headers,
request_body: builder
.body
.as_ref()
.map_or(crate::capsule::CapsuleBody::Absent, |body| {
encode_body(body, max_body_bytes)
}),
}
}
fn finish(self, result: &Result<Response, ClientError>) {
let (Some(scope), Some(slot)) = (self.scope, self.slot) else {
return;
};
let (status, response_headers, response_body, error, final_url) = match result {
Ok(response) => (
response.status.as_u16(),
response
.headers
.iter()
.map(|(name, value)| {
(
name.as_str().to_owned(),
value.to_str().map_or_else(
|_| String::from_utf8_lossy(value.as_bytes()).into_owned(),
ToOwned::to_owned,
),
)
})
.collect(),
encode_body(&response.body, self.max_body_bytes),
None,
response
.url()
.map(reqwest::Url::as_str)
.filter(|final_url| *final_url != self.url)
.map(ToOwned::to_owned),
),
Err(error) => (
0,
Vec::new(),
crate::capsule::CapsuleBody::Absent,
Some((error.to_string(), http_error_kind(error))),
None,
),
};
let (error, error_kind) = match error {
Some((text, kind)) => (Some(text), Some(kind)),
None => (None, None),
};
scope.fill_http(
slot,
crate::capsule::HttpEffect {
method: self.method,
url: self.url,
request_headers: self.request_headers,
request_body: self.request_body,
status,
response_headers,
final_url,
response_body,
error,
error_kind,
},
);
}
}
#[cfg(feature = "reporting")]
fn encode_body(bytes: &Bytes, max_body_bytes: usize) -> crate::capsule::CapsuleBody {
if bytes.is_empty() {
return crate::capsule::CapsuleBody::Absent;
}
if bytes.len() > max_body_bytes {
return crate::capsule::CapsuleBody::Skipped {
declared_len: Some(bytes.len()),
};
}
std::str::from_utf8(bytes).map_or_else(
|_| {
use base64::Engine as _;
crate::capsule::CapsuleBody::Base64(
base64::engine::general_purpose::STANDARD.encode(bytes),
)
},
|text| crate::capsule::CapsuleBody::Text(text.to_owned()),
)
}
#[cfg(not(feature = "reporting"))]
struct OutboundRecorder;
#[cfg(not(feature = "reporting"))]
impl OutboundRecorder {
const fn arm(_builder: &RequestBuilder) -> Self {
Self
}
const fn finish(self, _result: &Result<Response, ClientError>) {}
}
pub struct HttpMockRegistryExt(pub Arc<MockRegistry>);
#[derive(Clone)]
pub(crate) struct SharedReqwestClient {
pub(crate) client: reqwest::Client,
pub(crate) timeout_secs: u64,
}
pub struct MockHandle {
alias: String,
method: String,
path: String,
call_count: Arc<AtomicUsize>,
}
impl MockHandle {
pub fn expect_called(&self, expected: usize) {
let actual = self.call_count.load(Ordering::SeqCst);
assert_eq!(
actual, expected,
"http mock for {} {} {} expected {} call(s) but got {}",
self.alias, self.method, self.path, expected, actual,
);
}
#[must_use]
pub fn call_count(&self) -> usize {
self.call_count.load(Ordering::SeqCst)
}
}
pub struct MockSetupBuilder {
pub(crate) registry: Arc<MockRegistry>,
pub(crate) alias: String,
pub(crate) method: Option<Method>,
pub(crate) path: Option<String>,
}
impl MockSetupBuilder {
#[must_use]
pub fn get(mut self, path: &str) -> Self {
self.method = Some(Method::GET);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn post(mut self, path: &str) -> Self {
self.method = Some(Method::POST);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn put(mut self, path: &str) -> Self {
self.method = Some(Method::PUT);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn patch(mut self, path: &str) -> Self {
self.method = Some(Method::PATCH);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn delete(mut self, path: &str) -> Self {
self.method = Some(Method::DELETE);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn head(mut self, path: &str) -> Self {
self.method = Some(Method::HEAD);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn respond_with(self, status: u16, body: serde_json::Value) -> MockHandle {
let path = self.path.clone().unwrap_or_default();
let method_str = self
.method
.as_ref()
.map_or_else(|| "*".to_owned(), ToString::to_string);
let call_count = Arc::new(AtomicUsize::new(0));
self.registry.register(MockEntry {
method: self.method,
path: path.clone(),
alias: Some(self.alias.clone()),
status,
body: Some(body),
call_count: call_count.clone(),
});
MockHandle {
alias: self.alias,
method: method_str,
path,
call_count,
}
}
#[must_use]
pub fn respond_with_status(self, status: u16) -> MockHandle {
let path = self.path.clone().unwrap_or_default();
let method_str = self
.method
.as_ref()
.map_or_else(|| "*".to_owned(), ToString::to_string);
let call_count = Arc::new(AtomicUsize::new(0));
self.registry.register(MockEntry {
method: self.method,
path: path.clone(),
alias: Some(self.alias.clone()),
status,
body: None,
call_count: call_count.clone(),
});
MockHandle {
alias: self.alias,
method: method_str,
path,
call_count,
}
}
}
#[derive(Clone)]
pub struct Client {
inner: reqwest::Client,
alias: Option<String>,
base_url: Option<String>,
base_urls: HashMap<String, String>,
retry_policy: RetryPolicy,
mock: Option<Arc<MockRegistry>>,
resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
sim_net: Option<Arc<crate::sim::SimNet>>,
retry: RetrySettings,
entropy: Arc<dyn crate::entropy::Entropy>,
throttle: Option<Arc<crate::admission::AdaptiveThrottle>>,
}
#[derive(Clone)]
pub(crate) struct SharedThrottle {
throttle: Arc<crate::admission::AdaptiveThrottle>,
config: crate::config::AdaptiveThrottleConfig,
}
pub(crate) fn install_shared_throttle(
state: &crate::AppState,
config: &crate::config::HttpClientConfig,
) {
let _ = shared_throttle(state, config);
}
static THROTTLE_REPLACE: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn shared_throttle(
state: &crate::AppState,
config: &crate::config::HttpClientConfig,
) -> Option<Arc<crate::admission::AdaptiveThrottle>> {
let settings = config.adaptive_throttle;
if !settings.enabled {
return None;
}
let current = || {
state
.extension::<SharedThrottle>()
.filter(|shared| shared.config == settings)
.map(|shared| Arc::clone(&shared.throttle))
};
if let Some(throttle) = current() {
return Some(throttle);
}
let _guard = THROTTLE_REPLACE
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(throttle) = current() {
return Some(throttle);
}
let throttle = throttle_from_config(config)?;
state.insert_extension(SharedThrottle {
throttle: Arc::clone(&throttle),
config: settings,
});
Some(throttle)
}
fn throttle_from_config(
config: &crate::config::HttpClientConfig,
) -> Option<Arc<crate::admission::AdaptiveThrottle>> {
let t = config.adaptive_throttle;
t.enabled.then(|| {
Arc::new(crate::admission::AdaptiveThrottle::new(
t.k,
Duration::from_secs(t.window_secs),
))
})
}
fn os_entropy() -> Arc<dyn crate::entropy::Entropy> {
Arc::new(crate::entropy::OsEntropy)
}
impl Client {
#[must_use]
pub fn new() -> Self {
Self::with_timeout(Duration::from_secs(30))
}
#[must_use]
pub fn with_timeout(timeout: Duration) -> Self {
let inner = reqwest::ClientBuilder::new()
.timeout(timeout)
.redirect(pooled_redirect_policy())
.build()
.expect("failed to build reqwest client");
Self {
inner,
alias: None,
base_url: None,
base_urls: HashMap::new(),
retry_policy: RetryPolicy {
request_timeout: Some(timeout),
..RetryPolicy::default()
},
mock: None,
resilience_config: None,
sim_net: None,
retry: RetrySettings::standalone(),
entropy: os_entropy(),
throttle: None,
}
}
pub(crate) fn build_inner(config: &crate::config::HttpClientConfig) -> reqwest::Client {
reqwest::ClientBuilder::new()
.timeout(Duration::from_secs(config.timeout_secs))
.redirect(pooled_redirect_policy())
.build()
.expect("failed to build reqwest client")
}
fn from_config_with_inner(
inner: reqwest::Client,
config: &crate::config::HttpClientConfig,
) -> Self {
let timeout = Duration::from_secs(config.timeout_secs);
Self {
inner,
alias: None,
base_url: None,
base_urls: config.base_urls.clone(),
retry_policy: RetryPolicy {
max_retries: config.max_retries,
retry_idempotent_only: true,
max_retry_after: Duration::from_secs(config.max_retry_after_secs),
request_timeout: Some(timeout),
max_backoff: Duration::from_millis(config.max_backoff_ms),
},
mock: None,
resilience_config: None,
sim_net: None,
retry: RetrySettings {
budgets: None,
send_deadline_header: config.send_deadline_header,
},
entropy: os_entropy(),
throttle: throttle_from_config(config),
}
}
fn with_inner(inner: reqwest::Client) -> Self {
Self {
inner,
alias: None,
base_url: None,
base_urls: HashMap::new(),
retry_policy: RetryPolicy::default(),
mock: None,
resilience_config: None,
sim_net: None,
retry: RetrySettings {
budgets: None,
send_deadline_header: true,
},
entropy: os_entropy(),
throttle: None,
}
}
#[must_use]
pub fn from_config(config: &crate::config::HttpClientConfig) -> Self {
let mut client = Self::from_config_with_inner(Self::build_inner(config), config);
client.retry.budgets = config
.retry_budget
.enabled
.then(|| Arc::new(RetryBudgets::new(&config.retry_budget)));
client
}
pub(crate) fn with_mock(mut self, registry: Arc<MockRegistry>) -> Self {
self.mock = Some(registry);
self
}
#[must_use]
pub fn from_state(state: &crate::AppState) -> Self {
let autumn_config = state.extension::<crate::config::AutumnConfig>();
let config = state
.extension::<crate::config::HttpConfig>()
.or_else(|| autumn_config.as_ref().map(|c| Arc::new(c.http.clone())));
let effective_timeout_secs = config.as_ref().map_or_else(
|| crate::config::HttpClientConfig::default().timeout_secs,
|c| c.client.timeout_secs,
);
let shared = state.extension::<SharedReqwestClient>().and_then(|s| {
if s.timeout_secs == effective_timeout_secs {
Some(s.client.clone())
} else {
None
}
});
let budget_config = config
.as_ref()
.map_or_else(RetryBudgetConfig::default, |cfg| {
cfg.client.retry_budget.clone()
});
let throttle = config
.as_ref()
.and_then(|cfg| shared_throttle(state, &cfg.client));
let mut client = match (config, shared) {
(Some(cfg), Some(inner)) => Self::from_config_with_inner(inner, &cfg.client),
(Some(cfg), None) => Self::from_config(&cfg.client),
(None, Some(inner)) => Self::with_inner(inner),
(None, None) => Self::new(),
};
client.retry.budgets = shared_retry_budgets(state, &budget_config);
client.resilience_config = autumn_config.map(|c| Arc::new(c.resilience.clone()));
if let Some(ext) = state.extension::<HttpMockRegistryExt>() {
client = client.with_mock(ext.0.clone());
}
client.sim_net = state.extension::<crate::sim::SimNet>();
client.throttle = throttle;
let entropy = state.entropy_arc();
client.entropy = entropy.unrecorded().unwrap_or(entropy);
client
}
#[must_use]
pub fn named(&self, alias: &str) -> Self {
let base_url = self
.base_urls
.get(alias)
.cloned()
.or_else(|| self.base_url.clone());
Self {
inner: self.inner.clone(),
alias: Some(alias.to_owned()),
base_url,
base_urls: self.base_urls.clone(),
retry_policy: self.retry_policy.clone(),
mock: self.mock.clone(),
resilience_config: self.resilience_config.clone(),
sim_net: self.sim_net.clone(),
retry: self.retry.clone(),
entropy: self.entropy.clone(),
throttle: self.throttle.clone(),
}
}
#[must_use]
pub fn with_base_url(&self, base_url: impl Into<String>) -> Self {
Self {
inner: self.inner.clone(),
alias: self.alias.clone(),
base_url: Some(base_url.into()),
base_urls: self.base_urls.clone(),
retry_policy: self.retry_policy.clone(),
mock: self.mock.clone(),
resilience_config: self.resilience_config.clone(),
sim_net: self.sim_net.clone(),
retry: self.retry.clone(),
entropy: self.entropy.clone(),
throttle: self.throttle.clone(),
}
}
fn build_request(&self, method: Method, url: impl AsRef<str>) -> RequestBuilder {
let url_str = url.as_ref();
let full_url = if url_str.starts_with("http://") || url_str.starts_with("https://") {
url_str.to_owned()
} else if let Some(base) = &self.base_url {
format!(
"{}/{}",
base.trim_end_matches('/'),
url_str.trim_start_matches('/')
)
} else {
url_str.to_owned()
};
RequestBuilder {
client: self.inner.clone(),
method,
url: full_url,
extra_headers: HeaderMap::new(),
body: None,
retry_policy: self.retry_policy.clone(),
mock: self.mock.clone(),
alias: self.alias.clone(),
pending_error: None,
resilience_config: self.resilience_config.clone(),
redirect_mode: RedirectMode::Default,
pin_addr: None,
ssrf_safe: false,
discard_response_body: false,
breaker_scoped: false,
sim_net: self.sim_net.clone(),
forward_request_id: true,
retry: self.retry.clone(),
entropy: self.entropy.clone(),
throttle: self.throttle.clone(),
criticality: None,
web_bot_auth: None,
}
}
#[must_use]
pub fn get(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::GET, url)
}
#[must_use]
pub fn post(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::POST, url)
}
#[must_use]
pub fn put(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::PUT, url)
}
#[must_use]
pub fn patch(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::PATCH, url)
}
#[must_use]
pub fn delete(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::DELETE, url)
}
#[must_use]
pub fn head(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::HEAD, url)
}
#[must_use]
pub fn get_ssrf_safe(&self, url: impl Into<String>) -> RequestBuilder {
self.build_request(Method::GET, url.into()).ssrf_safe()
}
}
impl Default for Client {
fn default() -> Self {
Self::new()
}
}
impl axum::extract::FromRequestParts<crate::AppState> for Client {
type Rejection = std::convert::Infallible;
async fn from_request_parts(
_parts: &mut http::request::Parts,
state: &crate::AppState,
) -> Result<Self, std::convert::Infallible> {
Ok(Self::from_state(state))
}
}
type RedirectValidator = Arc<dyn Fn(&str) -> bool + Send + Sync>;
tokio::task_local! {
static FOLLOWED: std::cell::RefCell<Vec<String>>;
static MANUAL_REDIRECTS: bool;
}
async fn send_tracking_redirects(
req: reqwest::RequestBuilder,
) -> (Result<reqwest::Response, reqwest::Error>, Vec<String>) {
FOLLOWED
.scope(std::cell::RefCell::new(Vec::new()), async {
let sent = req.send().await;
(sent, FOLLOWED.with(std::cell::RefCell::take))
})
.await
}
fn set_referer(headers: &mut HeaderMap, next: &str, previous: &str) {
let (Ok(next), Ok(mut referer)) = (url::Url::parse(next), url::Url::parse(previous)) else {
return;
};
if next.scheme() == "http" && referer.scheme() == "https" {
headers.remove(reqwest::header::REFERER);
return;
}
let _ = referer.set_username("");
let _ = referer.set_password(None);
referer.set_fragment(None);
if let Ok(value) = HeaderValue::from_str(referer.as_str()) {
headers.insert(reqwest::header::REFERER, value);
}
}
const PLAIN_MAX_REDIRECTS: usize = 10;
fn pooled_redirect_policy() -> reqwest::redirect::Policy {
reqwest::redirect::Policy::custom(|attempt| {
if Deadline::current().is_some() || MANUAL_REDIRECTS.try_with(|m| *m).unwrap_or(false) {
attempt.stop()
} else if attempt.previous().len() > PLAIN_MAX_REDIRECTS {
attempt.error("too many redirects")
} else {
let _ = FOLLOWED.try_with(|hops| hops.borrow_mut().push(attempt.url().to_string()));
attempt.follow()
}
})
}
#[derive(Clone, Copy)]
enum HopClient<'a> {
Pooled(&'a reqwest::Client),
OneShot(Duration),
}
enum RedirectMode {
Default,
None,
Follow {
max: usize,
validator: RedirectValidator,
},
}
const SSRF_SAFE_MAX_REDIRECTS: usize = 5;
#[allow(
clippy::struct_excessive_bools,
reason = "independent per-request switches; each has its own builder method"
)]
pub struct RequestBuilder {
client: reqwest::Client,
method: Method,
url: String,
extra_headers: HeaderMap,
body: Option<Bytes>,
retry_policy: RetryPolicy,
mock: Option<Arc<MockRegistry>>,
alias: Option<String>,
pending_error: Option<ClientError>,
resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
redirect_mode: RedirectMode,
pin_addr: Option<Vec<SocketAddr>>,
ssrf_safe: bool,
discard_response_body: bool,
breaker_scoped: bool,
sim_net: Option<Arc<crate::sim::SimNet>>,
forward_request_id: bool,
retry: RetrySettings,
entropy: Arc<dyn crate::entropy::Entropy>,
throttle: Option<Arc<crate::admission::AdaptiveThrottle>>,
criticality: Option<crate::admission::Criticality>,
web_bot_auth: Option<Arc<crate::aeo::web_bot_auth::WebBotAuthSigner>>,
}
impl RequestBuilder {
#[must_use]
pub const fn without_request_id(mut self) -> Self {
self.forward_request_id = false;
self
}
#[must_use]
pub const fn criticality(mut self, criticality: crate::admission::Criticality) -> Self {
self.criticality = Some(criticality);
self
}
#[must_use]
pub fn header(mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> Self {
let name_str = name.as_ref();
let value_str = value.as_ref();
match (
HeaderName::from_bytes(name_str.as_bytes()),
HeaderValue::from_str(value_str),
) {
(Ok(n), Ok(v)) => {
self.extra_headers.insert(n, v);
}
(Err(e), _) => {
tracing::warn!(header.name = name_str, error = %e, "invalid header name — header skipped");
}
(_, Err(e)) => {
tracing::warn!(header.name = name_str, error = %e, "invalid header value — header skipped");
}
}
self
}
#[must_use]
pub fn sign_web_bot_auth(
mut self,
signer: &crate::aeo::web_bot_auth::WebBotAuthSigner,
) -> Self {
if let Some(h) = signer.sign_url(&self.url) {
self.web_bot_auth = Some(Arc::new(signer.clone()));
return self
.header("signature-agent", h.signature_agent)
.header("signature-input", h.signature_input)
.header("signature", h.signature);
}
self.pending_error = Some(ClientError::InvalidUrl(format!(
"Web Bot Auth needs an absolute URL, got {:?}",
self.url
)));
self
}
#[must_use]
pub fn json<T: Serialize>(mut self, body: &T) -> Self {
match serde_json::to_vec(body) {
Ok(bytes) => {
self.body = Some(Bytes::from(bytes));
self = self.header("content-type", "application/json");
}
Err(e) => {
self.pending_error = Some(ClientError::Json(e));
}
}
self
}
#[must_use]
pub fn text_body(mut self, body: impl Into<String>) -> Self {
self.body = Some(Bytes::from(body.into().into_bytes()));
self
}
#[must_use]
pub fn bytes_body(mut self, body: impl Into<Bytes>) -> Self {
self.body = Some(body.into());
self
}
#[must_use]
pub const fn retries(mut self, max: u32) -> Self {
self.retry_policy.max_retries = max;
self
}
#[must_use]
pub const fn retry_non_idempotent(mut self) -> Self {
self.retry_policy.retry_idempotent_only = false;
self
}
#[must_use]
pub const fn max_backoff(mut self, max: Duration) -> Self {
self.retry_policy.max_backoff = max;
self
}
#[must_use]
pub const fn max_retry_after(mut self, max: Duration) -> Self {
self.retry_policy.max_retry_after = max;
self
}
#[must_use]
pub const fn no_retry(mut self) -> Self {
self.retry_policy.max_retries = 0;
self
}
#[must_use]
pub fn no_redirect(mut self) -> Self {
self.redirect_mode = RedirectMode::None;
self
}
#[must_use]
pub const fn discard_response_body(mut self) -> Self {
self.discard_response_body = true;
self
}
#[must_use]
pub fn follow_redirects<F>(mut self, max: usize, validator: F) -> Self
where
F: Fn(&str) -> bool + Send + Sync + 'static,
{
self.redirect_mode = RedirectMode::Follow {
max,
validator: Arc::new(validator),
};
self
}
#[must_use]
pub fn pin_to(self, addr: SocketAddr) -> Self {
self.pin_to_addrs([addr])
}
#[must_use]
pub fn pin_to_addrs(mut self, addrs: impl IntoIterator<Item = SocketAddr>) -> Self {
self.pin_addr = Some(addrs.into_iter().collect());
self
}
#[must_use]
pub const fn ssrf_safe(mut self) -> Self {
self.ssrf_safe = true;
self
}
#[must_use]
pub(crate) const fn breaker_scoped(mut self) -> Self {
self.breaker_scoped = true;
self
}
pub async fn send(self) -> Result<Response, ClientError> {
if let Some(err) = self.pending_error {
return Err(err);
}
#[cfg(feature = "reporting")]
if let Some(tape) = crate::capsule::effects::current_tape() {
let method = self.method.to_string();
let headers = caller_headers(&self);
let body = self
.body
.as_ref()
.map_or(crate::capsule::CapsuleBody::Absent, |body| {
encode_body(body, usize::MAX)
});
return replayed_response(
&tape,
&crate::capsule::effects::OutboundRequest {
method: &method,
url: &self.url,
headers: &headers,
body: &body,
},
);
}
#[cfg(feature = "reporting")]
if outbound_blocked_for_replay() {
return Err(ClientError::BlockedDuringReplay(
self.method.to_string(),
crate::capsule::redact::hide_url_userinfo(&self.url),
));
}
let recorder = OutboundRecorder::arm(&self);
let result = self.send_recorded().await;
recorder.finish(&result);
result
}
async fn send_recorded(mut self) -> Result<Response, ClientError> {
crate::fault_injection::inject(crate::fault_injection::FaultTarget::Http)
.await
.map_err(|fault| ClientError::FaultInjected(fault.to_string()))?;
if self.forward_request_id {
add_current_request_id(&mut self.extra_headers);
}
self.ensure_idempotency_key();
self.ensure_criticality_header();
if let Some(net) = self.sim_net.clone() {
return self.send_sim(&net).await;
}
if self.mock.is_some() {
return self.send_inner(false).await;
}
if self.needs_custom_path() {
if self.breaker_scoped {
return self.send_custom_breaker_guarded().await;
}
return self.send_custom(false).await;
}
let breaker = breaker_for_url(self.resilience_config.as_ref(), &self.url);
let Ok(guard) = breaker.admit() else {
return Err(ClientError::CircuitBreakerOpen);
};
let is_half_open = breaker.state() == crate::circuit_breaker::CircuitState::HalfOpen;
let res = self.send_inner(is_half_open).await;
match &res {
Ok(resp) => {
let success = resp.status().as_u16() < 500;
if success {
guard.success();
} else {
guard.failure();
}
}
Err(ClientError::DeadlineExceeded | ClientError::ThrottledLocally { .. }) => {
drop(guard);
}
Err(_) => {
guard.failure();
}
}
res
}
async fn send_custom_breaker_guarded(self) -> Result<Response, ClientError> {
let breaker = breaker_for_url(self.resilience_config.as_ref(), &self.url);
let Ok(guard) = breaker.admit() else {
return Err(ClientError::CircuitBreakerOpen);
};
let ticket = match self.throttle_attempt(None) {
Ok(ticket) => ticket,
Err(error) => {
drop(guard);
return Err(error);
}
};
let is_half_open = breaker.state() == crate::circuit_breaker::CircuitState::HalfOpen;
let res = self.send_custom(is_half_open).await;
record_call(ticket, &res);
match &res {
Ok(resp) if resp.status().as_u16() < 500 => guard.success(),
Err(ClientError::DeadlineExceeded) => drop(guard),
_ => guard.failure(),
}
res
}
async fn send_inner(self, suppress_retries: bool) -> Result<Response, ClientError> {
if let Some(ref mock) = self.mock {
return self.mock_response(mock);
}
let start = crate::time::ambient_instant();
let max_attempts = self.max_attempts(suppress_retries);
let mut gate = self.retry_gate(url_host(&self.url).as_deref());
if gate.deadline.is_some() || self.follows_redirects_itself() {
return self.follow_pooled(suppress_retries, &gate).await;
}
let mut last_retry = None;
let mut delay = Duration::ZERO;
let mut refilled: std::collections::HashSet<String> =
url_host(&self.url).into_iter().collect();
for attempt in 0..max_attempts {
if attempt > 0 {
tokio::time::sleep(delay).await;
}
let ticket = self.begin_attempt(&gate, None)?;
let last = attempt + 1 == max_attempts;
let (span, sent, followed) =
self.send_plain_attempt(&gate, attempt, &mut refilled).await;
match sent {
Ok(resp) => {
let status = resp.status();
let headers = resp.headers().clone();
let url_used = resp.url().clone();
if is_retryable_response(status.as_u16()) && !last {
gate.rekey(url_used.as_str());
let wait = self.retry_policy.retry_delay(
&*self.entropy,
attempt,
retry_hint(status.as_u16(), &headers),
);
let kind = retry_kind(status.as_u16());
if gate.allow(kind, wait) {
if let Some(ticket) = ticket {
ticket.record(throttle_accepts(status.as_u16()));
}
delay = wait;
last_retry = Some(kind);
continue;
}
}
let body = if self.discard_response_body {
Ok(Bytes::new())
} else {
read_body_in_span(resp, &span, &gate).await
};
if let Some(ticket) = ticket {
ticket.record(body.is_ok() && throttle_accepts(status.as_u16()));
}
let body = body?;
gate.finish(last_retry, status.as_u16());
let elapsed = crate::time::ambient_instant().saturating_duration_since(start);
log_request(
self.method.as_str(),
&url_used,
status.as_u16(),
elapsed,
&self.extra_headers,
);
return Ok(Response {
status,
headers,
body,
url: Some(url_used),
});
}
Err(e) if e.is_timeout() && gate.expired() => {
return Err(ClientError::DeadlineExceeded);
}
Err(e) if (e.is_connect() || e.is_timeout()) && !last => {
if let Some(ticket) = ticket {
ticket.record(false);
}
gate.rekey(followed.last().map_or(self.url.as_str(), String::as_str));
let wait = self.retry_policy.retry_delay(&*self.entropy, attempt, None);
if !gate.allow(RetryKind::Transient, wait) {
return Err(ClientError::Request(e.without_url()));
}
delay = wait;
last_retry = Some(RetryKind::Transient);
}
Err(e) => {
if let Some(ticket) = ticket {
ticket.record(false);
}
return Err(ClientError::Request(e.without_url()));
}
}
}
unreachable!("retry loop exited without returning a result — this is a bug")
}
fn retry_gate(&self, host: Option<&str>) -> RetryGate {
let mut gate = RetryGate::start(
self.retry.budgets.clone(),
host,
self.retry.send_deadline_header,
);
gate.signer.clone_from(&self.web_bot_auth);
gate
}
fn follows_redirects_itself(&self) -> bool {
self.extra_headers
.get(DEADLINE_HEADER)
.and_then(crate::deadline::parse_header)
.is_some()
|| self.extra_headers.contains_key("signature")
}
async fn follow_pooled(
self,
suppress_retries: bool,
gate: &RetryGate,
) -> Result<Response, ClientError> {
let client = self.client.clone();
MANUAL_REDIRECTS
.scope(
true,
self.follow_loop(
PLAIN_MAX_REDIRECTS,
Arc::new(|_: &str| true),
HopClient::Pooled(&client),
suppress_retries,
gate,
),
)
.await
}
fn ensure_idempotency_key(&mut self) {
let retries_unsafe_method = !is_idempotent_method(&self.method)
&& !self.retry_policy.retry_idempotent_only
&& self.retry_policy.max_retries > 0;
if !retries_unsafe_method || self.extra_headers.contains_key(IDEMPOTENCY_KEY) {
return;
}
let key = self.entropy.uuid_v4().to_string();
if let Ok(value) = HeaderValue::from_str(&key) {
self.extra_headers
.insert(HeaderName::from_static(IDEMPOTENCY_KEY), value);
}
}
fn ensure_criticality_header(&mut self) {
if self
.extra_headers
.contains_key(crate::admission::CRITICALITY_HEADER)
{
return;
}
let inbound = crate::admission::current_criticality()
.filter(|c| *c != crate::admission::Criticality::Default);
if let Some(c) = self.criticality.or(inbound) {
self.extra_headers.insert(
HeaderName::from_static(crate::admission::CRITICALITY_HEADER),
HeaderValue::from_static(c.as_str()),
);
}
}
fn begin_attempt(
&self,
gate: &RetryGate,
host: Option<&str>,
) -> Result<Option<ThrottleTicket>, ClientError> {
if gate.expired() {
return Err(ClientError::DeadlineExceeded);
}
let ticket = self.throttle_attempt(host)?;
gate.check()?;
Ok(ticket)
}
async fn send_plain_attempt(
&self,
gate: &RetryGate,
attempt: u32,
refilled: &mut std::collections::HashSet<String>,
) -> (
tracing::Span,
Result<reqwest::Response, reqwest::Error>,
Vec<String>,
) {
let span = client_attempt_span(&self.method, &self.url, attempt);
let req = self.plain_attempt(gate, &span);
let (sent, followed) =
tracing::Instrument::instrument(send_tracking_redirects(req), span.clone()).await;
record_send_outcome(&span, &sent);
gate.record_destinations(&followed, refilled);
(span, sent, followed)
}
fn plain_attempt(&self, gate: &RetryGate, span: &tracing::Span) -> reqwest::RequestBuilder {
let timeout = gate.attempt_timeout(self.retry_policy.request_timeout);
let mut req = self.client.request(self.method.clone(), &self.url);
if gate.deadline.is_some()
&& let Some(timeout) = timeout
{
req = req.timeout(timeout);
}
req = span.in_scope(|| inject_trace_context(req, &self.extra_headers));
req = with_caller_headers(req, gate, timeout, &self.extra_headers, &self.url);
if let Some(body) = &self.body {
req = req.body(body.clone());
}
req
}
fn throttle_attempt(&self, host: Option<&str>) -> Result<Option<ThrottleTicket>, ClientError> {
let Some(throttle) = &self.throttle else {
return Ok(None);
};
let Some(host) = host.map(str::to_owned).or_else(|| throttle_host(&self.url)) else {
return Ok(None);
};
admit_throttle(throttle, host, &*self.entropy).map(Some)
}
const fn max_attempts(&self, suppress_retries: bool) -> u32 {
if suppress_retries {
1
} else if is_idempotent_method(&self.method) || !self.retry_policy.retry_idempotent_only {
self.retry_policy.max_retries.saturating_add(1)
} else {
1
}
}
fn mock_response(&self, mock: &MockRegistry) -> Result<Response, ClientError> {
let Some(mock_resp) = mock.find_match(&self.method, &self.url, self.alias.as_deref())
else {
return Err(ClientError::NoMock(
self.method.to_string(),
self.url.clone(),
));
};
let status =
reqwest::StatusCode::from_u16(mock_resp.status).unwrap_or(reqwest::StatusCode::OK);
let body_bytes = mock_resp
.body
.as_ref()
.map(|v| serde_json::to_vec(v).unwrap_or_default())
.unwrap_or_default();
tracing::info!(
http.method = %self.method,
http.url = %self.url,
http.status = mock_resp.status,
"[mock] outbound request intercepted"
);
Ok(Response {
status,
headers: HeaderMap::new(),
body: Bytes::from(body_bytes),
url: None,
})
}
async fn send_sim(self, net: &crate::sim::SimNet) -> Result<Response, ClientError> {
use tracing::Instrument as _;
let url = self.sim_url()?;
let host = url
.host_str()
.ok_or_else(|| ClientError::InvalidUrl(format!("{}: no host", self.url)))?
.to_owned();
let max_attempts = self.max_attempts(false);
let gate = self.retry_gate(url_host(url.as_str()).as_deref());
let mut last_retry = None;
let mut delay = Duration::ZERO;
for attempt in 0..max_attempts {
if attempt > 0 {
tokio::time::sleep(delay).await;
}
let ticket = self.begin_attempt(&gate, Some(&host))?;
let last = attempt + 1 == max_attempts;
let span = client_attempt_span(&self.method, url.as_str(), attempt);
let timeout = gate.attempt_timeout(self.retry_policy.request_timeout);
let deadline_header = gate.header(timeout, &self.extra_headers);
let exchange = self
.sim_attempt(net, &host, &url, timeout, deadline_header)
.instrument(span.clone());
let exchange = async {
match gate.deadline {
Some(deadline) => crate::deadline::Bounded::until(deadline, exchange)
.await
.unwrap_or_else(|crate::deadline::DeadlineExceeded| {
Err(SimAttemptError::Transient(format!(
"request to {host} reached the request deadline"
)))
}),
None => exchange.await,
}
};
let outcome = match timeout {
Some(limit) => {
tokio::time::timeout(limit, exchange)
.await
.unwrap_or_else(|_elapsed| {
Err(SimAttemptError::Transient(format!(
"request to {host} timed out after {limit:?}"
)))
})
}
None => exchange.await,
};
match &outcome {
Ok(response) => record_attempt_status(&span, response.status.as_u16()),
Err(SimAttemptError::Transient(_)) => record_attempt_error(&span, "network"),
Err(SimAttemptError::Fatal(_)) => record_attempt_error(&span, "request"),
}
drop(span);
let deadline_stop =
matches!(outcome, Err(SimAttemptError::Transient(_))) && gate.expired();
if let Some(ticket) = ticket
&& !deadline_stop
{
ticket.record(matches!(&outcome, Ok(r) if throttle_accepts(r.status.as_u16())));
}
let response = match outcome {
Ok(response) => response,
Err(SimAttemptError::Transient(_)) if gate.expired() => {
return Err(ClientError::DeadlineExceeded);
}
Err(SimAttemptError::Transient(message)) if !last => {
let wait = self.retry_policy.retry_delay(&*self.entropy, attempt, None);
if !gate.allow(RetryKind::Transient, wait) {
return Err(ClientError::SimNetwork(message));
}
delay = wait;
last_retry = Some(RetryKind::Transient);
continue;
}
Err(SimAttemptError::Transient(message)) => {
return Err(ClientError::SimNetwork(message));
}
Err(SimAttemptError::Fatal(error)) => return Err(error),
};
let status = response.status.as_u16();
if is_retryable_response(status) && !last {
let wait = self.retry_policy.retry_delay(
&*self.entropy,
attempt,
retry_hint(status, &response.headers),
);
let kind = retry_kind(status);
if gate.allow(kind, wait) {
delay = wait;
last_retry = Some(kind);
continue;
}
}
gate.finish(last_retry, status);
return Ok(response);
}
unreachable!("the sim retry loop returns on its last attempt")
}
fn sim_url(&self) -> Result<reqwest::Url, ClientError> {
match (reqwest::Url::parse(&self.url), self.alias.as_deref()) {
(Ok(url), _) => Ok(url),
(Err(url::ParseError::RelativeUrlWithoutBase), Some(alias)) => {
let path = self.url.trim_start_matches('/');
reqwest::Url::parse(&format!("http://{alias}/{path}"))
.map_err(|error| ClientError::InvalidUrl(format!("{}: {error}", self.url)))
}
(Err(error), _) => Err(ClientError::InvalidUrl(format!("{}: {error}", self.url))),
}
}
async fn sim_attempt(
&self,
net: &crate::sim::SimNet,
host: &str,
url: &reqwest::Url,
timeout: Option<Duration>,
deadline_header: Option<HeaderValue>,
) -> Result<Response, SimAttemptError> {
net.transmit(host, timeout)
.await
.map_err(|fault| SimAttemptError::Transient(format!("request to {host} {fault}")))?;
match (net.service(host), self.mock.as_ref()) {
(Some(router), _) => serve_sim_host(router, self, url.clone(), deadline_header)
.await
.map_err(SimAttemptError::Fatal),
(None, Some(mock)) => self.mock_response(mock).map_err(SimAttemptError::Fatal),
(None, None) => Err(SimAttemptError::Fatal(ClientError::SimNetwork(format!(
"no sim host named {host}"
)))),
}
}
const fn needs_custom_path(&self) -> bool {
self.ssrf_safe
|| self.pin_addr.is_some()
|| !matches!(self.redirect_mode, RedirectMode::Default)
}
async fn send_custom(self, is_half_open: bool) -> Result<Response, ClientError> {
if self.ssrf_safe && self.pin_addr.is_some() {
return Err(ClientError::PinNotAllowedWithSsrfSafe(
"get_ssrf_safe cannot be combined with pin_to: the SSRF-safe path \
performs its own per-hop resolve/validate/pin and never reads the \
pin_to address, so an explicit pin would be silently ignored. Use \
pin_to alone for a caller-chosen address, or get_ssrf_safe alone \
for guarded automatic per-hop pinning.",
));
}
if self.pin_addr.is_some() && matches!(self.redirect_mode, RedirectMode::Follow { .. }) {
return Err(ClientError::IncompatiblePinRedirect(
"pin_to cannot be combined with follow_redirects: the pin only \
covers the first hop and later redirect hops re-resolve via DNS, \
escaping the pin. Use get_ssrf_safe for pinned, per-hop-revalidated \
redirect following; pin_to alone (which returns the 3xx unfollowed); \
or follow_redirects without pin_to.",
));
}
if self.pin_addr.as_ref().is_some_and(Vec::is_empty) {
return Err(ClientError::InvalidUrl(
"pin_to_addrs needs at least one address".to_owned(),
));
}
if self.pin_addr.is_some() && url_host_is_ip_literal(&self.url)? {
return Err(ClientError::PinRequiresDomainHost(
"pin_to cannot be honored for an IP-literal URL host because the \
HTTP stack connects to the literal directly and skips the pinned \
address; put the desired IP directly in the URL, or use a domain host.",
));
}
let gate = self.retry_gate(url_host(&self.url).as_deref());
if gate.expired() {
return Err(ClientError::DeadlineExceeded);
}
let timeout = gate
.attempt_timeout(self.retry_policy.request_timeout)
.unwrap_or_else(|| Duration::from_secs(30));
if self.ssrf_safe {
return self.send_ssrf_safe(timeout, is_half_open, &gate).await;
}
let follow = match &self.redirect_mode {
RedirectMode::Follow { max, validator } => Some((*max, validator.clone())),
RedirectMode::None | RedirectMode::Default => None,
};
if let Some((max, validator)) = follow {
return self
.follow_loop(
max,
validator,
HopClient::OneShot(timeout),
is_half_open,
&gate,
)
.await;
}
let policy = reqwest::redirect::Policy::none();
let resolve = self.pin_resolve()?;
let client = build_oneshot_client(resolve, policy, timeout)?;
send_one(
&client,
&self.method,
&self.url,
&self.extra_headers,
self.body.as_ref(),
&self.retry_policy,
&*self.entropy,
self.discard_response_body,
None,
is_half_open,
&gate,
false,
None,
None,
)
.await
}
fn pin_resolve(&self) -> Result<Option<(String, Vec<SocketAddr>)>, ClientError> {
match &self.pin_addr {
Some(addrs) => Ok(Some((host_of(&self.url)?, addrs.clone()))),
None => Ok(None),
}
}
async fn follow_loop(
self,
max: usize,
validator: RedirectValidator,
hop_client: HopClient<'_>,
is_half_open: bool,
gate: &RetryGate,
) -> Result<Response, ClientError> {
let original =
url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let mut current = self.url.clone();
let mut method = self.method.clone();
let mut headers = self.extra_headers.clone();
let mut body = self.body.clone();
let chain_retries = matches!(hop_client, HopClient::Pooled(_))
.then(|| ChainRetries::new(self.max_attempts(is_half_open).saturating_sub(1)));
let mut refilled: std::collections::HashSet<String> =
url_host(&self.url).into_iter().collect();
for hop in 0.. {
let resolve = if hop == 0 {
match &self.pin_addr {
Some(addrs) => Some((host_of(¤t)?, addrs.clone())),
None => None,
}
} else {
None
};
if hop > 0 {
strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
}
let client = match hop_client {
HopClient::Pooled(client) => client.clone(),
HopClient::OneShot(timeout) => {
build_oneshot_client(resolve, reqwest::redirect::Policy::none(), timeout)?
}
};
let hop_gate;
let hop_gate = if hop == 0 {
gate
} else {
hop_gate = gate.for_hop(¤t, &mut refilled);
&hop_gate
};
let resp = send_one(
&client,
&method,
¤t,
&headers,
body.as_ref(),
&self.retry_policy,
&*self.entropy,
self.discard_response_body,
None,
is_half_open,
hop_gate,
true,
chain_retries.as_ref(),
matches!(hop_client, HopClient::Pooled(_))
.then_some(self.throttle.as_ref())
.flatten(),
)
.await?;
let next = match redirect_target(&resp, ¤t) {
Ok(Some(next)) => next,
Ok(None) => return Ok(resp),
Err(ClientError::InvalidUrl(_)) if matches!(hop_client, HopClient::Pooled(_)) => {
return Ok(resp);
}
Err(error) => return Err(error),
};
if hop >= max {
return Err(ClientError::TooManyRedirects(max));
}
if !validator(&next) {
return Err(ClientError::RedirectRejected(next));
}
rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
if matches!(hop_client, HopClient::Pooled(_)) {
set_referer(&mut headers, &next, ¤t);
}
current = next;
}
unreachable!("redirect loop is bounded by `max` and always returns")
}
const fn ssrf_redirect_plan(&self) -> (bool, usize) {
match &self.redirect_mode {
RedirectMode::Default => (true, SSRF_SAFE_MAX_REDIRECTS),
RedirectMode::None => (false, 0),
RedirectMode::Follow { max, .. } => (true, *max),
}
}
async fn send_ssrf_safe(
self,
timeout: Duration,
is_half_open: bool,
gate: &RetryGate,
) -> Result<Response, ClientError> {
let deadline = crate::time::ambient_instant() + timeout;
let (follow, max) = self.ssrf_redirect_plan();
let original =
url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let mut current = self.url.clone();
let mut method = self.method.clone();
let mut headers = self.extra_headers.clone();
let mut body = self.body.clone();
let mut refilled: std::collections::HashSet<String> =
url_host(&self.url).into_iter().collect();
for hop in 0.. {
let remaining_for_lookup = deadline_remaining_or_timeout(deadline, ¤t)
.map_err(|error| gate.classify(error))?;
let addrs = tokio::time::timeout(remaining_for_lookup, resolve_and_validate(¤t))
.await
.map_err(|_| gate.classify(ssrf_safe_deadline_error(¤t)))??;
let host = host_of(¤t)?;
let remaining_for_connect = deadline_remaining_or_timeout(deadline, ¤t)
.map_err(|error| gate.classify(error))?;
let client = build_oneshot_client(
Some((host, addrs)),
reqwest::redirect::Policy::none(),
remaining_for_connect,
)?;
if hop > 0 {
strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
}
let hop_gate;
let hop_gate = if hop == 0 {
gate
} else {
hop_gate = gate.for_hop(¤t, &mut refilled);
&hop_gate
};
let resp = send_one(
&client,
&method,
¤t,
&headers,
body.as_ref(),
&self.retry_policy,
&*self.entropy,
self.discard_response_body,
Some(deadline),
is_half_open,
hop_gate,
false,
None,
None,
)
.await?;
if !follow {
return Ok(resp);
}
let Some(next) = redirect_target(&resp, ¤t)? else {
return Ok(resp);
};
if hop >= max {
return Err(ClientError::TooManyRedirects(max));
}
if scheme_is_https(¤t)? && !scheme_is_https(&next)? {
return Err(ClientError::RedirectRejected(format!(
"https→http scheme downgrade on redirect to {next}"
)));
}
if let RedirectMode::Follow { validator, .. } = &self.redirect_mode
&& !validator(&next)
{
return Err(ClientError::RedirectRejected(next));
}
rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
current = next;
}
unreachable!("redirect loop is bounded by the SSRF-safe redirect plan")
}
}
const fn is_idempotent_method(method: &Method) -> bool {
matches!(
*method,
Method::GET | Method::HEAD | Method::PUT | Method::DELETE | Method::OPTIONS | Method::TRACE
)
}
const fn is_retryable_status(status: u16) -> bool {
matches!(status, 502..=504)
}
fn deadline_remaining_or_timeout(
deadline: std::time::Instant,
current: &str,
) -> Result<Duration, ClientError> {
let now = crate::time::ambient_instant();
if now >= deadline {
return Err(ssrf_safe_deadline_error(current));
}
Ok(deadline - now)
}
fn ssrf_safe_deadline_error(current: &str) -> ClientError {
ClientError::InvalidUrl(format!(
"SSRF-safe resolve/redirect operation exceeded its deadline resolving {current}"
))
}
fn breaker_for_url(
resilience_config: Option<&Arc<crate::config::ResilienceConfig>>,
url: &str,
) -> crate::circuit_breaker::CircuitBreaker {
let host = url::Url::parse(url).ok().map_or_else(
|| "unknown".to_owned(),
|u| {
let h = u.host_str().unwrap_or("unknown");
u.port()
.map_or_else(|| h.to_owned(), |port| format!("{h}:{port}"))
},
);
resilience_config.map_or_else(
|| {
crate::circuit_breaker::global_registry().get_or_create(
&host,
crate::circuit_breaker::CircuitBreakerPolicy::default(),
)
},
|rc| {
let policy = crate::circuit_breaker::CircuitBreakerPolicy::from_config(rc, &host);
crate::circuit_breaker::global_registry().get_or_create_with_config(&host, policy)
},
)
}
const fn retry_kind(status: u16) -> RetryKind {
if status == 429 {
RetryKind::Throttling
} else {
RetryKind::Transient
}
}
fn url_host(url: &str) -> Option<String> {
let parsed = reqwest::Url::parse(url).ok()?;
let host = parsed.host_str()?;
Some(
parsed
.port_or_known_default()
.map_or_else(|| host.to_owned(), |port| format!("{host}:{port}")),
)
}
#[derive(Clone)]
struct RetrySettings {
budgets: Option<Arc<RetryBudgets>>,
send_deadline_header: bool,
}
impl RetrySettings {
fn standalone() -> Self {
Self {
budgets: Some(Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()))),
send_deadline_header: true,
}
}
}
const MIN_ATTEMPT: Duration = Duration::from_millis(10);
#[derive(Clone)]
pub(crate) struct SharedRetryBudgets {
budgets: Arc<RetryBudgets>,
config: RetryBudgetConfig,
}
static BUDGETS_REPLACE: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn shared_retry_budgets(
state: &crate::AppState,
config: &RetryBudgetConfig,
) -> Option<Arc<RetryBudgets>> {
if !config.enabled {
return None;
}
let current = || {
state
.extension::<SharedRetryBudgets>()
.filter(|shared| shared.config == *config)
.map(|shared| Arc::clone(&shared.budgets))
};
if let Some(budgets) = current() {
return Some(budgets);
}
let _guard = BUDGETS_REPLACE
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(budgets) = current() {
return Some(budgets);
}
let budgets = Arc::new(RetryBudgets::new(config));
state.insert_extension(SharedRetryBudgets {
budgets: Arc::clone(&budgets),
config: config.clone(),
});
Some(budgets)
}
struct RetryGate {
deadline: Option<Deadline>,
budgets: Option<Arc<RetryBudgets>>,
budget: Option<Arc<RetryBudget>>,
host: Option<String>,
send_header: bool,
refill_pending: AtomicBool,
pending_retry: std::sync::Mutex<Option<(Arc<RetryBudget>, RetryKind)>>,
caller_deadline: std::sync::OnceLock<Option<Deadline>>,
signer: Option<Arc<crate::aeo::web_bot_auth::WebBotAuthSigner>>,
}
impl Drop for RetryGate {
fn drop(&mut self) {
let pending = self
.pending_retry
.get_mut()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
if let Some((budget, kind)) = pending {
budget.release(kind);
}
}
}
impl RetryGate {
fn start(budgets: Option<Arc<RetryBudgets>>, host: Option<&str>, send_header: bool) -> Self {
Self::with_deadline(Deadline::current(), budgets, host, send_header)
}
fn with_deadline(
deadline: Option<Deadline>,
budgets: Option<Arc<RetryBudgets>>,
host: Option<&str>,
send_header: bool,
) -> Self {
let budget = budgets
.as_deref()
.zip(host)
.map(|(budgets, host)| budgets.for_host(host));
Self {
deadline,
budgets,
budget,
host: host.map(str::to_owned),
send_header,
refill_pending: AtomicBool::new(true),
pending_retry: std::sync::Mutex::new(None),
caller_deadline: std::sync::OnceLock::new(),
signer: None,
}
}
fn rekey(&mut self, url: &str) {
let host = url_host(url);
if host.is_some() && host != self.host {
self.budget = self
.budgets
.as_deref()
.zip(host.as_deref())
.map(|(budgets, host)| budgets.for_host(host));
self.host = host;
}
}
fn record_destination(&self, url: &str) {
let host = url_host(url);
if host.is_none() || host == self.host {
return;
}
if let Some((budgets, host)) = self.budgets.as_deref().zip(host) {
budgets.for_host(&host).record_request();
}
}
fn record_destinations(&self, urls: &[String], seen: &mut std::collections::HashSet<String>) {
for url in urls {
if url_host(url).is_some_and(|host| seen.insert(host)) {
self.record_destination(url);
}
}
}
fn for_hop(&self, url: &str, refilled: &mut std::collections::HashSet<String>) -> Self {
let mut hop = Self {
deadline: self.deadline,
budgets: self.budgets.clone(),
budget: self.budget.clone(),
host: self.host.clone(),
send_header: self.send_header,
refill_pending: AtomicBool::new(false),
pending_retry: std::sync::Mutex::new(None),
caller_deadline: self.caller_deadline.clone(),
signer: self.signer.clone(),
};
hop.rekey(url);
if url_host(url).is_some_and(|host| refilled.insert(host)) {
hop.refill_pending = AtomicBool::new(true);
}
hop
}
fn body_error(&self, error: reqwest::Error) -> ClientError {
if error.is_timeout() && self.expired() {
ClientError::DeadlineExceeded
} else {
ClientError::Request(error.without_url())
}
}
fn classify(&self, error: ClientError) -> ClientError {
if self.expired() {
ClientError::DeadlineExceeded
} else {
error
}
}
fn check(&self) -> Result<(), ClientError> {
if self.expired() {
return Err(ClientError::DeadlineExceeded);
}
self.pending_retry
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
if self.refill_pending.swap(false, Ordering::Relaxed)
&& let Some(budget) = &self.budget
{
budget.record_request();
}
Ok(())
}
fn attempt_timeout(&self, per_try: Option<Duration>) -> Option<Duration> {
match (per_try, self.deadline) {
(Some(limit), Some(deadline)) => Some(deadline.clamp(limit)),
(None, Some(deadline)) => Some(deadline.remaining()),
(limit, None) => limit,
}
}
fn expired(&self) -> bool {
self.deadline.is_some_and(Deadline::is_expired)
}
fn allow(&self, kind: RetryKind, wait: Duration) -> bool {
if self
.deadline
.is_some_and(|deadline| deadline.remaining() <= wait + MIN_ATTEMPT)
{
return false;
}
let Some(budget) = &self.budget else {
return true;
};
if !budget.try_acquire(kind) {
return false;
}
*self
.pending_retry
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some((Arc::clone(budget), kind));
true
}
fn finish(&self, last_retry: Option<RetryKind>, status: u16) {
if let (Some(budget), Some(kind)) = (&self.budget, last_retry)
&& status < 400
{
budget.release(kind);
}
}
fn header(&self, timeout: Option<Duration>, caller: &HeaderMap) -> Option<HeaderValue> {
let millis = |d: Duration| u64::try_from(d.as_millis()).unwrap_or(u64::MAX);
let left = self.deadline.and(timeout).map(millis);
let Some(value) = caller.get(DEADLINE_HEADER) else {
return left.filter(|_| self.send_header).map(HeaderValue::from);
};
let mut first = None;
let caller_deadline = self.caller_deadline.get_or_init(|| {
first = crate::deadline::parse_header(value);
first.map(Deadline::after)
});
let theirs = first.map(millis).or_else(|| {
caller_deadline
.as_ref()
.map(|deadline| millis(deadline.remaining()))
});
match (left, theirs) {
(Some(left), Some(theirs)) => Some(HeaderValue::from(left.min(theirs))),
(Some(value), None) | (None, Some(value)) => Some(HeaderValue::from(value)),
(None, None) => None,
}
}
}
fn with_caller_headers(
mut req: reqwest::RequestBuilder,
gate: &RetryGate,
timeout: Option<Duration>,
caller: &HeaderMap,
url: &str,
) -> reqwest::RequestBuilder {
let deadline_value = gate.header(timeout, caller);
let sends_deadline = deadline_value.is_some();
if let Some(value) = deadline_value {
req = req.header(DEADLINE_HEADER, value);
}
let fresh = gate
.signer
.as_ref()
.filter(|_| caller.contains_key("signature"))
.and_then(|signer| signer.sign_url(url));
for (name, value) in caller {
if sends_deadline && name == DEADLINE_HEADER {
continue;
}
if fresh.is_some()
&& matches!(
name.as_str(),
"signature" | "signature-input" | "signature-agent"
)
{
continue;
}
req = req.header(name.clone(), value.clone());
}
if let Some(h) = fresh {
req = req
.header("signature-agent", h.signature_agent)
.header("signature-input", h.signature_input)
.header("signature", h.signature);
}
req
}
enum SimAttemptError {
Transient(String),
Fatal(ClientError),
}
async fn serve_sim_host(
router: axum::Router,
request: &RequestBuilder,
url: reqwest::Url,
deadline_header: Option<HeaderValue>,
) -> Result<Response, ClientError> {
let target = url.query().map_or_else(
|| url.path().to_owned(),
|query| format!("{}?{query}", url.path()),
);
let mut builder = axum::http::Request::builder()
.method(request.method.clone())
.uri(target);
if !request.extra_headers.contains_key(reqwest::header::HOST) {
let authority = &url[url::Position::BeforeHost..url::Position::AfterPort];
builder = builder.header(reqwest::header::HOST, authority);
}
for (name, value) in trace_headers_to_send(trace_context_headers(), &request.extra_headers) {
builder = builder.header(name, value);
}
let sends_deadline = deadline_header.is_some();
if let Some(value) = deadline_header {
builder = builder.header(DEADLINE_HEADER, value);
}
if let Some(body) = &request.body
&& !request
.extra_headers
.contains_key(reqwest::header::CONTENT_LENGTH)
{
builder = builder.header(reqwest::header::CONTENT_LENGTH, body.len());
}
let fresh = request
.web_bot_auth
.as_ref()
.filter(|_| request.extra_headers.contains_key("signature"))
.and_then(|signer| signer.sign_url(url.as_str()));
for (name, value) in &request.extra_headers {
if sends_deadline && name == DEADLINE_HEADER {
continue;
}
if fresh.is_some()
&& matches!(
name.as_str(),
"signature" | "signature-input" | "signature-agent"
)
{
continue;
}
builder = builder.header(name, value);
}
if let Some(h) = fresh {
builder = builder
.header("signature-agent", h.signature_agent)
.header("signature-input", h.signature_input)
.header("signature", h.signature);
}
let body = request.body.clone().unwrap_or_default();
let http_request = builder
.body(axum::body::Body::from(body))
.map_err(|error| ClientError::SimNetwork(error.to_string()))?;
let response = crate::deadline::unscoped(tower::ServiceExt::oneshot(router, http_request))
.await
.unwrap_or_else(|never| match never {});
let status = response.status();
let headers = response.headers().clone();
let body = if request.discard_response_body {
Bytes::new()
} else {
axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.map_err(|error| ClientError::SimNetwork(error.to_string()))?
};
Ok(Response {
status,
headers,
body,
url: Some(url),
})
}
fn build_oneshot_client(
resolve: Option<(String, Vec<SocketAddr>)>,
policy: reqwest::redirect::Policy,
timeout: Duration,
) -> Result<reqwest::Client, ClientError> {
let mut builder = reqwest::ClientBuilder::new()
.timeout(timeout)
.redirect(policy);
if let Some((host, addrs)) = resolve
&& !addrs.is_empty()
{
builder = builder.no_proxy().resolve_to_addrs(&host, &addrs);
}
builder.build().map_err(ClientError::Request)
}
struct ChainRetries {
left: AtomicU32,
used: AtomicU32,
attempt_end: Mutex<Option<Instant>>,
}
impl ChainRetries {
const fn new(retries: u32) -> Self {
Self {
left: AtomicU32::new(retries),
used: AtomicU32::new(0),
attempt_end: Mutex::new(None),
}
}
fn take_attempt_end(&self) -> Option<Instant> {
self.attempt_end
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take()
}
fn set_attempt_end(&self, end: Option<Instant>) {
*self
.attempt_end
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = end;
}
}
struct ThrottleTicket {
throttle: Arc<crate::admission::AdaptiveThrottle>,
host: String,
recorded: bool,
}
impl ThrottleTicket {
fn record(mut self, accepted: bool) {
self.recorded = true;
self.throttle
.record(&self.host, crate::time::ambient_instant(), accepted);
}
}
impl Drop for ThrottleTicket {
fn drop(&mut self) {
if !self.recorded {
self.throttle
.record(&self.host, crate::time::ambient_instant(), true);
}
}
}
fn admit_throttle(
throttle: &Arc<crate::admission::AdaptiveThrottle>,
host: String,
entropy: &dyn crate::entropy::Entropy,
) -> Result<ThrottleTicket, ClientError> {
let now = crate::time::ambient_instant();
if throttle.admit(&host, now, || entropy.next_u64()) {
Ok(ThrottleTicket {
throttle: Arc::clone(throttle),
host,
recorded: false,
})
} else {
tracing::debug!(host = %host, "outbound request throttled locally");
Err(ClientError::ThrottledLocally { host })
}
}
fn record_call(ticket: Option<ThrottleTicket>, res: &Result<Response, ClientError>) {
if let Some(ticket) = ticket
&& !matches!(res, Err(ClientError::DeadlineExceeded))
{
ticket.record(matches!(res, Ok(r) if throttle_accepts(r.status.as_u16())));
}
}
const fn throttle_accepts(status: u16) -> bool {
!matches!(status, 429 | 503)
}
fn throttle_host(url: &str) -> Option<String> {
let parsed = url::Url::parse(url).ok()?;
let host = parsed.host_str()?;
Some(
parsed
.port()
.map_or_else(|| host.to_owned(), |port| format!("{host}:{port}")),
)
}
#[allow(
clippy::too_many_arguments,
reason = "one seam shared by three call sites (plain custom path, follow_loop, send_ssrf_safe); \
splitting the request-shape fields into their own struct would still leave the \
retry/discard/deadline/half-open knobs alongside it, for no reduction in what a \
caller reasons about"
)]
#[allow(
clippy::too_many_lines,
reason = "one retry loop; the deadline and budget checks (#3058) sit at each decision"
)]
async fn send_one(
client: &reqwest::Client,
method: &Method,
url: &str,
extra_headers: &HeaderMap,
body: Option<&Bytes>,
retry_policy: &RetryPolicy,
entropy: &dyn crate::entropy::Entropy,
discard_response_body: bool,
deadline: Option<Instant>,
suppress_retries: bool,
gate: &RetryGate,
skip_redirect_body: bool,
chain: Option<&ChainRetries>,
throttle: Option<&Arc<crate::admission::AdaptiveThrottle>>,
) -> Result<Response, ClientError> {
let start = crate::time::ambient_instant();
let mut last_retry = None;
let mut max_attempts = if suppress_retries {
1
} else if is_idempotent_method(method) || !retry_policy.retry_idempotent_only {
retry_policy.max_retries.saturating_add(1)
} else {
1
};
let retries_before = chain.map_or(0, |chain| chain.used.load(Ordering::Relaxed));
if let Some(chain) = chain {
max_attempts = max_attempts.min(chain.left.load(Ordering::Relaxed).saturating_add(1));
}
let mut last_transient_err: Option<reqwest::Error> = None;
let deadline_exceeded_err = |last_transient_err: &mut Option<reqwest::Error>| {
last_transient_err.take().map_or_else(
|| ssrf_safe_deadline_error(url),
|e| ClientError::Request(e.without_url()),
)
};
let reaches_hop_deadline = |wait: Duration| {
deadline.is_some_and(|d| {
crate::time::ambient_instant()
.checked_add(wait.saturating_add(MIN_ATTEMPT))
.is_none_or(|resume| resume >= d)
})
};
let mut delay = Duration::ZERO;
for attempt in 0..max_attempts {
let last = attempt + 1 == max_attempts;
if attempt > 0 {
if let Some(chain) = chain {
chain.left.fetch_sub(1, Ordering::Relaxed);
chain.used.fetch_add(1, Ordering::Relaxed);
}
if deadline.is_some_and(|d| crate::time::ambient_instant() >= d) {
return Err(gate.classify(deadline_exceeded_err(&mut last_transient_err)));
}
let mut sleep_for = delay;
if let Some(d) = deadline {
sleep_for =
sleep_for.min(d.saturating_duration_since(crate::time::ambient_instant()));
}
tokio::time::sleep(sleep_for).await;
if deadline.is_some_and(|d| crate::time::ambient_instant() >= d) {
return Err(gate.classify(deadline_exceeded_err(&mut last_transient_err)));
}
}
if gate.expired() {
return Err(ClientError::DeadlineExceeded);
}
let ticket = match throttle.zip(throttle_host(url)) {
Some((throttle, host)) => Some(admit_throttle(throttle, host, entropy)?),
None => None,
};
gate.check()?;
let mut req = client.request(method.clone(), url);
let attempt_start = crate::time::ambient_instant();
let hop_timeout = deadline.map(|d| d.saturating_duration_since(attempt_start));
let carried_end = if attempt == 0 {
chain.and_then(ChainRetries::take_attempt_end)
} else {
None
};
let per_try = match (retry_policy.request_timeout, carried_end) {
(Some(_), Some(end)) => Some(end.saturating_duration_since(attempt_start)),
(limit, _) => limit,
};
let attempt_end = carried_end.or_else(|| {
retry_policy
.request_timeout
.and_then(|limit| attempt_start.checked_add(limit))
});
let attempt_timeout = if gate.deadline.is_some() {
gate.attempt_timeout(hop_timeout.or(per_try))
} else if carried_end.is_some() {
hop_timeout.or(per_try)
} else {
hop_timeout
};
if let Some(timeout) = attempt_timeout {
req = req.timeout(timeout);
}
let span = client_attempt_span(method, url, attempt);
req = span.in_scope(|| inject_trace_context(req, extra_headers));
req = with_caller_headers(req, gate, attempt_timeout, extra_headers, url);
if let Some(body) = body {
req = req.body(body.clone());
}
match send_in_span(req, &span).await {
Ok(resp) => {
let status = resp.status();
let headers = resp.headers().clone();
let url_used = resp.url().clone();
if is_retryable_response(status.as_u16())
&& !last
&& deadline.is_none_or(|d| crate::time::ambient_instant() < d)
{
let hint = retry_hint(status.as_u16(), &headers);
let next = retry_policy.retry_delay(
entropy,
retries_before.saturating_add(attempt),
hint,
);
let kind = retry_kind(status.as_u16());
if !reaches_hop_deadline(next) && gate.allow(kind, next) {
if let Some(ticket) = ticket {
ticket.record(throttle_accepts(status.as_u16()));
}
delay = next;
last_retry = Some(kind);
continue;
}
}
let followed = skip_redirect_body
&& matches!(redirect_location(status, &headers, url), Ok(Some(_)));
if followed && let Some(chain) = chain {
chain.set_attempt_end(attempt_end);
}
let body = if discard_response_body || followed {
Ok(Bytes::new())
} else {
read_body_in_span(resp, &span, gate).await
};
if let Some(ticket) = ticket
&& !matches!(body, Err(ClientError::DeadlineExceeded))
{
ticket.record(body.is_ok() && throttle_accepts(status.as_u16()));
}
let body = body?;
gate.finish(last_retry, status.as_u16());
log_request(
method.as_str(),
&url_used,
status.as_u16(),
crate::time::ambient_instant().saturating_duration_since(start),
extra_headers,
);
return Ok(Response {
status,
headers,
body,
url: Some(url_used),
});
}
Err(e) if e.is_timeout() && gate.expired() => {
return Err(ClientError::DeadlineExceeded);
}
Err(e) if (e.is_connect() || e.is_timeout()) && !last => {
if let Some(ticket) = ticket {
ticket.record(false);
}
let wait =
retry_policy.retry_delay(entropy, retries_before.saturating_add(attempt), None);
if reaches_hop_deadline(wait) || !gate.allow(RetryKind::Transient, wait) {
return Err(ClientError::Request(e.without_url()));
}
delay = wait;
last_retry = Some(RetryKind::Transient);
last_transient_err = Some(e);
}
Err(e) => {
if let Some(ticket) = ticket {
ticket.record(false);
}
return Err(ClientError::Request(e.without_url()));
}
}
}
unreachable!("retry loop exited without returning a result — this is a bug")
}
fn host_of(url: &str) -> Result<String, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
parsed
.host_str()
.map(str::to_owned)
.ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))
}
fn url_host_is_ip_literal(url: &str) -> Result<bool, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
match parsed.host() {
Some(url::Host::Ipv4(_) | url::Host::Ipv6(_)) => Ok(true),
Some(url::Host::Domain(_)) => Ok(false),
None => Err(ClientError::InvalidUrl(format!("URL has no host: {url}"))),
}
}
fn scheme_is_https(url: &str) -> Result<bool, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
Ok(parsed.scheme().eq_ignore_ascii_case("https"))
}
fn redirect_target(resp: &Response, base: &str) -> Result<Option<String>, ClientError> {
redirect_location(resp.status(), resp.headers(), base)
}
fn redirect_location(
status: reqwest::StatusCode,
headers: &HeaderMap,
base: &str,
) -> Result<Option<String>, ClientError> {
match status {
reqwest::StatusCode::MOVED_PERMANENTLY
| reqwest::StatusCode::FOUND
| reqwest::StatusCode::SEE_OTHER
| reqwest::StatusCode::TEMPORARY_REDIRECT
| reqwest::StatusCode::PERMANENT_REDIRECT => {}
_ => return Ok(None),
}
let Some(location) = headers
.get(reqwest::header::LOCATION)
.and_then(|v| v.to_str().ok())
else {
return Ok(None);
};
let base_url = url::Url::parse(base).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let joined = base_url
.join(location)
.map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
Ok(Some(joined.to_string()))
}
fn strip_sensitive_headers_if_cross_origin(
headers: &mut HeaderMap,
original: &url::Url,
current: &str,
) -> Result<(), ClientError> {
let current_url =
url::Url::parse(current).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
if current_url.origin() != original.origin() {
headers.remove(reqwest::header::AUTHORIZATION);
headers.remove(reqwest::header::COOKIE);
headers.remove("cookie2");
headers.remove(reqwest::header::PROXY_AUTHORIZATION);
headers.remove(reqwest::header::WWW_AUTHENTICATE);
headers.remove("signature");
headers.remove("signature-input");
headers.remove("signature-agent");
}
Ok(())
}
fn rewrite_after_redirect(
status: reqwest::StatusCode,
method: &mut Method,
body: &mut Option<Bytes>,
headers: &mut HeaderMap,
) {
match status.as_u16() {
303 => {
if *method != Method::HEAD {
*method = Method::GET;
}
*body = None;
strip_payload_headers(headers);
}
301 | 302 if *method == Method::POST => {
*method = Method::GET;
*body = None;
strip_payload_headers(headers);
}
_ => {}
}
}
fn strip_payload_headers(headers: &mut HeaderMap) {
use reqwest::header::{
CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH, CONTENT_TYPE, TRANSFER_ENCODING,
};
headers.remove(CONTENT_TYPE);
headers.remove(CONTENT_LENGTH);
headers.remove(TRANSFER_ENCODING);
headers.remove(CONTENT_ENCODING);
headers.remove(CONTENT_LANGUAGE);
}
fn validate_resolved_addrs(addrs: Vec<SocketAddr>) -> Result<Vec<SocketAddr>, ClientError> {
for addr in &addrs {
if is_blocked_ip(addr.ip()) {
return Err(ClientError::SsrfBlocked(addr.ip().to_string()));
}
}
Ok(addrs)
}
async fn resolve_and_validate(url: &str) -> Result<Vec<SocketAddr>, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let scheme = parsed.scheme();
if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
return Err(ClientError::InvalidUrl(format!(
"unsupported URL scheme `{scheme}` (only http/https are allowed): {url}"
)));
}
let port = parsed.port_or_known_default().ok_or_else(|| {
ClientError::InvalidUrl(format!("URL has no port and unknown scheme: {url}"))
})?;
let host = parsed
.host()
.ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))?;
match host {
url::Host::Ipv4(v4) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V4(v4), port)]),
url::Host::Ipv6(v6) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V6(v6), port)]),
url::Host::Domain(name) => {
let addrs: Vec<SocketAddr> = tokio::net::lookup_host((name, port))
.await
.map_err(|e| ClientError::InvalidUrl(format!("DNS lookup failed for {name}: {e}")))?
.collect();
if addrs.is_empty() {
return Err(ClientError::InvalidUrl(format!(
"DNS lookup for {name} returned no addresses"
)));
}
validate_resolved_addrs(addrs)
}
}
}
fn parse_retry_after(headers: &HeaderMap) -> Option<Duration> {
let value = headers.get("retry-after")?.to_str().ok()?;
if let Ok(secs) = value.parse::<u64>() {
return Some(Duration::from_secs(secs));
}
let dt = chrono::DateTime::parse_from_rfc2822(value).ok()?;
let now = crate::time::ambient_now();
let future = dt.with_timezone(&chrono::Utc);
let secs = u64::try_from((future - now).num_seconds().max(0)).unwrap_or(0);
Some(Duration::from_secs(secs))
}
const REDACTED_HEADERS: &[&str] = &["authorization", "cookie", "set-cookie"];
fn is_sensitive_header(name: &str) -> bool {
REDACTED_HEADERS
.iter()
.any(|h| h.eq_ignore_ascii_case(name))
}
fn log_request(
method: &str,
url: &reqwest::Url,
status: u16,
elapsed: Duration,
headers: &HeaderMap,
) {
let host = url.host_str().unwrap_or("unknown");
let path = url.path();
let sent_headers: Vec<&str> = headers
.keys()
.map(HeaderName::as_str)
.filter(|k| !is_sensitive_header(k))
.collect();
tracing::info!(
http.method = method,
http.host = host,
http.path = path,
http.status = status,
http.elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
http.sent_headers = ?sent_headers,
"outbound request"
);
}
fn client_attempt_span(method: &Method, url: &str, attempt: u32) -> tracing::Span {
let span = tracing::info_span!(
"http.client.request",
otel.name = %method,
otel.kind = "client",
http.request.method = %method,
server.address = tracing::field::Empty,
url.path = tracing::field::Empty,
http.request.resend_count = tracing::field::Empty,
http.response.status_code = tracing::field::Empty,
error.type = tracing::field::Empty,
otel.status_code = tracing::field::Empty,
);
if attempt > 0 {
span.record("http.request.resend_count", attempt);
}
if !span.is_disabled()
&& let Ok(parsed) = reqwest::Url::parse(url)
{
span.record("server.address", parsed.host_str().unwrap_or_default());
span.record("url.path", parsed.path());
}
span
}
fn record_attempt_status(span: &tracing::Span, status: u16) {
span.record("http.response.status_code", status);
if status >= 400 {
span.record("otel.status_code", "ERROR");
}
}
fn record_attempt_error(span: &tracing::Span, kind: &'static str) {
span.record("error.type", kind);
span.record("otel.status_code", "ERROR");
}
async fn read_body_in_span(
response: reqwest::Response,
span: &tracing::Span,
gate: &RetryGate,
) -> Result<Bytes, ClientError> {
use tracing::Instrument as _;
let body = response.bytes().instrument(span.clone()).await;
if body.is_err() {
record_attempt_error(span, "body");
}
body.map_err(|e| gate.body_error(e))
}
async fn send_in_span(
request: reqwest::RequestBuilder,
span: &tracing::Span,
) -> Result<reqwest::Response, reqwest::Error> {
use tracing::Instrument as _;
let sent = request.send().instrument(span.clone()).await;
record_send_outcome(span, &sent);
sent
}
fn record_send_outcome(span: &tracing::Span, sent: &Result<reqwest::Response, reqwest::Error>) {
match sent {
Ok(response) => record_attempt_status(span, response.status().as_u16()),
Err(error) => {
let kind = if error.is_timeout() {
"timeout"
} else if error.is_connect() {
"connect"
} else {
"request"
};
record_attempt_error(span, kind);
}
}
}
fn inject_trace_context(
builder: reqwest::RequestBuilder,
caller_headers: &HeaderMap,
) -> reqwest::RequestBuilder {
let mut builder = builder;
for (name, value) in trace_headers_to_send(trace_context_headers(), caller_headers) {
builder = builder.header(name, value);
}
builder
}
fn trace_headers_to_send(
injected: Vec<(String, HeaderValue)>,
caller_headers: &HeaderMap,
) -> Vec<(String, HeaderValue)> {
const W3C_PAIR: [&str; 2] = ["traceparent", "tracestate"];
let caller_has_context = W3C_PAIR
.iter()
.any(|name| caller_headers.contains_key(*name));
injected
.into_iter()
.filter(|(name, _)| {
let paired = caller_has_context && W3C_PAIR.contains(&name.as_str());
!paired && !caller_headers.contains_key(name.as_str())
})
.collect()
}
fn add_current_request_id(headers: &mut HeaderMap) {
static X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id");
if headers.contains_key(&X_REQUEST_ID) {
return;
}
if let Some(value) =
crate::log::context::current_request_id().and_then(|id| HeaderValue::from_str(&id).ok())
{
headers.insert(X_REQUEST_ID.clone(), value);
}
}
#[allow(clippy::missing_const_for_fn)]
fn trace_context_headers() -> Vec<(String, HeaderValue)> {
#[cfg(not(feature = "telemetry-otlp"))]
{
Vec::new()
}
#[cfg(feature = "telemetry-otlp")]
{
use std::collections::HashMap;
use tracing_opentelemetry::OpenTelemetrySpanExt as _;
let cx = tracing::Span::current().context();
let mut map = HashMap::<String, String>::new();
opentelemetry::global::get_text_map_propagator(|propagator| {
propagator.inject_context(&cx, &mut TraceHeaderInjector(&mut map));
});
map.into_iter()
.filter_map(|(name, value)| Some((name, HeaderValue::from_str(&value).ok()?)))
.collect()
}
}
#[cfg(feature = "telemetry-otlp")]
struct TraceHeaderInjector<'a>(&'a mut std::collections::HashMap<String, String>);
#[cfg(feature = "telemetry-otlp")]
impl opentelemetry::propagation::Injector for TraceHeaderInjector<'_> {
fn set(&mut self, key: &str, value: String) {
self.0.insert(key.to_owned(), value);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(feature = "reporting")]
#[test]
fn throttled_locally_round_trips_through_a_capsule() {
let err = ClientError::ThrottledLocally {
host: "api.example.com:8443".to_owned(),
};
let kind = http_error_kind(&err);
assert_eq!(
kind,
crate::capsule::schema::HttpErrorKind::ThrottledLocally
);
match rebuild_client_error(Some(kind), err.to_string()) {
ClientError::ThrottledLocally { host } => assert_eq!(host, "api.example.com:8443"),
other => panic!("rebuilt as {other:?}"),
}
}
#[cfg(feature = "reporting")]
#[test]
fn opaque_header_values_record_without_collision() {
let recorded = |value: HeaderValue| {
let mut builder = Client::new().get("http://example.test/");
builder.extra_headers.insert("x-sig", value);
caller_headers(&builder).remove(0).1
};
let first = recorded(HeaderValue::from_bytes(b"a\xfe").expect("value"));
let second = recorded(HeaderValue::from_bytes(b"a\xff").expect("value"));
assert_ne!(first, second);
assert_eq!(recorded(HeaderValue::from_static("plain")), "plain");
let look_alike = recorded(HeaderValue::from_static("[bytes] a\\xfe"));
assert_ne!(look_alike, first);
}
#[cfg(feature = "reporting")]
#[test]
fn fault_injected_round_trips_through_a_capsule() {
let err = ClientError::FaultInjected("fault injection: injected http error".to_owned());
let kind = http_error_kind(&err);
assert_eq!(kind, crate::capsule::schema::HttpErrorKind::FaultInjected);
match rebuild_client_error(Some(kind), err.to_string()) {
ClientError::FaultInjected(text) => {
assert_eq!(text, "fault injection: injected http error");
}
other => panic!("rebuilt as {other:?}"),
}
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn throttled_calls_do_not_open_the_circuit_breaker() {
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = axum::Router::new().route(
"/x",
axum::routing::get(|| async { axum::http::StatusCode::TOO_MANY_REQUESTS }),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let mut config = HttpClientConfig::default();
config.adaptive_throttle.enabled = true;
let client = Client::from_config(&config);
let url = format!("http://{addr}/x");
let mut throttled = 0;
for _ in 0..200 {
match client.get(&url).no_retry().send().await {
Err(ClientError::ThrottledLocally { .. }) => throttled += 1,
Ok(r) => assert_eq!(r.status().as_u16(), 429),
Err(e) => panic!("unexpected error: {e}"),
}
}
assert!(
throttled > 50,
"the throttle must reject locally: {throttled}"
);
let breaker = crate::circuit_breaker::global_registry().get_or_create(
&addr.to_string(),
crate::circuit_breaker::CircuitBreakerPolicy::default(),
);
assert_eq!(
breaker.state(),
crate::circuit_breaker::CircuitState::Closed,
"local rejects must not open the breaker"
);
crate::circuit_breaker::global_registry().clear();
}
#[test]
fn from_state_shares_a_throttle_built_from_the_effective_config() {
let state = crate::AppState::for_test();
let boot = HttpClientConfig::default();
install_shared_throttle(&state, &boot);
assert!(state.extension::<SharedThrottle>().is_none(), "off at boot");
assert!(Client::from_state(&state).throttle.is_none());
let mut replaced = crate::config::HttpConfig::default();
replaced.client.adaptive_throttle.enabled = true;
state.insert_extension(replaced.clone());
let a = Client::from_state(&state)
.throttle
.expect("on after replace");
let b = Client::from_state(&state)
.throttle
.expect("on after replace");
assert!(Arc::ptr_eq(&a, &b), "clients share one throttle");
replaced.client.adaptive_throttle.k = 3.0;
state.insert_extension(replaced.clone());
let c = Client::from_state(&state).throttle.expect("still on");
assert!(!Arc::ptr_eq(&a, &c), "new settings build a new throttle");
replaced.client.adaptive_throttle.enabled = false;
state.insert_extension(replaced);
assert!(Client::from_state(&state).throttle.is_none(), "off again");
}
#[test]
fn from_state_rebuilds_the_retry_budgets_when_their_config_changes() {
let state = crate::AppState::for_test();
let mut config = crate::config::HttpConfig::default();
state.insert_extension(config.clone());
let a = Client::from_state(&state).retry.budgets.expect("on");
let b = Client::from_state(&state).retry.budgets.expect("on");
assert!(Arc::ptr_eq(&a, &b), "clients share one set of budgets");
config.client.retry_budget.capacity = 50;
state.insert_extension(config.clone());
let c = Client::from_state(&state).retry.budgets.expect("still on");
assert!(!Arc::ptr_eq(&a, &c), "new settings build new budgets");
assert!((c.for_host("h:80").available() - 50.0).abs() < f64::EPSILON);
config.client.retry_budget.enabled = false;
state.insert_extension(config);
assert!(Client::from_state(&state).retry.budgets.is_none(), "off");
}
#[test]
fn concurrent_first_calls_share_one_throttle() {
let state = crate::AppState::for_test();
let mut config = crate::config::HttpConfig::default();
config.client.adaptive_throttle.enabled = true;
state.insert_extension(config);
let throttles: Vec<_> = std::thread::scope(|scope| {
let mut handles = Vec::with_capacity(8);
for _ in 0..8 {
handles.push(scope.spawn(|| Client::from_state(&state).throttle.expect("on")));
}
handles
.into_iter()
.map(|h| h.join().expect("thread"))
.collect()
});
assert!(
throttles.iter().all(|t| Arc::ptr_eq(t, &throttles[0])),
"every client must share the one throttle"
);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn a_truncated_body_is_not_an_accept() {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let Ok((mut socket, _)) = listener.accept().await else {
return;
};
let mut buf = [0_u8; 1024];
let _ = socket.read(&mut buf).await;
let _ = socket
.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 100\r\n\r\nabc")
.await;
drop(socket);
}
});
let mut config = HttpClientConfig::default();
config.adaptive_throttle.enabled = true;
let client = Client::from_config(&config);
let url = format!("http://{addr}/x");
for _ in 0..20 {
let res = client.get(&url).no_retry().send().await;
assert!(
matches!(
res,
Err(ClientError::Request(_) | ClientError::ThrottledLocally { .. })
),
"a truncated body is an error: {res:?}"
);
}
let throttle = client.throttle.as_ref().expect("on");
assert!(
throttle.reject_probability(&addr.to_string(), crate::time::ambient_instant()) > 0.5,
"truncated bodies must count as rejects"
);
crate::circuit_breaker::global_registry().clear();
}
#[test]
fn throttle_host_keeps_an_explicit_port() {
assert_eq!(
throttle_host("https://a.example/x").as_deref(),
Some("a.example")
);
assert_eq!(
throttle_host("http://a.example:8080/x").as_deref(),
Some("a.example:8080")
);
assert_eq!(throttle_host("not a url"), None);
assert!(throttle_accepts(200) && throttle_accepts(500));
assert!(!throttle_accepts(429) && !throttle_accepts(503));
}
use crate::config::HttpClientConfig;
#[test]
fn client_constructs_with_defaults() {
let client = Client::new();
assert!(client.alias.is_none());
assert!(client.base_url.is_none());
assert_eq!(client.retry_policy.max_retries, 3);
}
#[test]
fn request_builder_fluent_api_compiles() {
let client = Client::new();
let _builder = client
.post("https://example.com/api")
.header("x-api-key", "secret")
.json(&serde_json::json!({"key": "value"}))
.retries(2);
}
#[test]
fn response_accessors_work() {
let payload = serde_json::json!({"id": 42, "name": "Alice"});
let body = serde_json::to_vec(&payload).unwrap();
let resp = Response {
status: reqwest::StatusCode::OK,
headers: HeaderMap::new(),
body: Bytes::from(body),
url: None,
};
assert_eq!(resp.status().as_u16(), 200);
assert!(resp.is_success());
}
#[test]
fn response_json_deserialises() {
#[derive(serde::Deserialize, PartialEq, Debug)]
struct User {
id: i32,
name: String,
}
let payload = serde_json::json!({"id": 1, "name": "Bob"});
let resp = Response {
status: reqwest::StatusCode::OK,
headers: HeaderMap::new(),
body: Bytes::from(serde_json::to_vec(&payload).unwrap()),
url: None,
};
let user: User = resp.json().unwrap();
assert_eq!(user.id, 1);
assert_eq!(user.name, "Bob");
}
#[test]
fn response_text_returns_string() {
let resp = Response {
status: reqwest::StatusCode::OK,
headers: HeaderMap::new(),
body: Bytes::from_static(b"hello world"),
url: None,
};
assert_eq!(resp.text(), "hello world");
}
#[test]
fn response_bytes_returns_raw() {
let resp = Response {
status: reqwest::StatusCode::CREATED,
headers: HeaderMap::new(),
body: Bytes::from_static(b"\x00\x01\x02"),
url: None,
};
assert_eq!(resp.bytes(), Bytes::from_static(b"\x00\x01\x02"));
}
#[test]
fn config_deserialises_from_toml() {
let toml = r#"
[client]
timeout_secs = 60
max_retries = 5
[client.base_urls]
stripe = "https://api.stripe.com"
sendgrid = "https://api.sendgrid.com"
"#;
let http_cfg: crate::config::HttpConfig = toml::from_str(toml).unwrap();
let config = &http_cfg.client;
assert_eq!(config.timeout_secs, 60);
assert_eq!(config.max_retries, 5);
assert_eq!(
config.base_urls.get("stripe").map(String::as_str),
Some("https://api.stripe.com")
);
assert_eq!(
config.base_urls.get("sendgrid").map(String::as_str),
Some("https://api.sendgrid.com")
);
}
#[test]
fn config_has_correct_defaults() {
let config = HttpClientConfig::default();
assert_eq!(config.timeout_secs, 30);
assert_eq!(config.max_retries, 3);
assert!(config.base_urls.is_empty());
}
#[test]
fn idempotent_method_classification() {
assert!(is_idempotent_method(&Method::GET));
assert!(is_idempotent_method(&Method::HEAD));
assert!(is_idempotent_method(&Method::PUT));
assert!(is_idempotent_method(&Method::DELETE));
assert!(is_idempotent_method(&Method::OPTIONS));
assert!(is_idempotent_method(&Method::TRACE));
assert!(!is_idempotent_method(&Method::POST));
assert!(!is_idempotent_method(&Method::PATCH));
}
#[test]
fn retryable_status_classification() {
assert!(is_retryable_status(502));
assert!(is_retryable_status(503));
assert!(is_retryable_status(504));
assert!(!is_retryable_status(200));
assert!(!is_retryable_status(400));
assert!(!is_retryable_status(404));
assert!(!is_retryable_status(500));
assert!(!is_retryable_status(429));
}
#[test]
fn retry_after_header_parsing() {
let mut headers = HeaderMap::new();
headers.insert(
reqwest::header::HeaderName::from_static("retry-after"),
HeaderValue::from_static("5"),
);
assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(5)));
let empty = HeaderMap::new();
assert_eq!(parse_retry_after(&empty), None);
}
#[test]
fn sensitive_header_detection() {
assert!(is_sensitive_header("authorization"));
assert!(is_sensitive_header("Authorization"));
assert!(is_sensitive_header("AUTHORIZATION"));
assert!(is_sensitive_header("cookie"));
assert!(is_sensitive_header("set-cookie"));
assert!(!is_sensitive_header("content-type"));
assert!(!is_sensitive_header("x-api-key"));
}
#[tokio::test]
async fn mock_registry_captures_calls() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/charges".to_owned(),
alias: Some("stripe".to_owned()),
status: 200,
body: Some(serde_json::json!({"id": "ch_123"})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry).named("stripe");
let resp = client
.post("https://api.stripe.com/charges")
.json(&serde_json::json!({"amount": 1000}))
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let body: serde_json::Value = resp.json().unwrap();
assert_eq!(body["id"], "ch_123");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn mock_handle_expect_called_passes() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::GET),
path: "/users/1".to_owned(),
alias: None,
status: 200,
body: Some(serde_json::json!({"name": "Alice"})),
call_count: call_count.clone(),
});
let handle = MockHandle {
alias: "test".to_owned(),
method: "GET".to_owned(),
path: "/users/1".to_owned(),
call_count: call_count.clone(),
};
let client = Client::new().with_mock(registry);
client
.get("https://api.example.com/users/1")
.send()
.await
.unwrap();
handle.expect_called(1);
assert_eq!(handle.call_count(), 1);
}
#[tokio::test]
async fn mock_matches_by_path_suffix() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/v1/charges".to_owned(),
alias: None,
status: 201,
body: Some(serde_json::json!({"created": true})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.post("https://api.stripe.com/v1/charges")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 201);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn no_mock_error_when_unmatched() {
let registry = Arc::new(MockRegistry::new());
let client = Client::new().with_mock(registry);
let result = client.post("https://api.example.com/unknown").send().await;
assert!(matches!(result, Err(ClientError::NoMock(_, _))));
}
#[tokio::test]
async fn mock_setup_builder_registers_entry() {
let registry = Arc::new(MockRegistry::new());
let builder = MockSetupBuilder {
registry: registry.clone(),
alias: "myservice".to_owned(),
method: None,
path: None,
};
let handle = builder
.post("/api/resource")
.respond_with(201, serde_json::json!({"ok": true}));
let client = Client::new().with_mock(registry).named("myservice");
client
.post("https://myservice.example.com/api/resource")
.send()
.await
.unwrap();
handle.expect_called(1);
}
#[test]
fn client_from_config() {
let config = HttpClientConfig {
timeout_secs: 10,
max_retries: 1,
max_retry_after_secs: 10,
max_backoff_ms: 20_000,
base_urls: std::collections::HashMap::new(),
..HttpClientConfig::default()
};
let client = Client::from_config(&config);
assert_eq!(client.retry_policy.max_retries, 1);
}
#[test]
fn named_client_preserves_mock_registry() {
let registry = Arc::new(MockRegistry::new());
let client = Client::new().with_mock(registry);
let named = client.named("stripe");
assert!(named.mock.is_some());
assert_eq!(named.alias.as_deref(), Some("stripe"));
}
#[test]
fn base_url_prepended_to_relative_path() {
let client = Client::new();
let client = client.with_base_url("https://api.stripe.com");
let builder = client.post("/v1/charges");
assert_eq!(builder.url, "https://api.stripe.com/v1/charges");
}
#[test]
fn absolute_url_bypasses_base_url() {
let client = Client::new().with_base_url("https://ignored.example.com");
let builder = client.get("https://actual.example.com/path");
assert_eq!(builder.url, "https://actual.example.com/path");
}
#[test]
fn retry_override_per_request() {
let client = Client::new(); let builder = client.get("https://example.com").retries(0);
assert_eq!(builder.retry_policy.max_retries, 0);
let no_retry = client.get("https://example.com").no_retry();
assert_eq!(no_retry.retry_policy.max_retries, 0);
}
#[tokio::test]
async fn client_extracts_from_state() {
use axum::extract::FromRequestParts;
let state = crate::AppState::for_test();
let mut parts = axum::http::Request::new(axum::body::Body::empty())
.into_parts()
.0;
let client = Client::from_request_parts(&mut parts, &state)
.await
.unwrap();
assert!(client.mock.is_none());
assert!(client.alias.is_none());
}
#[test]
fn mock_registry_ext_round_trips_through_state() {
let registry = Arc::new(MockRegistry::new());
let ext = HttpMockRegistryExt(registry);
let state = crate::AppState::for_test();
state.insert_extension(ext);
let retrieved = state.extension::<HttpMockRegistryExt>();
assert!(retrieved.is_some());
}
#[test]
fn named_client_resolves_base_url_from_config() {
let mut base_urls = std::collections::HashMap::new();
base_urls.insert("stripe".to_owned(), "https://api.stripe.com".to_owned());
let config = HttpClientConfig {
timeout_secs: 30,
max_retries: 3,
max_retry_after_secs: 10,
max_backoff_ms: 20_000,
base_urls,
..HttpClientConfig::default()
};
let client = Client::from_config(&config);
let stripe = client.named("stripe");
assert_eq!(stripe.base_url.as_deref(), Some("https://api.stripe.com"));
assert_eq!(stripe.alias.as_deref(), Some("stripe"));
let other = client.named("sendgrid");
assert!(other.base_url.is_none());
}
#[test]
fn breaker_scoped_keys_by_resolved_url_not_alias() {
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut base_urls = std::collections::HashMap::new();
base_urls.insert(
"hook-service".to_owned(),
"http://mock-receiver/base".to_owned(),
);
let config = HttpClientConfig {
timeout_secs: 30,
max_retries: 3,
max_retry_after_secs: 10,
max_backoff_ms: 20_000,
base_urls,
..HttpClientConfig::default()
};
let client = Client::from_config(&config);
let req = client
.named("hook-service")
.post("hook-service")
.ssrf_safe()
.breaker_scoped();
assert_eq!(req.url, "http://mock-receiver/base/hook-service");
assert!(req.breaker_scoped);
assert!(url::Url::parse("hook-service").is_err());
let breaker = super::breaker_for_url(None, &req.url);
assert_eq!(breaker.name(), "mock-receiver");
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn breaker_scoped_custom_path_trips_and_fails_fast() {
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let mut rc = crate::config::ResilienceConfig::default();
rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
rc.circuit_breaker.defaults.open_duration_secs = Some(10);
let client = Client {
resilience_config: Some(Arc::new(rc)),
..Client::new()
};
let url = "http://127.0.0.1:1/blocked";
for _ in 0..3 {
let res = client.post(url).ssrf_safe().breaker_scoped().send().await;
assert!(
matches!(res, Err(ClientError::SsrfBlocked(_))),
"expected SsrfBlocked, got {res:?}"
);
}
let res = client.post(url).ssrf_safe().breaker_scoped().send().await;
assert!(matches!(res, Err(ClientError::CircuitBreakerOpen)));
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn open_breaker_calls_are_not_charged_to_the_throttle() {
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let mut rc = crate::config::ResilienceConfig::default();
rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
rc.circuit_breaker.defaults.open_duration_secs = Some(10);
let rc = Arc::new(rc);
let url = "http://127.0.0.1:1/blocked";
let plain = Client {
resilience_config: Some(Arc::clone(&rc)),
..Client::new()
};
for _ in 0..3 {
let _ = plain.post(url).ssrf_safe().breaker_scoped().send().await;
}
let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
2.0,
Duration::from_secs(120),
));
let throttled = Client {
resilience_config: Some(rc),
throttle: Some(Arc::clone(&throttle)),
..Client::new()
};
for _ in 0..50 {
let res = throttled
.post(url)
.ssrf_safe()
.breaker_scoped()
.send()
.await;
assert!(
matches!(res, Err(ClientError::CircuitBreakerOpen)),
"an open breaker fails fast, not through the throttle: {res:?}"
);
}
assert!(
throttle.reject_probability("127.0.0.1:1", crate::time::ambient_instant())
< f64::EPSILON,
"fail-fast calls must not be counted"
);
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn breaker_scoped_bypassed_for_mocked_client() {
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let registry = Arc::new(MockRegistry::new());
let mock = MockSetupBuilder {
registry: registry.clone(),
alias: "http://mock-receiver/hook".to_owned(),
method: None,
path: None,
}
.post("/hook")
.respond_with(500, serde_json::json!({ "error": "down" }));
let client = Client::new().with_mock(registry);
for _ in 0..12 {
let res = client
.named("http://mock-receiver/hook")
.post("http://mock-receiver/hook")
.ssrf_safe()
.breaker_scoped()
.send()
.await;
let res = res.expect("a mocked client must never see CircuitBreakerOpen");
assert_eq!(res.status().as_u16(), 500);
}
mock.expect_called(12);
assert!(
crate::circuit_breaker::global_registry()
.all_breakers()
.iter()
.all(|b| b.name() != "mock-receiver"),
"a mocked send must never create a real breaker entry"
);
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
async fn client_extracts_from_autumn_config_in_state() {
use axum::extract::FromRequestParts;
let mut cfg = crate::config::AutumnConfig::default();
cfg.http.client.max_retries = 7;
let state = crate::AppState::for_test();
state.insert_extension(cfg);
let mut parts = axum::http::Request::new(axum::body::Body::empty())
.into_parts()
.0;
let client = Client::from_request_parts(&mut parts, &state)
.await
.unwrap();
assert_eq!(client.retry_policy.max_retries, 7);
}
#[tokio::test]
async fn respond_with_status_produces_empty_body() {
let registry = Arc::new(MockRegistry::new());
let builder = MockSetupBuilder {
registry: registry.clone(),
alias: "svc".to_owned(),
method: None,
path: None,
};
let _handle = builder.delete("/items/1").respond_with_status(204);
let client = Client::new().with_mock(registry).named("svc");
let resp = client
.delete("https://svc.example.com/items/1")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 204);
assert_eq!(
resp.bytes(),
bytes::Bytes::new(),
"body must be empty, not \"null\""
);
}
#[test]
fn retry_after_http_date_parsing() {
let mut headers = HeaderMap::new();
headers.insert(
reqwest::header::HeaderName::from_static("retry-after"),
HeaderValue::from_static("Tue, 01 Jan 2030 00:00:00 GMT"),
);
let duration = parse_retry_after(&headers);
assert!(duration.is_some(), "should parse HTTP-date Retry-After");
assert!(
duration.unwrap().as_secs() > 0,
"future date should yield positive delay"
);
}
#[tokio::test]
async fn non_idempotent_post_no_retry() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/endpoint".to_owned(),
alias: None,
status: 503,
body: None,
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.post("https://example.com/endpoint")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 503);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn mock_strips_query_from_url_before_matching() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::GET),
path: "/v1/charges".to_owned(),
alias: None,
status: 200,
body: Some(serde_json::json!({"ok": true})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.get("https://api.stripe.com/v1/charges?expand[]=balance_transaction")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn mock_suffix_match_with_leading_slash_path() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/charges".to_owned(),
alias: None,
status: 201,
body: Some(serde_json::json!({"matched": true})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.post("https://api.stripe.com/v1/charges")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 201);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[test]
fn retries_does_not_clear_idempotent_only_flag() {
let client = Client::new();
let builder = client.post("https://example.com").retries(2);
assert_eq!(builder.retry_policy.max_retries, 2);
assert!(builder.retry_policy.retry_idempotent_only);
assert_eq!(builder.max_attempts(false), 1, "POST must not retry");
let builder = builder.retry_non_idempotent();
assert!(!builder.retry_policy.retry_idempotent_only);
assert_eq!(builder.max_attempts(false), 3);
}
#[test]
fn idempotency_key_is_added_only_for_opted_in_unsafe_retries() {
let client = Client::new();
let key_of = |builder: &RequestBuilder| {
builder
.extra_headers
.get(IDEMPOTENCY_KEY)
.map(|value| value.to_str().unwrap().to_owned())
};
let mut post = client.post("https://example.com").retry_non_idempotent();
post.ensure_idempotency_key();
let first = key_of(&post).expect("an opted-in POST gets a key");
post.ensure_idempotency_key();
assert_eq!(key_of(&post), Some(first), "the key does not change");
let mut caller = client
.post("https://example.com")
.header("Idempotency-Key", "mine")
.retry_non_idempotent();
caller.ensure_idempotency_key();
assert_eq!(key_of(&caller).as_deref(), Some("mine"));
for mut builder in [
client.post("https://example.com").retries(3),
client
.post("https://example.com")
.retry_non_idempotent()
.no_retry(),
client.get("https://example.com").retry_non_idempotent(),
] {
builder.ensure_idempotency_key();
assert_eq!(key_of(&builder), None);
}
}
#[test]
fn retry_delay_is_jittered_and_capped() {
let entropy = crate::entropy::SeededEntropy::new(9);
let policy = RetryPolicy {
max_backoff: Duration::from_secs(1),
..RetryPolicy::default()
};
for attempt in 0..40 {
let delay = policy.retry_delay(&entropy, attempt, None);
let ceiling = Duration::from_millis(crate::backoff::ceiling_ms(100, 1_000, attempt));
assert!(delay <= ceiling, "attempt {attempt}: {delay:?}");
}
}
#[test]
fn retry_delay_clamps_a_hint_to_the_backoff_window() {
let entropy = crate::entropy::SeededEntropy::new(10);
let policy = RetryPolicy::default();
let wait = policy.retry_delay(&entropy, 0, Some(Duration::from_secs(2)));
assert!(wait >= Duration::from_secs(2) && wait <= Duration::from_millis(2_100));
let wait = policy.retry_delay(&entropy, 0, Some(Duration::from_secs(3_600)));
assert!(wait >= Duration::from_secs(5) && wait <= Duration::from_millis(5_100));
}
#[test]
fn retry_hint_reads_retry_after_on_429_and_503_only() {
let mut headers = HeaderMap::new();
headers.insert("retry-after", HeaderValue::from_static("2"));
assert_eq!(retry_hint(429, &headers), Some(Duration::from_secs(2)));
assert_eq!(retry_hint(503, &headers), Some(Duration::from_secs(2)));
assert_eq!(retry_hint(502, &headers), None);
let empty = HeaderMap::new();
assert_eq!(retry_hint(429, &empty), Some(Duration::from_secs(1)));
assert_eq!(retry_hint(503, &empty), None);
}
#[test]
fn log_request_completes_with_sensitive_headers() {
let url = reqwest::Url::parse("https://api.example.com/v1/resource?q=1").unwrap();
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("content-type"),
HeaderValue::from_static("application/json"),
);
headers.insert(
HeaderName::from_static("authorization"),
HeaderValue::from_static("Bearer sk_test_xxx"),
);
log_request("POST", &url, 201, Duration::from_millis(12), &headers);
}
#[test]
fn sim_host_header_keeps_ipv6_brackets() {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let echo = axum::Router::new().route(
"/status",
axum::routing::get(|headers: HeaderMap| async move {
headers
.get(reqwest::header::HOST)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_owned()
}),
);
runtime.block_on(async {
for (url, host) in [
("http://[::1]:8080/status", "[::1]:8080"),
("http://[::1]/status", "[::1]"),
("http://payments:8443/status", "payments:8443"),
] {
let request = Client::new().get(url);
let parsed = reqwest::Url::parse(url).unwrap();
let seen = serve_sim_host(echo.clone(), &request, parsed, None)
.await
.unwrap()
.text();
assert_eq!(seen, host, "{url}");
}
});
}
#[test]
fn sim_host_gets_a_fresh_web_bot_auth_signature() {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let echo = axum::Router::new().fallback(|headers: HeaderMap| async move {
let count = headers.get_all("signature").iter().count();
let input = headers
.get("signature-input")
.and_then(|v| v.to_str().ok())
.unwrap_or_default()
.to_owned();
format!("{count} {input}")
});
let key = crate::aeo::web_bot_auth::WebBotAuthKey::from_seed_b64(
"nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A",
)
.unwrap();
let signer = crate::aeo::web_bot_auth::WebBotAuthSigner::new(key, "https://bot.example");
let created = |input: &str| -> u64 {
input
.split(";created=")
.nth(1)
.and_then(|r| r.split(';').next())
.and_then(|v| v.parse().ok())
.unwrap()
};
let url = "https://payments/charge";
let first = created(&signer.sign_url(url).unwrap().signature_input);
let request = Client::new().get(url).sign_web_bot_auth(&signer);
std::thread::sleep(Duration::from_millis(1100));
let parsed = reqwest::Url::parse(url).unwrap();
let seen = runtime.block_on(async {
serve_sim_host(echo, &request, parsed, None)
.await
.unwrap()
.text()
});
let (count, input) = seen.split_once(' ').unwrap();
assert_eq!(count, "1", "{seen}");
assert!(created(input) > first, "{seen}");
}
#[test]
fn sim_host_receives_content_length() {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let echo = axum::Router::new().fallback(|headers: HeaderMap| async move {
headers
.get(reqwest::header::CONTENT_LENGTH)
.and_then(|value| value.to_str().ok())
.unwrap_or("none")
.to_owned()
});
let url = "http://payments/charge";
let parsed = reqwest::Url::parse(url).unwrap();
runtime.block_on(async {
for (request, expected) in [
(Client::new().post(url).text_body("hello"), "5"),
(
Client::new().post(url).json(&serde_json::json!({"a": 1})),
"7",
),
(Client::new().post(url).bytes_body(Bytes::new()), "0"),
(Client::new().get(url), "none"),
] {
let seen = serve_sim_host(echo.clone(), &request, parsed.clone(), None)
.await
.unwrap()
.text();
assert_eq!(seen, expected);
}
});
}
#[cfg(feature = "telemetry-otlp")]
#[test]
fn sim_host_receives_trace_context() {
use opentelemetry::trace::TracerProvider as _;
use opentelemetry_sdk::propagation::TraceContextPropagator;
use opentelemetry_sdk::trace::SdkTracerProvider;
use tracing_subscriber::prelude::*;
opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
let provider = SdkTracerProvider::builder().build();
let subscriber = tracing_subscriber::registry()
.with(tracing_opentelemetry::layer().with_tracer(provider.tracer("test")));
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime");
let echo = || {
axum::Router::new().route(
"/echo",
axum::routing::get(|headers: HeaderMap| async move {
headers
.get("traceparent")
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_owned()
}),
)
};
let url = reqwest::Url::parse("http://payments/echo").unwrap();
tracing::subscriber::with_default(subscriber, || {
let span = tracing::info_span!("sim_trace_test");
let _guard = span.enter();
runtime.block_on(async {
let request = Client::new().get("http://payments/echo");
let traceparent = serve_sim_host(echo(), &request, url.clone(), None)
.await
.unwrap()
.text();
assert!(traceparent.starts_with("00-"), "{traceparent}");
let request = Client::new()
.get("http://payments/echo")
.header("traceparent", "caller-value");
let response = serve_sim_host(echo(), &request, url.clone(), None)
.await
.unwrap();
assert_eq!(response.text(), "caller-value");
});
});
}
#[test]
fn no_request_id_outside_a_request() {
let mut headers = HeaderMap::new();
add_current_request_id(&mut headers);
assert!(headers.is_empty());
}
#[tokio::test]
async fn current_request_id_is_added() {
let ctx = crate::log::context::LogContext::new(Some("rid-1".to_owned()));
let mut headers = HeaderMap::new();
crate::log::context::scope(ctx, async { add_current_request_id(&mut headers) }).await;
assert_eq!(headers.get("x-request-id").unwrap(), "rid-1");
}
#[tokio::test]
async fn caller_request_id_is_kept() {
let ctx = crate::log::context::LogContext::new(Some("rid-1".to_owned()));
let mut headers = HeaderMap::new();
headers.insert("x-request-id", HeaderValue::from_static("mine"));
crate::log::context::scope(ctx, async { add_current_request_id(&mut headers) }).await;
assert_eq!(headers.get_all("x-request-id").iter().count(), 1);
assert_eq!(headers.get("x-request-id").unwrap(), "mine");
}
#[test]
fn caller_trace_header_drops_the_whole_injected_pair() {
let injected = || {
vec![
("traceparent".to_owned(), HeaderValue::from_static("ours-p")),
("tracestate".to_owned(), HeaderValue::from_static("ours-s")),
("baggage".to_owned(), HeaderValue::from_static("ours-b")),
]
};
let names = |caller: &[&'static str]| {
let mut headers = HeaderMap::new();
for name in caller {
headers.insert(*name, HeaderValue::from_static("mine"));
}
let mut sent: Vec<String> = trace_headers_to_send(injected(), &headers)
.into_iter()
.map(|(name, _)| name)
.collect();
sent.sort();
sent
};
assert_eq!(names(&[]), ["baggage", "traceparent", "tracestate"]);
assert_eq!(names(&["traceparent"]), ["baggage"]);
assert_eq!(names(&["tracestate"]), ["baggage"]);
assert_eq!(names(&["baggage"]), ["traceparent", "tracestate"]);
}
#[test]
fn caller_trace_header_suppresses_the_injected_one() {
let mut caller = HeaderMap::new();
caller.insert("traceparent", HeaderValue::from_static("mine"));
let request =
inject_trace_context(reqwest::Client::new().get("https://example.com"), &caller)
.build()
.unwrap();
assert!(request.headers().get("traceparent").is_none());
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn without_request_id_sends_no_id() {
use axum::{Router, http::HeaderMap as AxumHeaders, routing::get};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new().route(
"/echo",
get(|headers: AxumHeaders| async move {
headers
.get("x-request-id")
.map_or_else(|| "none".to_owned(), |v| v.to_str().unwrap().to_owned())
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let ctx = crate::log::context::LogContext::new(Some("rid-1".to_owned()));
let url = format!("http://{addr}/echo");
let (forwarded, opted_out) = crate::log::context::scope(ctx, async {
let forwarded = Client::new().get(&url).send().await.unwrap().text();
let opted_out = Client::new()
.get(&url)
.without_request_id()
.send()
.await
.unwrap()
.text();
(forwarded, opted_out)
})
.await;
assert_eq!(forwarded, "rid-1");
assert_eq!(opted_out, "none");
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn real_get_request_covers_network_path() {
use axum::{Router, routing::get};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new().route("/ping", get(|| async { "pong" }));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = Client::new();
let resp = client
.get(format!("http://127.0.0.1:{}/ping", addr.port()))
.header("x-request-id", "test-35")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert!(resp.url().is_some());
assert_eq!(resp.text(), "pong");
crate::circuit_breaker::global_registry().clear();
}
#[derive(Clone, Default)]
struct ClientSpanCapture {
spans: std::sync::Arc<
std::sync::Mutex<
std::collections::HashMap<u64, std::collections::BTreeMap<String, String>>,
>,
>,
}
struct SpanFieldVisitor<'a>(&'a mut std::collections::BTreeMap<String, String>);
impl tracing::field::Visit for SpanFieldVisitor<'_> {
fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
self.0.insert(
field.name().to_owned(),
format!("{value:?}").trim_matches('"').to_owned(),
);
}
}
impl<S> tracing_subscriber::Layer<S> for ClientSpanCapture
where
S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
{
fn on_new_span(
&self,
attrs: &tracing::span::Attributes<'_>,
id: &tracing::span::Id,
_ctx: tracing_subscriber::layer::Context<'_, S>,
) {
if attrs.metadata().name() != "http.client.request" {
return;
}
let mut fields = std::collections::BTreeMap::new();
attrs.record(&mut SpanFieldVisitor(&mut fields));
self.spans.lock().unwrap().insert(id.into_u64(), fields);
}
fn on_record(
&self,
id: &tracing::span::Id,
values: &tracing::span::Record<'_>,
_ctx: tracing_subscriber::layer::Context<'_, S>,
) {
if let Some(fields) = self.spans.lock().unwrap().get_mut(&id.into_u64()) {
values.record(&mut SpanFieldVisitor(fields));
}
}
}
#[derive(Clone, Default)]
struct ClientSpanDurations {
opened: std::sync::Arc<std::sync::Mutex<std::collections::HashMap<u64, Instant>>>,
closed: std::sync::Arc<std::sync::Mutex<Vec<Duration>>>,
}
impl<S> tracing_subscriber::Layer<S> for ClientSpanDurations
where
S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
{
fn on_new_span(
&self,
attrs: &tracing::span::Attributes<'_>,
id: &tracing::span::Id,
_ctx: tracing_subscriber::layer::Context<'_, S>,
) {
if attrs.metadata().name() == "http.client.request" {
self.opened
.lock()
.unwrap()
.insert(id.into_u64(), Instant::now());
}
}
fn on_close(&self, id: tracing::span::Id, _ctx: tracing_subscriber::layer::Context<'_, S>) {
let start = self.opened.lock().unwrap().remove(&id.into_u64());
if let Some(start) = start {
self.closed.lock().unwrap().push(start.elapsed());
}
}
}
fn capture_client_spans() -> (
ClientSpanCapture,
ClientSpanDurations,
tracing::subscriber::DefaultGuard,
) {
use tracing_subscriber::layer::SubscriberExt as _;
let fields = ClientSpanCapture::default();
let durations = ClientSpanDurations::default();
let guard = tracing::subscriber::set_default(
tracing_subscriber::registry()
.with(fields.clone())
.with(durations.clone()),
);
drop(client_attempt_span(&Method::GET, "http://warm.up/", 0));
tracing::callsite::rebuild_interest_cache();
fields.spans.lock().unwrap().clear();
durations.closed.lock().unwrap().clear();
(fields, durations, guard)
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn attempt_span_covers_the_response_body() {
use axum::{Router, body::Body, routing::get};
use futures::StreamExt as _;
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new()
.route(
"/slow",
get(|| async {
let stream = futures::stream::once(async {
tokio::time::sleep(Duration::from_millis(300)).await;
Ok::<_, std::io::Error>(Bytes::from_static(b"late"))
});
Body::from_stream(stream)
}),
)
.route(
"/broken",
get(|| async {
let stream = futures::stream::iter([
Ok(Bytes::from_static(b"partial")),
Err(std::io::Error::other("cut")),
])
.then(|item| async move {
tokio::time::sleep(Duration::from_millis(50)).await;
item
});
Body::from_stream(stream)
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
for custom_path in [false, true] {
let (fields, durations, _guard) = capture_client_spans();
let mut slow = Client::new().get(format!("http://{addr}/slow"));
let mut broken = Client::new().get(format!("http://{addr}/broken"));
if custom_path {
slow = slow.no_redirect();
broken = broken.no_redirect();
}
assert_eq!(slow.send().await.unwrap().text(), "late");
let closed = durations.closed.lock().unwrap().clone();
assert_eq!(closed.len(), 1, "custom={custom_path}: {closed:?}");
assert!(
closed[0] >= Duration::from_millis(250),
"custom={custom_path}: the span ended before the body: {closed:?}"
);
fields.spans.lock().unwrap().clear();
assert!(broken.send().await.is_err());
let spans: Vec<_> = fields.spans.lock().unwrap().values().cloned().collect();
assert_eq!(spans.len(), 1, "custom={custom_path}: {spans:?}");
assert_eq!(
spans[0].get("error.type").map(String::as_str),
Some("body"),
"custom={custom_path}: {spans:?}"
);
assert_eq!(
spans[0].get("otel.status_code").map(String::as_str),
Some("ERROR"),
"custom={custom_path}: {spans:?}"
);
fields.spans.lock().unwrap().clear();
let mut missing = Client::new().get(format!("http://{addr}/missing"));
if custom_path {
missing = missing.no_redirect();
}
assert_eq!(missing.send().await.unwrap().status().as_u16(), 404);
let spans: Vec<_> = fields.spans.lock().unwrap().values().cloned().collect();
assert_eq!(spans.len(), 1, "custom={custom_path}: {spans:?}");
assert_eq!(
spans[0].get("otel.status_code").map(String::as_str),
Some("ERROR"),
"custom={custom_path}: {spans:?}"
);
}
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
async fn sim_attempts_open_client_spans() {
use axum::{Router, http::StatusCode, routing::get};
use std::sync::atomic::{AtomicUsize, Ordering};
let (fields, _durations, _guard) = capture_client_spans();
let calls = std::sync::Arc::new(AtomicUsize::new(0));
let payments = Router::new().route(
"/charge",
get(move || {
let calls = std::sync::Arc::clone(&calls);
async move {
if calls.fetch_add(1, Ordering::SeqCst) == 0 {
StatusCode::SERVICE_UNAVAILABLE
} else {
StatusCode::OK
}
}
}),
);
let mut client = Client::new();
client.sim_net = Some(Arc::new(
crate::sim::SimNet::new().host("payments", payments),
));
let response = client.get("http://payments/charge").send().await.unwrap();
assert_eq!(response.status().as_u16(), 200);
let mut attempts: Vec<_> = fields.spans.lock().unwrap().values().cloned().collect();
attempts.sort_by_key(|f| f.get("http.request.resend_count").cloned());
assert_eq!(attempts.len(), 2, "{attempts:?}");
assert_eq!(
attempts[0].get("server.address").map(String::as_str),
Some("payments")
);
assert_eq!(
attempts[0]
.get("http.response.status_code")
.map(String::as_str),
Some("503")
);
assert_eq!(
attempts[1]
.get("http.response.status_code")
.map(String::as_str),
Some("200")
);
assert_eq!(
attempts[0].get("otel.status_code").map(String::as_str),
Some("ERROR"),
"{attempts:?}"
);
assert_eq!(attempts[1].get("otel.status_code"), None, "{attempts:?}");
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn attempt_span_ends_before_the_retry_after_sleep() {
use axum::{Router, http::StatusCode, response::IntoResponse as _, routing::get};
use std::sync::atomic::{AtomicUsize, Ordering};
use tracing_subscriber::layer::SubscriberExt as _;
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
for custom_path in [false, true] {
let durations = ClientSpanDurations::default();
let _guard = tracing::subscriber::set_default(
tracing_subscriber::registry().with(durations.clone()),
);
drop(client_attempt_span(&Method::GET, "http://warm.up/", 0));
tracing::callsite::rebuild_interest_cache();
durations.closed.lock().unwrap().clear();
let calls = std::sync::Arc::new(AtomicUsize::new(0));
let app = Router::new().route(
"/limited",
get(move || {
let calls = std::sync::Arc::clone(&calls);
async move {
if calls.fetch_add(1, Ordering::SeqCst) == 0 {
(StatusCode::TOO_MANY_REQUESTS, [("retry-after", "1")]).into_response()
} else {
StatusCode::OK.into_response()
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let mut request = Client::new().get(format!("http://{addr}/limited"));
if custom_path {
request = request.no_redirect();
}
assert_eq!(request.send().await.unwrap().status().as_u16(), 200);
let closed = durations.closed.lock().unwrap().clone();
assert_eq!(closed.len(), 2, "custom={custom_path}: {closed:?}");
assert!(
closed[0] < Duration::from_millis(500),
"custom={custom_path}: the 429 attempt span held the 1s sleep: {closed:?}"
);
}
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn each_attempt_opens_one_client_span() {
use axum::{Router, http::StatusCode, routing::get};
use std::sync::atomic::{AtomicUsize, Ordering};
use tracing_subscriber::layer::SubscriberExt as _;
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
for custom_path in [false, true] {
let capture = ClientSpanCapture::default();
let _guard = tracing::subscriber::set_default(
tracing_subscriber::registry().with(capture.clone()),
);
drop(client_attempt_span(&Method::GET, "http://warm.up/", 0));
tracing::callsite::rebuild_interest_cache();
capture.spans.lock().unwrap().clear();
let calls = std::sync::Arc::new(AtomicUsize::new(0));
let app = Router::new().route(
"/flaky",
get(move || {
let calls = std::sync::Arc::clone(&calls);
async move {
if calls.fetch_add(1, Ordering::SeqCst) == 0 {
StatusCode::SERVICE_UNAVAILABLE
} else {
StatusCode::OK
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let mut request = Client::new().get(format!("http://{addr}/flaky"));
if custom_path {
request = request.no_redirect();
}
let response = request.send().await.unwrap();
assert_eq!(response.status().as_u16(), 200);
let mut attempts: Vec<_> = capture.spans.lock().unwrap().values().cloned().collect();
attempts.sort_by_key(|f| f.get("http.request.resend_count").cloned());
assert_eq!(attempts.len(), 2, "custom={custom_path}: {attempts:?}");
for (n, fields) in attempts.iter().enumerate() {
assert_eq!(fields.get("otel.kind").map(String::as_str), Some("client"));
assert_eq!(
fields.get("http.request.method").map(String::as_str),
Some("GET")
);
assert_eq!(
fields.get("server.address").map(String::as_str),
Some("127.0.0.1")
);
let resend = fields.get("http.request.resend_count");
assert_eq!(resend, (n > 0).then(|| n.to_string()).as_ref());
}
assert_eq!(
attempts[0]
.get("http.response.status_code")
.map(String::as_str),
Some("503")
);
assert_eq!(
attempts[1]
.get("http.response.status_code")
.map(String::as_str),
Some("200")
);
}
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn real_post_with_json_body_covers_body_path() {
use axum::{Json, Router, routing::post};
use serde_json::Value;
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new().route(
"/echo",
post(|Json(body): Json<Value>| async move { Json(body) }),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = Client::new();
let resp = client
.post(format!("http://127.0.0.1:{}/echo", addr.port()))
.json(&serde_json::json!({"hello": "world"}))
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let body: Value = resp.json().unwrap();
assert_eq!(body["hello"], "world");
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn real_get_retries_on_503_then_succeeds() {
use axum::{Router, routing::get};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let hit = Arc::new(AtomicU32::new(0));
let hit2 = hit.clone();
let app = Router::new().route(
"/flaky",
get(move || {
let c = hit2.clone();
async move {
if c.fetch_add(1, SeqOrdering::SeqCst) == 0 {
axum::http::StatusCode::SERVICE_UNAVAILABLE
} else {
axum::http::StatusCode::OK
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/flaky", addr.port()))
.retries(1)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(hit.load(SeqOrdering::SeqCst), 2);
crate::circuit_breaker::global_registry().clear();
}
async fn serve_down_once_with_retry_after() -> (SocketAddr, Arc<AtomicUsize>) {
use axum::{Router, response::IntoResponse as _, routing::get};
let hits = Arc::new(AtomicUsize::new(0));
let counter = hits.clone();
let app = Router::new().route(
"/down",
get(move || {
let counter = counter.clone();
async move {
if counter.fetch_add(1, Ordering::SeqCst) == 0 {
(
axum::http::StatusCode::SERVICE_UNAVAILABLE,
[("retry-after", "1")],
)
.into_response()
} else {
axum::http::StatusCode::OK.into_response()
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
(addr, hits)
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn real_paths_honour_retry_after_on_503() {
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
for custom_path in [false, true] {
let (addr, hits) = serve_down_once_with_retry_after().await;
let mut request = Client::new()
.get(format!("http://127.0.0.1:{}/down", addr.port()))
.retries(1);
if custom_path {
request = request.no_redirect();
}
let start = std::time::Instant::now();
let response = request.send().await.unwrap();
let waited = start.elapsed();
assert_eq!(response.status().as_u16(), 200, "custom_path={custom_path}");
assert_eq!(hits.load(Ordering::SeqCst), 2);
assert!(
waited >= Duration::from_secs(1),
"custom_path={custom_path}: waited only {waited:?}"
);
}
crate::circuit_breaker::global_registry().clear();
}
#[test]
fn from_config_reads_max_backoff_ms() {
let config = HttpClientConfig {
max_backoff_ms: 1_500,
..Default::default()
};
assert_eq!(
Client::from_config(&config).retry_policy.max_backoff,
Duration::from_millis(1_500)
);
assert_eq!(RetryPolicy::default().max_backoff, Duration::from_secs(20));
}
#[test]
fn text_body_sets_body() {
let client = Client::new();
let builder = client.post("https://example.com").text_body("hello");
assert_eq!(builder.body, Some(bytes::Bytes::from_static(b"hello")));
}
#[test]
fn client_error_display() {
let err = ClientError::NoMock("GET".to_owned(), "/path".to_owned());
assert!(err.to_string().contains("GET"));
assert!(err.to_string().contains("/path"));
}
#[test]
fn outbound_is_open_until_a_replay_blocks_it() {
assert!(
!outbound_blocked_for_replay(),
"a serving process must never start out blocked"
);
let error = ClientError::BlockedDuringReplay(
"GET".to_owned(),
"https://payments.example/charge".to_owned(),
)
.to_string();
assert!(
error.contains("blocked during replay") && error.contains("not recorded"),
"the error must say why the call did not happen, got {error}"
);
assert!(
error.contains("GET") && error.contains("https://payments.example/charge"),
"and which call it was, got {error}"
);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn test_http_client_circuit_breaker_integration() {
use axum::{Router, routing::get};
use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let hit = Arc::new(AtomicU32::new(0));
let hit2 = hit.clone();
let app = Router::new().route(
"/flaky",
get(move || {
let c = hit2.clone();
async move {
c.fetch_add(1, SeqOrdering::SeqCst);
axum::http::StatusCode::INTERNAL_SERVER_ERROR
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let mut rc = crate::config::ResilienceConfig::default();
rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
rc.circuit_breaker.defaults.open_duration_secs = Some(10);
let client = Client::new();
let client = Client {
resilience_config: Some(Arc::new(rc)),
..client
};
let url = format!("http://127.0.0.1:{}/flaky", addr.port());
for _ in 0..3 {
let res = client.get(&url).send().await;
let res = res.unwrap();
assert_eq!(res.status().as_u16(), 500);
}
let res = client.get(&url).send().await;
assert!(matches!(res, Err(ClientError::CircuitBreakerOpen)));
assert_eq!(hit.load(SeqOrdering::SeqCst), 3);
crate::circuit_breaker::global_registry().clear();
}
#[test]
fn shared_reqwest_client_ext_round_trips() {
let ext = SharedReqwestClient {
client: reqwest::Client::new(),
timeout_secs: 30,
};
let state = crate::AppState::for_test();
state.insert_extension(ext);
let retrieved = state.extension::<SharedReqwestClient>();
assert!(retrieved.is_some());
}
#[test]
fn client_head_method_builds_request_builder() {
let client = Client::new();
let _builder = client.head("https://example.com/resource");
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn from_state_reuses_shared_client() {
use axum::{Router, routing::get};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new().route(
"/ua",
get(|req: axum::http::Request<axum::body::Body>| async move {
req.headers()
.get("user-agent")
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_owned()
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let distinctive_inner = reqwest::ClientBuilder::new()
.user_agent("autumn-shared-pool-test")
.build()
.expect("failed to build inner client");
let state = crate::AppState::for_test();
state.insert_extension(SharedReqwestClient {
client: distinctive_inner,
timeout_secs: 30,
});
let client = Client::from_state(&state);
let resp = client
.get(format!("http://127.0.0.1:{}/ua", addr.port()))
.send()
.await
.expect("request should succeed");
assert_eq!(resp.text(), "autumn-shared-pool-test");
crate::circuit_breaker::global_registry().clear();
}
#[cfg(feature = "http-client")]
#[test]
fn from_state_falls_back_when_timeout_mismatches_shared_client() {
use crate::config::{AutumnConfig, HttpClientConfig};
use std::sync::Arc;
let mut config = AutumnConfig::default();
config.http.client = HttpClientConfig {
timeout_secs: 10,
..Default::default()
};
let state = crate::AppState::for_test();
state.insert_extension(SharedReqwestClient {
client: reqwest::Client::new(),
timeout_secs: 5, });
state.insert_extension(Arc::new(config));
let _client = Client::from_state(&state);
}
#[cfg(feature = "http-client")]
#[test]
fn from_state_reuses_shared_client_when_no_config() {
let state = crate::AppState::for_test();
let default_timeout = crate::config::HttpClientConfig::default().timeout_secs;
state.insert_extension(SharedReqwestClient {
client: reqwest::Client::new(),
timeout_secs: default_timeout,
});
let _client = Client::from_state(&state);
}
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
#[test]
fn ssrf_policy_blocks_private_and_reserved_ipv4() {
let blocked = [
"0.0.0.0",
"10.1.2.3",
"100.64.0.1", "127.0.0.1", "169.254.169.254", "172.16.5.4", "192.0.0.1", "192.0.2.5", "192.88.99.1", "192.168.1.1", "198.18.0.1", "198.51.100.7", "203.0.113.9", "224.0.0.1", "240.0.0.1", "255.255.255.255", ];
for s in blocked {
let ip: IpAddr = s.parse().unwrap();
assert!(is_blocked_ip(ip), "{s} should be blocked");
assert!(!is_public_ip(ip), "{s} should not be public");
}
}
#[test]
fn ssrf_policy_allows_public_ipv4() {
for s in ["1.1.1.1", "8.8.8.8", "93.184.216.34"] {
let ip: IpAddr = s.parse().unwrap();
assert!(is_public_ip(ip), "{s} should be public");
assert!(!is_blocked_ip(ip), "{s} should not be blocked");
}
}
#[test]
fn ssrf_policy_ipv6_and_mapped_forms() {
let blocked = [
"::", "::1", "fe80::1", "fc00::1", "ff02::1", "2001:db8::1", "fec0::1", "::ffff:169.254.169.254", "::ffff:127.0.0.1", "100::1", "100::dead:beef", "2001:2::1", "2001:10::1", "2001:20::1", "2001:20:abcd::1", "2001:2f::1", "3fff::1", "3fff:0fff::1", "5f00::1", "5f00:1234::1", "2620:4f:8000::1", ];
for s in blocked {
let ip: IpAddr = s.parse().unwrap();
assert!(is_blocked_ip(ip), "{s} should be blocked");
}
let compat = IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0x7f00, 0x0001));
assert!(
is_blocked_ip(compat),
"::7f00:1 (127.0.0.1) should be blocked"
);
let public: IpAddr = "2606:4700:4700::1111".parse().unwrap();
assert!(
is_public_ip(public),
"2606:4700:4700::1111 should be public"
);
let public_addrs = [
"2001:4860:4860::8888", "2606:4700:4700::1111", "2400:cb00:2048::1", "2620:0:2d0:200::7", "2001:2:1::1", "3fff:abcd::1", "4000::1", "5e00::1", "6000::1", ];
for s in public_addrs {
let ip: IpAddr = s.parse().unwrap();
assert!(is_public_ip(ip), "{s} should be public");
assert!(!is_blocked_ip(ip), "{s} should not be blocked");
}
}
#[test]
fn ssrf_policy_blocks_tunnelled_ipv4() {
let blocked = [
"64:ff9b::a9fe:a9fe", "64:ff9b::7f00:1", "64:ff9b:1::a9fe:a9fe", "64:ff9b:1::7f00:1", "64:ff9b:1::808:808", "2002:a9fe:a9fe::", "2002:7f00:1::", "2002:0a00:0001::", "::ffff:0:169.254.169.254", "::ffff:0:127.0.0.1", ];
for s in blocked {
let ip: IpAddr = s.parse().unwrap();
assert!(is_blocked_ip(ip), "{s} should be blocked");
assert!(!is_public_ip(ip), "{s} should not be public");
}
let anycast: IpAddr = "192.88.99.1".parse().unwrap();
assert!(is_blocked_ip(anycast), "192.88.99.1 should be blocked");
for s in ["2002:0808:0808::", "2606:4700::1111", "::ffff:0:8.8.8.8"] {
let ip: IpAddr = s.parse().unwrap();
assert!(is_public_ip(ip), "{s} should be public");
assert!(!is_blocked_ip(ip), "{s} should not be blocked");
}
}
#[test]
fn ssrf_policy_decimal_encoded_host_is_blocked() {
for raw in [
"http://2130706433/",
"http://0x7f000001/",
"http://127.0.0.1/",
] {
let parsed = url::Url::parse(raw).unwrap();
match parsed.host() {
Some(url::Host::Ipv4(v4)) => {
assert_eq!(
v4,
Ipv4Addr::LOCALHOST,
"{raw} should normalise to 127.0.0.1"
);
assert!(
is_blocked_ip(IpAddr::V4(v4)),
"{raw} host should be blocked"
);
}
other => panic!("{raw} did not parse to an Ipv4 host: {other:?}"),
}
}
}
async fn spawn(app: axum::Router) -> std::net::SocketAddr {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
addr
}
fn redirect_302(location: String) -> axum::response::Response {
axum::response::Response::builder()
.status(302)
.header("location", location)
.body(axum::body::Body::empty())
.unwrap()
}
fn redirect_307(location: String) -> axum::response::Response {
axum::response::Response::builder()
.status(307)
.header("location", location)
.body(axum::body::Body::empty())
.unwrap()
}
fn response_with_location(status: u16, location: String) -> axum::response::Response {
axum::response::Response::builder()
.status(status)
.header("location", location)
.body(axum::body::Body::empty())
.unwrap()
}
#[tokio::test]
async fn no_redirect_returns_3xx_unfollowed() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("http://127.0.0.1:1/never".to_owned()) }),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", addr.port()))
.no_redirect()
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 302);
assert_eq!(
resp.headers().get("location").and_then(|v| v.to_str().ok()),
Some("http://127.0.0.1:1/never")
);
}
#[tokio::test]
async fn no_redirect_returns_3xx_with_malformed_location() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("ht!tp://\\bad".to_owned()) }),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", addr.port()))
.no_redirect()
.send()
.await
.expect("no_redirect() must return the 3xx even with a malformed Location");
assert_eq!(resp.status().as_u16(), 302);
assert_eq!(
resp.headers().get("location").and_then(|v| v.to_str().ok()),
Some("ht!tp://\\bad")
);
}
#[tokio::test]
async fn follow_redirects_valid_chain_calls_validator() {
use axum::{Router, routing::get};
let b_addr = spawn(Router::new().route("/final", get(|| async { "final-body" }))).await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/final")) }),
))
.await;
let calls = Arc::new(AtomicUsize::new(0));
let calls2 = calls.clone();
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, move |_loc| {
calls2.fetch_add(1, Ordering::SeqCst);
true
})
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(resp.text(), "final-body");
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"validator called once per hop"
);
}
#[tokio::test]
async fn follow_redirects_rejects_private_target() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let priv_addr = spawn(Router::new().route(
"/secret",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"SECRET"
}
}),
))
.await;
let priv_port = priv_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(
move || async move { redirect_302(format!("http://127.0.0.1:{priv_port}/secret")) },
),
))
.await;
let validator = |u: &str| -> bool {
let Ok(p) = url::Url::parse(u) else {
return false;
};
match p.host() {
Some(url::Host::Ipv4(v4)) => is_public_ip(IpAddr::V4(v4)),
Some(url::Host::Ipv6(v6)) => is_public_ip(IpAddr::V6(v6)),
_ => true,
}
};
let result = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, validator)
.send()
.await;
assert!(
matches!(result, Err(ClientError::RedirectRejected(_))),
"expected RedirectRejected, got {result:?}"
);
assert!(
!touched.load(Ordering::SeqCst),
"the private target must never be connected to"
);
}
#[tokio::test]
async fn follow_redirects_cap_exceeded() {
use axum::{Router, routing::get};
let addr =
spawn(Router::new().route("/loop", get(|| async { redirect_302("/loop".to_owned()) })))
.await;
let result = Client::new()
.get(format!("http://127.0.0.1:{}/loop", addr.port()))
.follow_redirects(2, |_| true)
.send()
.await;
assert!(
matches!(result, Err(ClientError::TooManyRedirects(2))),
"expected TooManyRedirects(2), got {result:?}"
);
}
#[tokio::test]
async fn follow_redirects_zero_max_errors_on_first_3xx() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("http://127.0.0.1:1/x".to_owned()) }),
))
.await;
let result = Client::new()
.get(format!("http://127.0.0.1:{}/start", addr.port()))
.follow_redirects(0, |_| true)
.send()
.await;
assert!(matches!(result, Err(ClientError::TooManyRedirects(0))));
}
#[tokio::test]
async fn pin_to_bypasses_dns_and_uses_url_port() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route("/ping", get(|| async { "pong" }))).await;
let listener_port = addr.port();
let resp = Client::new()
.get(format!("http://pinned.invalid:{listener_port}/ping"))
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1))
.send()
.await
.expect("pinned request should reach the loopback listener");
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(resp.text(), "pong");
}
#[tokio::test]
async fn pin_to_addrs_falls_back_to_the_next_address_on_connect_failure() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route("/ping", get(|| async { "pong" }))).await;
let port = addr.port();
let resp = Client::new()
.get(format!("http://pinned.invalid:{port}/ping"))
.pin_to_addrs([
SocketAddr::new(IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), port),
SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
])
.send()
.await
.expect("the second pinned address must serve the request");
assert_eq!(resp.text(), "pong");
}
#[tokio::test]
async fn pin_to_addrs_with_no_address_fails_instead_of_sending_unpinned() {
let result = Client::new()
.get("http://pinned.invalid:9/ping")
.pin_to_addrs([])
.send()
.await;
assert!(
matches!(result, Err(ClientError::InvalidUrl(_))),
"{result:?}"
);
}
#[tokio::test]
async fn get_ssrf_safe_rejects_loopback_host_before_connecting() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let addr = spawn(Router::new().route(
"/x",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"reached"
}
}),
))
.await;
let result = Client::new()
.get_ssrf_safe(format!("http://localhost:{}/x", addr.port()))
.send()
.await;
assert!(
matches!(result, Err(ClientError::SsrfBlocked(_))),
"expected SsrfBlocked, got {result:?}"
);
assert!(
!touched.load(Ordering::SeqCst),
"SSRF guard must reject before any connection"
);
}
#[tokio::test]
async fn get_ssrf_safe_rejects_decimal_encoded_loopback() {
let result = Client::new()
.get_ssrf_safe("http://2130706433/")
.send()
.await;
assert!(
matches!(result, Err(ClientError::SsrfBlocked(_))),
"expected SsrfBlocked, got {result:?}"
);
}
#[tokio::test]
async fn resolve_and_validate_accepts_public_rejects_blocked() {
let ok = resolve_and_validate("http://8.8.8.8:8080/path")
.await
.unwrap();
assert_eq!(
ok,
vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)), 8080)]
);
let ok_https = resolve_and_validate("https://1.1.1.1/").await.unwrap();
assert_eq!(ok_https.len(), 1);
assert_eq!(ok_https[0].port(), 443);
for raw in [
"http://127.0.0.1/",
"http://169.254.169.254/latest/meta-data/",
"http://10.0.0.1/",
"http://2130706433/",
] {
let err = resolve_and_validate(raw).await;
assert!(
matches!(err, Err(ClientError::SsrfBlocked(_))),
"{raw} should be SsrfBlocked, got {err:?}"
);
}
}
#[test]
fn validate_resolved_addrs_returns_all_public_rejects_any_blocked() {
let v4 = |a, b, c, d, p| SocketAddr::new(IpAddr::V4(Ipv4Addr::new(a, b, c, d)), p);
let two = vec![v4(1, 1, 1, 1, 443), v4(8, 8, 8, 8, 443)];
assert_eq!(validate_resolved_addrs(two.clone()).unwrap(), two);
let mixed = vec![v4(1, 1, 1, 1, 443), v4(10, 0, 0, 1, 443)];
assert!(
matches!(
validate_resolved_addrs(mixed),
Err(ClientError::SsrfBlocked(_))
),
"a set containing a blocked address must be rejected"
);
let v6 = SocketAddr::new(
IpAddr::V6(Ipv6Addr::new(0x2606, 0x4700, 0x4700, 0, 0, 0, 0, 0x1111)),
443,
);
let mix = vec![v6, v4(8, 8, 8, 8, 443)];
assert_eq!(validate_resolved_addrs(mix.clone()).unwrap(), mix);
}
#[tokio::test]
async fn pin_to_does_not_follow_redirect_unpinned() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let onward = spawn(Router::new().route(
"/onward",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"REACHED"
}
}),
))
.await;
let onward_port = onward.port();
let start =
spawn(Router::new().route(
"/start",
get(move || async move {
redirect_302(format!("http://127.0.0.1:{onward_port}/onward"))
}),
))
.await;
let start_port = start.port();
let resp = Client::new()
.get(format!("http://pinned.invalid:{start_port}/start"))
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), start_port))
.send()
.await
.expect("pinned request should return the 302 unfollowed");
assert_eq!(
resp.status().as_u16(),
302,
"pin_to must return the redirect unfollowed"
);
assert!(
!touched.load(Ordering::SeqCst),
"pin_to must not silently follow the redirect onward"
);
}
#[tokio::test]
async fn get_ssrf_safe_rejects_non_http_scheme() {
for raw in ["ftp://public.example/resource", "gopher://public.example/"] {
let result = Client::new().get_ssrf_safe(raw).send().await;
assert!(
matches!(result, Err(ClientError::InvalidUrl(_))),
"{raw} should be rejected with InvalidUrl, got {result:?}"
);
}
}
#[tokio::test]
async fn follow_redirects_strips_sensitive_headers_cross_origin() {
use axum::{Router, routing::get};
let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let b_addr = spawn(Router::new().route(
"/dst",
get(move |headers: HeaderMap| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some(headers);
"ok"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/",
get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/", a_addr.port()))
.header("authorization", "secret")
.header("cookie", "session=abc")
.header("proxy-authorization", "Basic zzz")
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let headers = seen
.lock()
.unwrap()
.clone()
.expect("listener B must have been reached");
assert!(
headers.get("authorization").is_none(),
"authorization must be stripped on a cross-origin redirect"
);
assert!(
headers.get("cookie").is_none(),
"cookie must be stripped on a cross-origin redirect"
);
assert!(
headers.get("proxy-authorization").is_none(),
"proxy-authorization must be stripped on a cross-origin redirect"
);
}
#[tokio::test]
async fn default_redirects_strip_message_signatures_cross_origin() {
use axum::{Router, routing::get};
let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let b_addr = spawn(Router::new().route(
"/dst",
get(move |headers: HeaderMap| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some(headers);
"ok"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/",
get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/", a_addr.port()))
.header("signature-agent", "\"https://bot.example\"")
.header("signature-input", "sig1=(\"@authority\")")
.header("signature", "sig1=:AAAA:")
.send()
.await
.unwrap();
assert_eq!(
resp.status().as_u16(),
200,
"the redirect is still followed"
);
let headers = seen.lock().unwrap().clone().expect("B reached");
for name in ["signature", "signature-input", "signature-agent"] {
assert!(headers.get(name).is_none(), "{name} must be stripped");
}
}
#[tokio::test]
async fn web_bot_auth_signs_at_send_time() {
use axum::{Router, routing::get};
let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let addr = spawn(Router::new().route(
"/",
get(move |headers: HeaderMap| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some(headers);
"ok"
}
}),
))
.await;
let key = crate::aeo::web_bot_auth::WebBotAuthKey::from_seed_b64(
"nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A",
)
.unwrap();
let signer = crate::aeo::web_bot_auth::WebBotAuthSigner::new(key, "https://bot.example");
let created = |input: &str| -> u64 {
input
.split(";created=")
.nth(1)
.and_then(|r| r.split(';').next())
.and_then(|v| v.parse().ok())
.unwrap()
};
let url = format!("http://127.0.0.1:{}/", addr.port());
let first = created(&signer.sign_url(&url).unwrap().signature_input);
let builder = Client::new().get(&url).sign_web_bot_auth(&signer);
tokio::time::sleep(Duration::from_millis(1100)).await;
builder.send().await.unwrap();
let headers = seen.lock().unwrap().clone().expect("reached");
let sent = headers.get("signature-input").unwrap().to_str().unwrap();
assert!(created(sent) > first, "{sent}");
assert_eq!(headers.get_all("signature").iter().count(), 1);
}
#[cfg(feature = "reporting")]
#[test]
fn capsule_matching_skips_the_web_bot_auth_signature() {
let key = crate::aeo::web_bot_auth::WebBotAuthKey::from_seed_b64(
"nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A",
)
.unwrap();
let signer = crate::aeo::web_bot_auth::WebBotAuthSigner::new(key, "https://bot.example");
let builder = Client::new()
.get("https://api.example/x")
.header("x-kept", "1")
.sign_web_bot_auth(&signer);
let names: Vec<String> = caller_headers(&builder)
.into_iter()
.map(|(name, _)| name)
.collect();
assert!(names.contains(&"x-kept".to_owned()), "{names:?}");
assert!(names.contains(&"signature-agent".to_owned()), "{names:?}");
assert!(!names.contains(&"signature".to_owned()), "{names:?}");
assert!(!names.contains(&"signature-input".to_owned()), "{names:?}");
let plain = Client::new()
.get("https://api.example/x")
.header("signature", "mine");
assert_eq!(
caller_headers(&plain),
vec![("signature".to_owned(), "mine".to_owned())]
);
}
#[tokio::test]
async fn follow_redirects_keeps_sensitive_headers_same_origin() {
use axum::{Router, routing::get};
let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let addr = spawn(
Router::new()
.route("/", get(|| async { redirect_302("/next".to_owned()) }))
.route(
"/next",
get(move |headers: HeaderMap| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some(headers);
"ok"
}
}),
),
)
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/", addr.port()))
.header("authorization", "secret")
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let headers = seen
.lock()
.unwrap()
.clone()
.expect("/next must have been reached");
assert_eq!(
headers.get("authorization").and_then(|v| v.to_str().ok()),
Some("secret"),
"authorization must be preserved on a same-origin redirect"
);
}
#[tokio::test]
async fn follow_redirects_302_post_becomes_get() {
use axum::{
Router,
routing::{any, post},
};
let seen: Arc<Mutex<Option<(String, HeaderMap, Bytes)>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let b_addr = spawn(Router::new().route(
"/dst",
any(move |method: Method, headers: HeaderMap, body: Bytes| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some((method.to_string(), headers, body));
"ok"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/",
post(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
))
.await;
let resp = Client::new()
.post(format!("http://127.0.0.1:{}/", a_addr.port()))
.json(&serde_json::json!({"payload": true}))
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let (method, headers, body) = seen
.lock()
.unwrap()
.clone()
.expect("listener B must have been reached");
assert_eq!(method, "GET", "302 must rewrite POST → GET");
assert!(body.is_empty(), "302 POST→GET must drop the request body");
assert!(
!headers.contains_key(reqwest::header::CONTENT_TYPE),
"302 POST→GET must drop the Content-Type payload header"
);
assert!(
!headers.contains_key(reqwest::header::CONTENT_LENGTH),
"302 POST→GET must drop the Content-Length payload header"
);
}
#[tokio::test]
async fn follow_redirects_307_preserves_method_and_body() {
use axum::{
Router,
routing::{any, post},
};
let seen: Arc<Mutex<Option<(String, Bytes)>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let b_addr = spawn(Router::new().route(
"/dst",
any(move |method: Method, body: Bytes| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some((method.to_string(), body));
"ok"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/",
post(move || async move { redirect_307(format!("http://127.0.0.1:{b_port}/dst")) }),
))
.await;
let resp = Client::new()
.post(format!("http://127.0.0.1:{}/", a_addr.port()))
.text_body("payload")
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let (method, body) = seen
.lock()
.unwrap()
.clone()
.expect("listener B must have been reached");
assert_eq!(method, "POST", "307 must preserve the POST method");
assert_eq!(
&body[..],
b"payload",
"307 must preserve the request body verbatim"
);
}
#[test]
fn ssrf_redirect_plan_honours_chained_override() {
let client = Client::new();
let default = client.get_ssrf_safe("https://example.com/");
assert_eq!(
default.ssrf_redirect_plan(),
(true, SSRF_SAFE_MAX_REDIRECTS),
"default SSRF-safe path follows up to SSRF_SAFE_MAX_REDIRECTS"
);
let none = client.get_ssrf_safe("https://example.com/").no_redirect();
let (follow, _max) = none.ssrf_redirect_plan();
assert!(
!follow,
"no_redirect() must disable following on the safe path"
);
let follow3 = client
.get_ssrf_safe("https://example.com/")
.follow_redirects(3, |_| true);
assert_eq!(
follow3.ssrf_redirect_plan(),
(true, 3),
"follow_redirects(3, ..) must cap the safe path at 3 hops"
);
}
#[tokio::test]
async fn pin_then_follow_redirects_is_rejected() {
let result = Client::new()
.get("http://example.com/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.follow_redirects(2, |_| true)
.send()
.await;
assert!(
matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
"expected IncompatiblePinRedirect, got {result:?}"
);
}
#[tokio::test]
async fn follow_redirects_then_pin_is_rejected() {
let result = Client::new()
.get("http://example.com/")
.follow_redirects(2, |_| true)
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
"expected IncompatiblePinRedirect, got {result:?}"
);
}
#[tokio::test]
async fn pin_with_no_redirect_is_allowed() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("http://127.0.0.1:1/onward".to_owned()) }),
))
.await;
let port = addr.port();
let resp = Client::new()
.get(format!("http://pinned.invalid:{port}/start"))
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port))
.no_redirect()
.send()
.await
.expect("pin_to + no_redirect must return the 3xx unfollowed, not error");
assert_eq!(
resp.status().as_u16(),
302,
"pin_to + no_redirect returns the redirect verbatim"
);
}
#[tokio::test]
async fn pin_to_ipv4_literal_host_is_rejected() {
let result = Client::new()
.get("http://198.51.100.1:8080/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
"expected PinRequiresDomainHost, got {result:?}"
);
}
#[tokio::test]
async fn pin_to_ipv6_literal_host_is_rejected() {
let result = Client::new()
.get("http://[2606:4700::1111]/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
"expected PinRequiresDomainHost, got {result:?}"
);
}
#[tokio::test]
async fn get_ssrf_safe_with_pin_to_is_rejected() {
let result = Client::new()
.get_ssrf_safe("http://example.com/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::PinNotAllowedWithSsrfSafe(_))),
"expected PinNotAllowedWithSsrfSafe, got {result:?}"
);
}
#[tokio::test]
async fn follow_redirects_does_not_follow_304_with_location() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let b_addr = spawn(Router::new().route(
"/dst",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"SHOULD-NOT-BE-HIT"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(move || async move {
response_with_location(304, format!("http://127.0.0.1:{b_port}/dst"))
}),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, |_| true)
.send()
.await
.unwrap();
assert_eq!(
resp.status().as_u16(),
304,
"a 304 with a Location header must be returned verbatim, not followed"
);
assert!(
!touched.load(Ordering::SeqCst),
"the 304 Location target must never be requested"
);
}
#[tokio::test]
async fn follow_redirects_does_not_follow_300_with_location() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let b_addr = spawn(Router::new().route(
"/dst",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"SHOULD-NOT-BE-HIT"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(move || async move {
response_with_location(300, format!("http://127.0.0.1:{b_port}/dst"))
}),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, |_| true)
.send()
.await
.unwrap();
assert_eq!(
resp.status().as_u16(),
300,
"a 300 with a Location header must be returned verbatim, not followed"
);
assert!(
!touched.load(Ordering::SeqCst),
"the 300 Location target must never be requested"
);
}
#[allow(clippy::large_futures, reason = "test futures, awaited once")]
mod deadline_and_budget {
use super::super::*;
use std::sync::atomic::AtomicU32;
async fn counting(
status: Option<u16>,
headers: &'static [(&'static str, &'static str)],
) -> (String, Arc<AtomicU32>) {
let hits = Arc::new(AtomicU32::new(0));
let counter = Arc::clone(&hits);
let app = axum::Router::new().route(
"/x",
axum::routing::get(move || {
counter.fetch_add(1, Ordering::SeqCst);
async move {
let Some(code) = status else {
return std::future::pending().await;
};
let mut response = axum::response::Response::builder().status(code);
for (name, value) in headers {
response = response.header(*name, *value);
}
response.body(axum::body::Body::empty()).unwrap()
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
(format!("http://127.0.0.1:{}/x", addr.port()), hits)
}
fn one_retry_client() -> Client {
let mut config = crate::config::HttpClientConfig::default();
config.retry_budget.capacity = 14;
config.retry_budget.retry_ratio = 0.0;
Client::from_config(&config)
}
async fn with_deadline(
after: Duration,
send: impl std::future::Future<Output = Result<Response, ClientError>>,
) -> Result<Response, ClientError> {
tokio::time::timeout(Duration::from_secs(5), Deadline::after(after).scope(send))
.await
.expect("the deadline must stop the call")
}
fn lock() -> std::sync::MutexGuard<'static, ()> {
let guard = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
guard
}
async fn redirecting(delay: Duration) -> (String, Arc<Mutex<Vec<(String, u64)>>>) {
let seen = Arc::new(Mutex::new(Vec::new()));
let record = |path: &'static str, seen: Arc<Mutex<Vec<(String, u64)>>>| {
move |headers: HeaderMap| {
let millis = headers
.get(DEADLINE_HEADER)
.and_then(|value| value.to_str().ok()?.parse().ok())
.unwrap_or(u64::MAX);
seen.lock().unwrap().push((path.to_owned(), millis));
}
};
let at_r = record("/r", Arc::clone(&seen));
let at_t = record("/t", Arc::clone(&seen));
let app = axum::Router::new()
.route(
"/r",
axum::routing::get(move |headers: HeaderMap| async move {
at_r(headers);
tokio::time::sleep(delay).await;
axum::response::Redirect::temporary("/t")
}),
)
.route(
"/t",
axum::routing::get(move |headers: HeaderMap| async move {
at_t(headers);
"done"
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
(format!("http://127.0.0.1:{}/r", addr.port()), seen)
}
#[tokio::test]
async fn plain_path_sends_a_fresh_deadline_header_after_a_redirect() {
let (url, seen) = redirecting(Duration::from_millis(500)).await;
let response = with_deadline(Duration::from_secs(3), Client::new().get(&url).send())
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
let seen = seen.lock().unwrap().clone();
let [(first, at_origin), (second, at_target)] = seen.as_slice() else {
panic!("two hops: {seen:?}");
};
assert_eq!((first.as_str(), second.as_str()), ("/r", "/t"));
assert!(*at_origin <= 3_000, "{seen:?}");
assert!(
*at_target + 400 <= *at_origin,
"the target gets the time left after the redirect: {seen:?}"
);
}
#[tokio::test]
async fn a_redirected_connect_error_charges_the_destination_budget() {
let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let dead_port = closed.local_addr().unwrap().port();
drop(closed);
let target = format!("http://127.0.0.1:{dead_port}/t");
let app = axum::Router::new().route(
"/r",
axum::routing::get(move || {
let target = target.clone();
async move { axum::response::Redirect::temporary(&target) }
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let origin_port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = Client::from_config(&crate::config::HttpClientConfig::default());
let budgets = client.retry.budgets.clone().unwrap();
let result = client
.get(format!("http://127.0.0.1:{origin_port}/r"))
.retries(1)
.send()
.await;
assert!(result.is_err(), "{result:?}");
let full = f64::from(RetryBudgetConfig::default().capacity);
let origin = budgets.for_host(&format!("127.0.0.1:{origin_port}"));
let destination = budgets.for_host(&format!("127.0.0.1:{dead_port}"));
assert!((origin.available() - full).abs() < f64::EPSILON, "origin");
assert!(destination.available() < full, "the destination paid");
}
#[tokio::test]
async fn a_redirect_under_a_deadline_sets_referer() {
let seen = Arc::new(Mutex::new(None));
let record = Arc::clone(&seen);
let app = axum::Router::new()
.route(
"/r",
axum::routing::get(|| async { axum::response::Redirect::temporary("/t") }),
)
.route(
"/t",
axum::routing::get(move |headers: HeaderMap| async move {
*record.lock().unwrap() = headers
.get(reqwest::header::REFERER)
.map(|value| value.to_str().unwrap().to_owned());
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let origin = format!("http://127.0.0.1:{port}/r");
let response = with_deadline(Duration::from_secs(3), Client::new().get(&origin).send())
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
assert_eq!(seen.lock().unwrap().as_deref(), Some(origin.as_str()));
}
#[test]
fn referer_drops_credentials_and_is_not_sent_on_a_downgrade() {
let mut headers = HeaderMap::new();
set_referer(&mut headers, "https://b/x", "https://u:p@a/y#frag");
assert_eq!(headers[reqwest::header::REFERER], "https://a/y");
set_referer(&mut headers, "http://c/z", "https://b/x");
assert!(!headers.contains_key(reqwest::header::REFERER));
}
#[test]
fn every_host_of_a_redirect_chain_is_refilled() {
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let middle = budgets.for_host("b:443");
let last = budgets.for_host("c:443");
while middle.try_acquire(RetryKind::Transient) {}
while last.try_acquire(RetryKind::Transient) {}
let (middle_before, last_before) = (middle.available(), last.available());
let gate =
RetryGate::with_deadline(None, Some(Arc::clone(&budgets)), Some("a:443"), true);
let mut seen = std::collections::HashSet::new();
gate.record_destinations(
&[
"https://b/1".to_owned(),
"https://c/2".to_owned(),
"https://c/3".to_owned(),
],
&mut seen,
);
assert!(middle.available() > middle_before, "the middle host");
let one_refill = middle.available() - middle_before;
assert!(
(last.available() - last_before - one_refill).abs() < f64::EPSILON,
"one refill per host"
);
}
async fn redirect_server(location: &'static str, endless: bool) -> String {
let app = axum::Router::new()
.route(
"/r",
axum::routing::get(move || async move {
let body = if endless {
axum::body::Body::from_stream(futures::stream::pending::<
Result<Bytes, std::io::Error>,
>())
} else {
axum::body::Body::empty()
};
axum::response::Response::builder()
.status(302)
.header("location", location)
.body(body)
.unwrap()
}),
)
.route("/t", axum::routing::get(|| async { "done" }));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
format!("http://127.0.0.1:{port}/r")
}
#[tokio::test]
async fn a_redirect_under_a_deadline_does_not_read_the_redirect_body() {
let url = redirect_server("/t", true).await;
let response = with_deadline(Duration::from_secs(3), Client::new().get(&url).send())
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
assert_eq!(response.text(), "done");
}
#[tokio::test]
async fn a_bad_location_under_a_deadline_returns_the_redirect() {
let url = redirect_server("http://[::1", false).await;
let response = with_deadline(Duration::from_secs(3), Client::new().get(&url).send())
.await
.unwrap();
assert_eq!(response.status().as_u16(), 302);
let plain = Client::new().get(&url).send().await.unwrap();
assert_eq!(
plain.status().as_u16(),
302,
"the same as without a deadline"
);
}
#[tokio::test]
async fn a_redirect_chain_under_a_deadline_shares_one_retry_count() {
use axum::response::IntoResponse;
let origin_hits = Arc::new(AtomicU32::new(0));
let target_hits = Arc::new(AtomicU32::new(0));
let (origin, target) = (Arc::clone(&origin_hits), Arc::clone(&target_hits));
let app = axum::Router::new()
.route(
"/r",
axum::routing::get(move || async move {
if origin.fetch_add(1, Ordering::SeqCst) == 0 {
axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response()
} else {
axum::response::Redirect::temporary("/t").into_response()
}
}),
)
.route(
"/t",
axum::routing::get(move || async move {
if target.fetch_add(1, Ordering::SeqCst) == 0 {
axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response()
} else {
"done".into_response()
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let url = format!("http://127.0.0.1:{port}/r");
let response = with_deadline(
Duration::from_secs(3),
Client::new().get(&url).retries(1).send(),
)
.await
.unwrap();
assert_eq!(response.status().as_u16(), 503, "the one retry is spent");
assert_eq!(origin_hits.load(Ordering::SeqCst), 2);
assert_eq!(target_hits.load(Ordering::SeqCst), 1);
}
#[derive(Debug)]
struct FixedDraw(u64);
impl crate::entropy::Entropy for FixedDraw {
fn next_u64(&self) -> u64 {
self.0
}
fn fill_bytes(&self, dest: &mut [u8]) {
dest.fill(0);
}
}
#[tokio::test]
async fn a_cross_origin_redirect_under_a_deadline_strips_what_reqwest_strips() {
let seen: Arc<std::sync::Mutex<Option<HeaderMap>>> =
Arc::new(std::sync::Mutex::new(None));
let slot = Arc::clone(&seen);
let target = super::spawn(axum::Router::new().route(
"/dst",
axum::routing::get(move |headers: HeaderMap| {
let slot = Arc::clone(&slot);
async move {
*slot.lock().unwrap() = Some(headers);
"ok"
}
}),
))
.await;
let target_port = target.port();
let origin = super::spawn(axum::Router::new().route(
"/",
axum::routing::get(move || async move {
super::redirect_302(format!("http://127.0.0.1:{target_port}/dst"))
}),
))
.await;
let response = with_deadline(
Duration::from_secs(3),
Client::new()
.get(format!("http://127.0.0.1:{}/", origin.port()))
.header("authorization", "secret")
.header("cookie", "a=b")
.header("cookie2", "c=d")
.header("proxy-authorization", "Basic zzz")
.header("www-authenticate", "Basic realm=x")
.send(),
)
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
let headers = seen.lock().unwrap().clone().expect("target reached");
for name in [
"authorization",
"cookie",
"cookie2",
"proxy-authorization",
"www-authenticate",
] {
assert!(headers.get(name).is_none(), "{name} reached the target");
}
}
#[tokio::test]
async fn a_retry_cancelled_during_its_backoff_gives_its_tokens_back() {
let (url, hits) = counting(Some(502), &[]).await;
let host = url_host(&url).unwrap();
let mut client = Client::new();
client.entropy = Arc::new(FixedDraw(100));
let budget = client.retry.budgets.clone().unwrap().for_host(&host);
let cancelled = tokio::time::timeout(
Duration::from_millis(50),
client.get(&url).retries(3).send(),
)
.await;
assert!(cancelled.is_err(), "cancelled in the backoff");
assert_eq!(hits.load(Ordering::SeqCst), 1);
let full = f64::from(RetryBudgetConfig::default().capacity);
assert!(
(budget.available() - full).abs() < f64::EPSILON,
"no retry ran, so none is paid for: {}",
budget.available()
);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn a_deadline_stop_on_the_custom_path_is_not_a_throttle_reject() {
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let (url, _hits) = counting(None, &[]).await;
let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
2.0,
Duration::from_secs(120),
));
let client = Client {
resilience_config: Some(Arc::new(crate::config::ResilienceConfig::default())),
throttle: Some(Arc::clone(&throttle)),
..Client::new()
};
for _ in 0..10 {
let result = with_deadline(
Duration::from_millis(30),
client.get(&url).no_redirect().breaker_scoped().send(),
)
.await;
assert!(
matches!(result, Err(ClientError::DeadlineExceeded)),
"{result:?}"
);
}
let host = throttle_host(&url).unwrap();
assert!(
throttle.reject_probability(&host, crate::time::ambient_instant()) < f64::EPSILON,
"the caller's deadline is not the host's reject"
);
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
async fn the_throttle_applies_under_a_deadline() {
let (url, _hits) = counting(Some(503), &[]).await;
let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
2.0,
Duration::from_secs(120),
));
let client = Client {
throttle: Some(Arc::clone(&throttle)),
..Client::new()
};
let mut throttled = 0;
for _ in 0..100 {
let result =
with_deadline(Duration::from_secs(5), client.get(&url).no_retry().send()).await;
if matches!(result, Err(ClientError::ThrottledLocally { .. })) {
throttled += 1;
}
}
assert!(
throttled > 20,
"the deadline path must use the throttle: {throttled}"
);
}
#[tokio::test]
async fn each_retry_under_a_deadline_is_a_throttle_attempt() {
use axum::response::IntoResponse;
let hits = Arc::new(AtomicU32::new(0));
let counter = Arc::clone(&hits);
let upstream = super::spawn(axum::Router::new().route(
"/x",
axum::routing::get(move || {
let status = if counter.fetch_add(1, Ordering::SeqCst) % 3 < 2 {
axum::http::StatusCode::SERVICE_UNAVAILABLE
} else {
axum::http::StatusCode::OK
};
async move { status.into_response() }
}),
))
.await;
let url = format!("http://127.0.0.1:{}/x", upstream.port());
let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
2.0,
Duration::from_secs(120),
));
let mut client = Client {
throttle: Some(Arc::clone(&throttle)),
..Client::new()
};
client.entropy = Arc::new(FixedDraw(u64::MAX));
for _ in 0..10 {
let response =
with_deadline(Duration::from_secs(4), client.get(&url).retries(2).send())
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
}
assert_eq!(hits.load(Ordering::SeqCst), 30);
let host = throttle_host(&url).unwrap();
assert!(
throttle.reject_probability(&host, crate::time::ambient_instant()) > 0.3,
"each attempt counts"
);
}
#[tokio::test]
async fn a_retry_sends_what_is_left_of_the_callers_deadline_header() {
use axum::response::IntoResponse;
let seen: Arc<std::sync::Mutex<Vec<u64>>> = Arc::default();
let slot = Arc::clone(&seen);
let upstream = super::spawn(axum::Router::new().route(
"/x",
axum::routing::get(move |headers: HeaderMap| {
let slot = Arc::clone(&slot);
async move {
let value = headers
.get(DEADLINE_HEADER)
.and_then(|v| v.to_str().ok()?.parse().ok())
.unwrap_or(u64::MAX);
let first = {
let mut seen = slot.lock().unwrap();
seen.push(value);
seen.len() == 1
};
if first {
tokio::time::sleep(Duration::from_millis(60)).await;
axum::http::StatusCode::BAD_GATEWAY.into_response()
} else {
axum::http::StatusCode::OK.into_response()
}
}
}),
))
.await;
let mut client = Client::new();
client.entropy = Arc::new(FixedDraw(0));
let response = with_deadline(
Duration::from_secs(4),
client
.get(format!("http://127.0.0.1:{}/x", upstream.port()))
.header(DEADLINE_HEADER, "200")
.retries(1)
.send(),
)
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
let seen = seen.lock().unwrap().clone();
assert_eq!(seen.len(), 2);
assert!(seen[0] <= 200, "{seen:?}");
assert!(
seen[1] <= 140,
"the retry re-extended the deadline: {seen:?}"
);
}
#[tokio::test]
async fn a_redirect_chain_under_a_deadline_keeps_the_client_timeout() {
use axum::response::IntoResponse;
let app = axum::Router::new()
.route(
"/a",
axum::routing::get(|| async {
tokio::time::sleep(Duration::from_millis(200)).await;
axum::response::Redirect::temporary("/b").into_response()
}),
)
.route(
"/b",
axum::routing::get(|| async {
tokio::time::sleep(Duration::from_millis(200)).await;
"ok"
}),
);
let upstream = super::spawn(app).await;
let mut client = Client::new();
client.retry_policy.request_timeout = Some(Duration::from_millis(300));
let result = with_deadline(
Duration::from_secs(4),
client
.get(format!("http://127.0.0.1:{}/a", upstream.port()))
.no_retry()
.send(),
)
.await;
assert!(
matches!(&result, Err(ClientError::Request(e)) if e.is_timeout()),
"the chain must not get a new timeout per hop: {result:?}"
);
}
#[tokio::test]
async fn without_a_deadline_a_retry_sends_what_is_left_of_the_callers_header() {
use axum::response::IntoResponse;
let seen: Arc<std::sync::Mutex<Vec<u64>>> = Arc::default();
let slot = Arc::clone(&seen);
let upstream = super::spawn(axum::Router::new().route(
"/x",
axum::routing::get(move |headers: HeaderMap| {
let slot = Arc::clone(&slot);
async move {
let first = {
let mut seen = slot.lock().unwrap();
seen.push(header_millis(&headers));
seen.len() == 1
};
if first {
tokio::time::sleep(Duration::from_millis(60)).await;
axum::http::StatusCode::BAD_GATEWAY.into_response()
} else {
axum::http::StatusCode::OK.into_response()
}
}
}),
))
.await;
let mut client = Client::new();
client.entropy = Arc::new(FixedDraw(0));
let response = client
.get(format!("http://127.0.0.1:{}/x", upstream.port()))
.header(DEADLINE_HEADER, "200")
.retries(1)
.send()
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
let seen = seen.lock().unwrap().clone();
assert_eq!(seen[0], 200, "{seen:?}");
assert!(seen[1] <= 140, "the retry re-extended the header: {seen:?}");
}
#[tokio::test]
async fn without_a_deadline_a_redirect_hop_sends_what_is_left_of_the_callers_header() {
use axum::response::IntoResponse;
let seen: Arc<std::sync::Mutex<Vec<u64>>> = Arc::default();
let (at_a, at_b) = (Arc::clone(&seen), Arc::clone(&seen));
let app = axum::Router::new()
.route(
"/a",
axum::routing::get(move |headers: HeaderMap| async move {
at_a.lock().unwrap().push(header_millis(&headers));
tokio::time::sleep(Duration::from_millis(60)).await;
axum::response::Redirect::temporary("/b").into_response()
}),
)
.route(
"/b",
axum::routing::get(move |headers: HeaderMap| async move {
at_b.lock().unwrap().push(header_millis(&headers));
"ok"
}),
);
let upstream = super::spawn(app).await;
let response = Client::new()
.get(format!("http://127.0.0.1:{}/a", upstream.port()))
.header(DEADLINE_HEADER, "200")
.send()
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
let seen = seen.lock().unwrap().clone();
assert_eq!(seen.len(), 2, "{seen:?}");
assert_eq!(seen[0], 200, "{seen:?}");
assert!(seen[1] <= 140, "the hop re-extended the header: {seen:?}");
}
fn header_millis(headers: &HeaderMap) -> u64 {
headers
.get(DEADLINE_HEADER)
.and_then(|v| v.to_str().ok()?.parse().ok())
.unwrap_or(u64::MAX)
}
#[tokio::test]
async fn a_redirect_hop_continues_the_chain_backoff() {
use axum::response::IntoResponse;
let origin_hits = Arc::new(AtomicU32::new(0));
let target_hits = Arc::new(AtomicU32::new(0));
let (origin, target) = (Arc::clone(&origin_hits), Arc::clone(&target_hits));
let app = axum::Router::new()
.route(
"/r",
axum::routing::get(move || async move {
if origin.fetch_add(1, Ordering::SeqCst) == 0 {
axum::http::StatusCode::BAD_GATEWAY.into_response()
} else {
axum::response::Redirect::temporary("/t").into_response()
}
}),
)
.route(
"/t",
axum::routing::get(move || async move {
if target.fetch_add(1, Ordering::SeqCst) == 0 {
axum::http::StatusCode::BAD_GATEWAY.into_response()
} else {
"done".into_response()
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let url = format!("http://127.0.0.1:{port}/r");
let mut client = Client::new();
client.entropy = Arc::new(FixedDraw(20_099));
let start = std::time::Instant::now();
let response =
with_deadline(Duration::from_secs(5), client.get(&url).retries(2).send())
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
assert_eq!(origin_hits.load(Ordering::SeqCst), 2);
assert_eq!(target_hits.load(Ordering::SeqCst), 2);
assert!(
start.elapsed() >= Duration::from_millis(200),
"the hop's retry is the chain's second: {:?}",
start.elapsed()
);
}
#[tokio::test]
async fn a_post_redirected_to_get_under_a_deadline_gets_no_retry() {
use axum::response::IntoResponse;
let target_hits = Arc::new(AtomicU32::new(0));
let target = Arc::clone(&target_hits);
let app = axum::Router::new()
.route(
"/r",
axum::routing::post(|| async { axum::response::Redirect::to("/t") }),
)
.route(
"/t",
axum::routing::get(move || async move {
if target.fetch_add(1, Ordering::SeqCst) == 0 {
axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response()
} else {
"done".into_response()
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let url = format!("http://127.0.0.1:{port}/r");
let response = with_deadline(
Duration::from_secs(3),
Client::new().post(&url).retries(3).send(),
)
.await
.unwrap();
assert_eq!(response.status().as_u16(), 503, "as without a deadline");
assert_eq!(target_hits.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn a_backoff_past_the_hop_deadline_returns_the_response() {
let (url, hits) = counting(Some(502), &[]).await;
let policy = RetryPolicy {
max_retries: 20,
..RetryPolicy::default()
};
let gate = RetryGate::with_deadline(None, None, None, true);
let client = reqwest::Client::new();
let _ = client.get(&url).send().await;
let result = send_one(
&client,
&Method::GET,
&url,
&HeaderMap::new(),
None,
&policy,
&FixedDraw(400),
false,
Some(crate::time::ambient_instant() + Duration::from_secs(1)),
false,
&gate,
false,
None,
None,
)
.await;
assert_eq!(
result.map(|response| response.status().as_u16()).ok(),
Some(502),
"the last response, not a deadline error"
);
assert!(hits.load(Ordering::SeqCst) >= 3, "no retry ran");
}
#[tokio::test]
async fn a_transport_retry_past_the_hop_deadline_is_not_charged() {
let app = axum::Router::new().route(
"/hang",
axum::routing::get(|| async {
std::future::pending::<()>().await;
""
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let host = format!("127.0.0.1:{port}");
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let gate =
RetryGate::with_deadline(None, Some(Arc::clone(&budgets)), Some(&host), true);
let result = send_one(
&reqwest::Client::new(),
&Method::GET,
&format!("http://{host}/hang"),
&HeaderMap::new(),
None,
&RetryPolicy::default(),
&crate::entropy::SeededEntropy::new(1),
false,
Some(crate::time::ambient_instant() + Duration::from_millis(200)),
false,
&gate,
false,
None,
None,
)
.await;
assert!(result.is_err(), "{result:?}");
let full = f64::from(RetryBudgetConfig::default().capacity);
let available = budgets.for_host(&host).available();
assert!(
(full - available).abs() < f64::EPSILON,
"no tokens for a retry that cannot run: {available}"
);
}
#[tokio::test]
async fn a_retry_needs_a_minimum_attempt_before_the_hop_deadline() {
let (url, hits) = counting(Some(502), &[]).await;
let gate = RetryGate::with_deadline(None, None, None, true);
let result = send_one(
&reqwest::Client::new(),
&Method::GET,
&url,
&HeaderMap::new(),
None,
&RetryPolicy::default(),
&FixedDraw(100),
false,
Some(crate::time::ambient_instant() + Duration::from_millis(105)),
false,
&gate,
false,
None,
None,
)
.await;
assert_eq!(
result.map(|response| response.status().as_u16()).ok(),
Some(502),
"the response, not a retry that cannot run"
);
assert_eq!(hits.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn an_origin_failure_after_a_redirect_charges_the_origin() {
let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let dead_port = closed.local_addr().unwrap().port();
drop(closed);
let target = format!("http://127.0.0.1:{dead_port}/t");
let hits = Arc::new(AtomicU32::new(0));
let counter = Arc::clone(&hits);
let app = axum::Router::new().route(
"/r",
axum::routing::get(move || {
let target = target.clone();
let first = counter.fetch_add(1, Ordering::SeqCst) == 0;
async move {
if !first {
std::future::pending::<()>().await;
}
axum::response::Redirect::temporary(&target)
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let origin_port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = Client::with_timeout(Duration::from_millis(300));
let budgets = client.retry.budgets.clone().unwrap();
let result = client
.get(format!("http://127.0.0.1:{origin_port}/r"))
.retries(2)
.send()
.await;
assert!(result.is_err(), "{result:?}");
let full = f64::from(RetryBudgetConfig::default().capacity);
let cost = f64::from(RetryBudgetConfig::default().transient_cost);
let origin = budgets.for_host(&format!("127.0.0.1:{origin_port}"));
let dead = budgets.for_host(&format!("127.0.0.1:{dead_port}"));
assert!(
(full - dead.available() - cost).abs() < f64::EPSILON,
"the dead target pays its one retry: {}",
dead.available()
);
assert!(
(full - origin.available() - cost).abs() < f64::EPSILON,
"the origin pays the retry after its own timeout: {}",
origin.available()
);
}
#[tokio::test]
async fn a_redirect_host_is_refilled_when_the_final_body_fails() {
let target_app = axum::Router::new().route(
"/t",
axum::routing::get(|| async {
axum::body::Body::from_stream(futures::stream::once(async {
Err::<Bytes, _>(std::io::Error::other("broken body"))
}))
}),
);
let target_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let target_port = target_listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(target_listener, target_app).await.unwrap() });
let target = format!("http://127.0.0.1:{target_port}/t");
let origin_app = axum::Router::new().route(
"/r",
axum::routing::get(move || {
let target = target.clone();
async move { axum::response::Redirect::temporary(&target) }
}),
);
let origin_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let origin_port = origin_listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(origin_listener, origin_app).await.unwrap() });
let client = Client::new();
let budgets = client.retry.budgets.clone().unwrap();
let drained = budgets.for_host(&format!("127.0.0.1:{target_port}"));
while drained.try_acquire(RetryKind::Transient) {}
let before = drained.available();
let result = client
.get(format!("http://127.0.0.1:{origin_port}/r"))
.send()
.await;
assert!(result.is_err(), "{result:?}");
assert!(drained.available() > before, "the target got its refill");
}
#[tokio::test]
async fn ten_redirects_are_followed_with_and_without_a_deadline() {
let app = axum::Router::new().route(
"/r/{n}",
axum::routing::get(
|axum::extract::Path(n): axum::extract::Path<u32>| async move {
use axum::response::IntoResponse;
if n == 0 {
"done".into_response()
} else {
axum::response::Redirect::temporary(&format!("/r/{}", n - 1))
.into_response()
}
},
),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let ten = format!("http://127.0.0.1:{port}/r/10");
let eleven = format!("http://127.0.0.1:{port}/r/11");
let plain = Client::new().get(&ten).send().await.unwrap();
assert_eq!(plain.status().as_u16(), 200, "as reqwest's own limit");
let timed = with_deadline(Duration::from_secs(3), Client::new().get(&ten).send())
.await
.unwrap();
assert_eq!(timed.status().as_u16(), 200);
assert!(Client::new().get(&eleven).send().await.is_err());
assert!(
with_deadline(Duration::from_secs(3), Client::new().get(&eleven).send())
.await
.is_err()
);
}
#[tokio::test]
async fn a_caller_deadline_header_is_capped_at_the_time_left() {
let seen = Arc::new(Mutex::new(Vec::new()));
let record = Arc::clone(&seen);
let app = axum::Router::new().route(
"/x",
axum::routing::get(move |headers: HeaderMap| async move {
let values: Vec<String> = headers
.get_all(DEADLINE_HEADER)
.iter()
.map(|value| value.to_str().unwrap().to_owned())
.collect();
record.lock().unwrap().push(values);
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let url = format!("http://127.0.0.1:{port}/x");
let send = Client::new()
.get(&url)
.header(DEADLINE_HEADER, "5000")
.send();
with_deadline(Duration::from_secs(1), send).await.unwrap();
let send = Client::new()
.get(&url)
.header(DEADLINE_HEADER, "200")
.send();
with_deadline(Duration::from_secs(3), send).await.unwrap();
Client::new()
.get(&url)
.header(DEADLINE_HEADER, "5000")
.send()
.await
.unwrap();
let seen = seen.lock().unwrap().clone();
assert_eq!(seen.len(), 3);
assert_eq!(seen[0].len(), 1, "one header: {seen:?}");
assert!(seen[0][0].parse::<u64>().unwrap() <= 1_000, "{seen:?}");
assert_eq!(seen[1], vec!["200".to_owned()]);
assert_eq!(seen[2], vec!["5000".to_owned()]);
}
#[tokio::test]
async fn plain_path_follows_a_redirect_without_a_deadline() {
let (url, seen) = redirecting(Duration::ZERO).await;
let response = Client::new().get(&url).send().await.unwrap();
assert_eq!(response.status().as_u16(), 200);
assert_eq!(seen.lock().unwrap().len(), 2);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn plain_path_stops_a_hanging_call_at_the_deadline() {
let _lock = lock();
let (url, hits) = counting(None, &[]).await;
let result =
with_deadline(Duration::from_millis(300), Client::new().get(&url).send()).await;
assert!(
matches!(result, Err(ClientError::DeadlineExceeded)),
"{result:?}"
);
assert_eq!(
hits.load(Ordering::SeqCst),
1,
"no retry after the deadline"
);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn plain_path_does_not_start_a_retry_that_cannot_fit() {
let _lock = lock();
let (url, hits) = counting(Some(503), &[]).await;
let response = with_deadline(
Duration::from_millis(700),
Client::new().get(&url).retries(50).send(),
)
.await
.unwrap();
assert_eq!(response.status().as_u16(), 503);
let made = hits.load(Ordering::SeqCst);
assert!(
(2..51).contains(&made),
"retries stop at the deadline: {made}"
);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn plain_path_skips_a_retry_after_wait_past_the_deadline() {
let _lock = lock();
let (url, hits) = counting(Some(429), &[("retry-after", "10")]).await;
let response = with_deadline(Duration::from_secs(2), Client::new().get(&url).send())
.await
.unwrap();
assert_eq!(
response.status().as_u16(),
429,
"the 429 comes back at once"
);
assert_eq!(hits.load(Ordering::SeqCst), 1);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn plain_path_obeys_the_retry_budget() {
let _lock = lock();
let (url, hits) = counting(Some(503), &[]).await;
let client = one_retry_client();
let first = client.get(&url).send().await.unwrap();
assert_eq!(first.status().as_u16(), 503);
assert_eq!(hits.load(Ordering::SeqCst), 2, "one retry from the budget");
let second = client.get(&url).send().await.unwrap();
assert_eq!(second.status().as_u16(), 503);
assert_eq!(
hits.load(Ordering::SeqCst),
3,
"the empty budget blocks retries"
);
}
#[tokio::test]
async fn custom_path_obeys_the_deadline_and_the_budget() {
let (url, hits) = counting(Some(503), &[]).await;
let response = with_deadline(
Duration::from_millis(700),
Client::new().get(&url).retries(50).no_redirect().send(),
)
.await
.unwrap();
assert_eq!(response.status().as_u16(), 503);
let made = hits.load(Ordering::SeqCst);
assert!(
(2..51).contains(&made),
"retries stop at the deadline: {made}"
);
let client = one_retry_client();
client.get(&url).no_redirect().send().await.unwrap();
assert_eq!(
hits.load(Ordering::SeqCst),
made + 2,
"one retry from the budget"
);
client.get(&url).no_redirect().send().await.unwrap();
assert_eq!(
hits.load(Ordering::SeqCst),
made + 3,
"the empty budget blocks retries"
);
}
#[tokio::test]
async fn custom_path_reports_a_deadline_timeout_as_deadline_exceeded() {
let (url, hits) = counting(None, &[]).await;
let result = with_deadline(
Duration::from_millis(300),
Client::new().get(&url).no_redirect().send(),
)
.await;
assert!(
matches!(result, Err(ClientError::DeadlineExceeded)),
"{result:?}"
);
assert_eq!(
hits.load(Ordering::SeqCst),
1,
"no retry after the deadline"
);
}
async fn stalled_body() -> String {
let app = axum::Router::new().route(
"/x",
axum::routing::get(|| async {
let stream = futures::stream::pending::<Result<bytes::Bytes, std::io::Error>>();
axum::response::Response::new(axum::body::Body::from_stream(stream))
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
format!("http://127.0.0.1:{}/x", addr.port())
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn a_body_stalled_past_the_deadline_is_deadline_exceeded() {
let _lock = lock();
let url = stalled_body().await;
let plain =
with_deadline(Duration::from_millis(300), Client::new().get(&url).send()).await;
assert!(
matches!(plain, Err(ClientError::DeadlineExceeded)),
"{plain:?}"
);
let custom = with_deadline(
Duration::from_millis(300),
Client::new().get(&url).no_redirect().send(),
)
.await;
assert!(
matches!(custom, Err(ClientError::DeadlineExceeded)),
"{custom:?}"
);
}
#[tokio::test]
async fn ssrf_safe_path_fails_at_once_when_the_deadline_has_passed() {
let result = Deadline::at(tokio::time::Instant::now())
.scope(Client::new().get_ssrf_safe("https://example.com/").send())
.await;
assert!(
matches!(result, Err(ClientError::DeadlineExceeded)),
"{result:?}"
);
}
#[test]
fn a_redirect_hop_uses_the_budget_of_its_own_host() {
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let origin = RetryGate::with_deadline(
None,
Some(Arc::clone(&budgets)),
Some("origin:443"),
true,
);
let hop = origin.for_hop("https://target/x", &mut std::collections::HashSet::new());
assert!(Arc::ptr_eq(
hop.budget.as_ref().unwrap(),
&budgets.for_host("target:443")
));
assert!(!Arc::ptr_eq(
hop.budget.as_ref().unwrap(),
origin.budget.as_ref().unwrap()
));
}
#[test]
fn a_redirect_chain_refills_each_host_once() {
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let origin_budget = budgets.for_host("a:443");
let target_budget = budgets.for_host("b:443");
while origin_budget.try_acquire(RetryKind::Transient) {}
while target_budget.try_acquire(RetryKind::Transient) {}
let origin =
RetryGate::with_deadline(None, Some(Arc::clone(&budgets)), Some("a:443"), true);
let mut refilled = std::collections::HashSet::from(["a:443".to_owned()]);
origin.check().unwrap();
let after_first = origin_budget.available();
let before_b = target_budget.available();
origin
.for_hop("https://b/1", &mut refilled)
.check()
.unwrap();
let after_b = target_budget.available();
assert!(after_b > before_b, "the first visit to b refills it");
origin
.for_hop("https://a/2", &mut refilled)
.check()
.unwrap();
origin
.for_hop("https://b/3", &mut refilled)
.check()
.unwrap();
assert!(
(origin_budget.available() - after_first).abs() < f64::EPSILON,
"the origin is not credited again"
);
assert!(
(target_budget.available() - after_b).abs() < f64::EPSILON,
"the target is credited once"
);
}
#[test]
fn rekey_moves_to_the_budget_of_the_answering_host() {
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let mut gate = RetryGate::with_deadline(
None,
Some(Arc::clone(&budgets)),
Some("origin:443"),
true,
);
let origin = Arc::clone(gate.budget.as_ref().unwrap());
gate.rekey("https://origin/x");
assert!(
Arc::ptr_eq(gate.budget.as_ref().unwrap(), &origin),
"same host"
);
gate.rekey("https://target/x");
assert!(Arc::ptr_eq(
gate.budget.as_ref().unwrap(),
&budgets.for_host("target:443")
));
}
#[test]
fn a_redirected_answer_refills_the_destination_budget() {
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let target = budgets.for_host("target:443");
while target.try_acquire(RetryKind::Transient) {}
let gate = RetryGate::with_deadline(
None,
Some(Arc::clone(&budgets)),
Some("origin:443"),
true,
);
let origin = Arc::clone(gate.budget.as_ref().unwrap());
let (before, origin_before) = (target.available(), origin.available());
gate.record_destination("https://origin/x");
assert!(
(target.available() - before).abs() < f64::EPSILON,
"same host"
);
gate.record_destination("https://target/x");
assert!(target.available() > before, "the destination is refilled");
assert!(
(origin.available() - origin_before).abs() < f64::EPSILON,
"the origin budget is unchanged"
);
assert!(
Arc::ptr_eq(gate.budget.as_ref().unwrap(), &origin),
"no rekey"
);
}
#[tokio::test(start_paused = true)]
async fn an_expired_deadline_reclassifies_a_hop_error() {
let gate = Deadline::at(tokio::time::Instant::now())
.scope(async { RetryGate::start(None, None, true) })
.await;
let error = gate.classify(ClientError::InvalidUrl("dns stalled".into()));
assert!(matches!(error, ClientError::DeadlineExceeded));
let open = RetryGate::start(None, None, true);
let error = open.classify(ClientError::InvalidUrl("dns".into()));
assert!(matches!(error, ClientError::InvalidUrl(_)));
}
#[tokio::test]
async fn a_rejected_ssrf_safe_host_is_not_refilled() {
let client = Client::new();
let budget = client
.retry
.budgets
.clone()
.unwrap()
.for_host("127.0.0.1:80");
while budget.try_acquire(RetryKind::Transient) {}
let empty = budget.available();
let result = client.get_ssrf_safe("http://127.0.0.1/").send().await;
assert!(result.is_err(), "{result:?}");
assert!(
(budget.available() - empty).abs() < f64::EPSILON,
"no refill without an attempt: {}",
budget.available()
);
}
#[tokio::test(start_paused = true)]
async fn a_followed_redirect_host_is_refilled_after_the_deadline() {
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let target = budgets.for_host("t:80");
while target.try_acquire(RetryKind::Transient) {}
let empty = target.available();
let gate = RetryGate::with_deadline(
Some(Deadline::after(Duration::from_millis(1))),
Some(budgets),
Some("o:80"),
true,
);
tokio::time::advance(Duration::from_millis(5)).await;
let mut seen = std::collections::HashSet::from(["o:80".to_owned()]);
gate.record_destinations(&["http://t/".to_owned()], &mut seen);
assert!(
target.available() > empty,
"the target's attempt started, so it is refilled"
);
}
#[tokio::test(start_paused = true)]
async fn a_request_past_its_deadline_does_not_refill_the_budget() {
let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
let budget = budgets.for_host("h:80");
while budget.try_acquire(RetryKind::Transient) {}
let empty = budget.available();
let late = RetryGate::with_deadline(
Some(Deadline::at(tokio::time::Instant::now())),
Some(Arc::clone(&budgets)),
Some("h:80"),
true,
);
assert!(late.check().is_err());
assert!(
(budget.available() - empty).abs() < f64::EPSILON,
"no refill"
);
let short = RetryGate::with_deadline(
Some(Deadline::after(Duration::from_millis(1))),
Some(Arc::clone(&budgets)),
Some("h:80"),
true,
);
tokio::time::advance(Duration::from_millis(5)).await;
assert!(short.check().is_err());
assert!(
(budget.available() - empty).abs() < f64::EPSILON,
"no refill without an attempt"
);
let live = RetryGate::with_deadline(None, Some(budgets), Some("h:80"), true);
assert!(
(budget.available() - empty).abs() < f64::EPSILON,
"not when built"
);
live.check().unwrap();
let once = budget.available();
assert!(once > empty, "a live request refills at its first attempt");
live.check().unwrap();
assert!((budget.available() - once).abs() < f64::EPSILON, "once");
}
#[test]
fn budget_keys_keep_the_port() {
assert_eq!(url_host("http://svc:8001/a").as_deref(), Some("svc:8001"));
assert_eq!(url_host("https://svc/a").as_deref(), Some("svc:443"));
assert_eq!(url_host("http://svc/a").as_deref(), Some("svc:80"));
assert_eq!(url_host("/relative"), None);
}
#[test]
fn a_retry_ending_in_a_client_or_server_error_keeps_its_tokens_spent() {
let budget = Arc::new(RetryBudget::new(&RetryBudgetConfig::default()));
let gate = RetryGate {
deadline: None,
budgets: None,
budget: Some(Arc::clone(&budget)),
host: None,
send_header: true,
refill_pending: AtomicBool::new(false),
pending_retry: std::sync::Mutex::new(None),
caller_deadline: std::sync::OnceLock::new(),
signer: None,
};
assert!(gate.allow(RetryKind::Transient, Duration::ZERO));
gate.check().unwrap();
gate.finish(Some(RetryKind::Transient), 500);
assert!((budget.available() - 486.0).abs() < f64::EPSILON);
gate.finish(Some(RetryKind::Transient), 200);
assert!((budget.available() - 500.0).abs() < f64::EPSILON);
}
}
}