1#![cfg_attr(not(test), deny(clippy::disallowed_methods))]
56
57use std::collections::HashMap;
58use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
59use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
60use std::sync::{Arc, Mutex};
61use std::time::{Duration, Instant};
62
63use bytes::Bytes;
64use reqwest::Method;
65use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
66use serde::Serialize;
67use serde::de::DeserializeOwned;
68
69use crate::config::RetryBudgetConfig;
70use crate::deadline::{DEADLINE_HEADER, Deadline};
71use crate::retry_budget::{RetryBudget, RetryBudgets, RetryKind};
72
73#[non_exhaustive]
80#[derive(Debug, thiserror::Error)]
81pub enum ClientError {
82 #[error("outbound HTTP request failed: {0}")]
84 Request(#[from] reqwest::Error),
85 #[error("JSON error: {0}")]
87 Json(#[from] serde_json::Error),
88 #[error("no mock registered for {0} {1}")]
90 NoMock(String, String),
91 #[error("outbound circuit breaker is open")]
93 CircuitBreakerOpen,
94 #[error("{0}")]
97 FaultInjected(String),
98 #[error("outbound request to {host} throttled locally: the host rejects too many requests")]
102 ThrottledLocally {
103 host: String,
105 },
106 #[error("{0}")]
120 ReplayedRequestFailure(String),
121 #[error(
124 "outbound HTTP is not recorded in a failure capsule and is blocked during replay: {0} {1}"
125 )]
126 BlockedDuringReplay(String, String),
127 #[error("SSRF policy blocked address: {0}")]
131 SsrfBlocked(String),
132 #[error("too many redirects (max {0})")]
134 TooManyRedirects(usize),
135 #[error("redirect rejected: {0}")]
138 RedirectRejected(String),
139 #[error("invalid or unresolvable URL: {0}")]
142 InvalidUrl(String),
143 #[error("{0}")]
152 IncompatiblePinRedirect(&'static str),
153 #[error("{0}")]
162 PinRequiresDomainHost(&'static str),
163 #[error("{0}")]
170 PinNotAllowedWithSsrfSafe(&'static str),
171 #[error("simulated network: {0}")]
175 SimNetwork(String),
176
177 #[error("request deadline exceeded before the outbound request could start")]
180 DeadlineExceeded,
181}
182
183#[derive(Debug)]
190pub struct Response {
191 status: reqwest::StatusCode,
192 headers: HeaderMap,
193 body: Bytes,
194 url: Option<reqwest::Url>,
195}
196
197impl Response {
198 pub const fn status(&self) -> reqwest::StatusCode {
200 self.status
201 }
202
203 pub const fn headers(&self) -> &HeaderMap {
205 &self.headers
206 }
207
208 pub fn is_success(&self) -> bool {
210 self.status.is_success()
211 }
212
213 pub const fn url(&self) -> Option<&reqwest::Url> {
215 self.url.as_ref()
216 }
217
218 pub fn json<T: DeserializeOwned>(self) -> Result<T, ClientError> {
223 serde_json::from_slice(&self.body).map_err(ClientError::Json)
224 }
225
226 pub fn text(self) -> String {
228 String::from_utf8_lossy(&self.body).into_owned()
229 }
230
231 pub fn bytes(self) -> Bytes {
233 self.body
234 }
235}
236
237#[must_use]
254pub fn is_blocked_ip(ip: IpAddr) -> bool {
255 match ip {
256 IpAddr::V4(v4) => is_blocked_ipv4(v4),
257 IpAddr::V6(v6) => is_blocked_ipv6(v6),
258 }
259}
260
261#[must_use]
264pub fn is_public_ip(ip: IpAddr) -> bool {
265 !is_blocked_ip(ip)
266}
267
268fn is_blocked_ipv4(ip: Ipv4Addr) -> bool {
269 let [a, b, c, _d] = ip.octets();
270 if a == 0 {
272 return true;
273 }
274 if a == 10 {
276 return true;
277 }
278 if a == 100 && (64..=127).contains(&b) {
280 return true;
281 }
282 if a == 127 {
284 return true;
285 }
286 if a == 169 && b == 254 {
288 return true;
289 }
290 if a == 172 && (16..=31).contains(&b) {
292 return true;
293 }
294 if a == 192 && b == 0 && c == 0 {
296 return true;
297 }
298 if a == 192 && b == 0 && c == 2 {
300 return true;
301 }
302 if a == 192 && b == 88 && c == 99 {
304 return true;
305 }
306 if a == 192 && b == 168 {
308 return true;
309 }
310 if a == 198 && (18..=19).contains(&b) {
312 return true;
313 }
314 if a == 198 && b == 51 && c == 100 {
316 return true;
317 }
318 if a == 203 && b == 0 && c == 113 {
320 return true;
321 }
322 if a >= 224 {
324 return true;
325 }
326 false
327}
328
329fn embedded_ipv4(ip: Ipv6Addr) -> Option<Ipv4Addr> {
347 if let Some(v4) = ip.to_ipv4_mapped() {
348 return Some(v4);
349 }
350 let segs = ip.segments();
351 if segs[0] == 0
358 && segs[1] == 0
359 && segs[2] == 0
360 && segs[3] == 0
361 && segs[4] == 0xffff
362 && segs[5] == 0
363 {
364 let o = ip.octets();
365 return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
366 }
367 let o = ip.octets();
368 if o[0] == 0x00
371 && o[1] == 0x64
372 && o[2] == 0xff
373 && o[3] == 0x9b
374 && o[4..12].iter().all(|&b| b == 0)
375 {
376 return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
377 }
378 if o[0] == 0x20 && o[1] == 0x02 {
380 return Some(Ipv4Addr::new(o[2], o[3], o[4], o[5]));
381 }
382 ip.to_ipv4()
384}
385
386fn is_blocked_ipv6(ip: Ipv6Addr) -> bool {
387 if let Some(v4) = embedded_ipv4(ip)
394 && is_blocked_ipv4(v4)
395 {
396 return true;
397 }
398
399 let segs = ip.segments();
400 if segs[0] == 0x0064 && segs[1] == 0xff9b && segs[2] == 0x0001 {
407 return true;
408 }
409 if ip == Ipv6Addr::UNSPECIFIED {
411 return true;
412 }
413 if ip == Ipv6Addr::LOCALHOST {
415 return true;
416 }
417 if segs[0] & 0xfe00 == 0xfc00 {
419 return true;
420 }
421 if segs[0] & 0xffc0 == 0xfe80 {
423 return true;
424 }
425 if segs[0] & 0xffc0 == 0xfec0 {
427 return true;
428 }
429 if segs[0] & 0xff00 == 0xff00 {
431 return true;
432 }
433 if segs[0] == 0x2001 && segs[1] == 0x0db8 {
435 return true;
436 }
437
438 let s = segs;
454 if s[0] == 0x0100 && s[1] == 0 && s[2] == 0 && s[3] == 0 {
456 return true;
457 }
458 if s[0] == 0x2001 && s[1] == 0x0002 && s[2] == 0x0000 {
460 return true;
461 }
462 if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0010 {
464 return true;
465 }
466 if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0020 {
468 return true;
469 }
470 if s[0] == 0x3fff && (s[1] & 0xF000) == 0x0000 {
472 return true;
473 }
474 if s[0] == 0x5f00 {
476 return true;
477 }
478 if s[0] == 0x2620 && s[1] == 0x004f && s[2] == 0x8000 {
480 return true;
481 }
482
483 false
484}
485
486#[derive(Clone, Debug)]
493pub struct RetryPolicy {
494 pub max_retries: u32,
497 pub retry_idempotent_only: bool,
501 pub max_retry_after: Duration,
504 pub request_timeout: Option<Duration>,
506 pub max_backoff: Duration,
508}
509
510impl Default for RetryPolicy {
511 fn default() -> Self {
512 Self {
513 max_retries: 3,
514 retry_idempotent_only: true,
515 max_retry_after: Duration::from_secs(10),
516 request_timeout: Some(Duration::from_secs(30)),
517 max_backoff: Duration::from_millis(crate::backoff::DEFAULT_HTTP_MAX_BACKOFF_MS),
518 }
519 }
520}
521
522const BASE_BACKOFF: Duration = Duration::from_millis(100);
524
525const IDEMPOTENCY_KEY: &str = "idempotency-key";
527
528impl RetryPolicy {
529 fn retry_delay(
537 &self,
538 entropy: &dyn crate::entropy::Entropy,
539 attempt: u32,
540 hint: Option<Duration>,
541 ) -> Duration {
542 let backoff = crate::backoff::full_jitter(entropy, BASE_BACKOFF, self.max_backoff, attempt);
543 let Some(mut hint) = hint else {
544 return backoff;
545 };
546 hint = hint.min(self.max_retry_after);
547 if let Some(timeout) = self.request_timeout {
548 hint = hint.min(timeout);
549 }
550 crate::backoff::retry_after_wait(hint, backoff)
551 }
552}
553
554const fn is_retryable_response(status: u16) -> bool {
556 status == 429 || is_retryable_status(status)
557}
558
559fn retry_hint(status: u16, headers: &HeaderMap) -> Option<Duration> {
562 match status {
563 429 => Some(parse_retry_after(headers).unwrap_or(Duration::from_secs(1))),
564 503 => parse_retry_after(headers),
565 _ => None,
566 }
567}
568
569pub(crate) struct MockEntry {
573 pub(crate) method: Option<Method>,
574 pub(crate) path: String,
576 pub(crate) alias: Option<String>,
578 pub(crate) status: u16,
579 pub(crate) body: Option<serde_json::Value>,
580 pub(crate) call_count: Arc<AtomicUsize>,
581}
582
583pub(crate) struct MockResponse {
585 pub(crate) status: u16,
586 pub(crate) body: Option<serde_json::Value>,
587}
588
589pub struct MockRegistry {
595 entries: Mutex<Vec<MockEntry>>,
596}
597
598impl MockRegistry {
599 #[must_use]
601 pub const fn new() -> Self {
602 Self {
603 entries: Mutex::new(Vec::new()),
604 }
605 }
606
607 pub(crate) fn register(&self, entry: MockEntry) {
609 self.entries
610 .lock()
611 .expect("mock registry lock poisoned")
612 .push(entry);
613 }
614
615 pub(crate) fn find_match(
618 &self,
619 method: &Method,
620 url: &str,
621 alias: Option<&str>,
622 ) -> Option<MockResponse> {
623 let url_path_owned: String = reqwest::Url::parse(url).map_or_else(
629 |_| {
630 let s = url.split_once('?').map_or(url, |(p, _)| p);
631 s.split_once('#').map_or(s, |(p, _)| p).to_owned()
632 },
633 |parsed| parsed.path().to_owned(),
634 );
635 let url_path = url_path_owned.as_str();
636
637 let found = {
639 let entries = self.entries.lock().expect("mock registry lock poisoned");
640 entries.iter().find_map(|entry| {
641 let method_ok = entry.method.as_ref().is_none_or(|m| m == method);
642 let path_ok = url_path == entry.path.as_str()
647 || url_path
648 .strip_suffix(entry.path.as_str())
649 .is_some_and(|prefix| {
650 prefix.is_empty()
651 || prefix.ends_with('/')
652 || entry.path.starts_with('/')
653 });
654 let alias_ok = entry
655 .alias
656 .as_deref()
657 .is_none_or(|a| alias.is_some_and(|b| a == b));
658 if method_ok && path_ok && alias_ok {
659 Some((entry.call_count.clone(), entry.status, entry.body.clone()))
660 } else {
661 None
662 }
663 })
664 };
665
666 found.map(|(call_count, status, body)| {
667 call_count.fetch_add(1, Ordering::SeqCst);
668 MockResponse { status, body }
669 })
670 }
671}
672
673impl Default for MockRegistry {
674 fn default() -> Self {
675 Self::new()
676 }
677}
678
679static OUTBOUND_BLOCKED: AtomicBool = AtomicBool::new(false);
682
683#[cfg_attr(not(feature = "reporting"), allow(dead_code))]
692pub(crate) fn block_outbound_for_replay() {
693 OUTBOUND_BLOCKED.store(true, Ordering::SeqCst);
694}
695
696#[must_use]
698pub(crate) fn outbound_blocked_for_replay() -> bool {
699 OUTBOUND_BLOCKED.load(Ordering::SeqCst)
700}
701
702#[cfg(feature = "reporting")]
710const OPAQUE_HEADER_PREFIX: &str = "[bytes] ";
711
712#[cfg(feature = "reporting")]
722fn caller_headers(builder: &RequestBuilder) -> Vec<(String, String)> {
723 builder
724 .extra_headers
725 .iter()
726 .filter(|(name, _)| {
727 builder.web_bot_auth.is_none()
728 || !matches!(name.as_str(), "signature" | "signature-input")
729 })
730 .map(|(name, value)| {
731 (
732 name.as_str().to_owned(),
733 match value.to_str() {
737 Ok(text) if !text.starts_with(OPAQUE_HEADER_PREFIX) => text.to_owned(),
738 _ => format!("{OPAQUE_HEADER_PREFIX}{}", value.as_bytes().escape_ascii()),
739 },
740 )
741 })
742 .collect()
743}
744
745#[cfg(feature = "reporting")]
750const fn http_error_kind(error: &ClientError) -> crate::capsule::schema::HttpErrorKind {
751 use crate::capsule::schema::HttpErrorKind as Kind;
752 match error {
753 ClientError::Json(_) => Kind::Json,
754 ClientError::NoMock(..) => Kind::NoMock,
755 ClientError::CircuitBreakerOpen => Kind::CircuitBreakerOpen,
756 ClientError::ThrottledLocally { .. } => Kind::ThrottledLocally,
757 ClientError::SsrfBlocked(_) => Kind::SsrfBlocked,
758 ClientError::TooManyRedirects(_) => Kind::TooManyRedirects,
759 ClientError::RedirectRejected(_) => Kind::RedirectRejected,
760 ClientError::InvalidUrl(_) => Kind::InvalidUrl,
761 ClientError::FaultInjected(_) => Kind::FaultInjected,
762 _ => Kind::Transport,
765 }
766}
767
768#[cfg(feature = "reporting")]
776fn rebuild_client_error(
777 kind: Option<crate::capsule::schema::HttpErrorKind>,
778 text: String,
779) -> ClientError {
780 use crate::capsule::schema::HttpErrorKind as Kind;
781 match kind {
782 Some(Kind::CircuitBreakerOpen) => ClientError::CircuitBreakerOpen,
783 Some(Kind::ThrottledLocally) => text
784 .strip_prefix("outbound request to ")
785 .and_then(|rest| rest.split_once(" throttled locally"))
786 .map_or_else(
787 || ClientError::ReplayedRequestFailure(text.clone()),
788 |(host, _)| ClientError::ThrottledLocally {
789 host: host.to_owned(),
790 },
791 ),
792 Some(Kind::SsrfBlocked) => {
793 ClientError::SsrfBlocked(strip_prefix_payload(&text, "SSRF policy blocked address: "))
794 }
795 Some(Kind::RedirectRejected) => {
796 ClientError::RedirectRejected(strip_prefix_payload(&text, "redirect rejected: "))
797 }
798 Some(Kind::InvalidUrl) => {
799 ClientError::InvalidUrl(strip_prefix_payload(&text, "invalid or unresolvable URL: "))
800 }
801 Some(Kind::FaultInjected) => ClientError::FaultInjected(text),
802 Some(Kind::TooManyRedirects) => text
803 .trim_end_matches(')')
804 .rsplit_once("(max ")
805 .and_then(|(_, hops)| hops.parse().ok())
806 .map_or_else(
807 || ClientError::ReplayedRequestFailure(text.clone()),
808 ClientError::TooManyRedirects,
809 ),
810 Some(Kind::Json | Kind::NoMock | Kind::Transport) | None => {
814 ClientError::ReplayedRequestFailure(text)
815 }
816 }
817}
818
819#[cfg(feature = "reporting")]
825fn strip_prefix_payload(text: &str, prefix: &str) -> String {
826 text.strip_prefix(prefix).unwrap_or(text).to_owned()
827}
828
829#[cfg(feature = "reporting")]
833fn replayed_response(
834 tape: &crate::capsule::effects::ReplayEffects,
835 request: &crate::capsule::effects::OutboundRequest<'_>,
836) -> Result<Response, ClientError> {
837 let Some(recorded) = tape.next_http(request) else {
838 return Err(ClientError::BlockedDuringReplay(
840 request.method.to_owned(),
841 crate::capsule::redact::hide_url_userinfo(request.url),
842 ));
843 };
844 if let Some(error) = recorded.error {
845 return Err(rebuild_client_error(recorded.error_kind, error));
846 }
847 let mut headers = HeaderMap::new();
848 for (name, value) in &recorded.response_headers {
849 if let (Ok(name), Ok(value)) = (
850 HeaderName::from_bytes(name.as_bytes()),
851 HeaderValue::from_str(value),
852 ) {
853 headers.append(name, value);
854 }
855 }
856 Ok(Response {
857 status: reqwest::StatusCode::from_u16(recorded.status)
858 .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR),
859 headers,
860 body: Bytes::from(crate::capsule::effects::body_bytes(&recorded.response_body)),
861 url: recorded
864 .final_url
865 .as_deref()
866 .or(Some(request.url))
867 .and_then(|url| url::Url::parse(url).ok()),
868 })
869}
870
871#[cfg(feature = "reporting")]
879struct OutboundRecorder {
880 scope: Option<Arc<crate::capsule::CaptureScope>>,
881 slot: Option<usize>,
885 max_body_bytes: usize,
887 method: String,
888 url: String,
889 request_headers: Vec<(String, String)>,
890 request_body: crate::capsule::CapsuleBody,
891}
892
893#[cfg(feature = "reporting")]
894impl OutboundRecorder {
895 fn arm(builder: &RequestBuilder) -> Self {
903 let scope = crate::capsule::current_scope();
904 let Some(scope) = scope else {
905 return Self {
906 scope: None,
907 slot: None,
908 max_body_bytes: 0,
909 method: String::new(),
910 url: String::new(),
911 request_headers: Vec::new(),
912 request_body: crate::capsule::CapsuleBody::Absent,
913 };
914 };
915 let max_body_bytes = scope.settings().max_body_bytes;
916 let slot = scope.reserve_http();
917 let request_headers = caller_headers(builder);
918 Self {
919 scope: Some(scope),
920 slot,
921 max_body_bytes,
922 method: builder.method.to_string(),
923 url: builder.url.clone(),
924 request_headers,
925 request_body: builder
926 .body
927 .as_ref()
928 .map_or(crate::capsule::CapsuleBody::Absent, |body| {
929 encode_body(body, max_body_bytes)
930 }),
931 }
932 }
933
934 fn finish(self, result: &Result<Response, ClientError>) {
936 let (Some(scope), Some(slot)) = (self.scope, self.slot) else {
937 return;
938 };
939 let (status, response_headers, response_body, error, final_url) = match result {
940 Ok(response) => (
941 response.status.as_u16(),
942 response
943 .headers
944 .iter()
945 .map(|(name, value)| {
946 (
953 name.as_str().to_owned(),
954 value.to_str().map_or_else(
955 |_| String::from_utf8_lossy(value.as_bytes()).into_owned(),
956 ToOwned::to_owned,
957 ),
958 )
959 })
960 .collect(),
961 encode_body(&response.body, self.max_body_bytes),
962 None,
963 response
964 .url()
965 .map(reqwest::Url::as_str)
966 .filter(|final_url| *final_url != self.url)
967 .map(ToOwned::to_owned),
968 ),
969 Err(error) => (
970 0,
971 Vec::new(),
972 crate::capsule::CapsuleBody::Absent,
973 Some((error.to_string(), http_error_kind(error))),
974 None,
975 ),
976 };
977 let (error, error_kind) = match error {
978 Some((text, kind)) => (Some(text), Some(kind)),
979 None => (None, None),
980 };
981 scope.fill_http(
982 slot,
983 crate::capsule::HttpEffect {
984 method: self.method,
985 url: self.url,
986 request_headers: self.request_headers,
987 request_body: self.request_body,
988 status,
989 response_headers,
990 final_url,
991 response_body,
992 error,
993 error_kind,
994 },
995 );
996 }
997}
998
999#[cfg(feature = "reporting")]
1004fn encode_body(bytes: &Bytes, max_body_bytes: usize) -> crate::capsule::CapsuleBody {
1005 if bytes.is_empty() {
1006 return crate::capsule::CapsuleBody::Absent;
1007 }
1008 if bytes.len() > max_body_bytes {
1012 return crate::capsule::CapsuleBody::Skipped {
1013 declared_len: Some(bytes.len()),
1014 };
1015 }
1016 std::str::from_utf8(bytes).map_or_else(
1017 |_| {
1018 use base64::Engine as _;
1019 crate::capsule::CapsuleBody::Base64(
1020 base64::engine::general_purpose::STANDARD.encode(bytes),
1021 )
1022 },
1023 |text| crate::capsule::CapsuleBody::Text(text.to_owned()),
1024 )
1025}
1026
1027#[cfg(not(feature = "reporting"))]
1030struct OutboundRecorder;
1031
1032#[cfg(not(feature = "reporting"))]
1033impl OutboundRecorder {
1034 const fn arm(_builder: &RequestBuilder) -> Self {
1035 Self
1036 }
1037
1038 const fn finish(self, _result: &Result<Response, ClientError>) {}
1039}
1040
1041pub struct HttpMockRegistryExt(pub Arc<MockRegistry>);
1044
1045#[derive(Clone)]
1055pub(crate) struct SharedReqwestClient {
1056 pub(crate) client: reqwest::Client,
1057 pub(crate) timeout_secs: u64,
1058}
1059
1060pub struct MockHandle {
1063 alias: String,
1064 method: String,
1065 path: String,
1066 call_count: Arc<AtomicUsize>,
1067}
1068
1069impl MockHandle {
1070 pub fn expect_called(&self, expected: usize) {
1076 let actual = self.call_count.load(Ordering::SeqCst);
1077 assert_eq!(
1078 actual, expected,
1079 "http mock for {} {} {} expected {} call(s) but got {}",
1080 self.alias, self.method, self.path, expected, actual,
1081 );
1082 }
1083
1084 #[must_use]
1086 pub fn call_count(&self) -> usize {
1087 self.call_count.load(Ordering::SeqCst)
1088 }
1089}
1090
1091pub struct MockSetupBuilder {
1097 pub(crate) registry: Arc<MockRegistry>,
1098 pub(crate) alias: String,
1099 pub(crate) method: Option<Method>,
1100 pub(crate) path: Option<String>,
1101}
1102
1103impl MockSetupBuilder {
1104 #[must_use]
1106 pub fn get(mut self, path: &str) -> Self {
1107 self.method = Some(Method::GET);
1108 self.path = Some(path.to_owned());
1109 self
1110 }
1111 #[must_use]
1113 pub fn post(mut self, path: &str) -> Self {
1114 self.method = Some(Method::POST);
1115 self.path = Some(path.to_owned());
1116 self
1117 }
1118 #[must_use]
1120 pub fn put(mut self, path: &str) -> Self {
1121 self.method = Some(Method::PUT);
1122 self.path = Some(path.to_owned());
1123 self
1124 }
1125 #[must_use]
1127 pub fn patch(mut self, path: &str) -> Self {
1128 self.method = Some(Method::PATCH);
1129 self.path = Some(path.to_owned());
1130 self
1131 }
1132 #[must_use]
1134 pub fn delete(mut self, path: &str) -> Self {
1135 self.method = Some(Method::DELETE);
1136 self.path = Some(path.to_owned());
1137 self
1138 }
1139
1140 #[must_use]
1142 pub fn head(mut self, path: &str) -> Self {
1143 self.method = Some(Method::HEAD);
1144 self.path = Some(path.to_owned());
1145 self
1146 }
1147
1148 #[must_use]
1153 pub fn respond_with(self, status: u16, body: serde_json::Value) -> MockHandle {
1154 let path = self.path.clone().unwrap_or_default();
1155 let method_str = self
1156 .method
1157 .as_ref()
1158 .map_or_else(|| "*".to_owned(), ToString::to_string);
1159 let call_count = Arc::new(AtomicUsize::new(0));
1160
1161 self.registry.register(MockEntry {
1162 method: self.method,
1163 path: path.clone(),
1164 alias: Some(self.alias.clone()),
1165 status,
1166 body: Some(body),
1167 call_count: call_count.clone(),
1168 });
1169
1170 MockHandle {
1171 alias: self.alias,
1172 method: method_str,
1173 path,
1174 call_count,
1175 }
1176 }
1177
1178 #[must_use]
1183 pub fn respond_with_status(self, status: u16) -> MockHandle {
1184 let path = self.path.clone().unwrap_or_default();
1185 let method_str = self
1186 .method
1187 .as_ref()
1188 .map_or_else(|| "*".to_owned(), ToString::to_string);
1189 let call_count = Arc::new(AtomicUsize::new(0));
1190
1191 self.registry.register(MockEntry {
1192 method: self.method,
1193 path: path.clone(),
1194 alias: Some(self.alias.clone()),
1195 status,
1196 body: None,
1197 call_count: call_count.clone(),
1198 });
1199
1200 MockHandle {
1201 alias: self.alias,
1202 method: method_str,
1203 path,
1204 call_count,
1205 }
1206 }
1207}
1208
1209#[derive(Clone)]
1237pub struct Client {
1238 inner: reqwest::Client,
1239 alias: Option<String>,
1241 base_url: Option<String>,
1243 base_urls: HashMap<String, String>,
1245 retry_policy: RetryPolicy,
1246 mock: Option<Arc<MockRegistry>>,
1248 resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
1250 sim_net: Option<Arc<crate::sim::SimNet>>,
1253 retry: RetrySettings,
1255 entropy: Arc<dyn crate::entropy::Entropy>,
1258 throttle: Option<Arc<crate::admission::AdaptiveThrottle>>,
1260}
1261
1262#[derive(Clone)]
1266pub(crate) struct SharedThrottle {
1267 throttle: Arc<crate::admission::AdaptiveThrottle>,
1268 config: crate::config::AdaptiveThrottleConfig,
1269}
1270
1271pub(crate) fn install_shared_throttle(
1274 state: &crate::AppState,
1275 config: &crate::config::HttpClientConfig,
1276) {
1277 let _ = shared_throttle(state, config);
1278}
1279
1280static THROTTLE_REPLACE: std::sync::Mutex<()> = std::sync::Mutex::new(());
1282
1283fn shared_throttle(
1289 state: &crate::AppState,
1290 config: &crate::config::HttpClientConfig,
1291) -> Option<Arc<crate::admission::AdaptiveThrottle>> {
1292 let settings = config.adaptive_throttle;
1293 if !settings.enabled {
1294 return None;
1295 }
1296 let current = || {
1297 state
1298 .extension::<SharedThrottle>()
1299 .filter(|shared| shared.config == settings)
1300 .map(|shared| Arc::clone(&shared.throttle))
1301 };
1302 if let Some(throttle) = current() {
1303 return Some(throttle);
1304 }
1305 let _guard = THROTTLE_REPLACE
1308 .lock()
1309 .unwrap_or_else(std::sync::PoisonError::into_inner);
1310 if let Some(throttle) = current() {
1311 return Some(throttle);
1312 }
1313 let throttle = throttle_from_config(config)?;
1314 state.insert_extension(SharedThrottle {
1315 throttle: Arc::clone(&throttle),
1316 config: settings,
1317 });
1318 Some(throttle)
1319}
1320
1321fn throttle_from_config(
1323 config: &crate::config::HttpClientConfig,
1324) -> Option<Arc<crate::admission::AdaptiveThrottle>> {
1325 let t = config.adaptive_throttle;
1326 t.enabled.then(|| {
1327 Arc::new(crate::admission::AdaptiveThrottle::new(
1328 t.k,
1329 Duration::from_secs(t.window_secs),
1330 ))
1331 })
1332}
1333
1334fn os_entropy() -> Arc<dyn crate::entropy::Entropy> {
1336 Arc::new(crate::entropy::OsEntropy)
1337}
1338
1339impl Client {
1340 #[must_use]
1343 pub fn new() -> Self {
1344 Self::with_timeout(Duration::from_secs(30))
1345 }
1346
1347 #[must_use]
1354 pub fn with_timeout(timeout: Duration) -> Self {
1355 let inner = reqwest::ClientBuilder::new()
1356 .timeout(timeout)
1357 .redirect(pooled_redirect_policy())
1358 .build()
1359 .expect("failed to build reqwest client");
1360 Self {
1361 inner,
1362 alias: None,
1363 base_url: None,
1364 base_urls: HashMap::new(),
1365 retry_policy: RetryPolicy {
1366 request_timeout: Some(timeout),
1367 ..RetryPolicy::default()
1368 },
1369 mock: None,
1370 resilience_config: None,
1371 sim_net: None,
1372 retry: RetrySettings::standalone(),
1373 entropy: os_entropy(),
1374 throttle: None,
1375 }
1376 }
1377
1378 pub(crate) fn build_inner(config: &crate::config::HttpClientConfig) -> reqwest::Client {
1388 reqwest::ClientBuilder::new()
1389 .timeout(Duration::from_secs(config.timeout_secs))
1390 .redirect(pooled_redirect_policy())
1391 .build()
1392 .expect("failed to build reqwest client")
1393 }
1394
1395 fn from_config_with_inner(
1399 inner: reqwest::Client,
1400 config: &crate::config::HttpClientConfig,
1401 ) -> Self {
1402 let timeout = Duration::from_secs(config.timeout_secs);
1403 Self {
1404 inner,
1405 alias: None,
1406 base_url: None,
1407 base_urls: config.base_urls.clone(),
1408 retry_policy: RetryPolicy {
1409 max_retries: config.max_retries,
1410 retry_idempotent_only: true,
1411 max_retry_after: Duration::from_secs(config.max_retry_after_secs),
1412 request_timeout: Some(timeout),
1413 max_backoff: Duration::from_millis(config.max_backoff_ms),
1414 },
1415 mock: None,
1416 resilience_config: None,
1417 sim_net: None,
1418 retry: RetrySettings {
1420 budgets: None,
1421 send_deadline_header: config.send_deadline_header,
1422 },
1423 entropy: os_entropy(),
1424 throttle: throttle_from_config(config),
1425 }
1426 }
1427
1428 fn with_inner(inner: reqwest::Client) -> Self {
1432 Self {
1433 inner,
1434 alias: None,
1435 base_url: None,
1436 base_urls: HashMap::new(),
1437 retry_policy: RetryPolicy::default(),
1438 mock: None,
1439 resilience_config: None,
1440 sim_net: None,
1441 retry: RetrySettings {
1443 budgets: None,
1444 send_deadline_header: true,
1445 },
1446 entropy: os_entropy(),
1447 throttle: None,
1448 }
1449 }
1450
1451 #[must_use]
1458 pub fn from_config(config: &crate::config::HttpClientConfig) -> Self {
1459 let mut client = Self::from_config_with_inner(Self::build_inner(config), config);
1460 client.retry.budgets = config
1461 .retry_budget
1462 .enabled
1463 .then(|| Arc::new(RetryBudgets::new(&config.retry_budget)));
1464 client
1465 }
1466
1467 pub(crate) fn with_mock(mut self, registry: Arc<MockRegistry>) -> Self {
1469 self.mock = Some(registry);
1470 self
1471 }
1472
1473 #[must_use]
1484 pub fn from_state(state: &crate::AppState) -> Self {
1485 let autumn_config = state.extension::<crate::config::AutumnConfig>();
1486 let config = state
1487 .extension::<crate::config::HttpConfig>()
1488 .or_else(|| autumn_config.as_ref().map(|c| Arc::new(c.http.clone())));
1489
1490 let effective_timeout_secs = config.as_ref().map_or_else(
1496 || crate::config::HttpClientConfig::default().timeout_secs,
1497 |c| c.client.timeout_secs,
1498 );
1499 let shared = state.extension::<SharedReqwestClient>().and_then(|s| {
1500 if s.timeout_secs == effective_timeout_secs {
1501 Some(s.client.clone())
1502 } else {
1503 None
1504 }
1505 });
1506
1507 let budget_config = config
1508 .as_ref()
1509 .map_or_else(RetryBudgetConfig::default, |cfg| {
1510 cfg.client.retry_budget.clone()
1511 });
1512 let throttle = config
1514 .as_ref()
1515 .and_then(|cfg| shared_throttle(state, &cfg.client));
1516 let mut client = match (config, shared) {
1517 (Some(cfg), Some(inner)) => Self::from_config_with_inner(inner, &cfg.client),
1518 (Some(cfg), None) => Self::from_config(&cfg.client),
1519 (None, Some(inner)) => Self::with_inner(inner),
1520 (None, None) => Self::new(),
1521 };
1522
1523 client.retry.budgets = shared_retry_budgets(state, &budget_config);
1526
1527 client.resilience_config = autumn_config.map(|c| Arc::new(c.resilience.clone()));
1528
1529 if let Some(ext) = state.extension::<HttpMockRegistryExt>() {
1530 client = client.with_mock(ext.0.clone());
1531 }
1532 client.sim_net = state.extension::<crate::sim::SimNet>();
1533 client.throttle = throttle;
1534 let entropy = state.entropy_arc();
1537 client.entropy = entropy.unrecorded().unwrap_or(entropy);
1538
1539 client
1540 }
1541
1542 #[must_use]
1549 pub fn named(&self, alias: &str) -> Self {
1550 let base_url = self
1551 .base_urls
1552 .get(alias)
1553 .cloned()
1554 .or_else(|| self.base_url.clone());
1555 Self {
1556 inner: self.inner.clone(),
1557 alias: Some(alias.to_owned()),
1558 base_url,
1559 base_urls: self.base_urls.clone(),
1560 retry_policy: self.retry_policy.clone(),
1561 mock: self.mock.clone(),
1562 resilience_config: self.resilience_config.clone(),
1563 sim_net: self.sim_net.clone(),
1564 retry: self.retry.clone(),
1565 entropy: self.entropy.clone(),
1566 throttle: self.throttle.clone(),
1567 }
1568 }
1569
1570 #[must_use]
1572 pub fn with_base_url(&self, base_url: impl Into<String>) -> Self {
1573 Self {
1574 inner: self.inner.clone(),
1575 alias: self.alias.clone(),
1576 base_url: Some(base_url.into()),
1577 base_urls: self.base_urls.clone(),
1578 retry_policy: self.retry_policy.clone(),
1579 mock: self.mock.clone(),
1580 resilience_config: self.resilience_config.clone(),
1581 sim_net: self.sim_net.clone(),
1582 retry: self.retry.clone(),
1583 entropy: self.entropy.clone(),
1584 throttle: self.throttle.clone(),
1585 }
1586 }
1587
1588 fn build_request(&self, method: Method, url: impl AsRef<str>) -> RequestBuilder {
1589 let url_str = url.as_ref();
1590 let full_url = if url_str.starts_with("http://") || url_str.starts_with("https://") {
1591 url_str.to_owned()
1592 } else if let Some(base) = &self.base_url {
1593 format!(
1594 "{}/{}",
1595 base.trim_end_matches('/'),
1596 url_str.trim_start_matches('/')
1597 )
1598 } else {
1599 url_str.to_owned()
1600 };
1601
1602 RequestBuilder {
1603 client: self.inner.clone(),
1604 method,
1605 url: full_url,
1606 extra_headers: HeaderMap::new(),
1607 body: None,
1608 retry_policy: self.retry_policy.clone(),
1609 mock: self.mock.clone(),
1610 alias: self.alias.clone(),
1611 pending_error: None,
1612 resilience_config: self.resilience_config.clone(),
1613 redirect_mode: RedirectMode::Default,
1614 pin_addr: None,
1615 ssrf_safe: false,
1616 discard_response_body: false,
1617 breaker_scoped: false,
1618 sim_net: self.sim_net.clone(),
1619 forward_request_id: true,
1620 retry: self.retry.clone(),
1621 entropy: self.entropy.clone(),
1622 throttle: self.throttle.clone(),
1623 criticality: None,
1624 web_bot_auth: None,
1625 }
1626 }
1627
1628 #[must_use]
1630 pub fn get(&self, url: impl AsRef<str>) -> RequestBuilder {
1631 self.build_request(Method::GET, url)
1632 }
1633 #[must_use]
1635 pub fn post(&self, url: impl AsRef<str>) -> RequestBuilder {
1636 self.build_request(Method::POST, url)
1637 }
1638 #[must_use]
1640 pub fn put(&self, url: impl AsRef<str>) -> RequestBuilder {
1641 self.build_request(Method::PUT, url)
1642 }
1643 #[must_use]
1645 pub fn patch(&self, url: impl AsRef<str>) -> RequestBuilder {
1646 self.build_request(Method::PATCH, url)
1647 }
1648 #[must_use]
1650 pub fn delete(&self, url: impl AsRef<str>) -> RequestBuilder {
1651 self.build_request(Method::DELETE, url)
1652 }
1653
1654 #[must_use]
1656 pub fn head(&self, url: impl AsRef<str>) -> RequestBuilder {
1657 self.build_request(Method::HEAD, url)
1658 }
1659
1660 #[must_use]
1706 pub fn get_ssrf_safe(&self, url: impl Into<String>) -> RequestBuilder {
1707 self.build_request(Method::GET, url.into()).ssrf_safe()
1708 }
1709}
1710
1711impl Default for Client {
1712 fn default() -> Self {
1713 Self::new()
1714 }
1715}
1716
1717impl axum::extract::FromRequestParts<crate::AppState> for Client {
1718 type Rejection = std::convert::Infallible;
1719
1720 async fn from_request_parts(
1721 _parts: &mut http::request::Parts,
1722 state: &crate::AppState,
1723 ) -> Result<Self, std::convert::Infallible> {
1724 Ok(Self::from_state(state))
1725 }
1726}
1727
1728type RedirectValidator = Arc<dyn Fn(&str) -> bool + Send + Sync>;
1732
1733tokio::task_local! {
1734 static FOLLOWED: std::cell::RefCell<Vec<String>>;
1739 static MANUAL_REDIRECTS: bool;
1742}
1743
1744async fn send_tracking_redirects(
1747 req: reqwest::RequestBuilder,
1748) -> (Result<reqwest::Response, reqwest::Error>, Vec<String>) {
1749 FOLLOWED
1750 .scope(std::cell::RefCell::new(Vec::new()), async {
1751 let sent = req.send().await;
1752 (sent, FOLLOWED.with(std::cell::RefCell::take))
1753 })
1754 .await
1755}
1756
1757fn set_referer(headers: &mut HeaderMap, next: &str, previous: &str) {
1761 let (Ok(next), Ok(mut referer)) = (url::Url::parse(next), url::Url::parse(previous)) else {
1762 return;
1763 };
1764 if next.scheme() == "http" && referer.scheme() == "https" {
1765 headers.remove(reqwest::header::REFERER);
1766 return;
1767 }
1768 let _ = referer.set_username("");
1769 let _ = referer.set_password(None);
1770 referer.set_fragment(None);
1771 if let Ok(value) = HeaderValue::from_str(referer.as_str()) {
1772 headers.insert(reqwest::header::REFERER, value);
1773 }
1774}
1775
1776const PLAIN_MAX_REDIRECTS: usize = 10;
1778
1779fn pooled_redirect_policy() -> reqwest::redirect::Policy {
1785 reqwest::redirect::Policy::custom(|attempt| {
1786 if Deadline::current().is_some() || MANUAL_REDIRECTS.try_with(|m| *m).unwrap_or(false) {
1787 attempt.stop()
1788 } else if attempt.previous().len() > PLAIN_MAX_REDIRECTS {
1789 attempt.error("too many redirects")
1791 } else {
1792 let _ = FOLLOWED.try_with(|hops| hops.borrow_mut().push(attempt.url().to_string()));
1793 attempt.follow()
1794 }
1795 })
1796}
1797
1798#[derive(Clone, Copy)]
1800enum HopClient<'a> {
1801 Pooled(&'a reqwest::Client),
1803 OneShot(Duration),
1805}
1806
1807enum RedirectMode {
1813 Default,
1815 None,
1817 Follow {
1820 max: usize,
1821 validator: RedirectValidator,
1822 },
1823}
1824
1825const SSRF_SAFE_MAX_REDIRECTS: usize = 5;
1827
1828#[allow(
1830 clippy::struct_excessive_bools,
1831 reason = "independent per-request switches; each has its own builder method"
1832)]
1833pub struct RequestBuilder {
1834 client: reqwest::Client,
1835 method: Method,
1836 url: String,
1837 extra_headers: HeaderMap,
1838 body: Option<Bytes>,
1840 retry_policy: RetryPolicy,
1841 mock: Option<Arc<MockRegistry>>,
1842 alias: Option<String>,
1843 pending_error: Option<ClientError>,
1845 resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
1847 redirect_mode: RedirectMode,
1849 pin_addr: Option<Vec<SocketAddr>>,
1852 ssrf_safe: bool,
1855 discard_response_body: bool,
1858 breaker_scoped: bool,
1862 sim_net: Option<Arc<crate::sim::SimNet>>,
1864 forward_request_id: bool,
1867 retry: RetrySettings,
1869 entropy: Arc<dyn crate::entropy::Entropy>,
1871 throttle: Option<Arc<crate::admission::AdaptiveThrottle>>,
1873 criticality: Option<crate::admission::Criticality>,
1875 web_bot_auth: Option<Arc<crate::aeo::web_bot_auth::WebBotAuthSigner>>,
1877}
1878
1879impl RequestBuilder {
1880 #[must_use]
1885 pub const fn without_request_id(mut self) -> Self {
1886 self.forward_request_id = false;
1887 self
1888 }
1889
1890 #[must_use]
1894 pub const fn criticality(mut self, criticality: crate::admission::Criticality) -> Self {
1895 self.criticality = Some(criticality);
1896 self
1897 }
1898
1899 #[must_use]
1905 pub fn header(mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> Self {
1906 let name_str = name.as_ref();
1907 let value_str = value.as_ref();
1908 match (
1909 HeaderName::from_bytes(name_str.as_bytes()),
1910 HeaderValue::from_str(value_str),
1911 ) {
1912 (Ok(n), Ok(v)) => {
1913 self.extra_headers.insert(n, v);
1914 }
1915 (Err(e), _) => {
1916 tracing::warn!(header.name = name_str, error = %e, "invalid header name — header skipped");
1917 }
1918 (_, Err(e)) => {
1919 tracing::warn!(header.name = name_str, error = %e, "invalid header value — header skipped");
1920 }
1921 }
1922 self
1923 }
1924
1925 #[must_use]
1933 pub fn sign_web_bot_auth(
1934 mut self,
1935 signer: &crate::aeo::web_bot_auth::WebBotAuthSigner,
1936 ) -> Self {
1937 if let Some(h) = signer.sign_url(&self.url) {
1938 self.web_bot_auth = Some(Arc::new(signer.clone()));
1939 return self
1940 .header("signature-agent", h.signature_agent)
1941 .header("signature-input", h.signature_input)
1942 .header("signature", h.signature);
1943 }
1944 self.pending_error = Some(ClientError::InvalidUrl(format!(
1945 "Web Bot Auth needs an absolute URL, got {:?}",
1946 self.url
1947 )));
1948 self
1949 }
1950
1951 #[must_use]
1956 pub fn json<T: Serialize>(mut self, body: &T) -> Self {
1957 match serde_json::to_vec(body) {
1958 Ok(bytes) => {
1959 self.body = Some(Bytes::from(bytes));
1960 self = self.header("content-type", "application/json");
1961 }
1962 Err(e) => {
1963 self.pending_error = Some(ClientError::Json(e));
1964 }
1965 }
1966 self
1967 }
1968
1969 #[must_use]
1971 pub fn text_body(mut self, body: impl Into<String>) -> Self {
1972 self.body = Some(Bytes::from(body.into().into_bytes()));
1973 self
1974 }
1975
1976 #[must_use]
1983 pub fn bytes_body(mut self, body: impl Into<Bytes>) -> Self {
1984 self.body = Some(body.into());
1985 self
1986 }
1987
1988 #[must_use]
1993 pub const fn retries(mut self, max: u32) -> Self {
1994 self.retry_policy.max_retries = max;
1995 self
1996 }
1997
1998 #[must_use]
2005 pub const fn retry_non_idempotent(mut self) -> Self {
2006 self.retry_policy.retry_idempotent_only = false;
2007 self
2008 }
2009
2010 #[must_use]
2012 pub const fn max_backoff(mut self, max: Duration) -> Self {
2013 self.retry_policy.max_backoff = max;
2014 self
2015 }
2016
2017 #[must_use]
2019 pub const fn max_retry_after(mut self, max: Duration) -> Self {
2020 self.retry_policy.max_retry_after = max;
2021 self
2022 }
2023
2024 #[must_use]
2026 pub const fn no_retry(mut self) -> Self {
2027 self.retry_policy.max_retries = 0;
2028 self
2029 }
2030
2031 #[must_use]
2038 pub fn no_redirect(mut self) -> Self {
2039 self.redirect_mode = RedirectMode::None;
2040 self
2041 }
2042
2043 #[must_use]
2057 pub const fn discard_response_body(mut self) -> Self {
2058 self.discard_response_body = true;
2059 self
2060 }
2061
2062 #[must_use]
2082 pub fn follow_redirects<F>(mut self, max: usize, validator: F) -> Self
2083 where
2084 F: Fn(&str) -> bool + Send + Sync + 'static,
2085 {
2086 self.redirect_mode = RedirectMode::Follow {
2087 max,
2088 validator: Arc::new(validator),
2089 };
2090 self
2091 }
2092
2093 #[must_use]
2140 pub fn pin_to(self, addr: SocketAddr) -> Self {
2141 self.pin_to_addrs([addr])
2142 }
2143
2144 #[must_use]
2154 pub fn pin_to_addrs(mut self, addrs: impl IntoIterator<Item = SocketAddr>) -> Self {
2155 self.pin_addr = Some(addrs.into_iter().collect());
2156 self
2157 }
2158
2159 #[must_use]
2181 pub const fn ssrf_safe(mut self) -> Self {
2182 self.ssrf_safe = true;
2183 self
2184 }
2185
2186 #[must_use]
2217 pub(crate) const fn breaker_scoped(mut self) -> Self {
2218 self.breaker_scoped = true;
2219 self
2220 }
2221
2222 pub async fn send(self) -> Result<Response, ClientError> {
2236 if let Some(err) = self.pending_error {
2238 return Err(err);
2239 }
2240
2241 #[cfg(feature = "reporting")]
2250 if let Some(tape) = crate::capsule::effects::current_tape() {
2251 let method = self.method.to_string();
2252 let headers = caller_headers(&self);
2257 let body = self
2258 .body
2259 .as_ref()
2260 .map_or(crate::capsule::CapsuleBody::Absent, |body| {
2261 encode_body(body, usize::MAX)
2262 });
2263 return replayed_response(
2264 &tape,
2265 &crate::capsule::effects::OutboundRequest {
2266 method: &method,
2267 url: &self.url,
2268 headers: &headers,
2269 body: &body,
2270 },
2271 );
2272 }
2273 #[cfg(feature = "reporting")]
2275 if outbound_blocked_for_replay() {
2276 return Err(ClientError::BlockedDuringReplay(
2277 self.method.to_string(),
2278 crate::capsule::redact::hide_url_userinfo(&self.url),
2279 ));
2280 }
2281
2282 let recorder = OutboundRecorder::arm(&self);
2288 let result = self.send_recorded().await;
2289 recorder.finish(&result);
2290 result
2291 }
2292
2293 async fn send_recorded(mut self) -> Result<Response, ClientError> {
2295 crate::fault_injection::inject(crate::fault_injection::FaultTarget::Http)
2297 .await
2298 .map_err(|fault| ClientError::FaultInjected(fault.to_string()))?;
2299 if self.forward_request_id {
2302 add_current_request_id(&mut self.extra_headers);
2303 }
2304 self.ensure_idempotency_key();
2307 self.ensure_criticality_header();
2308
2309 if let Some(net) = self.sim_net.clone() {
2312 return self.send_sim(&net).await;
2313 }
2314
2315 if self.mock.is_some() {
2317 return self.send_inner(false).await;
2318 }
2319
2320 if self.needs_custom_path() {
2329 if self.breaker_scoped {
2330 return self.send_custom_breaker_guarded().await;
2331 }
2332 return self.send_custom(false).await;
2333 }
2334
2335 let breaker = breaker_for_url(self.resilience_config.as_ref(), &self.url);
2337
2338 let Ok(guard) = breaker.admit() else {
2340 return Err(ClientError::CircuitBreakerOpen);
2341 };
2342
2343 let is_half_open = breaker.state() == crate::circuit_breaker::CircuitState::HalfOpen;
2344 let res = self.send_inner(is_half_open).await;
2345 match &res {
2346 Ok(resp) => {
2347 let success = resp.status().as_u16() < 500;
2348 if success {
2349 guard.success();
2350 } else {
2351 guard.failure();
2352 }
2353 }
2354 Err(ClientError::DeadlineExceeded | ClientError::ThrottledLocally { .. }) => {
2360 drop(guard);
2361 }
2362 Err(_) => {
2363 guard.failure();
2364 }
2365 }
2366 res
2367 }
2368
2369 async fn send_custom_breaker_guarded(self) -> Result<Response, ClientError> {
2377 let breaker = breaker_for_url(self.resilience_config.as_ref(), &self.url);
2378 let Ok(guard) = breaker.admit() else {
2379 return Err(ClientError::CircuitBreakerOpen);
2380 };
2381 let ticket = match self.throttle_attempt(None) {
2389 Ok(ticket) => ticket,
2390 Err(error) => {
2391 drop(guard);
2392 return Err(error);
2393 }
2394 };
2395 let is_half_open = breaker.state() == crate::circuit_breaker::CircuitState::HalfOpen;
2404 let res = self.send_custom(is_half_open).await;
2405 record_call(ticket, &res);
2406 match &res {
2407 Ok(resp) if resp.status().as_u16() < 500 => guard.success(),
2408 Err(ClientError::DeadlineExceeded) => drop(guard),
2411 _ => guard.failure(),
2412 }
2413 res
2414 }
2415
2416 async fn send_inner(self, suppress_retries: bool) -> Result<Response, ClientError> {
2417 if let Some(ref mock) = self.mock {
2419 return self.mock_response(mock);
2420 }
2421
2422 let start = crate::time::ambient_instant();
2424 let max_attempts = self.max_attempts(suppress_retries);
2425 let mut gate = self.retry_gate(url_host(&self.url).as_deref());
2426 if gate.deadline.is_some() || self.follows_redirects_itself() {
2427 return self.follow_pooled(suppress_retries, &gate).await;
2428 }
2429 let mut last_retry = None;
2430 let mut delay = Duration::ZERO;
2431 let mut refilled: std::collections::HashSet<String> =
2433 url_host(&self.url).into_iter().collect();
2434
2435 for attempt in 0..max_attempts {
2436 if attempt > 0 {
2437 tokio::time::sleep(delay).await;
2438 }
2439 let ticket = self.begin_attempt(&gate, None)?;
2441 let last = attempt + 1 == max_attempts;
2442 let (span, sent, followed) =
2443 self.send_plain_attempt(&gate, attempt, &mut refilled).await;
2444 match sent {
2445 Ok(resp) => {
2446 let status = resp.status();
2447 let headers = resp.headers().clone();
2448 let url_used = resp.url().clone();
2449
2450 if is_retryable_response(status.as_u16()) && !last {
2454 gate.rekey(url_used.as_str());
2455 let wait = self.retry_policy.retry_delay(
2456 &*self.entropy,
2457 attempt,
2458 retry_hint(status.as_u16(), &headers),
2459 );
2460 let kind = retry_kind(status.as_u16());
2461 if gate.allow(kind, wait) {
2462 if let Some(ticket) = ticket {
2463 ticket.record(throttle_accepts(status.as_u16()));
2464 }
2465 delay = wait;
2466 last_retry = Some(kind);
2467 continue;
2468 }
2469 }
2470
2471 let body = if self.discard_response_body {
2472 Ok(Bytes::new())
2474 } else {
2475 read_body_in_span(resp, &span, &gate).await
2476 };
2477 if let Some(ticket) = ticket {
2480 ticket.record(body.is_ok() && throttle_accepts(status.as_u16()));
2481 }
2482 let body = body?;
2483 gate.finish(last_retry, status.as_u16());
2485 let elapsed = crate::time::ambient_instant().saturating_duration_since(start);
2486 log_request(
2487 self.method.as_str(),
2488 &url_used,
2489 status.as_u16(),
2490 elapsed,
2491 &self.extra_headers,
2492 );
2493
2494 return Ok(Response {
2495 status,
2496 headers,
2497 body,
2498 url: Some(url_used),
2499 });
2500 }
2501 Err(e) if e.is_timeout() && gate.expired() => {
2503 return Err(ClientError::DeadlineExceeded);
2504 }
2505 Err(e) if (e.is_connect() || e.is_timeout()) && !last => {
2508 if let Some(ticket) = ticket {
2509 ticket.record(false);
2510 }
2511 gate.rekey(followed.last().map_or(self.url.as_str(), String::as_str));
2516 let wait = self.retry_policy.retry_delay(&*self.entropy, attempt, None);
2517 if !gate.allow(RetryKind::Transient, wait) {
2518 return Err(ClientError::Request(e.without_url()));
2519 }
2520 delay = wait;
2521 last_retry = Some(RetryKind::Transient);
2522 }
2523 Err(e) => {
2524 if let Some(ticket) = ticket {
2525 ticket.record(false);
2526 }
2527 return Err(ClientError::Request(e.without_url()));
2528 }
2529 }
2530 }
2531
2532 unreachable!("retry loop exited without returning a result — this is a bug")
2534 }
2535
2536 fn retry_gate(&self, host: Option<&str>) -> RetryGate {
2538 let mut gate = RetryGate::start(
2539 self.retry.budgets.clone(),
2540 host,
2541 self.retry.send_deadline_header,
2542 );
2543 gate.signer.clone_from(&self.web_bot_auth);
2544 gate
2545 }
2546
2547 fn follows_redirects_itself(&self) -> bool {
2555 self.extra_headers
2556 .get(DEADLINE_HEADER)
2557 .and_then(crate::deadline::parse_header)
2558 .is_some()
2559 || self.extra_headers.contains_key("signature")
2560 }
2561
2562 async fn follow_pooled(
2566 self,
2567 suppress_retries: bool,
2568 gate: &RetryGate,
2569 ) -> Result<Response, ClientError> {
2570 let client = self.client.clone();
2573 MANUAL_REDIRECTS
2574 .scope(
2575 true,
2576 self.follow_loop(
2577 PLAIN_MAX_REDIRECTS,
2578 Arc::new(|_: &str| true),
2579 HopClient::Pooled(&client),
2580 suppress_retries,
2581 gate,
2582 ),
2583 )
2584 .await
2585 }
2586
2587 fn ensure_idempotency_key(&mut self) {
2591 let retries_unsafe_method = !is_idempotent_method(&self.method)
2592 && !self.retry_policy.retry_idempotent_only
2593 && self.retry_policy.max_retries > 0;
2594 if !retries_unsafe_method || self.extra_headers.contains_key(IDEMPOTENCY_KEY) {
2595 return;
2596 }
2597 let key = self.entropy.uuid_v4().to_string();
2598 if let Ok(value) = HeaderValue::from_str(&key) {
2599 self.extra_headers
2600 .insert(HeaderName::from_static(IDEMPOTENCY_KEY), value);
2601 }
2602 }
2603
2604 fn ensure_criticality_header(&mut self) {
2609 if self
2610 .extra_headers
2611 .contains_key(crate::admission::CRITICALITY_HEADER)
2612 {
2613 return;
2614 }
2615 let inbound = crate::admission::current_criticality()
2616 .filter(|c| *c != crate::admission::Criticality::Default);
2617 if let Some(c) = self.criticality.or(inbound) {
2618 self.extra_headers.insert(
2619 HeaderName::from_static(crate::admission::CRITICALITY_HEADER),
2620 HeaderValue::from_static(c.as_str()),
2621 );
2622 }
2623 }
2624
2625 fn begin_attempt(
2629 &self,
2630 gate: &RetryGate,
2631 host: Option<&str>,
2632 ) -> Result<Option<ThrottleTicket>, ClientError> {
2633 if gate.expired() {
2634 return Err(ClientError::DeadlineExceeded);
2635 }
2636 let ticket = self.throttle_attempt(host)?;
2637 gate.check()?;
2638 Ok(ticket)
2639 }
2640
2641 async fn send_plain_attempt(
2648 &self,
2649 gate: &RetryGate,
2650 attempt: u32,
2651 refilled: &mut std::collections::HashSet<String>,
2652 ) -> (
2653 tracing::Span,
2654 Result<reqwest::Response, reqwest::Error>,
2655 Vec<String>,
2656 ) {
2657 let span = client_attempt_span(&self.method, &self.url, attempt);
2658 let req = self.plain_attempt(gate, &span);
2659 let (sent, followed) =
2660 tracing::Instrument::instrument(send_tracking_redirects(req), span.clone()).await;
2661 record_send_outcome(&span, &sent);
2662 gate.record_destinations(&followed, refilled);
2663 (span, sent, followed)
2664 }
2665
2666 fn plain_attempt(&self, gate: &RetryGate, span: &tracing::Span) -> reqwest::RequestBuilder {
2670 let timeout = gate.attempt_timeout(self.retry_policy.request_timeout);
2671 let mut req = self.client.request(self.method.clone(), &self.url);
2672 if gate.deadline.is_some()
2673 && let Some(timeout) = timeout
2674 {
2675 req = req.timeout(timeout);
2676 }
2677 req = span.in_scope(|| inject_trace_context(req, &self.extra_headers));
2678 req = with_caller_headers(req, gate, timeout, &self.extra_headers, &self.url);
2679 if let Some(body) = &self.body {
2680 req = req.body(body.clone());
2681 }
2682 req
2683 }
2684
2685 fn throttle_attempt(&self, host: Option<&str>) -> Result<Option<ThrottleTicket>, ClientError> {
2689 let Some(throttle) = &self.throttle else {
2690 return Ok(None);
2691 };
2692 let Some(host) = host.map(str::to_owned).or_else(|| throttle_host(&self.url)) else {
2693 return Ok(None);
2694 };
2695 admit_throttle(throttle, host, &*self.entropy).map(Some)
2696 }
2697
2698 const fn max_attempts(&self, suppress_retries: bool) -> u32 {
2700 if suppress_retries {
2701 1
2702 } else if is_idempotent_method(&self.method) || !self.retry_policy.retry_idempotent_only {
2703 self.retry_policy.max_retries.saturating_add(1)
2704 } else {
2705 1
2706 }
2707 }
2708
2709 fn mock_response(&self, mock: &MockRegistry) -> Result<Response, ClientError> {
2712 let Some(mock_resp) = mock.find_match(&self.method, &self.url, self.alias.as_deref())
2713 else {
2714 return Err(ClientError::NoMock(
2717 self.method.to_string(),
2718 self.url.clone(),
2719 ));
2720 };
2721 let status =
2722 reqwest::StatusCode::from_u16(mock_resp.status).unwrap_or(reqwest::StatusCode::OK);
2723 let body_bytes = mock_resp
2724 .body
2725 .as_ref()
2726 .map(|v| serde_json::to_vec(v).unwrap_or_default())
2727 .unwrap_or_default();
2728
2729 tracing::info!(
2730 http.method = %self.method,
2731 http.url = %self.url,
2732 http.status = mock_resp.status,
2733 "[mock] outbound request intercepted"
2734 );
2735
2736 Ok(Response {
2737 status,
2738 headers: HeaderMap::new(),
2739 body: Bytes::from(body_bytes),
2740 url: None,
2741 })
2742 }
2743
2744 async fn send_sim(self, net: &crate::sim::SimNet) -> Result<Response, ClientError> {
2748 use tracing::Instrument as _;
2749
2750 let url = self.sim_url()?;
2751 let host = url
2752 .host_str()
2753 .ok_or_else(|| ClientError::InvalidUrl(format!("{}: no host", self.url)))?
2754 .to_owned();
2755 let max_attempts = self.max_attempts(false);
2756 let gate = self.retry_gate(url_host(url.as_str()).as_deref());
2757 let mut last_retry = None;
2758 let mut delay = Duration::ZERO;
2759 for attempt in 0..max_attempts {
2760 if attempt > 0 {
2761 tokio::time::sleep(delay).await;
2762 }
2763 let ticket = self.begin_attempt(&gate, Some(&host))?;
2764 let last = attempt + 1 == max_attempts;
2765 let span = client_attempt_span(&self.method, url.as_str(), attempt);
2768 let timeout = gate.attempt_timeout(self.retry_policy.request_timeout);
2769 let deadline_header = gate.header(timeout, &self.extra_headers);
2770 let exchange = self
2771 .sim_attempt(net, &host, &url, timeout, deadline_header)
2772 .instrument(span.clone());
2773 let exchange = async {
2778 match gate.deadline {
2779 Some(deadline) => crate::deadline::Bounded::until(deadline, exchange)
2780 .await
2781 .unwrap_or_else(|crate::deadline::DeadlineExceeded| {
2782 Err(SimAttemptError::Transient(format!(
2783 "request to {host} reached the request deadline"
2784 )))
2785 }),
2786 None => exchange.await,
2787 }
2788 };
2789 let outcome = match timeout {
2790 Some(limit) => {
2791 tokio::time::timeout(limit, exchange)
2792 .await
2793 .unwrap_or_else(|_elapsed| {
2794 Err(SimAttemptError::Transient(format!(
2795 "request to {host} timed out after {limit:?}"
2796 )))
2797 })
2798 }
2799 None => exchange.await,
2800 };
2801 match &outcome {
2802 Ok(response) => record_attempt_status(&span, response.status.as_u16()),
2803 Err(SimAttemptError::Transient(_)) => record_attempt_error(&span, "network"),
2804 Err(SimAttemptError::Fatal(_)) => record_attempt_error(&span, "request"),
2805 }
2806 drop(span);
2807 let deadline_stop =
2810 matches!(outcome, Err(SimAttemptError::Transient(_))) && gate.expired();
2811 if let Some(ticket) = ticket
2812 && !deadline_stop
2813 {
2814 ticket.record(matches!(&outcome, Ok(r) if throttle_accepts(r.status.as_u16())));
2815 }
2816 let response = match outcome {
2817 Ok(response) => response,
2818 Err(SimAttemptError::Transient(_)) if gate.expired() => {
2821 return Err(ClientError::DeadlineExceeded);
2822 }
2823 Err(SimAttemptError::Transient(message)) if !last => {
2824 let wait = self.retry_policy.retry_delay(&*self.entropy, attempt, None);
2825 if !gate.allow(RetryKind::Transient, wait) {
2826 return Err(ClientError::SimNetwork(message));
2827 }
2828 delay = wait;
2829 last_retry = Some(RetryKind::Transient);
2830 continue;
2831 }
2832 Err(SimAttemptError::Transient(message)) => {
2833 return Err(ClientError::SimNetwork(message));
2834 }
2835 Err(SimAttemptError::Fatal(error)) => return Err(error),
2836 };
2837 let status = response.status.as_u16();
2838 if is_retryable_response(status) && !last {
2839 let wait = self.retry_policy.retry_delay(
2840 &*self.entropy,
2841 attempt,
2842 retry_hint(status, &response.headers),
2843 );
2844 let kind = retry_kind(status);
2845 if gate.allow(kind, wait) {
2846 delay = wait;
2847 last_retry = Some(kind);
2848 continue;
2849 }
2850 }
2851 gate.finish(last_retry, status);
2852 return Ok(response);
2853 }
2854 unreachable!("the sim retry loop returns on its last attempt")
2855 }
2856
2857 fn sim_url(&self) -> Result<reqwest::Url, ClientError> {
2861 match (reqwest::Url::parse(&self.url), self.alias.as_deref()) {
2862 (Ok(url), _) => Ok(url),
2863 (Err(url::ParseError::RelativeUrlWithoutBase), Some(alias)) => {
2864 let path = self.url.trim_start_matches('/');
2865 reqwest::Url::parse(&format!("http://{alias}/{path}"))
2866 .map_err(|error| ClientError::InvalidUrl(format!("{}: {error}", self.url)))
2867 }
2868 (Err(error), _) => Err(ClientError::InvalidUrl(format!("{}: {error}", self.url))),
2869 }
2870 }
2871
2872 async fn sim_attempt(
2875 &self,
2876 net: &crate::sim::SimNet,
2877 host: &str,
2878 url: &reqwest::Url,
2879 timeout: Option<Duration>,
2880 deadline_header: Option<HeaderValue>,
2881 ) -> Result<Response, SimAttemptError> {
2882 net.transmit(host, timeout)
2883 .await
2884 .map_err(|fault| SimAttemptError::Transient(format!("request to {host} {fault}")))?;
2885 match (net.service(host), self.mock.as_ref()) {
2886 (Some(router), _) => serve_sim_host(router, self, url.clone(), deadline_header)
2887 .await
2888 .map_err(SimAttemptError::Fatal),
2889 (None, Some(mock)) => self.mock_response(mock).map_err(SimAttemptError::Fatal),
2890 (None, None) => Err(SimAttemptError::Fatal(ClientError::SimNetwork(format!(
2891 "no sim host named {host}"
2892 )))),
2893 }
2894 }
2895
2896 const fn needs_custom_path(&self) -> bool {
2898 self.ssrf_safe
2899 || self.pin_addr.is_some()
2900 || !matches!(self.redirect_mode, RedirectMode::Default)
2901 }
2902
2903 async fn send_custom(self, is_half_open: bool) -> Result<Response, ClientError> {
2912 if self.ssrf_safe && self.pin_addr.is_some() {
2921 return Err(ClientError::PinNotAllowedWithSsrfSafe(
2922 "get_ssrf_safe cannot be combined with pin_to: the SSRF-safe path \
2923 performs its own per-hop resolve/validate/pin and never reads the \
2924 pin_to address, so an explicit pin would be silently ignored. Use \
2925 pin_to alone for a caller-chosen address, or get_ssrf_safe alone \
2926 for guarded automatic per-hop pinning.",
2927 ));
2928 }
2929
2930 if self.pin_addr.is_some() && matches!(self.redirect_mode, RedirectMode::Follow { .. }) {
2937 return Err(ClientError::IncompatiblePinRedirect(
2938 "pin_to cannot be combined with follow_redirects: the pin only \
2939 covers the first hop and later redirect hops re-resolve via DNS, \
2940 escaping the pin. Use get_ssrf_safe for pinned, per-hop-revalidated \
2941 redirect following; pin_to alone (which returns the 3xx unfollowed); \
2942 or follow_redirects without pin_to.",
2943 ));
2944 }
2945
2946 if self.pin_addr.as_ref().is_some_and(Vec::is_empty) {
2948 return Err(ClientError::InvalidUrl(
2949 "pin_to_addrs needs at least one address".to_owned(),
2950 ));
2951 }
2952
2953 if self.pin_addr.is_some() && url_host_is_ip_literal(&self.url)? {
2962 return Err(ClientError::PinRequiresDomainHost(
2963 "pin_to cannot be honored for an IP-literal URL host because the \
2964 HTTP stack connects to the literal directly and skips the pinned \
2965 address; put the desired IP directly in the URL, or use a domain host.",
2966 ));
2967 }
2968
2969 let gate = self.retry_gate(url_host(&self.url).as_deref());
2972 if gate.expired() {
2975 return Err(ClientError::DeadlineExceeded);
2976 }
2977 let timeout = gate
2978 .attempt_timeout(self.retry_policy.request_timeout)
2979 .unwrap_or_else(|| Duration::from_secs(30));
2980
2981 if self.ssrf_safe {
2982 return self.send_ssrf_safe(timeout, is_half_open, &gate).await;
2983 }
2984
2985 let follow = match &self.redirect_mode {
2987 RedirectMode::Follow { max, validator } => Some((*max, validator.clone())),
2988 RedirectMode::None | RedirectMode::Default => None,
2989 };
2990 if let Some((max, validator)) = follow {
2991 return self
2992 .follow_loop(
2993 max,
2994 validator,
2995 HopClient::OneShot(timeout),
2996 is_half_open,
2997 &gate,
2998 )
2999 .await;
3000 }
3001
3002 let policy = reqwest::redirect::Policy::none();
3011 let resolve = self.pin_resolve()?;
3012 let client = build_oneshot_client(resolve, policy, timeout)?;
3013 send_one(
3014 &client,
3015 &self.method,
3016 &self.url,
3017 &self.extra_headers,
3018 self.body.as_ref(),
3019 &self.retry_policy,
3020 &*self.entropy,
3021 self.discard_response_body,
3022 None,
3023 is_half_open,
3024 &gate,
3025 false,
3026 None,
3027 None,
3028 )
3029 .await
3030 }
3031
3032 fn pin_resolve(&self) -> Result<Option<(String, Vec<SocketAddr>)>, ClientError> {
3034 match &self.pin_addr {
3035 Some(addrs) => Ok(Some((host_of(&self.url)?, addrs.clone()))),
3036 None => Ok(None),
3037 }
3038 }
3039
3040 async fn follow_loop(
3042 self,
3043 max: usize,
3044 validator: RedirectValidator,
3045 hop_client: HopClient<'_>,
3046 is_half_open: bool,
3047 gate: &RetryGate,
3048 ) -> Result<Response, ClientError> {
3049 let original =
3050 url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
3051 let mut current = self.url.clone();
3052 let mut method = self.method.clone();
3055 let mut headers = self.extra_headers.clone();
3056 let mut body = self.body.clone();
3057 let chain_retries = matches!(hop_client, HopClient::Pooled(_))
3062 .then(|| ChainRetries::new(self.max_attempts(is_half_open).saturating_sub(1)));
3063 let mut refilled: std::collections::HashSet<String> =
3065 url_host(&self.url).into_iter().collect();
3066 for hop in 0.. {
3067 let resolve = if hop == 0 {
3069 match &self.pin_addr {
3070 Some(addrs) => Some((host_of(¤t)?, addrs.clone())),
3071 None => None,
3072 }
3073 } else {
3074 None
3075 };
3076 if hop > 0 {
3079 strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
3080 }
3081 let client = match hop_client {
3082 HopClient::Pooled(client) => client.clone(),
3083 HopClient::OneShot(timeout) => {
3084 build_oneshot_client(resolve, reqwest::redirect::Policy::none(), timeout)?
3085 }
3086 };
3087 let hop_gate;
3089 let hop_gate = if hop == 0 {
3090 gate
3091 } else {
3092 hop_gate = gate.for_hop(¤t, &mut refilled);
3093 &hop_gate
3094 };
3095 let resp = send_one(
3096 &client,
3097 &method,
3098 ¤t,
3099 &headers,
3100 body.as_ref(),
3101 &self.retry_policy,
3102 &*self.entropy,
3103 self.discard_response_body,
3104 None,
3105 is_half_open,
3106 hop_gate,
3107 true,
3108 chain_retries.as_ref(),
3109 matches!(hop_client, HopClient::Pooled(_))
3112 .then_some(self.throttle.as_ref())
3113 .flatten(),
3114 )
3115 .await?;
3116
3117 let next = match redirect_target(&resp, ¤t) {
3118 Ok(Some(next)) => next,
3119 Ok(None) => return Ok(resp),
3120 Err(ClientError::InvalidUrl(_)) if matches!(hop_client, HopClient::Pooled(_)) => {
3122 return Ok(resp);
3123 }
3124 Err(error) => return Err(error),
3125 };
3126 if hop >= max {
3127 return Err(ClientError::TooManyRedirects(max));
3128 }
3129 if !validator(&next) {
3130 return Err(ClientError::RedirectRejected(next));
3131 }
3132 rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
3134 if matches!(hop_client, HopClient::Pooled(_)) {
3137 set_referer(&mut headers, &next, ¤t);
3138 }
3139 current = next;
3140 }
3141 unreachable!("redirect loop is bounded by `max` and always returns")
3142 }
3143
3144 const fn ssrf_redirect_plan(&self) -> (bool, usize) {
3155 match &self.redirect_mode {
3156 RedirectMode::Default => (true, SSRF_SAFE_MAX_REDIRECTS),
3157 RedirectMode::None => (false, 0),
3158 RedirectMode::Follow { max, .. } => (true, *max),
3159 }
3160 }
3161
3162 async fn send_ssrf_safe(
3201 self,
3202 timeout: Duration,
3203 is_half_open: bool,
3204 gate: &RetryGate,
3205 ) -> Result<Response, ClientError> {
3206 let deadline = crate::time::ambient_instant() + timeout;
3207 let (follow, max) = self.ssrf_redirect_plan();
3208 let original =
3209 url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
3210 let mut current = self.url.clone();
3211 let mut method = self.method.clone();
3214 let mut headers = self.extra_headers.clone();
3215 let mut body = self.body.clone();
3216 let mut refilled: std::collections::HashSet<String> =
3218 url_host(&self.url).into_iter().collect();
3219 for hop in 0.. {
3220 let remaining_for_lookup = deadline_remaining_or_timeout(deadline, ¤t)
3221 .map_err(|error| gate.classify(error))?;
3222 let addrs = tokio::time::timeout(remaining_for_lookup, resolve_and_validate(¤t))
3229 .await
3230 .map_err(|_| gate.classify(ssrf_safe_deadline_error(¤t)))??;
3231 let host = host_of(¤t)?;
3232 let remaining_for_connect = deadline_remaining_or_timeout(deadline, ¤t)
3236 .map_err(|error| gate.classify(error))?;
3237 let client = build_oneshot_client(
3238 Some((host, addrs)),
3239 reqwest::redirect::Policy::none(),
3240 remaining_for_connect,
3241 )?;
3242 if hop > 0 {
3245 strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
3246 }
3247 let hop_gate;
3249 let hop_gate = if hop == 0 {
3250 gate
3251 } else {
3252 hop_gate = gate.for_hop(¤t, &mut refilled);
3253 &hop_gate
3254 };
3255 let resp = send_one(
3256 &client,
3257 &method,
3258 ¤t,
3259 &headers,
3260 body.as_ref(),
3261 &self.retry_policy,
3262 &*self.entropy,
3263 self.discard_response_body,
3264 Some(deadline),
3265 is_half_open,
3266 hop_gate,
3267 false,
3268 None,
3269 None,
3270 )
3271 .await?;
3272
3273 if !follow {
3281 return Ok(resp);
3282 }
3283 let Some(next) = redirect_target(&resp, ¤t)? else {
3284 return Ok(resp);
3285 };
3286 if hop >= max {
3287 return Err(ClientError::TooManyRedirects(max));
3288 }
3289 if scheme_is_https(¤t)? && !scheme_is_https(&next)? {
3291 return Err(ClientError::RedirectRejected(format!(
3292 "https→http scheme downgrade on redirect to {next}"
3293 )));
3294 }
3295 if let RedirectMode::Follow { validator, .. } = &self.redirect_mode
3299 && !validator(&next)
3300 {
3301 return Err(ClientError::RedirectRejected(next));
3302 }
3303 rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
3305 current = next;
3306 }
3310 unreachable!("redirect loop is bounded by the SSRF-safe redirect plan")
3311 }
3312}
3313
3314const fn is_idempotent_method(method: &Method) -> bool {
3317 matches!(
3318 *method,
3319 Method::GET | Method::HEAD | Method::PUT | Method::DELETE | Method::OPTIONS | Method::TRACE
3320 )
3321}
3322
3323const fn is_retryable_status(status: u16) -> bool {
3324 matches!(status, 502..=504)
3325}
3326
3327fn deadline_remaining_or_timeout(
3332 deadline: std::time::Instant,
3333 current: &str,
3334) -> Result<Duration, ClientError> {
3335 let now = crate::time::ambient_instant();
3336 if now >= deadline {
3337 return Err(ssrf_safe_deadline_error(current));
3338 }
3339 Ok(deadline - now)
3340}
3341
3342fn ssrf_safe_deadline_error(current: &str) -> ClientError {
3348 ClientError::InvalidUrl(format!(
3349 "SSRF-safe resolve/redirect operation exceeded its deadline resolving {current}"
3350 ))
3351}
3352
3353fn breaker_for_url(
3359 resilience_config: Option<&Arc<crate::config::ResilienceConfig>>,
3360 url: &str,
3361) -> crate::circuit_breaker::CircuitBreaker {
3362 let host = url::Url::parse(url).ok().map_or_else(
3363 || "unknown".to_owned(),
3364 |u| {
3365 let h = u.host_str().unwrap_or("unknown");
3366 u.port()
3367 .map_or_else(|| h.to_owned(), |port| format!("{h}:{port}"))
3368 },
3369 );
3370
3371 resilience_config.map_or_else(
3372 || {
3373 crate::circuit_breaker::global_registry().get_or_create(
3374 &host,
3375 crate::circuit_breaker::CircuitBreakerPolicy::default(),
3376 )
3377 },
3378 |rc| {
3379 let policy = crate::circuit_breaker::CircuitBreakerPolicy::from_config(rc, &host);
3380 crate::circuit_breaker::global_registry().get_or_create_with_config(&host, policy)
3381 },
3382 )
3383}
3384
3385const fn retry_kind(status: u16) -> RetryKind {
3389 if status == 429 {
3390 RetryKind::Throttling
3391 } else {
3392 RetryKind::Transient
3393 }
3394}
3395
3396fn url_host(url: &str) -> Option<String> {
3399 let parsed = reqwest::Url::parse(url).ok()?;
3400 let host = parsed.host_str()?;
3401 Some(
3402 parsed
3403 .port_or_known_default()
3404 .map_or_else(|| host.to_owned(), |port| format!("{host}:{port}")),
3405 )
3406}
3407
3408#[derive(Clone)]
3410struct RetrySettings {
3411 budgets: Option<Arc<RetryBudgets>>,
3413 send_deadline_header: bool,
3415}
3416
3417impl RetrySettings {
3418 fn standalone() -> Self {
3420 Self {
3421 budgets: Some(Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()))),
3422 send_deadline_header: true,
3423 }
3424 }
3425}
3426
3427const MIN_ATTEMPT: Duration = Duration::from_millis(10);
3430
3431#[derive(Clone)]
3434pub(crate) struct SharedRetryBudgets {
3435 budgets: Arc<RetryBudgets>,
3436 config: RetryBudgetConfig,
3437}
3438
3439static BUDGETS_REPLACE: std::sync::Mutex<()> = std::sync::Mutex::new(());
3441
3442fn shared_retry_budgets(
3446 state: &crate::AppState,
3447 config: &RetryBudgetConfig,
3448) -> Option<Arc<RetryBudgets>> {
3449 if !config.enabled {
3450 return None;
3451 }
3452 let current = || {
3453 state
3454 .extension::<SharedRetryBudgets>()
3455 .filter(|shared| shared.config == *config)
3456 .map(|shared| Arc::clone(&shared.budgets))
3457 };
3458 if let Some(budgets) = current() {
3459 return Some(budgets);
3460 }
3461 let _guard = BUDGETS_REPLACE
3464 .lock()
3465 .unwrap_or_else(std::sync::PoisonError::into_inner);
3466 if let Some(budgets) = current() {
3467 return Some(budgets);
3468 }
3469 let budgets = Arc::new(RetryBudgets::new(config));
3470 state.insert_extension(SharedRetryBudgets {
3471 budgets: Arc::clone(&budgets),
3472 config: config.clone(),
3473 });
3474 Some(budgets)
3475}
3476
3477struct RetryGate {
3481 deadline: Option<Deadline>,
3482 budgets: Option<Arc<RetryBudgets>>,
3484 budget: Option<Arc<RetryBudget>>,
3486 host: Option<String>,
3488 send_header: bool,
3489 refill_pending: AtomicBool,
3493 pending_retry: std::sync::Mutex<Option<(Arc<RetryBudget>, RetryKind)>>,
3497 caller_deadline: std::sync::OnceLock<Option<Deadline>>,
3501 signer: Option<Arc<crate::aeo::web_bot_auth::WebBotAuthSigner>>,
3503}
3504
3505impl Drop for RetryGate {
3506 fn drop(&mut self) {
3507 let pending = self
3508 .pending_retry
3509 .get_mut()
3510 .unwrap_or_else(std::sync::PoisonError::into_inner)
3511 .take();
3512 if let Some((budget, kind)) = pending {
3513 budget.release(kind);
3514 }
3515 }
3516}
3517
3518impl RetryGate {
3519 fn start(budgets: Option<Arc<RetryBudgets>>, host: Option<&str>, send_header: bool) -> Self {
3522 Self::with_deadline(Deadline::current(), budgets, host, send_header)
3523 }
3524
3525 fn with_deadline(
3526 deadline: Option<Deadline>,
3527 budgets: Option<Arc<RetryBudgets>>,
3528 host: Option<&str>,
3529 send_header: bool,
3530 ) -> Self {
3531 let budget = budgets
3532 .as_deref()
3533 .zip(host)
3534 .map(|(budgets, host)| budgets.for_host(host));
3535 Self {
3536 deadline,
3537 budgets,
3538 budget,
3539 host: host.map(str::to_owned),
3540 send_header,
3541 refill_pending: AtomicBool::new(true),
3542 pending_retry: std::sync::Mutex::new(None),
3543 caller_deadline: std::sync::OnceLock::new(),
3544 signer: None,
3545 }
3546 }
3547
3548 fn rekey(&mut self, url: &str) {
3554 let host = url_host(url);
3555 if host.is_some() && host != self.host {
3556 self.budget = self
3557 .budgets
3558 .as_deref()
3559 .zip(host.as_deref())
3560 .map(|(budgets, host)| budgets.for_host(host));
3561 self.host = host;
3562 }
3563 }
3564
3565 fn record_destination(&self, url: &str) {
3570 let host = url_host(url);
3571 if host.is_none() || host == self.host {
3573 return;
3574 }
3575 if let Some((budgets, host)) = self.budgets.as_deref().zip(host) {
3576 budgets.for_host(&host).record_request();
3577 }
3578 }
3579
3580 fn record_destinations(&self, urls: &[String], seen: &mut std::collections::HashSet<String>) {
3583 for url in urls {
3584 if url_host(url).is_some_and(|host| seen.insert(host)) {
3585 self.record_destination(url);
3586 }
3587 }
3588 }
3589
3590 fn for_hop(&self, url: &str, refilled: &mut std::collections::HashSet<String>) -> Self {
3596 let mut hop = Self {
3597 deadline: self.deadline,
3598 budgets: self.budgets.clone(),
3599 budget: self.budget.clone(),
3600 host: self.host.clone(),
3601 send_header: self.send_header,
3602 refill_pending: AtomicBool::new(false),
3603 pending_retry: std::sync::Mutex::new(None),
3604 caller_deadline: self.caller_deadline.clone(),
3605 signer: self.signer.clone(),
3606 };
3607 hop.rekey(url);
3608 if url_host(url).is_some_and(|host| refilled.insert(host)) {
3609 hop.refill_pending = AtomicBool::new(true);
3610 }
3611 hop
3612 }
3613
3614 fn body_error(&self, error: reqwest::Error) -> ClientError {
3617 if error.is_timeout() && self.expired() {
3618 ClientError::DeadlineExceeded
3619 } else {
3620 ClientError::Request(error.without_url())
3621 }
3622 }
3623
3624 fn classify(&self, error: ClientError) -> ClientError {
3627 if self.expired() {
3628 ClientError::DeadlineExceeded
3629 } else {
3630 error
3631 }
3632 }
3633
3634 fn check(&self) -> Result<(), ClientError> {
3638 if self.expired() {
3639 return Err(ClientError::DeadlineExceeded);
3640 }
3641 self.pending_retry
3643 .lock()
3644 .unwrap_or_else(std::sync::PoisonError::into_inner)
3645 .take();
3646 if self.refill_pending.swap(false, Ordering::Relaxed)
3647 && let Some(budget) = &self.budget
3648 {
3649 budget.record_request();
3650 }
3651 Ok(())
3652 }
3653
3654 fn attempt_timeout(&self, per_try: Option<Duration>) -> Option<Duration> {
3657 match (per_try, self.deadline) {
3658 (Some(limit), Some(deadline)) => Some(deadline.clamp(limit)),
3659 (None, Some(deadline)) => Some(deadline.remaining()),
3660 (limit, None) => limit,
3661 }
3662 }
3663
3664 fn expired(&self) -> bool {
3666 self.deadline.is_some_and(Deadline::is_expired)
3667 }
3668
3669 fn allow(&self, kind: RetryKind, wait: Duration) -> bool {
3673 if self
3674 .deadline
3675 .is_some_and(|deadline| deadline.remaining() <= wait + MIN_ATTEMPT)
3676 {
3677 return false;
3678 }
3679 let Some(budget) = &self.budget else {
3680 return true;
3681 };
3682 if !budget.try_acquire(kind) {
3683 return false;
3684 }
3685 *self
3686 .pending_retry
3687 .lock()
3688 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some((Arc::clone(budget), kind));
3689 true
3690 }
3691
3692 fn finish(&self, last_retry: Option<RetryKind>, status: u16) {
3695 if let (Some(budget), Some(kind)) = (&self.budget, last_retry)
3696 && status < 400
3697 {
3698 budget.release(kind);
3699 }
3700 }
3701
3702 fn header(&self, timeout: Option<Duration>, caller: &HeaderMap) -> Option<HeaderValue> {
3709 let millis = |d: Duration| u64::try_from(d.as_millis()).unwrap_or(u64::MAX);
3710 let left = self.deadline.and(timeout).map(millis);
3712 let Some(value) = caller.get(DEADLINE_HEADER) else {
3713 return left.filter(|_| self.send_header).map(HeaderValue::from);
3714 };
3715 let mut first = None;
3718 let caller_deadline = self.caller_deadline.get_or_init(|| {
3719 first = crate::deadline::parse_header(value);
3720 first.map(Deadline::after)
3721 });
3722 let theirs = first.map(millis).or_else(|| {
3723 caller_deadline
3724 .as_ref()
3725 .map(|deadline| millis(deadline.remaining()))
3726 });
3727 match (left, theirs) {
3728 (Some(left), Some(theirs)) => Some(HeaderValue::from(left.min(theirs))),
3729 (Some(value), None) | (None, Some(value)) => Some(HeaderValue::from(value)),
3730 (None, None) => None,
3732 }
3733 }
3734}
3735
3736fn with_caller_headers(
3740 mut req: reqwest::RequestBuilder,
3741 gate: &RetryGate,
3742 timeout: Option<Duration>,
3743 caller: &HeaderMap,
3744 url: &str,
3745) -> reqwest::RequestBuilder {
3746 let deadline_value = gate.header(timeout, caller);
3747 let sends_deadline = deadline_value.is_some();
3748 if let Some(value) = deadline_value {
3749 req = req.header(DEADLINE_HEADER, value);
3750 }
3751 let fresh = gate
3754 .signer
3755 .as_ref()
3756 .filter(|_| caller.contains_key("signature"))
3757 .and_then(|signer| signer.sign_url(url));
3758 for (name, value) in caller {
3759 if sends_deadline && name == DEADLINE_HEADER {
3760 continue;
3761 }
3762 if fresh.is_some()
3763 && matches!(
3764 name.as_str(),
3765 "signature" | "signature-input" | "signature-agent"
3766 )
3767 {
3768 continue;
3769 }
3770 req = req.header(name.clone(), value.clone());
3771 }
3772 if let Some(h) = fresh {
3773 req = req
3774 .header("signature-agent", h.signature_agent)
3775 .header("signature-input", h.signature_input)
3776 .header("signature", h.signature);
3777 }
3778 req
3779}
3780
3781enum SimAttemptError {
3785 Transient(String),
3787 Fatal(ClientError),
3789}
3790
3791async fn serve_sim_host(
3793 router: axum::Router,
3794 request: &RequestBuilder,
3795 url: reqwest::Url,
3796 deadline_header: Option<HeaderValue>,
3797) -> Result<Response, ClientError> {
3798 let target = url.query().map_or_else(
3799 || url.path().to_owned(),
3800 |query| format!("{}?{query}", url.path()),
3801 );
3802 let mut builder = axum::http::Request::builder()
3803 .method(request.method.clone())
3804 .uri(target);
3805 if !request.extra_headers.contains_key(reqwest::header::HOST) {
3806 let authority = &url[url::Position::BeforeHost..url::Position::AfterPort];
3809 builder = builder.header(reqwest::header::HOST, authority);
3810 }
3811 for (name, value) in trace_headers_to_send(trace_context_headers(), &request.extra_headers) {
3813 builder = builder.header(name, value);
3814 }
3815 let sends_deadline = deadline_header.is_some();
3816 if let Some(value) = deadline_header {
3817 builder = builder.header(DEADLINE_HEADER, value);
3818 }
3819 if let Some(body) = &request.body
3822 && !request
3823 .extra_headers
3824 .contains_key(reqwest::header::CONTENT_LENGTH)
3825 {
3826 builder = builder.header(reqwest::header::CONTENT_LENGTH, body.len());
3827 }
3828 let fresh = request
3831 .web_bot_auth
3832 .as_ref()
3833 .filter(|_| request.extra_headers.contains_key("signature"))
3834 .and_then(|signer| signer.sign_url(url.as_str()));
3835 for (name, value) in &request.extra_headers {
3836 if sends_deadline && name == DEADLINE_HEADER {
3837 continue;
3838 }
3839 if fresh.is_some()
3840 && matches!(
3841 name.as_str(),
3842 "signature" | "signature-input" | "signature-agent"
3843 )
3844 {
3845 continue;
3846 }
3847 builder = builder.header(name, value);
3848 }
3849 if let Some(h) = fresh {
3850 builder = builder
3851 .header("signature-agent", h.signature_agent)
3852 .header("signature-input", h.signature_input)
3853 .header("signature", h.signature);
3854 }
3855 let body = request.body.clone().unwrap_or_default();
3856 let http_request = builder
3857 .body(axum::body::Body::from(body))
3858 .map_err(|error| ClientError::SimNetwork(error.to_string()))?;
3859 let response = crate::deadline::unscoped(tower::ServiceExt::oneshot(router, http_request))
3862 .await
3863 .unwrap_or_else(|never| match never {});
3864 let status = response.status();
3865 let headers = response.headers().clone();
3866 let body = if request.discard_response_body {
3867 Bytes::new()
3868 } else {
3869 axum::body::to_bytes(response.into_body(), usize::MAX)
3870 .await
3871 .map_err(|error| ClientError::SimNetwork(error.to_string()))?
3872 };
3873 Ok(Response {
3874 status,
3875 headers,
3876 body,
3877 url: Some(url),
3878 })
3879}
3880
3881fn build_oneshot_client(
3893 resolve: Option<(String, Vec<SocketAddr>)>,
3894 policy: reqwest::redirect::Policy,
3895 timeout: Duration,
3896) -> Result<reqwest::Client, ClientError> {
3897 let mut builder = reqwest::ClientBuilder::new()
3898 .timeout(timeout)
3899 .redirect(policy);
3900 if let Some((host, addrs)) = resolve
3901 && !addrs.is_empty()
3902 {
3903 builder = builder.no_proxy().resolve_to_addrs(&host, &addrs);
3909 }
3910 builder.build().map_err(ClientError::Request)
3911}
3912
3913struct ChainRetries {
3915 left: AtomicU32,
3917 used: AtomicU32,
3919 attempt_end: Mutex<Option<Instant>>,
3923}
3924
3925impl ChainRetries {
3926 const fn new(retries: u32) -> Self {
3927 Self {
3928 left: AtomicU32::new(retries),
3929 used: AtomicU32::new(0),
3930 attempt_end: Mutex::new(None),
3931 }
3932 }
3933
3934 fn take_attempt_end(&self) -> Option<Instant> {
3936 self.attempt_end
3937 .lock()
3938 .unwrap_or_else(std::sync::PoisonError::into_inner)
3939 .take()
3940 }
3941
3942 fn set_attempt_end(&self, end: Option<Instant>) {
3944 *self
3945 .attempt_end
3946 .lock()
3947 .unwrap_or_else(std::sync::PoisonError::into_inner) = end;
3948 }
3949}
3950
3951struct ThrottleTicket {
3955 throttle: Arc<crate::admission::AdaptiveThrottle>,
3956 host: String,
3957 recorded: bool,
3958}
3959
3960impl ThrottleTicket {
3961 fn record(mut self, accepted: bool) {
3962 self.recorded = true;
3963 self.throttle
3964 .record(&self.host, crate::time::ambient_instant(), accepted);
3965 }
3966}
3967
3968impl Drop for ThrottleTicket {
3969 fn drop(&mut self) {
3970 if !self.recorded {
3971 self.throttle
3972 .record(&self.host, crate::time::ambient_instant(), true);
3973 }
3974 }
3975}
3976
3977fn admit_throttle(
3980 throttle: &Arc<crate::admission::AdaptiveThrottle>,
3981 host: String,
3982 entropy: &dyn crate::entropy::Entropy,
3983) -> Result<ThrottleTicket, ClientError> {
3984 let now = crate::time::ambient_instant();
3985 if throttle.admit(&host, now, || entropy.next_u64()) {
3986 Ok(ThrottleTicket {
3987 throttle: Arc::clone(throttle),
3988 host,
3989 recorded: false,
3990 })
3991 } else {
3992 tracing::debug!(host = %host, "outbound request throttled locally");
3993 Err(ClientError::ThrottledLocally { host })
3994 }
3995}
3996
3997fn record_call(ticket: Option<ThrottleTicket>, res: &Result<Response, ClientError>) {
4001 if let Some(ticket) = ticket
4002 && !matches!(res, Err(ClientError::DeadlineExceeded))
4003 {
4004 ticket.record(matches!(res, Ok(r) if throttle_accepts(r.status.as_u16())));
4005 }
4006}
4007
4008const fn throttle_accepts(status: u16) -> bool {
4010 !matches!(status, 429 | 503)
4011}
4012
4013fn throttle_host(url: &str) -> Option<String> {
4015 let parsed = url::Url::parse(url).ok()?;
4016 let host = parsed.host_str()?;
4017 Some(
4018 parsed
4019 .port()
4020 .map_or_else(|| host.to_owned(), |port| format!("{host}:{port}")),
4021 )
4022}
4023
4024#[allow(
4067 clippy::too_many_arguments,
4068 reason = "one seam shared by three call sites (plain custom path, follow_loop, send_ssrf_safe); \
4069 splitting the request-shape fields into their own struct would still leave the \
4070 retry/discard/deadline/half-open knobs alongside it, for no reduction in what a \
4071 caller reasons about"
4072)]
4073#[allow(
4074 clippy::too_many_lines,
4075 reason = "one retry loop; the deadline and budget checks (#3058) sit at each decision"
4076)]
4077async fn send_one(
4078 client: &reqwest::Client,
4079 method: &Method,
4080 url: &str,
4081 extra_headers: &HeaderMap,
4082 body: Option<&Bytes>,
4083 retry_policy: &RetryPolicy,
4084 entropy: &dyn crate::entropy::Entropy,
4085 discard_response_body: bool,
4086 deadline: Option<Instant>,
4087 suppress_retries: bool,
4088 gate: &RetryGate,
4089 skip_redirect_body: bool,
4090 chain: Option<&ChainRetries>,
4091 throttle: Option<&Arc<crate::admission::AdaptiveThrottle>>,
4092) -> Result<Response, ClientError> {
4093 let start = crate::time::ambient_instant();
4094 let mut last_retry = None;
4095 let mut max_attempts = if suppress_retries {
4096 1
4097 } else if is_idempotent_method(method) || !retry_policy.retry_idempotent_only {
4098 retry_policy.max_retries.saturating_add(1)
4099 } else {
4100 1
4101 };
4102 let retries_before = chain.map_or(0, |chain| chain.used.load(Ordering::Relaxed));
4105 if let Some(chain) = chain {
4106 max_attempts = max_attempts.min(chain.left.load(Ordering::Relaxed).saturating_add(1));
4107 }
4108 let mut last_transient_err: Option<reqwest::Error> = None;
4109
4110 let deadline_exceeded_err = |last_transient_err: &mut Option<reqwest::Error>| {
4115 last_transient_err.take().map_or_else(
4116 || ssrf_safe_deadline_error(url),
4117 |e| ClientError::Request(e.without_url()),
4118 )
4119 };
4120
4121 let reaches_hop_deadline = |wait: Duration| {
4125 deadline.is_some_and(|d| {
4126 crate::time::ambient_instant()
4127 .checked_add(wait.saturating_add(MIN_ATTEMPT))
4128 .is_none_or(|resume| resume >= d)
4129 })
4130 };
4131
4132 let mut delay = Duration::ZERO;
4133 for attempt in 0..max_attempts {
4134 let last = attempt + 1 == max_attempts;
4135 if attempt > 0 {
4136 if let Some(chain) = chain {
4137 chain.left.fetch_sub(1, Ordering::Relaxed);
4138 chain.used.fetch_add(1, Ordering::Relaxed);
4139 }
4140 if deadline.is_some_and(|d| crate::time::ambient_instant() >= d) {
4141 return Err(gate.classify(deadline_exceeded_err(&mut last_transient_err)));
4142 }
4143 let mut sleep_for = delay;
4144 if let Some(d) = deadline {
4145 sleep_for =
4146 sleep_for.min(d.saturating_duration_since(crate::time::ambient_instant()));
4147 }
4148 tokio::time::sleep(sleep_for).await;
4149 if deadline.is_some_and(|d| crate::time::ambient_instant() >= d) {
4150 return Err(gate.classify(deadline_exceeded_err(&mut last_transient_err)));
4151 }
4152 }
4153
4154 if gate.expired() {
4158 return Err(ClientError::DeadlineExceeded);
4159 }
4160 let ticket = match throttle.zip(throttle_host(url)) {
4161 Some((throttle, host)) => Some(admit_throttle(throttle, host, entropy)?),
4162 None => None,
4163 };
4164 gate.check()?;
4165 let mut req = client.request(method.clone(), url);
4166 let attempt_start = crate::time::ambient_instant();
4172 let hop_timeout = deadline.map(|d| d.saturating_duration_since(attempt_start));
4173 let carried_end = if attempt == 0 {
4178 chain.and_then(ChainRetries::take_attempt_end)
4179 } else {
4180 None
4181 };
4182 let per_try = match (retry_policy.request_timeout, carried_end) {
4183 (Some(_), Some(end)) => Some(end.saturating_duration_since(attempt_start)),
4184 (limit, _) => limit,
4185 };
4186 let attempt_end = carried_end.or_else(|| {
4187 retry_policy
4188 .request_timeout
4189 .and_then(|limit| attempt_start.checked_add(limit))
4190 });
4191 let attempt_timeout = if gate.deadline.is_some() {
4193 gate.attempt_timeout(hop_timeout.or(per_try))
4194 } else if carried_end.is_some() {
4195 hop_timeout.or(per_try)
4197 } else {
4198 hop_timeout
4199 };
4200 if let Some(timeout) = attempt_timeout {
4201 req = req.timeout(timeout);
4202 }
4203 let span = client_attempt_span(method, url, attempt);
4204 req = span.in_scope(|| inject_trace_context(req, extra_headers));
4205 req = with_caller_headers(req, gate, attempt_timeout, extra_headers, url);
4206 if let Some(body) = body {
4207 req = req.body(body.clone());
4208 }
4209
4210 match send_in_span(req, &span).await {
4211 Ok(resp) => {
4212 let status = resp.status();
4213 let headers = resp.headers().clone();
4214 let url_used = resp.url().clone();
4215
4216 if is_retryable_response(status.as_u16())
4217 && !last
4218 && deadline.is_none_or(|d| crate::time::ambient_instant() < d)
4219 {
4220 let hint = retry_hint(status.as_u16(), &headers);
4221 let next = retry_policy.retry_delay(
4222 entropy,
4223 retries_before.saturating_add(attempt),
4224 hint,
4225 );
4226 let kind = retry_kind(status.as_u16());
4229 if !reaches_hop_deadline(next) && gate.allow(kind, next) {
4230 if let Some(ticket) = ticket {
4231 ticket.record(throttle_accepts(status.as_u16()));
4232 }
4233 delay = next;
4234 last_retry = Some(kind);
4235 continue;
4236 }
4237 }
4238 let followed = skip_redirect_body
4241 && matches!(redirect_location(status, &headers, url), Ok(Some(_)));
4242 if followed && let Some(chain) = chain {
4243 chain.set_attempt_end(attempt_end);
4244 }
4245 let body = if discard_response_body || followed {
4246 Ok(Bytes::new())
4248 } else {
4249 read_body_in_span(resp, &span, gate).await
4250 };
4251 if let Some(ticket) = ticket
4255 && !matches!(body, Err(ClientError::DeadlineExceeded))
4256 {
4257 ticket.record(body.is_ok() && throttle_accepts(status.as_u16()));
4258 }
4259 let body = body?;
4260 gate.finish(last_retry, status.as_u16());
4262 log_request(
4263 method.as_str(),
4264 &url_used,
4265 status.as_u16(),
4266 crate::time::ambient_instant().saturating_duration_since(start),
4267 extra_headers,
4268 );
4269 return Ok(Response {
4270 status,
4271 headers,
4272 body,
4273 url: Some(url_used),
4274 });
4275 }
4276 Err(e) if e.is_timeout() && gate.expired() => {
4278 return Err(ClientError::DeadlineExceeded);
4279 }
4280 Err(e) if (e.is_connect() || e.is_timeout()) && !last => {
4281 if let Some(ticket) = ticket {
4282 ticket.record(false);
4283 }
4284 let wait =
4285 retry_policy.retry_delay(entropy, retries_before.saturating_add(attempt), None);
4286 if reaches_hop_deadline(wait) || !gate.allow(RetryKind::Transient, wait) {
4288 return Err(ClientError::Request(e.without_url()));
4289 }
4290 delay = wait;
4291 last_retry = Some(RetryKind::Transient);
4292 last_transient_err = Some(e);
4293 }
4294 Err(e) => {
4295 if let Some(ticket) = ticket {
4296 ticket.record(false);
4297 }
4298 return Err(ClientError::Request(e.without_url()));
4299 }
4300 }
4301 }
4302
4303 unreachable!("retry loop exited without returning a result — this is a bug")
4304}
4305
4306fn host_of(url: &str) -> Result<String, ClientError> {
4308 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
4309 parsed
4310 .host_str()
4311 .map(str::to_owned)
4312 .ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))
4313}
4314
4315fn url_host_is_ip_literal(url: &str) -> Result<bool, ClientError> {
4319 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
4320 match parsed.host() {
4321 Some(url::Host::Ipv4(_) | url::Host::Ipv6(_)) => Ok(true),
4322 Some(url::Host::Domain(_)) => Ok(false),
4323 None => Err(ClientError::InvalidUrl(format!("URL has no host: {url}"))),
4324 }
4325}
4326
4327fn scheme_is_https(url: &str) -> Result<bool, ClientError> {
4329 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
4330 Ok(parsed.scheme().eq_ignore_ascii_case("https"))
4331}
4332
4333fn redirect_target(resp: &Response, base: &str) -> Result<Option<String>, ClientError> {
4340 redirect_location(resp.status(), resp.headers(), base)
4341}
4342
4343fn redirect_location(
4346 status: reqwest::StatusCode,
4347 headers: &HeaderMap,
4348 base: &str,
4349) -> Result<Option<String>, ClientError> {
4350 match status {
4356 reqwest::StatusCode::MOVED_PERMANENTLY
4357 | reqwest::StatusCode::FOUND
4358 | reqwest::StatusCode::SEE_OTHER
4359 | reqwest::StatusCode::TEMPORARY_REDIRECT
4360 | reqwest::StatusCode::PERMANENT_REDIRECT => {}
4361 _ => return Ok(None),
4362 }
4363 let Some(location) = headers
4364 .get(reqwest::header::LOCATION)
4365 .and_then(|v| v.to_str().ok())
4366 else {
4367 return Ok(None);
4368 };
4369 let base_url = url::Url::parse(base).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
4370 let joined = base_url
4371 .join(location)
4372 .map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
4373 Ok(Some(joined.to_string()))
4374}
4375
4376fn strip_sensitive_headers_if_cross_origin(
4387 headers: &mut HeaderMap,
4388 original: &url::Url,
4389 current: &str,
4390) -> Result<(), ClientError> {
4391 let current_url =
4392 url::Url::parse(current).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
4393 if current_url.origin() != original.origin() {
4394 headers.remove(reqwest::header::AUTHORIZATION);
4396 headers.remove(reqwest::header::COOKIE);
4397 headers.remove("cookie2");
4398 headers.remove(reqwest::header::PROXY_AUTHORIZATION);
4399 headers.remove(reqwest::header::WWW_AUTHENTICATE);
4400 headers.remove("signature");
4402 headers.remove("signature-input");
4403 headers.remove("signature-agent");
4404 }
4405 Ok(())
4406}
4407
4408fn rewrite_after_redirect(
4428 status: reqwest::StatusCode,
4429 method: &mut Method,
4430 body: &mut Option<Bytes>,
4431 headers: &mut HeaderMap,
4432) {
4433 match status.as_u16() {
4434 303 => {
4435 if *method != Method::HEAD {
4436 *method = Method::GET;
4437 }
4438 *body = None;
4439 strip_payload_headers(headers);
4440 }
4441 301 | 302 if *method == Method::POST => {
4442 *method = Method::GET;
4443 *body = None;
4444 strip_payload_headers(headers);
4445 }
4446 _ => {}
4447 }
4448}
4449
4450fn strip_payload_headers(headers: &mut HeaderMap) {
4454 use reqwest::header::{
4455 CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH, CONTENT_TYPE, TRANSFER_ENCODING,
4456 };
4457 headers.remove(CONTENT_TYPE);
4458 headers.remove(CONTENT_LENGTH);
4459 headers.remove(TRANSFER_ENCODING);
4460 headers.remove(CONTENT_ENCODING);
4461 headers.remove(CONTENT_LANGUAGE);
4462}
4463
4464fn validate_resolved_addrs(addrs: Vec<SocketAddr>) -> Result<Vec<SocketAddr>, ClientError> {
4473 for addr in &addrs {
4474 if is_blocked_ip(addr.ip()) {
4475 return Err(ClientError::SsrfBlocked(addr.ip().to_string()));
4476 }
4477 }
4478 Ok(addrs)
4479}
4480
4481async fn resolve_and_validate(url: &str) -> Result<Vec<SocketAddr>, ClientError> {
4493 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
4494 let scheme = parsed.scheme();
4499 if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
4500 return Err(ClientError::InvalidUrl(format!(
4501 "unsupported URL scheme `{scheme}` (only http/https are allowed): {url}"
4502 )));
4503 }
4504 let port = parsed.port_or_known_default().ok_or_else(|| {
4505 ClientError::InvalidUrl(format!("URL has no port and unknown scheme: {url}"))
4506 })?;
4507 let host = parsed
4508 .host()
4509 .ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))?;
4510
4511 match host {
4512 url::Host::Ipv4(v4) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V4(v4), port)]),
4513 url::Host::Ipv6(v6) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V6(v6), port)]),
4514 url::Host::Domain(name) => {
4515 let addrs: Vec<SocketAddr> = tokio::net::lookup_host((name, port))
4516 .await
4517 .map_err(|e| ClientError::InvalidUrl(format!("DNS lookup failed for {name}: {e}")))?
4518 .collect();
4519 if addrs.is_empty() {
4520 return Err(ClientError::InvalidUrl(format!(
4521 "DNS lookup for {name} returned no addresses"
4522 )));
4523 }
4524 validate_resolved_addrs(addrs)
4527 }
4528 }
4529}
4530
4531fn parse_retry_after(headers: &HeaderMap) -> Option<Duration> {
4532 let value = headers.get("retry-after")?.to_str().ok()?;
4533 if let Ok(secs) = value.parse::<u64>() {
4535 return Some(Duration::from_secs(secs));
4536 }
4537 let dt = chrono::DateTime::parse_from_rfc2822(value).ok()?;
4539 let now = crate::time::ambient_now();
4540 let future = dt.with_timezone(&chrono::Utc);
4541 let secs = u64::try_from((future - now).num_seconds().max(0)).unwrap_or(0);
4542 Some(Duration::from_secs(secs))
4543}
4544
4545const REDACTED_HEADERS: &[&str] = &["authorization", "cookie", "set-cookie"];
4546
4547fn is_sensitive_header(name: &str) -> bool {
4548 REDACTED_HEADERS
4549 .iter()
4550 .any(|h| h.eq_ignore_ascii_case(name))
4551}
4552
4553fn log_request(
4554 method: &str,
4555 url: &reqwest::Url,
4556 status: u16,
4557 elapsed: Duration,
4558 headers: &HeaderMap,
4559) {
4560 let host = url.host_str().unwrap_or("unknown");
4561 let path = url.path();
4562
4563 let sent_headers: Vec<&str> = headers
4565 .keys()
4566 .map(HeaderName::as_str)
4567 .filter(|k| !is_sensitive_header(k))
4568 .collect();
4569
4570 tracing::info!(
4571 http.method = method,
4572 http.host = host,
4573 http.path = path,
4574 http.status = status,
4575 http.elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
4576 http.sent_headers = ?sent_headers,
4577 "outbound request"
4578 );
4579}
4580
4581fn client_attempt_span(method: &Method, url: &str, attempt: u32) -> tracing::Span {
4586 let span = tracing::info_span!(
4587 "http.client.request",
4588 otel.name = %method,
4589 otel.kind = "client",
4590 http.request.method = %method,
4591 server.address = tracing::field::Empty,
4592 url.path = tracing::field::Empty,
4593 http.request.resend_count = tracing::field::Empty,
4594 http.response.status_code = tracing::field::Empty,
4595 error.type = tracing::field::Empty,
4596 otel.status_code = tracing::field::Empty,
4597 );
4598 if attempt > 0 {
4600 span.record("http.request.resend_count", attempt);
4601 }
4602 if !span.is_disabled()
4604 && let Ok(parsed) = reqwest::Url::parse(url)
4605 {
4606 span.record("server.address", parsed.host_str().unwrap_or_default());
4607 span.record("url.path", parsed.path());
4608 }
4609 span
4610}
4611
4612fn record_attempt_status(span: &tracing::Span, status: u16) {
4615 span.record("http.response.status_code", status);
4616 if status >= 400 {
4617 span.record("otel.status_code", "ERROR");
4618 }
4619}
4620
4621fn record_attempt_error(span: &tracing::Span, kind: &'static str) {
4623 span.record("error.type", kind);
4624 span.record("otel.status_code", "ERROR");
4625}
4626
4627async fn read_body_in_span(
4631 response: reqwest::Response,
4632 span: &tracing::Span,
4633 gate: &RetryGate,
4634) -> Result<Bytes, ClientError> {
4635 use tracing::Instrument as _;
4636
4637 let body = response.bytes().instrument(span.clone()).await;
4638 if body.is_err() {
4639 record_attempt_error(span, "body");
4640 }
4641 body.map_err(|e| gate.body_error(e))
4642}
4643
4644async fn send_in_span(
4650 request: reqwest::RequestBuilder,
4651 span: &tracing::Span,
4652) -> Result<reqwest::Response, reqwest::Error> {
4653 use tracing::Instrument as _;
4654
4655 let sent = request.send().instrument(span.clone()).await;
4656 record_send_outcome(span, &sent);
4657 sent
4658}
4659
4660fn record_send_outcome(span: &tracing::Span, sent: &Result<reqwest::Response, reqwest::Error>) {
4663 match sent {
4664 Ok(response) => record_attempt_status(span, response.status().as_u16()),
4665 Err(error) => {
4666 let kind = if error.is_timeout() {
4667 "timeout"
4668 } else if error.is_connect() {
4669 "connect"
4670 } else {
4671 "request"
4672 };
4673 record_attempt_error(span, kind);
4674 }
4675 }
4676}
4677
4678fn inject_trace_context(
4681 builder: reqwest::RequestBuilder,
4682 caller_headers: &HeaderMap,
4683) -> reqwest::RequestBuilder {
4684 let mut builder = builder;
4685 for (name, value) in trace_headers_to_send(trace_context_headers(), caller_headers) {
4686 builder = builder.header(name, value);
4687 }
4688 builder
4689}
4690
4691fn trace_headers_to_send(
4698 injected: Vec<(String, HeaderValue)>,
4699 caller_headers: &HeaderMap,
4700) -> Vec<(String, HeaderValue)> {
4701 const W3C_PAIR: [&str; 2] = ["traceparent", "tracestate"];
4702 let caller_has_context = W3C_PAIR
4703 .iter()
4704 .any(|name| caller_headers.contains_key(*name));
4705 injected
4706 .into_iter()
4707 .filter(|(name, _)| {
4708 let paired = caller_has_context && W3C_PAIR.contains(&name.as_str());
4709 !paired && !caller_headers.contains_key(name.as_str())
4710 })
4711 .collect()
4712}
4713
4714fn add_current_request_id(headers: &mut HeaderMap) {
4717 static X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id");
4718 if headers.contains_key(&X_REQUEST_ID) {
4719 return;
4720 }
4721 if let Some(value) =
4722 crate::log::context::current_request_id().and_then(|id| HeaderValue::from_str(&id).ok())
4723 {
4724 headers.insert(X_REQUEST_ID.clone(), value);
4725 }
4726}
4727
4728#[allow(clippy::missing_const_for_fn)]
4731fn trace_context_headers() -> Vec<(String, HeaderValue)> {
4732 #[cfg(not(feature = "telemetry-otlp"))]
4733 {
4734 Vec::new()
4735 }
4736 #[cfg(feature = "telemetry-otlp")]
4737 {
4738 use std::collections::HashMap;
4739 use tracing_opentelemetry::OpenTelemetrySpanExt as _;
4740 let cx = tracing::Span::current().context();
4741 let mut map = HashMap::<String, String>::new();
4742 opentelemetry::global::get_text_map_propagator(|propagator| {
4743 propagator.inject_context(&cx, &mut TraceHeaderInjector(&mut map));
4744 });
4745 map.into_iter()
4746 .filter_map(|(name, value)| Some((name, HeaderValue::from_str(&value).ok()?)))
4747 .collect()
4748 }
4749}
4750
4751#[cfg(feature = "telemetry-otlp")]
4752struct TraceHeaderInjector<'a>(&'a mut std::collections::HashMap<String, String>);
4753
4754#[cfg(feature = "telemetry-otlp")]
4755impl opentelemetry::propagation::Injector for TraceHeaderInjector<'_> {
4756 fn set(&mut self, key: &str, value: String) {
4757 self.0.insert(key.to_owned(), value);
4758 }
4759}
4760
4761#[cfg(test)]
4764mod tests {
4765 use super::*;
4766
4767 #[cfg(feature = "reporting")]
4770 #[test]
4771 fn throttled_locally_round_trips_through_a_capsule() {
4772 let err = ClientError::ThrottledLocally {
4773 host: "api.example.com:8443".to_owned(),
4774 };
4775 let kind = http_error_kind(&err);
4776 assert_eq!(
4777 kind,
4778 crate::capsule::schema::HttpErrorKind::ThrottledLocally
4779 );
4780 match rebuild_client_error(Some(kind), err.to_string()) {
4781 ClientError::ThrottledLocally { host } => assert_eq!(host, "api.example.com:8443"),
4782 other => panic!("rebuilt as {other:?}"),
4783 }
4784 }
4785
4786 #[cfg(feature = "reporting")]
4789 #[test]
4790 fn opaque_header_values_record_without_collision() {
4791 let recorded = |value: HeaderValue| {
4792 let mut builder = Client::new().get("http://example.test/");
4793 builder.extra_headers.insert("x-sig", value);
4794 caller_headers(&builder).remove(0).1
4795 };
4796 let first = recorded(HeaderValue::from_bytes(b"a\xfe").expect("value"));
4797 let second = recorded(HeaderValue::from_bytes(b"a\xff").expect("value"));
4798 assert_ne!(first, second);
4799 assert_eq!(recorded(HeaderValue::from_static("plain")), "plain");
4800 let look_alike = recorded(HeaderValue::from_static("[bytes] a\\xfe"));
4801 assert_ne!(look_alike, first);
4802 }
4803
4804 #[cfg(feature = "reporting")]
4806 #[test]
4807 fn fault_injected_round_trips_through_a_capsule() {
4808 let err = ClientError::FaultInjected("fault injection: injected http error".to_owned());
4809 let kind = http_error_kind(&err);
4810 assert_eq!(kind, crate::capsule::schema::HttpErrorKind::FaultInjected);
4811 match rebuild_client_error(Some(kind), err.to_string()) {
4812 ClientError::FaultInjected(text) => {
4813 assert_eq!(text, "fault injection: injected http error");
4814 }
4815 other => panic!("rebuilt as {other:?}"),
4816 }
4817 }
4818
4819 #[tokio::test]
4822 #[allow(clippy::await_holding_lock)]
4823 async fn throttled_calls_do_not_open_the_circuit_breaker() {
4824 let _lock = crate::circuit_breaker::TEST_LOCK
4825 .lock()
4826 .unwrap_or_else(std::sync::PoisonError::into_inner);
4827 crate::circuit_breaker::global_registry().clear();
4828
4829 let app = axum::Router::new().route(
4831 "/x",
4832 axum::routing::get(|| async { axum::http::StatusCode::TOO_MANY_REQUESTS }),
4833 );
4834 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4835 let addr = listener.local_addr().unwrap();
4836 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
4837
4838 let mut config = HttpClientConfig::default();
4839 config.adaptive_throttle.enabled = true;
4840 let client = Client::from_config(&config);
4841 let url = format!("http://{addr}/x");
4842 let mut throttled = 0;
4843 for _ in 0..200 {
4844 match client.get(&url).no_retry().send().await {
4845 Err(ClientError::ThrottledLocally { .. }) => throttled += 1,
4846 Ok(r) => assert_eq!(r.status().as_u16(), 429),
4847 Err(e) => panic!("unexpected error: {e}"),
4848 }
4849 }
4850 assert!(
4851 throttled > 50,
4852 "the throttle must reject locally: {throttled}"
4853 );
4854 let breaker = crate::circuit_breaker::global_registry().get_or_create(
4855 &addr.to_string(),
4856 crate::circuit_breaker::CircuitBreakerPolicy::default(),
4857 );
4858 assert_eq!(
4859 breaker.state(),
4860 crate::circuit_breaker::CircuitState::Closed,
4861 "local rejects must not open the breaker"
4862 );
4863 crate::circuit_breaker::global_registry().clear();
4864 }
4865
4866 #[test]
4869 fn from_state_shares_a_throttle_built_from_the_effective_config() {
4870 let state = crate::AppState::for_test();
4871 let boot = HttpClientConfig::default();
4872 install_shared_throttle(&state, &boot);
4873 assert!(state.extension::<SharedThrottle>().is_none(), "off at boot");
4874 assert!(Client::from_state(&state).throttle.is_none());
4875
4876 let mut replaced = crate::config::HttpConfig::default();
4877 replaced.client.adaptive_throttle.enabled = true;
4878 state.insert_extension(replaced.clone());
4879 let a = Client::from_state(&state)
4880 .throttle
4881 .expect("on after replace");
4882 let b = Client::from_state(&state)
4883 .throttle
4884 .expect("on after replace");
4885 assert!(Arc::ptr_eq(&a, &b), "clients share one throttle");
4886
4887 replaced.client.adaptive_throttle.k = 3.0;
4888 state.insert_extension(replaced.clone());
4889 let c = Client::from_state(&state).throttle.expect("still on");
4890 assert!(!Arc::ptr_eq(&a, &c), "new settings build a new throttle");
4891
4892 replaced.client.adaptive_throttle.enabled = false;
4893 state.insert_extension(replaced);
4894 assert!(Client::from_state(&state).throttle.is_none(), "off again");
4895 }
4896
4897 #[test]
4900 fn from_state_rebuilds_the_retry_budgets_when_their_config_changes() {
4901 let state = crate::AppState::for_test();
4902 let mut config = crate::config::HttpConfig::default();
4903 state.insert_extension(config.clone());
4904 let a = Client::from_state(&state).retry.budgets.expect("on");
4905 let b = Client::from_state(&state).retry.budgets.expect("on");
4906 assert!(Arc::ptr_eq(&a, &b), "clients share one set of budgets");
4907
4908 config.client.retry_budget.capacity = 50;
4909 state.insert_extension(config.clone());
4910 let c = Client::from_state(&state).retry.budgets.expect("still on");
4911 assert!(!Arc::ptr_eq(&a, &c), "new settings build new budgets");
4912 assert!((c.for_host("h:80").available() - 50.0).abs() < f64::EPSILON);
4913
4914 config.client.retry_budget.enabled = false;
4915 state.insert_extension(config);
4916 assert!(Client::from_state(&state).retry.budgets.is_none(), "off");
4917 }
4918
4919 #[test]
4922 fn concurrent_first_calls_share_one_throttle() {
4923 let state = crate::AppState::for_test();
4924 let mut config = crate::config::HttpConfig::default();
4925 config.client.adaptive_throttle.enabled = true;
4926 state.insert_extension(config);
4927 let throttles: Vec<_> = std::thread::scope(|scope| {
4928 let mut handles = Vec::with_capacity(8);
4930 for _ in 0..8 {
4931 handles.push(scope.spawn(|| Client::from_state(&state).throttle.expect("on")));
4932 }
4933 handles
4934 .into_iter()
4935 .map(|h| h.join().expect("thread"))
4936 .collect()
4937 });
4938 assert!(
4939 throttles.iter().all(|t| Arc::ptr_eq(t, &throttles[0])),
4940 "every client must share the one throttle"
4941 );
4942 }
4943
4944 #[tokio::test]
4947 #[allow(clippy::await_holding_lock)]
4948 async fn a_truncated_body_is_not_an_accept() {
4949 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
4950 let _lock = crate::circuit_breaker::TEST_LOCK
4951 .lock()
4952 .unwrap_or_else(std::sync::PoisonError::into_inner);
4953 crate::circuit_breaker::global_registry().clear();
4954
4955 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4958 let addr = listener.local_addr().unwrap();
4959 tokio::spawn(async move {
4960 loop {
4961 let Ok((mut socket, _)) = listener.accept().await else {
4962 return;
4963 };
4964 let mut buf = [0_u8; 1024];
4965 let _ = socket.read(&mut buf).await;
4966 let _ = socket
4967 .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 100\r\n\r\nabc")
4968 .await;
4969 drop(socket);
4970 }
4971 });
4972
4973 let mut config = HttpClientConfig::default();
4974 config.adaptive_throttle.enabled = true;
4975 let client = Client::from_config(&config);
4976 let url = format!("http://{addr}/x");
4977 for _ in 0..20 {
4978 let res = client.get(&url).no_retry().send().await;
4979 assert!(
4980 matches!(
4981 res,
4982 Err(ClientError::Request(_) | ClientError::ThrottledLocally { .. })
4983 ),
4984 "a truncated body is an error: {res:?}"
4985 );
4986 }
4987 let throttle = client.throttle.as_ref().expect("on");
4988 assert!(
4989 throttle.reject_probability(&addr.to_string(), crate::time::ambient_instant()) > 0.5,
4990 "truncated bodies must count as rejects"
4991 );
4992 crate::circuit_breaker::global_registry().clear();
4993 }
4994
4995 #[test]
4996 fn throttle_host_keeps_an_explicit_port() {
4997 assert_eq!(
4998 throttle_host("https://a.example/x").as_deref(),
4999 Some("a.example")
5000 );
5001 assert_eq!(
5002 throttle_host("http://a.example:8080/x").as_deref(),
5003 Some("a.example:8080")
5004 );
5005 assert_eq!(throttle_host("not a url"), None);
5006 assert!(throttle_accepts(200) && throttle_accepts(500));
5007 assert!(!throttle_accepts(429) && !throttle_accepts(503));
5008 }
5009 use crate::config::HttpClientConfig;
5010
5011 #[test]
5013 fn client_constructs_with_defaults() {
5014 let client = Client::new();
5015 assert!(client.alias.is_none());
5016 assert!(client.base_url.is_none());
5017 assert_eq!(client.retry_policy.max_retries, 3);
5018 }
5019
5020 #[test]
5022 fn request_builder_fluent_api_compiles() {
5023 let client = Client::new();
5024 let _builder = client
5025 .post("https://example.com/api")
5026 .header("x-api-key", "secret")
5027 .json(&serde_json::json!({"key": "value"}))
5028 .retries(2);
5029 }
5030
5031 #[test]
5033 fn response_accessors_work() {
5034 let payload = serde_json::json!({"id": 42, "name": "Alice"});
5035 let body = serde_json::to_vec(&payload).unwrap();
5036 let resp = Response {
5037 status: reqwest::StatusCode::OK,
5038 headers: HeaderMap::new(),
5039 body: Bytes::from(body),
5040 url: None,
5041 };
5042 assert_eq!(resp.status().as_u16(), 200);
5043 assert!(resp.is_success());
5044 }
5045
5046 #[test]
5048 fn response_json_deserialises() {
5049 #[derive(serde::Deserialize, PartialEq, Debug)]
5050 struct User {
5051 id: i32,
5052 name: String,
5053 }
5054 let payload = serde_json::json!({"id": 1, "name": "Bob"});
5055 let resp = Response {
5056 status: reqwest::StatusCode::OK,
5057 headers: HeaderMap::new(),
5058 body: Bytes::from(serde_json::to_vec(&payload).unwrap()),
5059 url: None,
5060 };
5061 let user: User = resp.json().unwrap();
5062 assert_eq!(user.id, 1);
5063 assert_eq!(user.name, "Bob");
5064 }
5065
5066 #[test]
5068 fn response_text_returns_string() {
5069 let resp = Response {
5070 status: reqwest::StatusCode::OK,
5071 headers: HeaderMap::new(),
5072 body: Bytes::from_static(b"hello world"),
5073 url: None,
5074 };
5075 assert_eq!(resp.text(), "hello world");
5076 }
5077
5078 #[test]
5080 fn response_bytes_returns_raw() {
5081 let resp = Response {
5082 status: reqwest::StatusCode::CREATED,
5083 headers: HeaderMap::new(),
5084 body: Bytes::from_static(b"\x00\x01\x02"),
5085 url: None,
5086 };
5087 assert_eq!(resp.bytes(), Bytes::from_static(b"\x00\x01\x02"));
5088 }
5089
5090 #[test]
5092 fn config_deserialises_from_toml() {
5093 let toml = r#"
5095 [client]
5096 timeout_secs = 60
5097 max_retries = 5
5098 [client.base_urls]
5099 stripe = "https://api.stripe.com"
5100 sendgrid = "https://api.sendgrid.com"
5101 "#;
5102 let http_cfg: crate::config::HttpConfig = toml::from_str(toml).unwrap();
5103 let config = &http_cfg.client;
5104 assert_eq!(config.timeout_secs, 60);
5105 assert_eq!(config.max_retries, 5);
5106 assert_eq!(
5107 config.base_urls.get("stripe").map(String::as_str),
5108 Some("https://api.stripe.com")
5109 );
5110 assert_eq!(
5111 config.base_urls.get("sendgrid").map(String::as_str),
5112 Some("https://api.sendgrid.com")
5113 );
5114 }
5115
5116 #[test]
5118 fn config_has_correct_defaults() {
5119 let config = HttpClientConfig::default();
5120 assert_eq!(config.timeout_secs, 30);
5121 assert_eq!(config.max_retries, 3);
5122 assert!(config.base_urls.is_empty());
5123 }
5124
5125 #[test]
5127 fn idempotent_method_classification() {
5128 assert!(is_idempotent_method(&Method::GET));
5129 assert!(is_idempotent_method(&Method::HEAD));
5130 assert!(is_idempotent_method(&Method::PUT));
5131 assert!(is_idempotent_method(&Method::DELETE));
5132 assert!(is_idempotent_method(&Method::OPTIONS));
5133 assert!(is_idempotent_method(&Method::TRACE));
5134 assert!(!is_idempotent_method(&Method::POST));
5135 assert!(!is_idempotent_method(&Method::PATCH));
5136 }
5137
5138 #[test]
5140 fn retryable_status_classification() {
5141 assert!(is_retryable_status(502));
5142 assert!(is_retryable_status(503));
5143 assert!(is_retryable_status(504));
5144 assert!(!is_retryable_status(200));
5145 assert!(!is_retryable_status(400));
5146 assert!(!is_retryable_status(404));
5147 assert!(!is_retryable_status(500));
5148 assert!(!is_retryable_status(429));
5149 }
5150
5151 #[test]
5153 fn retry_after_header_parsing() {
5154 let mut headers = HeaderMap::new();
5155 headers.insert(
5156 reqwest::header::HeaderName::from_static("retry-after"),
5157 HeaderValue::from_static("5"),
5158 );
5159 assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(5)));
5160
5161 let empty = HeaderMap::new();
5162 assert_eq!(parse_retry_after(&empty), None);
5163 }
5164
5165 #[test]
5167 fn sensitive_header_detection() {
5168 assert!(is_sensitive_header("authorization"));
5169 assert!(is_sensitive_header("Authorization"));
5170 assert!(is_sensitive_header("AUTHORIZATION"));
5171 assert!(is_sensitive_header("cookie"));
5172 assert!(is_sensitive_header("set-cookie"));
5173 assert!(!is_sensitive_header("content-type"));
5174 assert!(!is_sensitive_header("x-api-key"));
5175 }
5176
5177 #[tokio::test]
5179 async fn mock_registry_captures_calls() {
5180 let registry = Arc::new(MockRegistry::new());
5181 let call_count = Arc::new(AtomicUsize::new(0));
5182
5183 registry.register(MockEntry {
5184 method: Some(Method::POST),
5185 path: "/charges".to_owned(),
5186 alias: Some("stripe".to_owned()),
5187 status: 200,
5188 body: Some(serde_json::json!({"id": "ch_123"})),
5189 call_count: call_count.clone(),
5190 });
5191
5192 let client = Client::new().with_mock(registry).named("stripe");
5193
5194 let resp = client
5195 .post("https://api.stripe.com/charges")
5196 .json(&serde_json::json!({"amount": 1000}))
5197 .send()
5198 .await
5199 .unwrap();
5200
5201 assert_eq!(resp.status().as_u16(), 200);
5202 let body: serde_json::Value = resp.json().unwrap();
5203 assert_eq!(body["id"], "ch_123");
5204 assert_eq!(call_count.load(Ordering::SeqCst), 1);
5205 }
5206
5207 #[tokio::test]
5209 async fn mock_handle_expect_called_passes() {
5210 let registry = Arc::new(MockRegistry::new());
5211 let call_count = Arc::new(AtomicUsize::new(0));
5212
5213 registry.register(MockEntry {
5214 method: Some(Method::GET),
5215 path: "/users/1".to_owned(),
5216 alias: None,
5217 status: 200,
5218 body: Some(serde_json::json!({"name": "Alice"})),
5219 call_count: call_count.clone(),
5220 });
5221
5222 let handle = MockHandle {
5223 alias: "test".to_owned(),
5224 method: "GET".to_owned(),
5225 path: "/users/1".to_owned(),
5226 call_count: call_count.clone(),
5227 };
5228
5229 let client = Client::new().with_mock(registry);
5230 client
5231 .get("https://api.example.com/users/1")
5232 .send()
5233 .await
5234 .unwrap();
5235
5236 handle.expect_called(1);
5237 assert_eq!(handle.call_count(), 1);
5238 }
5239
5240 #[tokio::test]
5242 async fn mock_matches_by_path_suffix() {
5243 let registry = Arc::new(MockRegistry::new());
5244 let call_count = Arc::new(AtomicUsize::new(0));
5245
5246 registry.register(MockEntry {
5247 method: Some(Method::POST),
5248 path: "/v1/charges".to_owned(),
5249 alias: None,
5250 status: 201,
5251 body: Some(serde_json::json!({"created": true})),
5252 call_count: call_count.clone(),
5253 });
5254
5255 let client = Client::new().with_mock(registry);
5256 let resp = client
5257 .post("https://api.stripe.com/v1/charges")
5258 .send()
5259 .await
5260 .unwrap();
5261
5262 assert_eq!(resp.status().as_u16(), 201);
5263 assert_eq!(call_count.load(Ordering::SeqCst), 1);
5264 }
5265
5266 #[tokio::test]
5268 async fn no_mock_error_when_unmatched() {
5269 let registry = Arc::new(MockRegistry::new());
5270 let client = Client::new().with_mock(registry);
5271 let result = client.post("https://api.example.com/unknown").send().await;
5272 assert!(matches!(result, Err(ClientError::NoMock(_, _))));
5273 }
5274
5275 #[tokio::test]
5277 async fn mock_setup_builder_registers_entry() {
5278 let registry = Arc::new(MockRegistry::new());
5279 let builder = MockSetupBuilder {
5280 registry: registry.clone(),
5281 alias: "myservice".to_owned(),
5282 method: None,
5283 path: None,
5284 };
5285
5286 let handle = builder
5287 .post("/api/resource")
5288 .respond_with(201, serde_json::json!({"ok": true}));
5289
5290 let client = Client::new().with_mock(registry).named("myservice");
5291 client
5292 .post("https://myservice.example.com/api/resource")
5293 .send()
5294 .await
5295 .unwrap();
5296
5297 handle.expect_called(1);
5298 }
5299
5300 #[test]
5302 fn client_from_config() {
5303 let config = HttpClientConfig {
5304 timeout_secs: 10,
5305 max_retries: 1,
5306 max_retry_after_secs: 10,
5307 max_backoff_ms: 20_000,
5308 base_urls: std::collections::HashMap::new(),
5309 ..HttpClientConfig::default()
5310 };
5311 let client = Client::from_config(&config);
5312 assert_eq!(client.retry_policy.max_retries, 1);
5313 }
5314
5315 #[test]
5317 fn named_client_preserves_mock_registry() {
5318 let registry = Arc::new(MockRegistry::new());
5319 let client = Client::new().with_mock(registry);
5320 let named = client.named("stripe");
5321 assert!(named.mock.is_some());
5322 assert_eq!(named.alias.as_deref(), Some("stripe"));
5323 }
5324
5325 #[test]
5327 fn base_url_prepended_to_relative_path() {
5328 let client = Client::new();
5329 let client = client.with_base_url("https://api.stripe.com");
5330 let builder = client.post("/v1/charges");
5331 assert_eq!(builder.url, "https://api.stripe.com/v1/charges");
5332 }
5333
5334 #[test]
5336 fn absolute_url_bypasses_base_url() {
5337 let client = Client::new().with_base_url("https://ignored.example.com");
5338 let builder = client.get("https://actual.example.com/path");
5339 assert_eq!(builder.url, "https://actual.example.com/path");
5340 }
5341
5342 #[test]
5344 fn retry_override_per_request() {
5345 let client = Client::new(); let builder = client.get("https://example.com").retries(0);
5347 assert_eq!(builder.retry_policy.max_retries, 0);
5348
5349 let no_retry = client.get("https://example.com").no_retry();
5350 assert_eq!(no_retry.retry_policy.max_retries, 0);
5351 }
5352
5353 #[tokio::test]
5355 async fn client_extracts_from_state() {
5356 use axum::extract::FromRequestParts;
5357 let state = crate::AppState::for_test();
5358 let mut parts = axum::http::Request::new(axum::body::Body::empty())
5359 .into_parts()
5360 .0;
5361 let client = Client::from_request_parts(&mut parts, &state)
5362 .await
5363 .unwrap();
5364 assert!(client.mock.is_none());
5366 assert!(client.alias.is_none());
5367 }
5368
5369 #[test]
5371 fn mock_registry_ext_round_trips_through_state() {
5372 let registry = Arc::new(MockRegistry::new());
5373 let ext = HttpMockRegistryExt(registry);
5374 let state = crate::AppState::for_test();
5375 state.insert_extension(ext);
5376 let retrieved = state.extension::<HttpMockRegistryExt>();
5377 assert!(retrieved.is_some());
5378 }
5379
5380 #[test]
5382 fn named_client_resolves_base_url_from_config() {
5383 let mut base_urls = std::collections::HashMap::new();
5384 base_urls.insert("stripe".to_owned(), "https://api.stripe.com".to_owned());
5385 let config = HttpClientConfig {
5386 timeout_secs: 30,
5387 max_retries: 3,
5388 max_retry_after_secs: 10,
5389 max_backoff_ms: 20_000,
5390 base_urls,
5391 ..HttpClientConfig::default()
5392 };
5393 let client = Client::from_config(&config);
5394 let stripe = client.named("stripe");
5395 assert_eq!(stripe.base_url.as_deref(), Some("https://api.stripe.com"));
5396 assert_eq!(stripe.alias.as_deref(), Some("stripe"));
5397
5398 let other = client.named("sendgrid");
5400 assert!(other.base_url.is_none());
5401 }
5402
5403 #[test]
5416 fn breaker_scoped_keys_by_resolved_url_not_alias() {
5417 let _lock = crate::circuit_breaker::TEST_LOCK
5423 .lock()
5424 .unwrap_or_else(std::sync::PoisonError::into_inner);
5425
5426 let mut base_urls = std::collections::HashMap::new();
5427 base_urls.insert(
5428 "hook-service".to_owned(),
5429 "http://mock-receiver/base".to_owned(),
5430 );
5431 let config = HttpClientConfig {
5432 timeout_secs: 30,
5433 max_retries: 3,
5434 max_retry_after_secs: 10,
5435 max_backoff_ms: 20_000,
5436 base_urls,
5437 ..HttpClientConfig::default()
5438 };
5439 let client = Client::from_config(&config);
5440
5441 let req = client
5442 .named("hook-service")
5443 .post("hook-service")
5444 .ssrf_safe()
5445 .breaker_scoped();
5446 assert_eq!(req.url, "http://mock-receiver/base/hook-service");
5447 assert!(req.breaker_scoped);
5448
5449 assert!(url::Url::parse("hook-service").is_err());
5452 let breaker = super::breaker_for_url(None, &req.url);
5453 assert_eq!(breaker.name(), "mock-receiver");
5454
5455 crate::circuit_breaker::global_registry().clear();
5456 }
5457
5458 #[tokio::test]
5470 #[allow(clippy::await_holding_lock)]
5471 async fn breaker_scoped_custom_path_trips_and_fails_fast() {
5472 let _lock = crate::circuit_breaker::TEST_LOCK
5473 .lock()
5474 .unwrap_or_else(std::sync::PoisonError::into_inner);
5475 crate::circuit_breaker::global_registry().clear();
5476
5477 let mut rc = crate::config::ResilienceConfig::default();
5478 rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
5479 rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
5480 rc.circuit_breaker.defaults.open_duration_secs = Some(10);
5481 let client = Client {
5482 resilience_config: Some(Arc::new(rc)),
5483 ..Client::new()
5484 };
5485
5486 let url = "http://127.0.0.1:1/blocked";
5487
5488 for _ in 0..3 {
5489 let res = client.post(url).ssrf_safe().breaker_scoped().send().await;
5490 assert!(
5491 matches!(res, Err(ClientError::SsrfBlocked(_))),
5492 "expected SsrfBlocked, got {res:?}"
5493 );
5494 }
5495
5496 let res = client.post(url).ssrf_safe().breaker_scoped().send().await;
5501 assert!(matches!(res, Err(ClientError::CircuitBreakerOpen)));
5502
5503 crate::circuit_breaker::global_registry().clear();
5504 }
5505
5506 #[tokio::test]
5510 #[allow(clippy::await_holding_lock)]
5511 async fn open_breaker_calls_are_not_charged_to_the_throttle() {
5512 let _lock = crate::circuit_breaker::TEST_LOCK
5513 .lock()
5514 .unwrap_or_else(std::sync::PoisonError::into_inner);
5515 crate::circuit_breaker::global_registry().clear();
5516
5517 let mut rc = crate::config::ResilienceConfig::default();
5518 rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
5519 rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
5520 rc.circuit_breaker.defaults.open_duration_secs = Some(10);
5521 let rc = Arc::new(rc);
5522 let url = "http://127.0.0.1:1/blocked";
5523
5524 let plain = Client {
5526 resilience_config: Some(Arc::clone(&rc)),
5527 ..Client::new()
5528 };
5529 for _ in 0..3 {
5530 let _ = plain.post(url).ssrf_safe().breaker_scoped().send().await;
5531 }
5532
5533 let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
5534 2.0,
5535 Duration::from_secs(120),
5536 ));
5537 let throttled = Client {
5538 resilience_config: Some(rc),
5539 throttle: Some(Arc::clone(&throttle)),
5540 ..Client::new()
5541 };
5542 for _ in 0..50 {
5543 let res = throttled
5544 .post(url)
5545 .ssrf_safe()
5546 .breaker_scoped()
5547 .send()
5548 .await;
5549 assert!(
5550 matches!(res, Err(ClientError::CircuitBreakerOpen)),
5551 "an open breaker fails fast, not through the throttle: {res:?}"
5552 );
5553 }
5554 assert!(
5555 throttle.reject_probability("127.0.0.1:1", crate::time::ambient_instant())
5556 < f64::EPSILON,
5557 "fail-fast calls must not be counted"
5558 );
5559 crate::circuit_breaker::global_registry().clear();
5560 }
5561
5562 #[tokio::test]
5569 #[allow(clippy::await_holding_lock)]
5570 async fn breaker_scoped_bypassed_for_mocked_client() {
5571 let _lock = crate::circuit_breaker::TEST_LOCK
5572 .lock()
5573 .unwrap_or_else(std::sync::PoisonError::into_inner);
5574 crate::circuit_breaker::global_registry().clear();
5575
5576 let registry = Arc::new(MockRegistry::new());
5577 let mock = MockSetupBuilder {
5578 registry: registry.clone(),
5579 alias: "http://mock-receiver/hook".to_owned(),
5580 method: None,
5581 path: None,
5582 }
5583 .post("/hook")
5584 .respond_with(500, serde_json::json!({ "error": "down" }));
5585 let client = Client::new().with_mock(registry);
5586
5587 for _ in 0..12 {
5591 let res = client
5592 .named("http://mock-receiver/hook")
5593 .post("http://mock-receiver/hook")
5594 .ssrf_safe()
5595 .breaker_scoped()
5596 .send()
5597 .await;
5598 let res = res.expect("a mocked client must never see CircuitBreakerOpen");
5599 assert_eq!(res.status().as_u16(), 500);
5600 }
5601 mock.expect_called(12);
5602
5603 assert!(
5605 crate::circuit_breaker::global_registry()
5606 .all_breakers()
5607 .iter()
5608 .all(|b| b.name() != "mock-receiver"),
5609 "a mocked send must never create a real breaker entry"
5610 );
5611
5612 crate::circuit_breaker::global_registry().clear();
5613 }
5614
5615 #[tokio::test]
5617 async fn client_extracts_from_autumn_config_in_state() {
5618 use axum::extract::FromRequestParts;
5619 let mut cfg = crate::config::AutumnConfig::default();
5620 cfg.http.client.max_retries = 7;
5621 let state = crate::AppState::for_test();
5622 state.insert_extension(cfg);
5623
5624 let mut parts = axum::http::Request::new(axum::body::Body::empty())
5625 .into_parts()
5626 .0;
5627 let client = Client::from_request_parts(&mut parts, &state)
5628 .await
5629 .unwrap();
5630 assert_eq!(client.retry_policy.max_retries, 7);
5631 }
5632
5633 #[tokio::test]
5635 async fn respond_with_status_produces_empty_body() {
5636 let registry = Arc::new(MockRegistry::new());
5637 let builder = MockSetupBuilder {
5638 registry: registry.clone(),
5639 alias: "svc".to_owned(),
5640 method: None,
5641 path: None,
5642 };
5643 let _handle = builder.delete("/items/1").respond_with_status(204);
5644
5645 let client = Client::new().with_mock(registry).named("svc");
5646 let resp = client
5647 .delete("https://svc.example.com/items/1")
5648 .send()
5649 .await
5650 .unwrap();
5651
5652 assert_eq!(resp.status().as_u16(), 204);
5653 assert_eq!(
5654 resp.bytes(),
5655 bytes::Bytes::new(),
5656 "body must be empty, not \"null\""
5657 );
5658 }
5659
5660 #[test]
5662 fn retry_after_http_date_parsing() {
5663 let mut headers = HeaderMap::new();
5664 headers.insert(
5666 reqwest::header::HeaderName::from_static("retry-after"),
5667 HeaderValue::from_static("Tue, 01 Jan 2030 00:00:00 GMT"),
5668 );
5669 let duration = parse_retry_after(&headers);
5670 assert!(duration.is_some(), "should parse HTTP-date Retry-After");
5671 assert!(
5672 duration.unwrap().as_secs() > 0,
5673 "future date should yield positive delay"
5674 );
5675 }
5676
5677 #[tokio::test]
5679 async fn non_idempotent_post_no_retry() {
5680 let registry = Arc::new(MockRegistry::new());
5681 let call_count = Arc::new(AtomicUsize::new(0));
5682 registry.register(MockEntry {
5683 method: Some(Method::POST),
5684 path: "/endpoint".to_owned(),
5685 alias: None,
5686 status: 503,
5687 body: None,
5688 call_count: call_count.clone(),
5689 });
5690
5691 let client = Client::new().with_mock(registry);
5693 let resp = client
5694 .post("https://example.com/endpoint")
5695 .send()
5696 .await
5697 .unwrap();
5698
5699 assert_eq!(resp.status().as_u16(), 503);
5700 assert_eq!(call_count.load(Ordering::SeqCst), 1);
5702 }
5703
5704 #[tokio::test]
5706 async fn mock_strips_query_from_url_before_matching() {
5707 let registry = Arc::new(MockRegistry::new());
5708 let call_count = Arc::new(AtomicUsize::new(0));
5709 registry.register(MockEntry {
5710 method: Some(Method::GET),
5711 path: "/v1/charges".to_owned(),
5712 alias: None,
5713 status: 200,
5714 body: Some(serde_json::json!({"ok": true})),
5715 call_count: call_count.clone(),
5716 });
5717
5718 let client = Client::new().with_mock(registry);
5720 let resp = client
5721 .get("https://api.stripe.com/v1/charges?expand[]=balance_transaction")
5722 .send()
5723 .await
5724 .unwrap();
5725
5726 assert_eq!(resp.status().as_u16(), 200);
5727 assert_eq!(call_count.load(Ordering::SeqCst), 1);
5728 }
5729
5730 #[tokio::test]
5732 async fn mock_suffix_match_with_leading_slash_path() {
5733 let registry = Arc::new(MockRegistry::new());
5734 let call_count = Arc::new(AtomicUsize::new(0));
5735 registry.register(MockEntry {
5737 method: Some(Method::POST),
5738 path: "/charges".to_owned(),
5739 alias: None,
5740 status: 201,
5741 body: Some(serde_json::json!({"matched": true})),
5742 call_count: call_count.clone(),
5743 });
5744
5745 let client = Client::new().with_mock(registry);
5746 let resp = client
5748 .post("https://api.stripe.com/v1/charges")
5749 .send()
5750 .await
5751 .unwrap();
5752
5753 assert_eq!(resp.status().as_u16(), 201);
5754 assert_eq!(call_count.load(Ordering::SeqCst), 1);
5755 }
5756
5757 #[test]
5760 fn retries_does_not_clear_idempotent_only_flag() {
5761 let client = Client::new();
5762 let builder = client.post("https://example.com").retries(2);
5763 assert_eq!(builder.retry_policy.max_retries, 2);
5764 assert!(builder.retry_policy.retry_idempotent_only);
5765 assert_eq!(builder.max_attempts(false), 1, "POST must not retry");
5766
5767 let builder = builder.retry_non_idempotent();
5768 assert!(!builder.retry_policy.retry_idempotent_only);
5769 assert_eq!(builder.max_attempts(false), 3);
5770 }
5771
5772 #[test]
5773 fn idempotency_key_is_added_only_for_opted_in_unsafe_retries() {
5774 let client = Client::new();
5775 let key_of = |builder: &RequestBuilder| {
5776 builder
5777 .extra_headers
5778 .get(IDEMPOTENCY_KEY)
5779 .map(|value| value.to_str().unwrap().to_owned())
5780 };
5781
5782 let mut post = client.post("https://example.com").retry_non_idempotent();
5783 post.ensure_idempotency_key();
5784 let first = key_of(&post).expect("an opted-in POST gets a key");
5785 post.ensure_idempotency_key();
5786 assert_eq!(key_of(&post), Some(first), "the key does not change");
5787
5788 let mut caller = client
5789 .post("https://example.com")
5790 .header("Idempotency-Key", "mine")
5791 .retry_non_idempotent();
5792 caller.ensure_idempotency_key();
5793 assert_eq!(key_of(&caller).as_deref(), Some("mine"));
5794
5795 for mut builder in [
5796 client.post("https://example.com").retries(3),
5797 client
5798 .post("https://example.com")
5799 .retry_non_idempotent()
5800 .no_retry(),
5801 client.get("https://example.com").retry_non_idempotent(),
5802 ] {
5803 builder.ensure_idempotency_key();
5804 assert_eq!(key_of(&builder), None);
5805 }
5806 }
5807
5808 #[test]
5809 fn retry_delay_is_jittered_and_capped() {
5810 let entropy = crate::entropy::SeededEntropy::new(9);
5811 let policy = RetryPolicy {
5812 max_backoff: Duration::from_secs(1),
5813 ..RetryPolicy::default()
5814 };
5815 for attempt in 0..40 {
5816 let delay = policy.retry_delay(&entropy, attempt, None);
5817 let ceiling = Duration::from_millis(crate::backoff::ceiling_ms(100, 1_000, attempt));
5818 assert!(delay <= ceiling, "attempt {attempt}: {delay:?}");
5819 }
5820 }
5821
5822 #[test]
5823 fn retry_delay_clamps_a_hint_to_the_backoff_window() {
5824 let entropy = crate::entropy::SeededEntropy::new(10);
5825 let policy = RetryPolicy::default();
5826 let wait = policy.retry_delay(&entropy, 0, Some(Duration::from_secs(2)));
5828 assert!(wait >= Duration::from_secs(2) && wait <= Duration::from_millis(2_100));
5829 let wait = policy.retry_delay(&entropy, 0, Some(Duration::from_secs(3_600)));
5831 assert!(wait >= Duration::from_secs(5) && wait <= Duration::from_millis(5_100));
5832 }
5833
5834 #[test]
5835 fn retry_hint_reads_retry_after_on_429_and_503_only() {
5836 let mut headers = HeaderMap::new();
5837 headers.insert("retry-after", HeaderValue::from_static("2"));
5838 assert_eq!(retry_hint(429, &headers), Some(Duration::from_secs(2)));
5839 assert_eq!(retry_hint(503, &headers), Some(Duration::from_secs(2)));
5840 assert_eq!(retry_hint(502, &headers), None);
5841 let empty = HeaderMap::new();
5842 assert_eq!(retry_hint(429, &empty), Some(Duration::from_secs(1)));
5843 assert_eq!(retry_hint(503, &empty), None);
5844 }
5845
5846 #[test]
5848 fn log_request_completes_with_sensitive_headers() {
5849 let url = reqwest::Url::parse("https://api.example.com/v1/resource?q=1").unwrap();
5850 let mut headers = HeaderMap::new();
5851 headers.insert(
5852 HeaderName::from_static("content-type"),
5853 HeaderValue::from_static("application/json"),
5854 );
5855 headers.insert(
5856 HeaderName::from_static("authorization"),
5857 HeaderValue::from_static("Bearer sk_test_xxx"),
5858 );
5859 log_request("POST", &url, 201, Duration::from_millis(12), &headers);
5861 }
5862
5863 #[test]
5865 fn sim_host_header_keeps_ipv6_brackets() {
5866 let runtime = tokio::runtime::Builder::new_current_thread()
5867 .enable_all()
5868 .build()
5869 .expect("runtime");
5870 let echo = axum::Router::new().route(
5871 "/status",
5872 axum::routing::get(|headers: HeaderMap| async move {
5873 headers
5874 .get(reqwest::header::HOST)
5875 .and_then(|value| value.to_str().ok())
5876 .unwrap_or_default()
5877 .to_owned()
5878 }),
5879 );
5880 runtime.block_on(async {
5881 for (url, host) in [
5882 ("http://[::1]:8080/status", "[::1]:8080"),
5883 ("http://[::1]/status", "[::1]"),
5884 ("http://payments:8443/status", "payments:8443"),
5885 ] {
5886 let request = Client::new().get(url);
5887 let parsed = reqwest::Url::parse(url).unwrap();
5888 let seen = serve_sim_host(echo.clone(), &request, parsed, None)
5889 .await
5890 .unwrap()
5891 .text();
5892 assert_eq!(seen, host, "{url}");
5893 }
5894 });
5895 }
5896
5897 #[test]
5900 fn sim_host_gets_a_fresh_web_bot_auth_signature() {
5901 let runtime = tokio::runtime::Builder::new_current_thread()
5902 .enable_all()
5903 .build()
5904 .expect("runtime");
5905 let echo = axum::Router::new().fallback(|headers: HeaderMap| async move {
5906 let count = headers.get_all("signature").iter().count();
5907 let input = headers
5908 .get("signature-input")
5909 .and_then(|v| v.to_str().ok())
5910 .unwrap_or_default()
5911 .to_owned();
5912 format!("{count} {input}")
5913 });
5914 let key = crate::aeo::web_bot_auth::WebBotAuthKey::from_seed_b64(
5915 "nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A",
5916 )
5917 .unwrap();
5918 let signer = crate::aeo::web_bot_auth::WebBotAuthSigner::new(key, "https://bot.example");
5919 let created = |input: &str| -> u64 {
5920 input
5921 .split(";created=")
5922 .nth(1)
5923 .and_then(|r| r.split(';').next())
5924 .and_then(|v| v.parse().ok())
5925 .unwrap()
5926 };
5927 let url = "https://payments/charge";
5928 let first = created(&signer.sign_url(url).unwrap().signature_input);
5929 let request = Client::new().get(url).sign_web_bot_auth(&signer);
5930 std::thread::sleep(Duration::from_millis(1100));
5931 let parsed = reqwest::Url::parse(url).unwrap();
5932 let seen = runtime.block_on(async {
5933 serve_sim_host(echo, &request, parsed, None)
5934 .await
5935 .unwrap()
5936 .text()
5937 });
5938 let (count, input) = seen.split_once(' ').unwrap();
5939 assert_eq!(count, "1", "{seen}");
5940 assert!(created(input) > first, "{seen}");
5941 }
5942
5943 #[test]
5946 fn sim_host_receives_content_length() {
5947 let runtime = tokio::runtime::Builder::new_current_thread()
5948 .enable_all()
5949 .build()
5950 .expect("runtime");
5951 let echo = axum::Router::new().fallback(|headers: HeaderMap| async move {
5952 headers
5953 .get(reqwest::header::CONTENT_LENGTH)
5954 .and_then(|value| value.to_str().ok())
5955 .unwrap_or("none")
5956 .to_owned()
5957 });
5958 let url = "http://payments/charge";
5959 let parsed = reqwest::Url::parse(url).unwrap();
5960 runtime.block_on(async {
5961 for (request, expected) in [
5962 (Client::new().post(url).text_body("hello"), "5"),
5963 (
5964 Client::new().post(url).json(&serde_json::json!({"a": 1})),
5965 "7",
5966 ),
5967 (Client::new().post(url).bytes_body(Bytes::new()), "0"),
5968 (Client::new().get(url), "none"),
5969 ] {
5970 let seen = serve_sim_host(echo.clone(), &request, parsed.clone(), None)
5971 .await
5972 .unwrap()
5973 .text();
5974 assert_eq!(seen, expected);
5975 }
5976 });
5977 }
5978
5979 #[cfg(feature = "telemetry-otlp")]
5982 #[test]
5983 fn sim_host_receives_trace_context() {
5984 use opentelemetry::trace::TracerProvider as _;
5985 use opentelemetry_sdk::propagation::TraceContextPropagator;
5986 use opentelemetry_sdk::trace::SdkTracerProvider;
5987 use tracing_subscriber::prelude::*;
5988
5989 opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
5990 let provider = SdkTracerProvider::builder().build();
5991 let subscriber = tracing_subscriber::registry()
5992 .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("test")));
5993 let runtime = tokio::runtime::Builder::new_current_thread()
5994 .enable_all()
5995 .build()
5996 .expect("runtime");
5997 let echo = || {
5998 axum::Router::new().route(
5999 "/echo",
6000 axum::routing::get(|headers: HeaderMap| async move {
6001 headers
6002 .get("traceparent")
6003 .and_then(|value| value.to_str().ok())
6004 .unwrap_or_default()
6005 .to_owned()
6006 }),
6007 )
6008 };
6009 let url = reqwest::Url::parse("http://payments/echo").unwrap();
6010
6011 tracing::subscriber::with_default(subscriber, || {
6012 let span = tracing::info_span!("sim_trace_test");
6013 let _guard = span.enter();
6014 runtime.block_on(async {
6015 let request = Client::new().get("http://payments/echo");
6016 let traceparent = serve_sim_host(echo(), &request, url.clone(), None)
6017 .await
6018 .unwrap()
6019 .text();
6020 assert!(traceparent.starts_with("00-"), "{traceparent}");
6021
6022 let request = Client::new()
6023 .get("http://payments/echo")
6024 .header("traceparent", "caller-value");
6025 let response = serve_sim_host(echo(), &request, url.clone(), None)
6026 .await
6027 .unwrap();
6028 assert_eq!(response.text(), "caller-value");
6029 });
6030 });
6031 }
6032
6033 #[test]
6035 fn no_request_id_outside_a_request() {
6036 let mut headers = HeaderMap::new();
6037 add_current_request_id(&mut headers);
6038 assert!(headers.is_empty());
6039 }
6040
6041 #[tokio::test]
6042 async fn current_request_id_is_added() {
6043 let ctx = crate::log::context::LogContext::new(Some("rid-1".to_owned()));
6044 let mut headers = HeaderMap::new();
6045 crate::log::context::scope(ctx, async { add_current_request_id(&mut headers) }).await;
6046 assert_eq!(headers.get("x-request-id").unwrap(), "rid-1");
6047 }
6048
6049 #[tokio::test]
6050 async fn caller_request_id_is_kept() {
6051 let ctx = crate::log::context::LogContext::new(Some("rid-1".to_owned()));
6052 let mut headers = HeaderMap::new();
6053 headers.insert("x-request-id", HeaderValue::from_static("mine"));
6054 crate::log::context::scope(ctx, async { add_current_request_id(&mut headers) }).await;
6055 assert_eq!(headers.get_all("x-request-id").iter().count(), 1);
6056 assert_eq!(headers.get("x-request-id").unwrap(), "mine");
6057 }
6058
6059 #[test]
6063 fn caller_trace_header_drops_the_whole_injected_pair() {
6064 let injected = || {
6065 vec![
6066 ("traceparent".to_owned(), HeaderValue::from_static("ours-p")),
6067 ("tracestate".to_owned(), HeaderValue::from_static("ours-s")),
6068 ("baggage".to_owned(), HeaderValue::from_static("ours-b")),
6069 ]
6070 };
6071 let names = |caller: &[&'static str]| {
6072 let mut headers = HeaderMap::new();
6073 for name in caller {
6074 headers.insert(*name, HeaderValue::from_static("mine"));
6075 }
6076 let mut sent: Vec<String> = trace_headers_to_send(injected(), &headers)
6077 .into_iter()
6078 .map(|(name, _)| name)
6079 .collect();
6080 sent.sort();
6081 sent
6082 };
6083 assert_eq!(names(&[]), ["baggage", "traceparent", "tracestate"]);
6084 assert_eq!(names(&["traceparent"]), ["baggage"]);
6085 assert_eq!(names(&["tracestate"]), ["baggage"]);
6086 assert_eq!(names(&["baggage"]), ["traceparent", "tracestate"]);
6087 }
6088
6089 #[test]
6090 fn caller_trace_header_suppresses_the_injected_one() {
6091 let mut caller = HeaderMap::new();
6092 caller.insert("traceparent", HeaderValue::from_static("mine"));
6093 let request =
6094 inject_trace_context(reqwest::Client::new().get("https://example.com"), &caller)
6095 .build()
6096 .unwrap();
6097 assert!(request.headers().get("traceparent").is_none());
6098 }
6099
6100 #[tokio::test]
6102 #[allow(clippy::await_holding_lock)]
6103 async fn without_request_id_sends_no_id() {
6104 use axum::{Router, http::HeaderMap as AxumHeaders, routing::get};
6105
6106 let _lock = crate::circuit_breaker::TEST_LOCK
6107 .lock()
6108 .unwrap_or_else(std::sync::PoisonError::into_inner);
6109 crate::circuit_breaker::global_registry().clear();
6110
6111 let app = Router::new().route(
6112 "/echo",
6113 get(|headers: AxumHeaders| async move {
6114 headers
6115 .get("x-request-id")
6116 .map_or_else(|| "none".to_owned(), |v| v.to_str().unwrap().to_owned())
6117 }),
6118 );
6119 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6120 let addr = listener.local_addr().unwrap();
6121 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6122
6123 let ctx = crate::log::context::LogContext::new(Some("rid-1".to_owned()));
6124 let url = format!("http://{addr}/echo");
6125 let (forwarded, opted_out) = crate::log::context::scope(ctx, async {
6126 let forwarded = Client::new().get(&url).send().await.unwrap().text();
6127 let opted_out = Client::new()
6128 .get(&url)
6129 .without_request_id()
6130 .send()
6131 .await
6132 .unwrap()
6133 .text();
6134 (forwarded, opted_out)
6135 })
6136 .await;
6137 assert_eq!(forwarded, "rid-1");
6138 assert_eq!(opted_out, "none");
6139
6140 crate::circuit_breaker::global_registry().clear();
6141 }
6142
6143 #[tokio::test]
6146 #[allow(clippy::await_holding_lock)]
6147 async fn real_get_request_covers_network_path() {
6148 use axum::{Router, routing::get};
6149
6150 let _lock = crate::circuit_breaker::TEST_LOCK
6151 .lock()
6152 .unwrap_or_else(std::sync::PoisonError::into_inner);
6153 crate::circuit_breaker::global_registry().clear();
6154
6155 let app = Router::new().route("/ping", get(|| async { "pong" }));
6156 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6157 let addr = listener.local_addr().unwrap();
6158 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6159
6160 let client = Client::new();
6161 let resp = client
6162 .get(format!("http://127.0.0.1:{}/ping", addr.port()))
6163 .header("x-request-id", "test-35")
6164 .send()
6165 .await
6166 .unwrap();
6167
6168 assert_eq!(resp.status().as_u16(), 200);
6169 assert!(resp.url().is_some());
6170 assert_eq!(resp.text(), "pong");
6171
6172 crate::circuit_breaker::global_registry().clear();
6173 }
6174
6175 #[derive(Clone, Default)]
6177 struct ClientSpanCapture {
6178 spans: std::sync::Arc<
6179 std::sync::Mutex<
6180 std::collections::HashMap<u64, std::collections::BTreeMap<String, String>>,
6181 >,
6182 >,
6183 }
6184
6185 struct SpanFieldVisitor<'a>(&'a mut std::collections::BTreeMap<String, String>);
6186
6187 impl tracing::field::Visit for SpanFieldVisitor<'_> {
6188 fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
6189 self.0.insert(
6190 field.name().to_owned(),
6191 format!("{value:?}").trim_matches('"').to_owned(),
6192 );
6193 }
6194 }
6195
6196 impl<S> tracing_subscriber::Layer<S> for ClientSpanCapture
6197 where
6198 S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
6199 {
6200 fn on_new_span(
6201 &self,
6202 attrs: &tracing::span::Attributes<'_>,
6203 id: &tracing::span::Id,
6204 _ctx: tracing_subscriber::layer::Context<'_, S>,
6205 ) {
6206 if attrs.metadata().name() != "http.client.request" {
6207 return;
6208 }
6209 let mut fields = std::collections::BTreeMap::new();
6210 attrs.record(&mut SpanFieldVisitor(&mut fields));
6211 self.spans.lock().unwrap().insert(id.into_u64(), fields);
6212 }
6213
6214 fn on_record(
6215 &self,
6216 id: &tracing::span::Id,
6217 values: &tracing::span::Record<'_>,
6218 _ctx: tracing_subscriber::layer::Context<'_, S>,
6219 ) {
6220 if let Some(fields) = self.spans.lock().unwrap().get_mut(&id.into_u64()) {
6221 values.record(&mut SpanFieldVisitor(fields));
6222 }
6223 }
6224 }
6225
6226 #[derive(Clone, Default)]
6228 struct ClientSpanDurations {
6229 opened: std::sync::Arc<std::sync::Mutex<std::collections::HashMap<u64, Instant>>>,
6230 closed: std::sync::Arc<std::sync::Mutex<Vec<Duration>>>,
6231 }
6232
6233 impl<S> tracing_subscriber::Layer<S> for ClientSpanDurations
6234 where
6235 S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
6236 {
6237 fn on_new_span(
6238 &self,
6239 attrs: &tracing::span::Attributes<'_>,
6240 id: &tracing::span::Id,
6241 _ctx: tracing_subscriber::layer::Context<'_, S>,
6242 ) {
6243 if attrs.metadata().name() == "http.client.request" {
6244 self.opened
6245 .lock()
6246 .unwrap()
6247 .insert(id.into_u64(), Instant::now());
6248 }
6249 }
6250
6251 fn on_close(&self, id: tracing::span::Id, _ctx: tracing_subscriber::layer::Context<'_, S>) {
6252 let start = self.opened.lock().unwrap().remove(&id.into_u64());
6253 if let Some(start) = start {
6254 self.closed.lock().unwrap().push(start.elapsed());
6255 }
6256 }
6257 }
6258
6259 fn capture_client_spans() -> (
6262 ClientSpanCapture,
6263 ClientSpanDurations,
6264 tracing::subscriber::DefaultGuard,
6265 ) {
6266 use tracing_subscriber::layer::SubscriberExt as _;
6267
6268 let fields = ClientSpanCapture::default();
6269 let durations = ClientSpanDurations::default();
6270 let guard = tracing::subscriber::set_default(
6271 tracing_subscriber::registry()
6272 .with(fields.clone())
6273 .with(durations.clone()),
6274 );
6275 drop(client_attempt_span(&Method::GET, "http://warm.up/", 0));
6276 tracing::callsite::rebuild_interest_cache();
6277 fields.spans.lock().unwrap().clear();
6278 durations.closed.lock().unwrap().clear();
6279 (fields, durations, guard)
6280 }
6281
6282 #[tokio::test]
6285 #[allow(clippy::await_holding_lock)]
6286 async fn attempt_span_covers_the_response_body() {
6287 use axum::{Router, body::Body, routing::get};
6288 use futures::StreamExt as _;
6289
6290 let _lock = crate::circuit_breaker::TEST_LOCK
6291 .lock()
6292 .unwrap_or_else(std::sync::PoisonError::into_inner);
6293 crate::circuit_breaker::global_registry().clear();
6294
6295 let app = Router::new()
6296 .route(
6297 "/slow",
6298 get(|| async {
6299 let stream = futures::stream::once(async {
6300 tokio::time::sleep(Duration::from_millis(300)).await;
6301 Ok::<_, std::io::Error>(Bytes::from_static(b"late"))
6302 });
6303 Body::from_stream(stream)
6304 }),
6305 )
6306 .route(
6307 "/broken",
6308 get(|| async {
6309 let stream = futures::stream::iter([
6311 Ok(Bytes::from_static(b"partial")),
6312 Err(std::io::Error::other("cut")),
6313 ])
6314 .then(|item| async move {
6315 tokio::time::sleep(Duration::from_millis(50)).await;
6316 item
6317 });
6318 Body::from_stream(stream)
6319 }),
6320 );
6321 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6322 let addr = listener.local_addr().unwrap();
6323 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6324
6325 for custom_path in [false, true] {
6326 let (fields, durations, _guard) = capture_client_spans();
6327 let mut slow = Client::new().get(format!("http://{addr}/slow"));
6328 let mut broken = Client::new().get(format!("http://{addr}/broken"));
6329 if custom_path {
6330 slow = slow.no_redirect();
6331 broken = broken.no_redirect();
6332 }
6333 assert_eq!(slow.send().await.unwrap().text(), "late");
6334 let closed = durations.closed.lock().unwrap().clone();
6335 assert_eq!(closed.len(), 1, "custom={custom_path}: {closed:?}");
6336 assert!(
6337 closed[0] >= Duration::from_millis(250),
6338 "custom={custom_path}: the span ended before the body: {closed:?}"
6339 );
6340
6341 fields.spans.lock().unwrap().clear();
6342 assert!(broken.send().await.is_err());
6343 let spans: Vec<_> = fields.spans.lock().unwrap().values().cloned().collect();
6344 assert_eq!(spans.len(), 1, "custom={custom_path}: {spans:?}");
6345 assert_eq!(
6346 spans[0].get("error.type").map(String::as_str),
6347 Some("body"),
6348 "custom={custom_path}: {spans:?}"
6349 );
6350 assert_eq!(
6351 spans[0].get("otel.status_code").map(String::as_str),
6352 Some("ERROR"),
6353 "custom={custom_path}: {spans:?}"
6354 );
6355
6356 fields.spans.lock().unwrap().clear();
6358 let mut missing = Client::new().get(format!("http://{addr}/missing"));
6359 if custom_path {
6360 missing = missing.no_redirect();
6361 }
6362 assert_eq!(missing.send().await.unwrap().status().as_u16(), 404);
6363 let spans: Vec<_> = fields.spans.lock().unwrap().values().cloned().collect();
6364 assert_eq!(spans.len(), 1, "custom={custom_path}: {spans:?}");
6365 assert_eq!(
6366 spans[0].get("otel.status_code").map(String::as_str),
6367 Some("ERROR"),
6368 "custom={custom_path}: {spans:?}"
6369 );
6370 }
6371
6372 crate::circuit_breaker::global_registry().clear();
6373 }
6374
6375 #[tokio::test]
6377 async fn sim_attempts_open_client_spans() {
6378 use axum::{Router, http::StatusCode, routing::get};
6379 use std::sync::atomic::{AtomicUsize, Ordering};
6380
6381 let (fields, _durations, _guard) = capture_client_spans();
6382 let calls = std::sync::Arc::new(AtomicUsize::new(0));
6383 let payments = Router::new().route(
6384 "/charge",
6385 get(move || {
6386 let calls = std::sync::Arc::clone(&calls);
6387 async move {
6388 if calls.fetch_add(1, Ordering::SeqCst) == 0 {
6389 StatusCode::SERVICE_UNAVAILABLE
6390 } else {
6391 StatusCode::OK
6392 }
6393 }
6394 }),
6395 );
6396 let mut client = Client::new();
6397 client.sim_net = Some(Arc::new(
6398 crate::sim::SimNet::new().host("payments", payments),
6399 ));
6400
6401 let response = client.get("http://payments/charge").send().await.unwrap();
6402 assert_eq!(response.status().as_u16(), 200);
6403
6404 let mut attempts: Vec<_> = fields.spans.lock().unwrap().values().cloned().collect();
6405 attempts.sort_by_key(|f| f.get("http.request.resend_count").cloned());
6406 assert_eq!(attempts.len(), 2, "{attempts:?}");
6407 assert_eq!(
6408 attempts[0].get("server.address").map(String::as_str),
6409 Some("payments")
6410 );
6411 assert_eq!(
6412 attempts[0]
6413 .get("http.response.status_code")
6414 .map(String::as_str),
6415 Some("503")
6416 );
6417 assert_eq!(
6418 attempts[1]
6419 .get("http.response.status_code")
6420 .map(String::as_str),
6421 Some("200")
6422 );
6423 assert_eq!(
6425 attempts[0].get("otel.status_code").map(String::as_str),
6426 Some("ERROR"),
6427 "{attempts:?}"
6428 );
6429 assert_eq!(attempts[1].get("otel.status_code"), None, "{attempts:?}");
6430 }
6431
6432 #[tokio::test]
6435 #[allow(clippy::await_holding_lock)]
6436 async fn attempt_span_ends_before_the_retry_after_sleep() {
6437 use axum::{Router, http::StatusCode, response::IntoResponse as _, routing::get};
6438 use std::sync::atomic::{AtomicUsize, Ordering};
6439 use tracing_subscriber::layer::SubscriberExt as _;
6440
6441 let _lock = crate::circuit_breaker::TEST_LOCK
6442 .lock()
6443 .unwrap_or_else(std::sync::PoisonError::into_inner);
6444 crate::circuit_breaker::global_registry().clear();
6445
6446 for custom_path in [false, true] {
6447 let durations = ClientSpanDurations::default();
6448 let _guard = tracing::subscriber::set_default(
6449 tracing_subscriber::registry().with(durations.clone()),
6450 );
6451 drop(client_attempt_span(&Method::GET, "http://warm.up/", 0));
6452 tracing::callsite::rebuild_interest_cache();
6453 durations.closed.lock().unwrap().clear();
6454
6455 let calls = std::sync::Arc::new(AtomicUsize::new(0));
6457 let app = Router::new().route(
6458 "/limited",
6459 get(move || {
6460 let calls = std::sync::Arc::clone(&calls);
6461 async move {
6462 if calls.fetch_add(1, Ordering::SeqCst) == 0 {
6463 (StatusCode::TOO_MANY_REQUESTS, [("retry-after", "1")]).into_response()
6464 } else {
6465 StatusCode::OK.into_response()
6466 }
6467 }
6468 }),
6469 );
6470 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6471 let addr = listener.local_addr().unwrap();
6472 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6473
6474 let mut request = Client::new().get(format!("http://{addr}/limited"));
6475 if custom_path {
6476 request = request.no_redirect();
6477 }
6478 assert_eq!(request.send().await.unwrap().status().as_u16(), 200);
6479
6480 let closed = durations.closed.lock().unwrap().clone();
6481 assert_eq!(closed.len(), 2, "custom={custom_path}: {closed:?}");
6482 assert!(
6483 closed[0] < Duration::from_millis(500),
6484 "custom={custom_path}: the 429 attempt span held the 1s sleep: {closed:?}"
6485 );
6486 }
6487
6488 crate::circuit_breaker::global_registry().clear();
6489 }
6490
6491 #[tokio::test]
6494 #[allow(clippy::await_holding_lock)]
6495 async fn each_attempt_opens_one_client_span() {
6496 use axum::{Router, http::StatusCode, routing::get};
6497 use std::sync::atomic::{AtomicUsize, Ordering};
6498 use tracing_subscriber::layer::SubscriberExt as _;
6499
6500 let _lock = crate::circuit_breaker::TEST_LOCK
6501 .lock()
6502 .unwrap_or_else(std::sync::PoisonError::into_inner);
6503 crate::circuit_breaker::global_registry().clear();
6504
6505 for custom_path in [false, true] {
6506 let capture = ClientSpanCapture::default();
6507 let _guard = tracing::subscriber::set_default(
6508 tracing_subscriber::registry().with(capture.clone()),
6509 );
6510 drop(client_attempt_span(&Method::GET, "http://warm.up/", 0));
6514 tracing::callsite::rebuild_interest_cache();
6515 capture.spans.lock().unwrap().clear();
6516
6517 let calls = std::sync::Arc::new(AtomicUsize::new(0));
6519 let app = Router::new().route(
6520 "/flaky",
6521 get(move || {
6522 let calls = std::sync::Arc::clone(&calls);
6523 async move {
6524 if calls.fetch_add(1, Ordering::SeqCst) == 0 {
6525 StatusCode::SERVICE_UNAVAILABLE
6526 } else {
6527 StatusCode::OK
6528 }
6529 }
6530 }),
6531 );
6532 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6533 let addr = listener.local_addr().unwrap();
6534 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6535
6536 let mut request = Client::new().get(format!("http://{addr}/flaky"));
6537 if custom_path {
6538 request = request.no_redirect();
6539 }
6540 let response = request.send().await.unwrap();
6541 assert_eq!(response.status().as_u16(), 200);
6542
6543 let mut attempts: Vec<_> = capture.spans.lock().unwrap().values().cloned().collect();
6544 attempts.sort_by_key(|f| f.get("http.request.resend_count").cloned());
6545 assert_eq!(attempts.len(), 2, "custom={custom_path}: {attempts:?}");
6546 for (n, fields) in attempts.iter().enumerate() {
6547 assert_eq!(fields.get("otel.kind").map(String::as_str), Some("client"));
6548 assert_eq!(
6549 fields.get("http.request.method").map(String::as_str),
6550 Some("GET")
6551 );
6552 assert_eq!(
6553 fields.get("server.address").map(String::as_str),
6554 Some("127.0.0.1")
6555 );
6556 let resend = fields.get("http.request.resend_count");
6557 assert_eq!(resend, (n > 0).then(|| n.to_string()).as_ref());
6558 }
6559 assert_eq!(
6560 attempts[0]
6561 .get("http.response.status_code")
6562 .map(String::as_str),
6563 Some("503")
6564 );
6565 assert_eq!(
6566 attempts[1]
6567 .get("http.response.status_code")
6568 .map(String::as_str),
6569 Some("200")
6570 );
6571 }
6572
6573 crate::circuit_breaker::global_registry().clear();
6574 }
6575
6576 #[tokio::test]
6578 #[allow(clippy::await_holding_lock)]
6579 async fn real_post_with_json_body_covers_body_path() {
6580 use axum::{Json, Router, routing::post};
6581 use serde_json::Value;
6582
6583 let _lock = crate::circuit_breaker::TEST_LOCK
6584 .lock()
6585 .unwrap_or_else(std::sync::PoisonError::into_inner);
6586 crate::circuit_breaker::global_registry().clear();
6587
6588 let app = Router::new().route(
6589 "/echo",
6590 post(|Json(body): Json<Value>| async move { Json(body) }),
6591 );
6592 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6593 let addr = listener.local_addr().unwrap();
6594 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6595
6596 let client = Client::new();
6597 let resp = client
6598 .post(format!("http://127.0.0.1:{}/echo", addr.port()))
6599 .json(&serde_json::json!({"hello": "world"}))
6600 .send()
6601 .await
6602 .unwrap();
6603
6604 assert_eq!(resp.status().as_u16(), 200);
6605 let body: Value = resp.json().unwrap();
6606 assert_eq!(body["hello"], "world");
6607
6608 crate::circuit_breaker::global_registry().clear();
6609 }
6610
6611 #[tokio::test]
6613 #[allow(clippy::await_holding_lock)]
6614 async fn real_get_retries_on_503_then_succeeds() {
6615 use axum::{Router, routing::get};
6616 use std::sync::Arc;
6617 use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
6618
6619 let _lock = crate::circuit_breaker::TEST_LOCK
6620 .lock()
6621 .unwrap_or_else(std::sync::PoisonError::into_inner);
6622 crate::circuit_breaker::global_registry().clear();
6623
6624 let hit = Arc::new(AtomicU32::new(0));
6625 let hit2 = hit.clone();
6626 let app = Router::new().route(
6627 "/flaky",
6628 get(move || {
6629 let c = hit2.clone();
6630 async move {
6631 if c.fetch_add(1, SeqOrdering::SeqCst) == 0 {
6632 axum::http::StatusCode::SERVICE_UNAVAILABLE
6633 } else {
6634 axum::http::StatusCode::OK
6635 }
6636 }
6637 }),
6638 );
6639 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6640 let addr = listener.local_addr().unwrap();
6641 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6642
6643 let resp = Client::new()
6645 .get(format!("http://127.0.0.1:{}/flaky", addr.port()))
6646 .retries(1)
6647 .send()
6648 .await
6649 .unwrap();
6650
6651 assert_eq!(resp.status().as_u16(), 200);
6652 assert_eq!(hit.load(SeqOrdering::SeqCst), 2);
6653
6654 crate::circuit_breaker::global_registry().clear();
6655 }
6656
6657 async fn serve_down_once_with_retry_after() -> (SocketAddr, Arc<AtomicUsize>) {
6660 use axum::{Router, response::IntoResponse as _, routing::get};
6661 let hits = Arc::new(AtomicUsize::new(0));
6662 let counter = hits.clone();
6663 let app = Router::new().route(
6664 "/down",
6665 get(move || {
6666 let counter = counter.clone();
6667 async move {
6668 if counter.fetch_add(1, Ordering::SeqCst) == 0 {
6669 (
6670 axum::http::StatusCode::SERVICE_UNAVAILABLE,
6671 [("retry-after", "1")],
6672 )
6673 .into_response()
6674 } else {
6675 axum::http::StatusCode::OK.into_response()
6676 }
6677 }
6678 }),
6679 );
6680 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6681 let addr = listener.local_addr().unwrap();
6682 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6683 (addr, hits)
6684 }
6685
6686 #[tokio::test]
6690 #[allow(clippy::await_holding_lock)]
6691 async fn real_paths_honour_retry_after_on_503() {
6692 let _lock = crate::circuit_breaker::TEST_LOCK
6693 .lock()
6694 .unwrap_or_else(std::sync::PoisonError::into_inner);
6695 crate::circuit_breaker::global_registry().clear();
6696
6697 for custom_path in [false, true] {
6698 let (addr, hits) = serve_down_once_with_retry_after().await;
6699 let mut request = Client::new()
6700 .get(format!("http://127.0.0.1:{}/down", addr.port()))
6701 .retries(1);
6702 if custom_path {
6703 request = request.no_redirect();
6704 }
6705 let start = std::time::Instant::now();
6706 let response = request.send().await.unwrap();
6707 let waited = start.elapsed();
6708 assert_eq!(response.status().as_u16(), 200, "custom_path={custom_path}");
6709 assert_eq!(hits.load(Ordering::SeqCst), 2);
6710 assert!(
6711 waited >= Duration::from_secs(1),
6712 "custom_path={custom_path}: waited only {waited:?}"
6713 );
6714 }
6715
6716 crate::circuit_breaker::global_registry().clear();
6717 }
6718
6719 #[test]
6720 fn from_config_reads_max_backoff_ms() {
6721 let config = HttpClientConfig {
6722 max_backoff_ms: 1_500,
6723 ..Default::default()
6724 };
6725 assert_eq!(
6726 Client::from_config(&config).retry_policy.max_backoff,
6727 Duration::from_millis(1_500)
6728 );
6729 assert_eq!(RetryPolicy::default().max_backoff, Duration::from_secs(20));
6730 }
6731
6732 #[test]
6734 fn text_body_sets_body() {
6735 let client = Client::new();
6736 let builder = client.post("https://example.com").text_body("hello");
6737 assert_eq!(builder.body, Some(bytes::Bytes::from_static(b"hello")));
6738 }
6739
6740 #[test]
6742 fn client_error_display() {
6743 let err = ClientError::NoMock("GET".to_owned(), "/path".to_owned());
6744 assert!(err.to_string().contains("GET"));
6745 assert!(err.to_string().contains("/path"));
6746 }
6747
6748 #[test]
6755 fn outbound_is_open_until_a_replay_blocks_it() {
6756 assert!(
6757 !outbound_blocked_for_replay(),
6758 "a serving process must never start out blocked"
6759 );
6760 let error = ClientError::BlockedDuringReplay(
6761 "GET".to_owned(),
6762 "https://payments.example/charge".to_owned(),
6763 )
6764 .to_string();
6765 assert!(
6766 error.contains("blocked during replay") && error.contains("not recorded"),
6767 "the error must say why the call did not happen, got {error}"
6768 );
6769 assert!(
6770 error.contains("GET") && error.contains("https://payments.example/charge"),
6771 "and which call it was, got {error}"
6772 );
6773 }
6774
6775 #[tokio::test]
6777 #[allow(clippy::await_holding_lock)]
6778 async fn test_http_client_circuit_breaker_integration() {
6779 use axum::{Router, routing::get};
6780 use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
6781
6782 let _lock = crate::circuit_breaker::TEST_LOCK
6783 .lock()
6784 .unwrap_or_else(std::sync::PoisonError::into_inner);
6785 crate::circuit_breaker::global_registry().clear();
6786
6787 let hit = Arc::new(AtomicU32::new(0));
6788 let hit2 = hit.clone();
6789 let app = Router::new().route(
6790 "/flaky",
6791 get(move || {
6792 let c = hit2.clone();
6793 async move {
6794 c.fetch_add(1, SeqOrdering::SeqCst);
6795 axum::http::StatusCode::INTERNAL_SERVER_ERROR
6796 }
6797 }),
6798 );
6799 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6800 let addr = listener.local_addr().unwrap();
6801 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6802
6803 let mut rc = crate::config::ResilienceConfig::default();
6805 rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
6806 rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
6807 rc.circuit_breaker.defaults.open_duration_secs = Some(10);
6808
6809 let client = Client::new();
6810 let client = Client {
6812 resilience_config: Some(Arc::new(rc)),
6813 ..client
6814 };
6815
6816 let url = format!("http://127.0.0.1:{}/flaky", addr.port());
6817
6818 for _ in 0..3 {
6820 let res = client.get(&url).send().await;
6821 let res = res.unwrap();
6822 assert_eq!(res.status().as_u16(), 500);
6823 }
6824
6825 let res = client.get(&url).send().await;
6827 assert!(matches!(res, Err(ClientError::CircuitBreakerOpen)));
6828
6829 assert_eq!(hit.load(SeqOrdering::SeqCst), 3);
6831 crate::circuit_breaker::global_registry().clear();
6832 }
6833
6834 #[test]
6836 fn shared_reqwest_client_ext_round_trips() {
6837 let ext = SharedReqwestClient {
6838 client: reqwest::Client::new(),
6839 timeout_secs: 30,
6840 };
6841 let state = crate::AppState::for_test();
6842 state.insert_extension(ext);
6843 let retrieved = state.extension::<SharedReqwestClient>();
6844 assert!(retrieved.is_some());
6845 }
6846
6847 #[test]
6849 fn client_head_method_builds_request_builder() {
6850 let client = Client::new();
6851 let _builder = client.head("https://example.com/resource");
6852 }
6853
6854 #[tokio::test]
6860 #[allow(clippy::await_holding_lock)]
6861 async fn from_state_reuses_shared_client() {
6862 use axum::{Router, routing::get};
6863
6864 let _lock = crate::circuit_breaker::TEST_LOCK
6865 .lock()
6866 .unwrap_or_else(std::sync::PoisonError::into_inner);
6867 crate::circuit_breaker::global_registry().clear();
6868
6869 let app = Router::new().route(
6870 "/ua",
6871 get(|req: axum::http::Request<axum::body::Body>| async move {
6872 req.headers()
6873 .get("user-agent")
6874 .and_then(|v| v.to_str().ok())
6875 .unwrap_or("")
6876 .to_owned()
6877 }),
6878 );
6879 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6880 let addr = listener.local_addr().unwrap();
6881 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
6882
6883 let distinctive_inner = reqwest::ClientBuilder::new()
6884 .user_agent("autumn-shared-pool-test")
6885 .build()
6886 .expect("failed to build inner client");
6887 let state = crate::AppState::for_test();
6888 state.insert_extension(SharedReqwestClient {
6889 client: distinctive_inner,
6890 timeout_secs: 30,
6891 });
6892
6893 let client = Client::from_state(&state);
6894 let resp = client
6895 .get(format!("http://127.0.0.1:{}/ua", addr.port()))
6896 .send()
6897 .await
6898 .expect("request should succeed");
6899
6900 assert_eq!(resp.text(), "autumn-shared-pool-test");
6901 crate::circuit_breaker::global_registry().clear();
6902 }
6903
6904 #[cfg(feature = "http-client")]
6905 #[test]
6906 fn from_state_falls_back_when_timeout_mismatches_shared_client() {
6907 use crate::config::{AutumnConfig, HttpClientConfig};
6911 use std::sync::Arc;
6912
6913 let mut config = AutumnConfig::default();
6914 config.http.client = HttpClientConfig {
6915 timeout_secs: 10,
6916 ..Default::default()
6917 };
6918
6919 let state = crate::AppState::for_test();
6920 state.insert_extension(SharedReqwestClient {
6921 client: reqwest::Client::new(),
6922 timeout_secs: 5, });
6924 state.insert_extension(Arc::new(config));
6925
6926 let _client = Client::from_state(&state);
6928 }
6929
6930 #[cfg(feature = "http-client")]
6931 #[test]
6932 fn from_state_reuses_shared_client_when_no_config() {
6933 let state = crate::AppState::for_test();
6936 let default_timeout = crate::config::HttpClientConfig::default().timeout_secs;
6939 state.insert_extension(SharedReqwestClient {
6940 client: reqwest::Client::new(),
6941 timeout_secs: default_timeout,
6942 });
6943
6944 let _client = Client::from_state(&state);
6945 }
6946
6947 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
6950
6951 #[test]
6953 fn ssrf_policy_blocks_private_and_reserved_ipv4() {
6954 let blocked = [
6955 "0.0.0.0",
6956 "10.1.2.3",
6957 "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", ];
6972 for s in blocked {
6973 let ip: IpAddr = s.parse().unwrap();
6974 assert!(is_blocked_ip(ip), "{s} should be blocked");
6975 assert!(!is_public_ip(ip), "{s} should not be public");
6976 }
6977 }
6978
6979 #[test]
6981 fn ssrf_policy_allows_public_ipv4() {
6982 for s in ["1.1.1.1", "8.8.8.8", "93.184.216.34"] {
6983 let ip: IpAddr = s.parse().unwrap();
6984 assert!(is_public_ip(ip), "{s} should be public");
6985 assert!(!is_blocked_ip(ip), "{s} should not be blocked");
6986 }
6987 }
6988
6989 #[test]
6991 fn ssrf_policy_ipv6_and_mapped_forms() {
6992 let blocked = [
6993 "::", "::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", ];
7016 for s in blocked {
7017 let ip: IpAddr = s.parse().unwrap();
7018 assert!(is_blocked_ip(ip), "{s} should be blocked");
7019 }
7020 let compat = IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0x7f00, 0x0001));
7022 assert!(
7023 is_blocked_ip(compat),
7024 "::7f00:1 (127.0.0.1) should be blocked"
7025 );
7026
7027 let public: IpAddr = "2606:4700:4700::1111".parse().unwrap();
7029 assert!(
7030 is_public_ip(public),
7031 "2606:4700:4700::1111 should be public"
7032 );
7033
7034 let public_addrs = [
7037 "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", ];
7047 for s in public_addrs {
7048 let ip: IpAddr = s.parse().unwrap();
7049 assert!(is_public_ip(ip), "{s} should be public");
7050 assert!(!is_blocked_ip(ip), "{s} should not be blocked");
7051 }
7052 }
7053
7054 #[test]
7058 fn ssrf_policy_blocks_tunnelled_ipv4() {
7059 let blocked = [
7060 "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", ];
7075 for s in blocked {
7076 let ip: IpAddr = s.parse().unwrap();
7077 assert!(is_blocked_ip(ip), "{s} should be blocked");
7078 assert!(!is_public_ip(ip), "{s} should not be public");
7079 }
7080
7081 let anycast: IpAddr = "192.88.99.1".parse().unwrap();
7083 assert!(is_blocked_ip(anycast), "192.88.99.1 should be blocked");
7084
7085 for s in ["2002:0808:0808::", "2606:4700::1111", "::ffff:0:8.8.8.8"] {
7092 let ip: IpAddr = s.parse().unwrap();
7093 assert!(is_public_ip(ip), "{s} should be public");
7094 assert!(!is_blocked_ip(ip), "{s} should not be blocked");
7095 }
7096 }
7097
7098 #[test]
7101 fn ssrf_policy_decimal_encoded_host_is_blocked() {
7102 for raw in [
7103 "http://2130706433/",
7104 "http://0x7f000001/",
7105 "http://127.0.0.1/",
7106 ] {
7107 let parsed = url::Url::parse(raw).unwrap();
7108 match parsed.host() {
7109 Some(url::Host::Ipv4(v4)) => {
7110 assert_eq!(
7111 v4,
7112 Ipv4Addr::LOCALHOST,
7113 "{raw} should normalise to 127.0.0.1"
7114 );
7115 assert!(
7116 is_blocked_ip(IpAddr::V4(v4)),
7117 "{raw} host should be blocked"
7118 );
7119 }
7120 other => panic!("{raw} did not parse to an Ipv4 host: {other:?}"),
7121 }
7122 }
7123 }
7124
7125 async fn spawn(app: axum::Router) -> std::net::SocketAddr {
7127 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
7128 let addr = listener.local_addr().unwrap();
7129 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
7130 addr
7131 }
7132
7133 fn redirect_302(location: String) -> axum::response::Response {
7134 axum::response::Response::builder()
7135 .status(302)
7136 .header("location", location)
7137 .body(axum::body::Body::empty())
7138 .unwrap()
7139 }
7140
7141 fn redirect_307(location: String) -> axum::response::Response {
7142 axum::response::Response::builder()
7143 .status(307)
7144 .header("location", location)
7145 .body(axum::body::Body::empty())
7146 .unwrap()
7147 }
7148
7149 fn response_with_location(status: u16, location: String) -> axum::response::Response {
7153 axum::response::Response::builder()
7154 .status(status)
7155 .header("location", location)
7156 .body(axum::body::Body::empty())
7157 .unwrap()
7158 }
7159
7160 #[tokio::test]
7162 async fn no_redirect_returns_3xx_unfollowed() {
7163 use axum::{Router, routing::get};
7164 let addr = spawn(Router::new().route(
7165 "/start",
7166 get(|| async { redirect_302("http://127.0.0.1:1/never".to_owned()) }),
7167 ))
7168 .await;
7169
7170 let resp = Client::new()
7171 .get(format!("http://127.0.0.1:{}/start", addr.port()))
7172 .no_redirect()
7173 .send()
7174 .await
7175 .unwrap();
7176
7177 assert_eq!(resp.status().as_u16(), 302);
7178 assert_eq!(
7179 resp.headers().get("location").and_then(|v| v.to_str().ok()),
7180 Some("http://127.0.0.1:1/never")
7181 );
7182 }
7183
7184 #[tokio::test]
7198 async fn no_redirect_returns_3xx_with_malformed_location() {
7199 use axum::{Router, routing::get};
7200 let addr = spawn(Router::new().route(
7201 "/start",
7202 get(|| async { redirect_302("ht!tp://\\bad".to_owned()) }),
7205 ))
7206 .await;
7207
7208 let resp = Client::new()
7209 .get(format!("http://127.0.0.1:{}/start", addr.port()))
7210 .no_redirect()
7211 .send()
7212 .await
7213 .expect("no_redirect() must return the 3xx even with a malformed Location");
7214
7215 assert_eq!(resp.status().as_u16(), 302);
7216 assert_eq!(
7217 resp.headers().get("location").and_then(|v| v.to_str().ok()),
7218 Some("ht!tp://\\bad")
7219 );
7220 }
7221
7222 #[tokio::test]
7224 async fn follow_redirects_valid_chain_calls_validator() {
7225 use axum::{Router, routing::get};
7226
7227 let b_addr = spawn(Router::new().route("/final", get(|| async { "final-body" }))).await;
7228 let b_port = b_addr.port();
7229 let a_addr = spawn(Router::new().route(
7230 "/start",
7231 get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/final")) }),
7232 ))
7233 .await;
7234
7235 let calls = Arc::new(AtomicUsize::new(0));
7236 let calls2 = calls.clone();
7237 let resp = Client::new()
7238 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
7239 .follow_redirects(5, move |_loc| {
7240 calls2.fetch_add(1, Ordering::SeqCst);
7241 true
7242 })
7243 .send()
7244 .await
7245 .unwrap();
7246
7247 assert_eq!(resp.status().as_u16(), 200);
7248 assert_eq!(resp.text(), "final-body");
7249 assert_eq!(
7250 calls.load(Ordering::SeqCst),
7251 1,
7252 "validator called once per hop"
7253 );
7254 }
7255
7256 #[tokio::test]
7260 async fn follow_redirects_rejects_private_target() {
7261 use axum::{Router, routing::get};
7262 use std::sync::atomic::AtomicBool;
7263
7264 let touched = Arc::new(AtomicBool::new(false));
7266 let touched2 = touched.clone();
7267 let priv_addr = spawn(Router::new().route(
7268 "/secret",
7269 get(move || {
7270 let t = touched2.clone();
7271 async move {
7272 t.store(true, Ordering::SeqCst);
7273 "SECRET"
7274 }
7275 }),
7276 ))
7277 .await;
7278 let priv_port = priv_addr.port();
7279
7280 let a_addr = spawn(Router::new().route(
7282 "/start",
7283 get(
7284 move || async move { redirect_302(format!("http://127.0.0.1:{priv_port}/secret")) },
7285 ),
7286 ))
7287 .await;
7288
7289 let validator = |u: &str| -> bool {
7290 let Ok(p) = url::Url::parse(u) else {
7291 return false;
7292 };
7293 match p.host() {
7294 Some(url::Host::Ipv4(v4)) => is_public_ip(IpAddr::V4(v4)),
7295 Some(url::Host::Ipv6(v6)) => is_public_ip(IpAddr::V6(v6)),
7296 _ => true,
7297 }
7298 };
7299
7300 let result = Client::new()
7301 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
7302 .follow_redirects(5, validator)
7303 .send()
7304 .await;
7305
7306 assert!(
7307 matches!(result, Err(ClientError::RedirectRejected(_))),
7308 "expected RedirectRejected, got {result:?}"
7309 );
7310 assert!(
7311 !touched.load(Ordering::SeqCst),
7312 "the private target must never be connected to"
7313 );
7314 }
7315
7316 #[tokio::test]
7318 async fn follow_redirects_cap_exceeded() {
7319 use axum::{Router, routing::get};
7320
7321 let addr =
7323 spawn(Router::new().route("/loop", get(|| async { redirect_302("/loop".to_owned()) })))
7324 .await;
7325
7326 let result = Client::new()
7327 .get(format!("http://127.0.0.1:{}/loop", addr.port()))
7328 .follow_redirects(2, |_| true)
7329 .send()
7330 .await;
7331
7332 assert!(
7333 matches!(result, Err(ClientError::TooManyRedirects(2))),
7334 "expected TooManyRedirects(2), got {result:?}"
7335 );
7336 }
7337
7338 #[tokio::test]
7340 async fn follow_redirects_zero_max_errors_on_first_3xx() {
7341 use axum::{Router, routing::get};
7342 let addr = spawn(Router::new().route(
7343 "/start",
7344 get(|| async { redirect_302("http://127.0.0.1:1/x".to_owned()) }),
7345 ))
7346 .await;
7347
7348 let result = Client::new()
7349 .get(format!("http://127.0.0.1:{}/start", addr.port()))
7350 .follow_redirects(0, |_| true)
7351 .send()
7352 .await;
7353
7354 assert!(matches!(result, Err(ClientError::TooManyRedirects(0))));
7355 }
7356
7357 #[tokio::test]
7365 async fn pin_to_bypasses_dns_and_uses_url_port() {
7366 use axum::{Router, routing::get};
7367 let addr = spawn(Router::new().route("/ping", get(|| async { "pong" }))).await;
7368 let listener_port = addr.port();
7369
7370 let resp = Client::new()
7371 .get(format!("http://pinned.invalid:{listener_port}/ping"))
7372 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1))
7374 .send()
7375 .await
7376 .expect("pinned request should reach the loopback listener");
7377
7378 assert_eq!(resp.status().as_u16(), 200);
7379 assert_eq!(resp.text(), "pong");
7380 }
7381
7382 #[tokio::test]
7385 async fn pin_to_addrs_falls_back_to_the_next_address_on_connect_failure() {
7386 use axum::{Router, routing::get};
7387 let addr = spawn(Router::new().route("/ping", get(|| async { "pong" }))).await;
7388 let port = addr.port();
7389
7390 let resp = Client::new()
7391 .get(format!("http://pinned.invalid:{port}/ping"))
7392 .pin_to_addrs([
7393 SocketAddr::new(IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), port),
7396 SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
7397 ])
7398 .send()
7399 .await
7400 .expect("the second pinned address must serve the request");
7401
7402 assert_eq!(resp.text(), "pong");
7403 }
7404
7405 #[tokio::test]
7406 async fn pin_to_addrs_with_no_address_fails_instead_of_sending_unpinned() {
7407 let result = Client::new()
7408 .get("http://pinned.invalid:9/ping")
7409 .pin_to_addrs([])
7410 .send()
7411 .await;
7412 assert!(
7413 matches!(result, Err(ClientError::InvalidUrl(_))),
7414 "{result:?}"
7415 );
7416 }
7417
7418 #[tokio::test]
7422 async fn get_ssrf_safe_rejects_loopback_host_before_connecting() {
7423 use axum::{Router, routing::get};
7424 use std::sync::atomic::AtomicBool;
7425
7426 let touched = Arc::new(AtomicBool::new(false));
7427 let touched2 = touched.clone();
7428 let addr = spawn(Router::new().route(
7429 "/x",
7430 get(move || {
7431 let t = touched2.clone();
7432 async move {
7433 t.store(true, Ordering::SeqCst);
7434 "reached"
7435 }
7436 }),
7437 ))
7438 .await;
7439
7440 let result = Client::new()
7441 .get_ssrf_safe(format!("http://localhost:{}/x", addr.port()))
7442 .send()
7443 .await;
7444
7445 assert!(
7446 matches!(result, Err(ClientError::SsrfBlocked(_))),
7447 "expected SsrfBlocked, got {result:?}"
7448 );
7449 assert!(
7450 !touched.load(Ordering::SeqCst),
7451 "SSRF guard must reject before any connection"
7452 );
7453 }
7454
7455 #[tokio::test]
7458 async fn get_ssrf_safe_rejects_decimal_encoded_loopback() {
7459 let result = Client::new()
7460 .get_ssrf_safe("http://2130706433/")
7461 .send()
7462 .await;
7463 assert!(
7464 matches!(result, Err(ClientError::SsrfBlocked(_))),
7465 "expected SsrfBlocked, got {result:?}"
7466 );
7467 }
7468
7469 #[tokio::test]
7476 async fn resolve_and_validate_accepts_public_rejects_blocked() {
7477 let ok = resolve_and_validate("http://8.8.8.8:8080/path")
7480 .await
7481 .unwrap();
7482 assert_eq!(
7483 ok,
7484 vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)), 8080)]
7485 );
7486
7487 let ok_https = resolve_and_validate("https://1.1.1.1/").await.unwrap();
7489 assert_eq!(ok_https.len(), 1);
7490 assert_eq!(ok_https[0].port(), 443);
7491
7492 for raw in [
7494 "http://127.0.0.1/",
7495 "http://169.254.169.254/latest/meta-data/",
7496 "http://10.0.0.1/",
7497 "http://2130706433/",
7498 ] {
7499 let err = resolve_and_validate(raw).await;
7500 assert!(
7501 matches!(err, Err(ClientError::SsrfBlocked(_))),
7502 "{raw} should be SsrfBlocked, got {err:?}"
7503 );
7504 }
7505 }
7506
7507 #[test]
7514 fn validate_resolved_addrs_returns_all_public_rejects_any_blocked() {
7515 let v4 = |a, b, c, d, p| SocketAddr::new(IpAddr::V4(Ipv4Addr::new(a, b, c, d)), p);
7516
7517 let two = vec![v4(1, 1, 1, 1, 443), v4(8, 8, 8, 8, 443)];
7519 assert_eq!(validate_resolved_addrs(two.clone()).unwrap(), two);
7520
7521 let mixed = vec![v4(1, 1, 1, 1, 443), v4(10, 0, 0, 1, 443)];
7523 assert!(
7524 matches!(
7525 validate_resolved_addrs(mixed),
7526 Err(ClientError::SsrfBlocked(_))
7527 ),
7528 "a set containing a blocked address must be rejected"
7529 );
7530
7531 let v6 = SocketAddr::new(
7534 IpAddr::V6(Ipv6Addr::new(0x2606, 0x4700, 0x4700, 0, 0, 0, 0, 0x1111)),
7535 443,
7536 );
7537 let mix = vec![v6, v4(8, 8, 8, 8, 443)];
7538 assert_eq!(validate_resolved_addrs(mix.clone()).unwrap(), mix);
7539 }
7540
7541 #[tokio::test]
7546 async fn pin_to_does_not_follow_redirect_unpinned() {
7547 use axum::{Router, routing::get};
7548 use std::sync::atomic::AtomicBool;
7549
7550 let touched = Arc::new(AtomicBool::new(false));
7552 let touched2 = touched.clone();
7553 let onward = spawn(Router::new().route(
7554 "/onward",
7555 get(move || {
7556 let t = touched2.clone();
7557 async move {
7558 t.store(true, Ordering::SeqCst);
7559 "REACHED"
7560 }
7561 }),
7562 ))
7563 .await;
7564 let onward_port = onward.port();
7565
7566 let start =
7567 spawn(Router::new().route(
7568 "/start",
7569 get(move || async move {
7570 redirect_302(format!("http://127.0.0.1:{onward_port}/onward"))
7571 }),
7572 ))
7573 .await;
7574 let start_port = start.port();
7575
7576 let resp = Client::new()
7581 .get(format!("http://pinned.invalid:{start_port}/start"))
7582 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), start_port))
7583 .send()
7584 .await
7585 .expect("pinned request should return the 302 unfollowed");
7586
7587 assert_eq!(
7588 resp.status().as_u16(),
7589 302,
7590 "pin_to must return the redirect unfollowed"
7591 );
7592 assert!(
7593 !touched.load(Ordering::SeqCst),
7594 "pin_to must not silently follow the redirect onward"
7595 );
7596 }
7597
7598 #[tokio::test]
7601 async fn get_ssrf_safe_rejects_non_http_scheme() {
7602 for raw in ["ftp://public.example/resource", "gopher://public.example/"] {
7603 let result = Client::new().get_ssrf_safe(raw).send().await;
7604 assert!(
7605 matches!(result, Err(ClientError::InvalidUrl(_))),
7606 "{raw} should be rejected with InvalidUrl, got {result:?}"
7607 );
7608 }
7609 }
7610
7611 #[tokio::test]
7616 async fn follow_redirects_strips_sensitive_headers_cross_origin() {
7617 use axum::{Router, routing::get};
7618
7619 let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
7620 let seen2 = seen.clone();
7621 let b_addr = spawn(Router::new().route(
7624 "/dst",
7625 get(move |headers: HeaderMap| {
7626 let slot = seen2.clone();
7627 async move {
7628 *slot.lock().unwrap() = Some(headers);
7629 "ok"
7630 }
7631 }),
7632 ))
7633 .await;
7634 let b_port = b_addr.port();
7635
7636 let a_addr = spawn(Router::new().route(
7638 "/",
7639 get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
7640 ))
7641 .await;
7642
7643 let resp = Client::new()
7644 .get(format!("http://127.0.0.1:{}/", a_addr.port()))
7645 .header("authorization", "secret")
7646 .header("cookie", "session=abc")
7647 .header("proxy-authorization", "Basic zzz")
7648 .follow_redirects(3, |_| true)
7649 .send()
7650 .await
7651 .unwrap();
7652
7653 assert_eq!(resp.status().as_u16(), 200);
7654 let headers = seen
7655 .lock()
7656 .unwrap()
7657 .clone()
7658 .expect("listener B must have been reached");
7659 assert!(
7660 headers.get("authorization").is_none(),
7661 "authorization must be stripped on a cross-origin redirect"
7662 );
7663 assert!(
7664 headers.get("cookie").is_none(),
7665 "cookie must be stripped on a cross-origin redirect"
7666 );
7667 assert!(
7668 headers.get("proxy-authorization").is_none(),
7669 "proxy-authorization must be stripped on a cross-origin redirect"
7670 );
7671 }
7672
7673 #[tokio::test]
7676 async fn default_redirects_strip_message_signatures_cross_origin() {
7677 use axum::{Router, routing::get};
7678
7679 let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
7680 let seen2 = seen.clone();
7681 let b_addr = spawn(Router::new().route(
7682 "/dst",
7683 get(move |headers: HeaderMap| {
7684 let slot = seen2.clone();
7685 async move {
7686 *slot.lock().unwrap() = Some(headers);
7687 "ok"
7688 }
7689 }),
7690 ))
7691 .await;
7692 let b_port = b_addr.port();
7693 let a_addr = spawn(Router::new().route(
7694 "/",
7695 get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
7696 ))
7697 .await;
7698
7699 let resp = Client::new()
7700 .get(format!("http://127.0.0.1:{}/", a_addr.port()))
7701 .header("signature-agent", "\"https://bot.example\"")
7702 .header("signature-input", "sig1=(\"@authority\")")
7703 .header("signature", "sig1=:AAAA:")
7704 .send()
7705 .await
7706 .unwrap();
7707
7708 assert_eq!(
7709 resp.status().as_u16(),
7710 200,
7711 "the redirect is still followed"
7712 );
7713 let headers = seen.lock().unwrap().clone().expect("B reached");
7714 for name in ["signature", "signature-input", "signature-agent"] {
7715 assert!(headers.get(name).is_none(), "{name} must be stripped");
7716 }
7717 }
7718
7719 #[tokio::test]
7722 async fn web_bot_auth_signs_at_send_time() {
7723 use axum::{Router, routing::get};
7724
7725 let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
7726 let seen2 = seen.clone();
7727 let addr = spawn(Router::new().route(
7728 "/",
7729 get(move |headers: HeaderMap| {
7730 let slot = seen2.clone();
7731 async move {
7732 *slot.lock().unwrap() = Some(headers);
7733 "ok"
7734 }
7735 }),
7736 ))
7737 .await;
7738 let key = crate::aeo::web_bot_auth::WebBotAuthKey::from_seed_b64(
7739 "nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A",
7740 )
7741 .unwrap();
7742 let signer = crate::aeo::web_bot_auth::WebBotAuthSigner::new(key, "https://bot.example");
7743 let created = |input: &str| -> u64 {
7744 input
7745 .split(";created=")
7746 .nth(1)
7747 .and_then(|r| r.split(';').next())
7748 .and_then(|v| v.parse().ok())
7749 .unwrap()
7750 };
7751 let url = format!("http://127.0.0.1:{}/", addr.port());
7752 let first = created(&signer.sign_url(&url).unwrap().signature_input);
7753 let builder = Client::new().get(&url).sign_web_bot_auth(&signer);
7754 tokio::time::sleep(Duration::from_millis(1100)).await;
7755 builder.send().await.unwrap();
7756
7757 let headers = seen.lock().unwrap().clone().expect("reached");
7758 let sent = headers.get("signature-input").unwrap().to_str().unwrap();
7759 assert!(created(sent) > first, "{sent}");
7760 assert_eq!(headers.get_all("signature").iter().count(), 1);
7761 }
7762
7763 #[cfg(feature = "reporting")]
7767 #[test]
7768 fn capsule_matching_skips_the_web_bot_auth_signature() {
7769 let key = crate::aeo::web_bot_auth::WebBotAuthKey::from_seed_b64(
7770 "nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A",
7771 )
7772 .unwrap();
7773 let signer = crate::aeo::web_bot_auth::WebBotAuthSigner::new(key, "https://bot.example");
7774 let builder = Client::new()
7775 .get("https://api.example/x")
7776 .header("x-kept", "1")
7777 .sign_web_bot_auth(&signer);
7778 let names: Vec<String> = caller_headers(&builder)
7779 .into_iter()
7780 .map(|(name, _)| name)
7781 .collect();
7782 assert!(names.contains(&"x-kept".to_owned()), "{names:?}");
7783 assert!(names.contains(&"signature-agent".to_owned()), "{names:?}");
7784 assert!(!names.contains(&"signature".to_owned()), "{names:?}");
7785 assert!(!names.contains(&"signature-input".to_owned()), "{names:?}");
7786
7787 let plain = Client::new()
7789 .get("https://api.example/x")
7790 .header("signature", "mine");
7791 assert_eq!(
7792 caller_headers(&plain),
7793 vec![("signature".to_owned(), "mine".to_owned())]
7794 );
7795 }
7796
7797 #[tokio::test]
7800 async fn follow_redirects_keeps_sensitive_headers_same_origin() {
7801 use axum::{Router, routing::get};
7802
7803 let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
7804 let seen2 = seen.clone();
7805 let addr = spawn(
7806 Router::new()
7807 .route("/", get(|| async { redirect_302("/next".to_owned()) }))
7808 .route(
7809 "/next",
7810 get(move |headers: HeaderMap| {
7811 let slot = seen2.clone();
7812 async move {
7813 *slot.lock().unwrap() = Some(headers);
7814 "ok"
7815 }
7816 }),
7817 ),
7818 )
7819 .await;
7820
7821 let resp = Client::new()
7822 .get(format!("http://127.0.0.1:{}/", addr.port()))
7823 .header("authorization", "secret")
7824 .follow_redirects(3, |_| true)
7825 .send()
7826 .await
7827 .unwrap();
7828
7829 assert_eq!(resp.status().as_u16(), 200);
7830 let headers = seen
7831 .lock()
7832 .unwrap()
7833 .clone()
7834 .expect("/next must have been reached");
7835 assert_eq!(
7836 headers.get("authorization").and_then(|v| v.to_str().ok()),
7837 Some("secret"),
7838 "authorization must be preserved on a same-origin redirect"
7839 );
7840 }
7841
7842 #[tokio::test]
7845 async fn follow_redirects_302_post_becomes_get() {
7846 use axum::{
7847 Router,
7848 routing::{any, post},
7849 };
7850
7851 let seen: Arc<Mutex<Option<(String, HeaderMap, Bytes)>>> = Arc::new(Mutex::new(None));
7852 let seen2 = seen.clone();
7853 let b_addr = spawn(Router::new().route(
7855 "/dst",
7856 any(move |method: Method, headers: HeaderMap, body: Bytes| {
7857 let slot = seen2.clone();
7858 async move {
7859 *slot.lock().unwrap() = Some((method.to_string(), headers, body));
7860 "ok"
7861 }
7862 }),
7863 ))
7864 .await;
7865 let b_port = b_addr.port();
7866
7867 let a_addr = spawn(Router::new().route(
7868 "/",
7869 post(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
7870 ))
7871 .await;
7872
7873 let resp = Client::new()
7878 .post(format!("http://127.0.0.1:{}/", a_addr.port()))
7879 .json(&serde_json::json!({"payload": true}))
7880 .follow_redirects(3, |_| true)
7881 .send()
7882 .await
7883 .unwrap();
7884
7885 assert_eq!(resp.status().as_u16(), 200);
7886 let (method, headers, body) = seen
7887 .lock()
7888 .unwrap()
7889 .clone()
7890 .expect("listener B must have been reached");
7891 assert_eq!(method, "GET", "302 must rewrite POST → GET");
7892 assert!(body.is_empty(), "302 POST→GET must drop the request body");
7893 assert!(
7894 !headers.contains_key(reqwest::header::CONTENT_TYPE),
7895 "302 POST→GET must drop the Content-Type payload header"
7896 );
7897 assert!(
7898 !headers.contains_key(reqwest::header::CONTENT_LENGTH),
7899 "302 POST→GET must drop the Content-Length payload header"
7900 );
7901 }
7902
7903 #[tokio::test]
7907 async fn follow_redirects_307_preserves_method_and_body() {
7908 use axum::{
7909 Router,
7910 routing::{any, post},
7911 };
7912
7913 let seen: Arc<Mutex<Option<(String, Bytes)>>> = Arc::new(Mutex::new(None));
7914 let seen2 = seen.clone();
7915 let b_addr = spawn(Router::new().route(
7916 "/dst",
7917 any(move |method: Method, body: Bytes| {
7918 let slot = seen2.clone();
7919 async move {
7920 *slot.lock().unwrap() = Some((method.to_string(), body));
7921 "ok"
7922 }
7923 }),
7924 ))
7925 .await;
7926 let b_port = b_addr.port();
7927
7928 let a_addr = spawn(Router::new().route(
7929 "/",
7930 post(move || async move { redirect_307(format!("http://127.0.0.1:{b_port}/dst")) }),
7931 ))
7932 .await;
7933
7934 let resp = Client::new()
7935 .post(format!("http://127.0.0.1:{}/", a_addr.port()))
7936 .text_body("payload")
7937 .follow_redirects(3, |_| true)
7938 .send()
7939 .await
7940 .unwrap();
7941
7942 assert_eq!(resp.status().as_u16(), 200);
7943 let (method, body) = seen
7944 .lock()
7945 .unwrap()
7946 .clone()
7947 .expect("listener B must have been reached");
7948 assert_eq!(method, "POST", "307 must preserve the POST method");
7949 assert_eq!(
7950 &body[..],
7951 b"payload",
7952 "307 must preserve the request body verbatim"
7953 );
7954 }
7955
7956 #[test]
7965 fn ssrf_redirect_plan_honours_chained_override() {
7966 let client = Client::new();
7967
7968 let default = client.get_ssrf_safe("https://example.com/");
7970 assert_eq!(
7971 default.ssrf_redirect_plan(),
7972 (true, SSRF_SAFE_MAX_REDIRECTS),
7973 "default SSRF-safe path follows up to SSRF_SAFE_MAX_REDIRECTS"
7974 );
7975
7976 let none = client.get_ssrf_safe("https://example.com/").no_redirect();
7978 let (follow, _max) = none.ssrf_redirect_plan();
7979 assert!(
7980 !follow,
7981 "no_redirect() must disable following on the safe path"
7982 );
7983
7984 let follow3 = client
7986 .get_ssrf_safe("https://example.com/")
7987 .follow_redirects(3, |_| true);
7988 assert_eq!(
7989 follow3.ssrf_redirect_plan(),
7990 (true, 3),
7991 "follow_redirects(3, ..) must cap the safe path at 3 hops"
7992 );
7993 }
7994
7995 #[tokio::test]
8000 async fn pin_then_follow_redirects_is_rejected() {
8001 let result = Client::new()
8002 .get("http://example.com/")
8003 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
8004 .follow_redirects(2, |_| true)
8005 .send()
8006 .await;
8007
8008 assert!(
8009 matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
8010 "expected IncompatiblePinRedirect, got {result:?}"
8011 );
8012 }
8013
8014 #[tokio::test]
8017 async fn follow_redirects_then_pin_is_rejected() {
8018 let result = Client::new()
8019 .get("http://example.com/")
8020 .follow_redirects(2, |_| true)
8021 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
8022 .send()
8023 .await;
8024
8025 assert!(
8026 matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
8027 "expected IncompatiblePinRedirect, got {result:?}"
8028 );
8029 }
8030
8031 #[tokio::test]
8034 async fn pin_with_no_redirect_is_allowed() {
8035 use axum::{Router, routing::get};
8036
8037 let addr = spawn(Router::new().route(
8038 "/start",
8039 get(|| async { redirect_302("http://127.0.0.1:1/onward".to_owned()) }),
8040 ))
8041 .await;
8042 let port = addr.port();
8043
8044 let resp = Client::new()
8047 .get(format!("http://pinned.invalid:{port}/start"))
8048 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port))
8049 .no_redirect()
8050 .send()
8051 .await
8052 .expect("pin_to + no_redirect must return the 3xx unfollowed, not error");
8053
8054 assert_eq!(
8055 resp.status().as_u16(),
8056 302,
8057 "pin_to + no_redirect returns the redirect verbatim"
8058 );
8059 }
8060
8061 #[tokio::test]
8068 async fn pin_to_ipv4_literal_host_is_rejected() {
8069 let result = Client::new()
8070 .get("http://198.51.100.1:8080/")
8071 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
8072 .send()
8073 .await;
8074
8075 assert!(
8076 matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
8077 "expected PinRequiresDomainHost, got {result:?}"
8078 );
8079 }
8080
8081 #[tokio::test]
8083 async fn pin_to_ipv6_literal_host_is_rejected() {
8084 let result = Client::new()
8085 .get("http://[2606:4700::1111]/")
8086 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
8087 .send()
8088 .await;
8089
8090 assert!(
8091 matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
8092 "expected PinRequiresDomainHost, got {result:?}"
8093 );
8094 }
8095
8096 #[tokio::test]
8102 async fn get_ssrf_safe_with_pin_to_is_rejected() {
8103 let result = Client::new()
8104 .get_ssrf_safe("http://example.com/")
8105 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
8106 .send()
8107 .await;
8108
8109 assert!(
8110 matches!(result, Err(ClientError::PinNotAllowedWithSsrfSafe(_))),
8111 "expected PinNotAllowedWithSsrfSafe, got {result:?}"
8112 );
8113 }
8114
8115 #[tokio::test]
8120 async fn follow_redirects_does_not_follow_304_with_location() {
8121 use axum::{Router, routing::get};
8122 use std::sync::atomic::AtomicBool;
8123
8124 let touched = Arc::new(AtomicBool::new(false));
8126 let touched2 = touched.clone();
8127 let b_addr = spawn(Router::new().route(
8128 "/dst",
8129 get(move || {
8130 let t = touched2.clone();
8131 async move {
8132 t.store(true, Ordering::SeqCst);
8133 "SHOULD-NOT-BE-HIT"
8134 }
8135 }),
8136 ))
8137 .await;
8138 let b_port = b_addr.port();
8139
8140 let a_addr = spawn(Router::new().route(
8142 "/start",
8143 get(move || async move {
8144 response_with_location(304, format!("http://127.0.0.1:{b_port}/dst"))
8145 }),
8146 ))
8147 .await;
8148
8149 let resp = Client::new()
8150 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
8151 .follow_redirects(5, |_| true)
8152 .send()
8153 .await
8154 .unwrap();
8155
8156 assert_eq!(
8157 resp.status().as_u16(),
8158 304,
8159 "a 304 with a Location header must be returned verbatim, not followed"
8160 );
8161 assert!(
8162 !touched.load(Ordering::SeqCst),
8163 "the 304 Location target must never be requested"
8164 );
8165 }
8166
8167 #[tokio::test]
8171 async fn follow_redirects_does_not_follow_300_with_location() {
8172 use axum::{Router, routing::get};
8173 use std::sync::atomic::AtomicBool;
8174
8175 let touched = Arc::new(AtomicBool::new(false));
8176 let touched2 = touched.clone();
8177 let b_addr = spawn(Router::new().route(
8178 "/dst",
8179 get(move || {
8180 let t = touched2.clone();
8181 async move {
8182 t.store(true, Ordering::SeqCst);
8183 "SHOULD-NOT-BE-HIT"
8184 }
8185 }),
8186 ))
8187 .await;
8188 let b_port = b_addr.port();
8189
8190 let a_addr = spawn(Router::new().route(
8191 "/start",
8192 get(move || async move {
8193 response_with_location(300, format!("http://127.0.0.1:{b_port}/dst"))
8194 }),
8195 ))
8196 .await;
8197
8198 let resp = Client::new()
8199 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
8200 .follow_redirects(5, |_| true)
8201 .send()
8202 .await
8203 .unwrap();
8204
8205 assert_eq!(
8206 resp.status().as_u16(),
8207 300,
8208 "a 300 with a Location header must be returned verbatim, not followed"
8209 );
8210 assert!(
8211 !touched.load(Ordering::SeqCst),
8212 "the 300 Location target must never be requested"
8213 );
8214 }
8215
8216 #[allow(clippy::large_futures, reason = "test futures, awaited once")]
8218 mod deadline_and_budget {
8219 use super::super::*;
8220 use std::sync::atomic::AtomicU32;
8221
8222 async fn counting(
8225 status: Option<u16>,
8226 headers: &'static [(&'static str, &'static str)],
8227 ) -> (String, Arc<AtomicU32>) {
8228 let hits = Arc::new(AtomicU32::new(0));
8229 let counter = Arc::clone(&hits);
8230 let app = axum::Router::new().route(
8231 "/x",
8232 axum::routing::get(move || {
8233 counter.fetch_add(1, Ordering::SeqCst);
8234 async move {
8235 let Some(code) = status else {
8236 return std::future::pending().await;
8237 };
8238 let mut response = axum::response::Response::builder().status(code);
8239 for (name, value) in headers {
8240 response = response.header(*name, *value);
8241 }
8242 response.body(axum::body::Body::empty()).unwrap()
8243 }
8244 }),
8245 );
8246 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8247 let addr = listener.local_addr().unwrap();
8248 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8249 (format!("http://127.0.0.1:{}/x", addr.port()), hits)
8250 }
8251
8252 fn one_retry_client() -> Client {
8254 let mut config = crate::config::HttpClientConfig::default();
8255 config.retry_budget.capacity = 14;
8256 config.retry_budget.retry_ratio = 0.0;
8257 Client::from_config(&config)
8258 }
8259
8260 async fn with_deadline(
8263 after: Duration,
8264 send: impl std::future::Future<Output = Result<Response, ClientError>>,
8265 ) -> Result<Response, ClientError> {
8266 tokio::time::timeout(Duration::from_secs(5), Deadline::after(after).scope(send))
8267 .await
8268 .expect("the deadline must stop the call")
8269 }
8270
8271 fn lock() -> std::sync::MutexGuard<'static, ()> {
8272 let guard = crate::circuit_breaker::TEST_LOCK
8273 .lock()
8274 .unwrap_or_else(std::sync::PoisonError::into_inner);
8275 crate::circuit_breaker::global_registry().clear();
8276 guard
8277 }
8278
8279 async fn redirecting(delay: Duration) -> (String, Arc<Mutex<Vec<(String, u64)>>>) {
8282 let seen = Arc::new(Mutex::new(Vec::new()));
8283 let record = |path: &'static str, seen: Arc<Mutex<Vec<(String, u64)>>>| {
8284 move |headers: HeaderMap| {
8285 let millis = headers
8286 .get(DEADLINE_HEADER)
8287 .and_then(|value| value.to_str().ok()?.parse().ok())
8288 .unwrap_or(u64::MAX);
8289 seen.lock().unwrap().push((path.to_owned(), millis));
8290 }
8291 };
8292 let at_r = record("/r", Arc::clone(&seen));
8293 let at_t = record("/t", Arc::clone(&seen));
8294 let app = axum::Router::new()
8295 .route(
8296 "/r",
8297 axum::routing::get(move |headers: HeaderMap| async move {
8298 at_r(headers);
8299 tokio::time::sleep(delay).await;
8300 axum::response::Redirect::temporary("/t")
8301 }),
8302 )
8303 .route(
8304 "/t",
8305 axum::routing::get(move |headers: HeaderMap| async move {
8306 at_t(headers);
8307 "done"
8308 }),
8309 );
8310 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8311 let addr = listener.local_addr().unwrap();
8312 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8313 (format!("http://127.0.0.1:{}/r", addr.port()), seen)
8314 }
8315
8316 #[tokio::test]
8317 async fn plain_path_sends_a_fresh_deadline_header_after_a_redirect() {
8318 let (url, seen) = redirecting(Duration::from_millis(500)).await;
8319 let response = with_deadline(Duration::from_secs(3), Client::new().get(&url).send())
8320 .await
8321 .unwrap();
8322 assert_eq!(response.status().as_u16(), 200);
8323 let seen = seen.lock().unwrap().clone();
8324 let [(first, at_origin), (second, at_target)] = seen.as_slice() else {
8325 panic!("two hops: {seen:?}");
8326 };
8327 assert_eq!((first.as_str(), second.as_str()), ("/r", "/t"));
8328 assert!(*at_origin <= 3_000, "{seen:?}");
8329 assert!(
8330 *at_target + 400 <= *at_origin,
8331 "the target gets the time left after the redirect: {seen:?}"
8332 );
8333 }
8334
8335 #[tokio::test]
8336 async fn a_redirected_connect_error_charges_the_destination_budget() {
8337 let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8339 let dead_port = closed.local_addr().unwrap().port();
8340 drop(closed);
8341 let target = format!("http://127.0.0.1:{dead_port}/t");
8342 let app = axum::Router::new().route(
8343 "/r",
8344 axum::routing::get(move || {
8345 let target = target.clone();
8346 async move { axum::response::Redirect::temporary(&target) }
8347 }),
8348 );
8349 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8350 let origin_port = listener.local_addr().unwrap().port();
8351 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8352
8353 let client = Client::from_config(&crate::config::HttpClientConfig::default());
8354 let budgets = client.retry.budgets.clone().unwrap();
8355 let result = client
8356 .get(format!("http://127.0.0.1:{origin_port}/r"))
8357 .retries(1)
8358 .send()
8359 .await;
8360 assert!(result.is_err(), "{result:?}");
8361 let full = f64::from(RetryBudgetConfig::default().capacity);
8362 let origin = budgets.for_host(&format!("127.0.0.1:{origin_port}"));
8363 let destination = budgets.for_host(&format!("127.0.0.1:{dead_port}"));
8364 assert!((origin.available() - full).abs() < f64::EPSILON, "origin");
8365 assert!(destination.available() < full, "the destination paid");
8366 }
8367
8368 #[tokio::test]
8369 async fn a_redirect_under_a_deadline_sets_referer() {
8370 let seen = Arc::new(Mutex::new(None));
8371 let record = Arc::clone(&seen);
8372 let app = axum::Router::new()
8373 .route(
8374 "/r",
8375 axum::routing::get(|| async { axum::response::Redirect::temporary("/t") }),
8376 )
8377 .route(
8378 "/t",
8379 axum::routing::get(move |headers: HeaderMap| async move {
8380 *record.lock().unwrap() = headers
8381 .get(reqwest::header::REFERER)
8382 .map(|value| value.to_str().unwrap().to_owned());
8383 }),
8384 );
8385 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8386 let port = listener.local_addr().unwrap().port();
8387 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8388 let origin = format!("http://127.0.0.1:{port}/r");
8389 let response = with_deadline(Duration::from_secs(3), Client::new().get(&origin).send())
8390 .await
8391 .unwrap();
8392 assert_eq!(response.status().as_u16(), 200);
8393 assert_eq!(seen.lock().unwrap().as_deref(), Some(origin.as_str()));
8394 }
8395
8396 #[test]
8397 fn referer_drops_credentials_and_is_not_sent_on_a_downgrade() {
8398 let mut headers = HeaderMap::new();
8399 set_referer(&mut headers, "https://b/x", "https://u:p@a/y#frag");
8400 assert_eq!(headers[reqwest::header::REFERER], "https://a/y");
8401 set_referer(&mut headers, "http://c/z", "https://b/x");
8402 assert!(!headers.contains_key(reqwest::header::REFERER));
8403 }
8404
8405 #[test]
8406 fn every_host_of_a_redirect_chain_is_refilled() {
8407 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
8408 let middle = budgets.for_host("b:443");
8409 let last = budgets.for_host("c:443");
8410 while middle.try_acquire(RetryKind::Transient) {}
8411 while last.try_acquire(RetryKind::Transient) {}
8412 let (middle_before, last_before) = (middle.available(), last.available());
8413 let gate =
8414 RetryGate::with_deadline(None, Some(Arc::clone(&budgets)), Some("a:443"), true);
8415 let mut seen = std::collections::HashSet::new();
8416 gate.record_destinations(
8417 &[
8418 "https://b/1".to_owned(),
8419 "https://c/2".to_owned(),
8420 "https://c/3".to_owned(),
8421 ],
8422 &mut seen,
8423 );
8424 assert!(middle.available() > middle_before, "the middle host");
8425 let one_refill = middle.available() - middle_before;
8426 assert!(
8427 (last.available() - last_before - one_refill).abs() < f64::EPSILON,
8428 "one refill per host"
8429 );
8430 }
8431
8432 async fn redirect_server(location: &'static str, endless: bool) -> String {
8435 let app = axum::Router::new()
8436 .route(
8437 "/r",
8438 axum::routing::get(move || async move {
8439 let body = if endless {
8440 axum::body::Body::from_stream(futures::stream::pending::<
8441 Result<Bytes, std::io::Error>,
8442 >())
8443 } else {
8444 axum::body::Body::empty()
8445 };
8446 axum::response::Response::builder()
8447 .status(302)
8448 .header("location", location)
8449 .body(body)
8450 .unwrap()
8451 }),
8452 )
8453 .route("/t", axum::routing::get(|| async { "done" }));
8454 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8455 let port = listener.local_addr().unwrap().port();
8456 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8457 format!("http://127.0.0.1:{port}/r")
8458 }
8459
8460 #[tokio::test]
8461 async fn a_redirect_under_a_deadline_does_not_read_the_redirect_body() {
8462 let url = redirect_server("/t", true).await;
8463 let response = with_deadline(Duration::from_secs(3), Client::new().get(&url).send())
8464 .await
8465 .unwrap();
8466 assert_eq!(response.status().as_u16(), 200);
8467 assert_eq!(response.text(), "done");
8468 }
8469
8470 #[tokio::test]
8471 async fn a_bad_location_under_a_deadline_returns_the_redirect() {
8472 let url = redirect_server("http://[::1", false).await;
8473 let response = with_deadline(Duration::from_secs(3), Client::new().get(&url).send())
8474 .await
8475 .unwrap();
8476 assert_eq!(response.status().as_u16(), 302);
8477 let plain = Client::new().get(&url).send().await.unwrap();
8478 assert_eq!(
8479 plain.status().as_u16(),
8480 302,
8481 "the same as without a deadline"
8482 );
8483 }
8484
8485 #[tokio::test]
8486 async fn a_redirect_chain_under_a_deadline_shares_one_retry_count() {
8487 use axum::response::IntoResponse;
8488 let origin_hits = Arc::new(AtomicU32::new(0));
8489 let target_hits = Arc::new(AtomicU32::new(0));
8490 let (origin, target) = (Arc::clone(&origin_hits), Arc::clone(&target_hits));
8491 let app = axum::Router::new()
8493 .route(
8494 "/r",
8495 axum::routing::get(move || async move {
8496 if origin.fetch_add(1, Ordering::SeqCst) == 0 {
8497 axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response()
8498 } else {
8499 axum::response::Redirect::temporary("/t").into_response()
8500 }
8501 }),
8502 )
8503 .route(
8504 "/t",
8505 axum::routing::get(move || async move {
8506 if target.fetch_add(1, Ordering::SeqCst) == 0 {
8507 axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response()
8508 } else {
8509 "done".into_response()
8510 }
8511 }),
8512 );
8513 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8514 let port = listener.local_addr().unwrap().port();
8515 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8516 let url = format!("http://127.0.0.1:{port}/r");
8517 let response = with_deadline(
8518 Duration::from_secs(3),
8519 Client::new().get(&url).retries(1).send(),
8520 )
8521 .await
8522 .unwrap();
8523 assert_eq!(response.status().as_u16(), 503, "the one retry is spent");
8524 assert_eq!(origin_hits.load(Ordering::SeqCst), 2);
8525 assert_eq!(target_hits.load(Ordering::SeqCst), 1);
8526 }
8527
8528 #[derive(Debug)]
8530 struct FixedDraw(u64);
8531
8532 impl crate::entropy::Entropy for FixedDraw {
8533 fn next_u64(&self) -> u64 {
8534 self.0
8535 }
8536 fn fill_bytes(&self, dest: &mut [u8]) {
8537 dest.fill(0);
8538 }
8539 }
8540
8541 #[tokio::test]
8542 async fn a_cross_origin_redirect_under_a_deadline_strips_what_reqwest_strips() {
8543 let seen: Arc<std::sync::Mutex<Option<HeaderMap>>> =
8544 Arc::new(std::sync::Mutex::new(None));
8545 let slot = Arc::clone(&seen);
8546 let target = super::spawn(axum::Router::new().route(
8547 "/dst",
8548 axum::routing::get(move |headers: HeaderMap| {
8549 let slot = Arc::clone(&slot);
8550 async move {
8551 *slot.lock().unwrap() = Some(headers);
8552 "ok"
8553 }
8554 }),
8555 ))
8556 .await;
8557 let target_port = target.port();
8558 let origin = super::spawn(axum::Router::new().route(
8559 "/",
8560 axum::routing::get(move || async move {
8561 super::redirect_302(format!("http://127.0.0.1:{target_port}/dst"))
8562 }),
8563 ))
8564 .await;
8565 let response = with_deadline(
8566 Duration::from_secs(3),
8567 Client::new()
8568 .get(format!("http://127.0.0.1:{}/", origin.port()))
8569 .header("authorization", "secret")
8570 .header("cookie", "a=b")
8571 .header("cookie2", "c=d")
8572 .header("proxy-authorization", "Basic zzz")
8573 .header("www-authenticate", "Basic realm=x")
8574 .send(),
8575 )
8576 .await
8577 .unwrap();
8578 assert_eq!(response.status().as_u16(), 200);
8579 let headers = seen.lock().unwrap().clone().expect("target reached");
8580 for name in [
8581 "authorization",
8582 "cookie",
8583 "cookie2",
8584 "proxy-authorization",
8585 "www-authenticate",
8586 ] {
8587 assert!(headers.get(name).is_none(), "{name} reached the target");
8588 }
8589 }
8590
8591 #[tokio::test]
8592 async fn a_retry_cancelled_during_its_backoff_gives_its_tokens_back() {
8593 let (url, hits) = counting(Some(502), &[]).await;
8594 let host = url_host(&url).unwrap();
8595 let mut client = Client::new();
8597 client.entropy = Arc::new(FixedDraw(100));
8598 let budget = client.retry.budgets.clone().unwrap().for_host(&host);
8599 let cancelled = tokio::time::timeout(
8600 Duration::from_millis(50),
8601 client.get(&url).retries(3).send(),
8602 )
8603 .await;
8604 assert!(cancelled.is_err(), "cancelled in the backoff");
8605 assert_eq!(hits.load(Ordering::SeqCst), 1);
8606 let full = f64::from(RetryBudgetConfig::default().capacity);
8607 assert!(
8608 (budget.available() - full).abs() < f64::EPSILON,
8609 "no retry ran, so none is paid for: {}",
8610 budget.available()
8611 );
8612 }
8613
8614 #[tokio::test]
8615 #[allow(clippy::await_holding_lock)]
8616 async fn a_deadline_stop_on_the_custom_path_is_not_a_throttle_reject() {
8617 let _lock = crate::circuit_breaker::TEST_LOCK
8618 .lock()
8619 .unwrap_or_else(std::sync::PoisonError::into_inner);
8620 crate::circuit_breaker::global_registry().clear();
8621 let (url, _hits) = counting(None, &[]).await;
8622 let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
8623 2.0,
8624 Duration::from_secs(120),
8625 ));
8626 let client = Client {
8627 resilience_config: Some(Arc::new(crate::config::ResilienceConfig::default())),
8628 throttle: Some(Arc::clone(&throttle)),
8629 ..Client::new()
8630 };
8631 for _ in 0..10 {
8632 let result = with_deadline(
8633 Duration::from_millis(30),
8634 client.get(&url).no_redirect().breaker_scoped().send(),
8635 )
8636 .await;
8637 assert!(
8638 matches!(result, Err(ClientError::DeadlineExceeded)),
8639 "{result:?}"
8640 );
8641 }
8642 let host = throttle_host(&url).unwrap();
8643 assert!(
8644 throttle.reject_probability(&host, crate::time::ambient_instant()) < f64::EPSILON,
8645 "the caller's deadline is not the host's reject"
8646 );
8647 crate::circuit_breaker::global_registry().clear();
8648 }
8649
8650 #[tokio::test]
8651 async fn the_throttle_applies_under_a_deadline() {
8652 let (url, _hits) = counting(Some(503), &[]).await;
8653 let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
8654 2.0,
8655 Duration::from_secs(120),
8656 ));
8657 let client = Client {
8658 throttle: Some(Arc::clone(&throttle)),
8659 ..Client::new()
8660 };
8661 let mut throttled = 0;
8662 for _ in 0..100 {
8663 let result =
8664 with_deadline(Duration::from_secs(5), client.get(&url).no_retry().send()).await;
8665 if matches!(result, Err(ClientError::ThrottledLocally { .. })) {
8666 throttled += 1;
8667 }
8668 }
8669 assert!(
8670 throttled > 20,
8671 "the deadline path must use the throttle: {throttled}"
8672 );
8673 }
8674
8675 #[tokio::test]
8676 async fn each_retry_under_a_deadline_is_a_throttle_attempt() {
8677 use axum::response::IntoResponse;
8678 let hits = Arc::new(AtomicU32::new(0));
8680 let counter = Arc::clone(&hits);
8681 let upstream = super::spawn(axum::Router::new().route(
8682 "/x",
8683 axum::routing::get(move || {
8684 let status = if counter.fetch_add(1, Ordering::SeqCst) % 3 < 2 {
8685 axum::http::StatusCode::SERVICE_UNAVAILABLE
8686 } else {
8687 axum::http::StatusCode::OK
8688 };
8689 async move { status.into_response() }
8690 }),
8691 ))
8692 .await;
8693 let url = format!("http://127.0.0.1:{}/x", upstream.port());
8694 let throttle = Arc::new(crate::admission::AdaptiveThrottle::new(
8695 2.0,
8696 Duration::from_secs(120),
8697 ));
8698 let mut client = Client {
8701 throttle: Some(Arc::clone(&throttle)),
8702 ..Client::new()
8703 };
8704 client.entropy = Arc::new(FixedDraw(u64::MAX));
8705 for _ in 0..10 {
8706 let response =
8707 with_deadline(Duration::from_secs(4), client.get(&url).retries(2).send())
8708 .await
8709 .unwrap();
8710 assert_eq!(response.status().as_u16(), 200);
8711 }
8712 assert_eq!(hits.load(Ordering::SeqCst), 30);
8713 let host = throttle_host(&url).unwrap();
8716 assert!(
8717 throttle.reject_probability(&host, crate::time::ambient_instant()) > 0.3,
8718 "each attempt counts"
8719 );
8720 }
8721
8722 #[tokio::test]
8723 async fn a_retry_sends_what_is_left_of_the_callers_deadline_header() {
8724 use axum::response::IntoResponse;
8725 let seen: Arc<std::sync::Mutex<Vec<u64>>> = Arc::default();
8727 let slot = Arc::clone(&seen);
8728 let upstream = super::spawn(axum::Router::new().route(
8729 "/x",
8730 axum::routing::get(move |headers: HeaderMap| {
8731 let slot = Arc::clone(&slot);
8732 async move {
8733 let value = headers
8734 .get(DEADLINE_HEADER)
8735 .and_then(|v| v.to_str().ok()?.parse().ok())
8736 .unwrap_or(u64::MAX);
8737 let first = {
8738 let mut seen = slot.lock().unwrap();
8739 seen.push(value);
8740 seen.len() == 1
8741 };
8742 if first {
8743 tokio::time::sleep(Duration::from_millis(60)).await;
8744 axum::http::StatusCode::BAD_GATEWAY.into_response()
8745 } else {
8746 axum::http::StatusCode::OK.into_response()
8747 }
8748 }
8749 }),
8750 ))
8751 .await;
8752 let mut client = Client::new();
8754 client.entropy = Arc::new(FixedDraw(0));
8755 let response = with_deadline(
8756 Duration::from_secs(4),
8757 client
8758 .get(format!("http://127.0.0.1:{}/x", upstream.port()))
8759 .header(DEADLINE_HEADER, "200")
8760 .retries(1)
8761 .send(),
8762 )
8763 .await
8764 .unwrap();
8765 assert_eq!(response.status().as_u16(), 200);
8766 let seen = seen.lock().unwrap().clone();
8767 assert_eq!(seen.len(), 2);
8768 assert!(seen[0] <= 200, "{seen:?}");
8769 assert!(
8771 seen[1] <= 140,
8772 "the retry re-extended the deadline: {seen:?}"
8773 );
8774 }
8775
8776 #[tokio::test]
8777 async fn a_redirect_chain_under_a_deadline_keeps_the_client_timeout() {
8778 use axum::response::IntoResponse;
8779 let app = axum::Router::new()
8781 .route(
8782 "/a",
8783 axum::routing::get(|| async {
8784 tokio::time::sleep(Duration::from_millis(200)).await;
8785 axum::response::Redirect::temporary("/b").into_response()
8786 }),
8787 )
8788 .route(
8789 "/b",
8790 axum::routing::get(|| async {
8791 tokio::time::sleep(Duration::from_millis(200)).await;
8792 "ok"
8793 }),
8794 );
8795 let upstream = super::spawn(app).await;
8796 let mut client = Client::new();
8797 client.retry_policy.request_timeout = Some(Duration::from_millis(300));
8798 let result = with_deadline(
8799 Duration::from_secs(4),
8800 client
8801 .get(format!("http://127.0.0.1:{}/a", upstream.port()))
8802 .no_retry()
8803 .send(),
8804 )
8805 .await;
8806 assert!(
8808 matches!(&result, Err(ClientError::Request(e)) if e.is_timeout()),
8809 "the chain must not get a new timeout per hop: {result:?}"
8810 );
8811 }
8812
8813 #[tokio::test]
8814 async fn without_a_deadline_a_retry_sends_what_is_left_of_the_callers_header() {
8815 use axum::response::IntoResponse;
8816 let seen: Arc<std::sync::Mutex<Vec<u64>>> = Arc::default();
8818 let slot = Arc::clone(&seen);
8819 let upstream = super::spawn(axum::Router::new().route(
8820 "/x",
8821 axum::routing::get(move |headers: HeaderMap| {
8822 let slot = Arc::clone(&slot);
8823 async move {
8824 let first = {
8825 let mut seen = slot.lock().unwrap();
8826 seen.push(header_millis(&headers));
8827 seen.len() == 1
8828 };
8829 if first {
8830 tokio::time::sleep(Duration::from_millis(60)).await;
8831 axum::http::StatusCode::BAD_GATEWAY.into_response()
8832 } else {
8833 axum::http::StatusCode::OK.into_response()
8834 }
8835 }
8836 }),
8837 ))
8838 .await;
8839 let mut client = Client::new();
8840 client.entropy = Arc::new(FixedDraw(0));
8841 let response = client
8843 .get(format!("http://127.0.0.1:{}/x", upstream.port()))
8844 .header(DEADLINE_HEADER, "200")
8845 .retries(1)
8846 .send()
8847 .await
8848 .unwrap();
8849 assert_eq!(response.status().as_u16(), 200);
8850 let seen = seen.lock().unwrap().clone();
8851 assert_eq!(seen[0], 200, "{seen:?}");
8852 assert!(seen[1] <= 140, "the retry re-extended the header: {seen:?}");
8853 }
8854
8855 #[tokio::test]
8856 async fn without_a_deadline_a_redirect_hop_sends_what_is_left_of_the_callers_header() {
8857 use axum::response::IntoResponse;
8858 let seen: Arc<std::sync::Mutex<Vec<u64>>> = Arc::default();
8859 let (at_a, at_b) = (Arc::clone(&seen), Arc::clone(&seen));
8860 let app = axum::Router::new()
8861 .route(
8862 "/a",
8863 axum::routing::get(move |headers: HeaderMap| async move {
8864 at_a.lock().unwrap().push(header_millis(&headers));
8865 tokio::time::sleep(Duration::from_millis(60)).await;
8866 axum::response::Redirect::temporary("/b").into_response()
8867 }),
8868 )
8869 .route(
8870 "/b",
8871 axum::routing::get(move |headers: HeaderMap| async move {
8872 at_b.lock().unwrap().push(header_millis(&headers));
8873 "ok"
8874 }),
8875 );
8876 let upstream = super::spawn(app).await;
8877 let response = Client::new()
8878 .get(format!("http://127.0.0.1:{}/a", upstream.port()))
8879 .header(DEADLINE_HEADER, "200")
8880 .send()
8881 .await
8882 .unwrap();
8883 assert_eq!(response.status().as_u16(), 200);
8884 let seen = seen.lock().unwrap().clone();
8885 assert_eq!(seen.len(), 2, "{seen:?}");
8886 assert_eq!(seen[0], 200, "{seen:?}");
8887 assert!(seen[1] <= 140, "the hop re-extended the header: {seen:?}");
8888 }
8889
8890 fn header_millis(headers: &HeaderMap) -> u64 {
8892 headers
8893 .get(DEADLINE_HEADER)
8894 .and_then(|v| v.to_str().ok()?.parse().ok())
8895 .unwrap_or(u64::MAX)
8896 }
8897
8898 #[tokio::test]
8899 async fn a_redirect_hop_continues_the_chain_backoff() {
8900 use axum::response::IntoResponse;
8901 let origin_hits = Arc::new(AtomicU32::new(0));
8902 let target_hits = Arc::new(AtomicU32::new(0));
8903 let (origin, target) = (Arc::clone(&origin_hits), Arc::clone(&target_hits));
8904 let app = axum::Router::new()
8906 .route(
8907 "/r",
8908 axum::routing::get(move || async move {
8909 if origin.fetch_add(1, Ordering::SeqCst) == 0 {
8910 axum::http::StatusCode::BAD_GATEWAY.into_response()
8911 } else {
8912 axum::response::Redirect::temporary("/t").into_response()
8913 }
8914 }),
8915 )
8916 .route(
8917 "/t",
8918 axum::routing::get(move || async move {
8919 if target.fetch_add(1, Ordering::SeqCst) == 0 {
8920 axum::http::StatusCode::BAD_GATEWAY.into_response()
8921 } else {
8922 "done".into_response()
8923 }
8924 }),
8925 );
8926 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8927 let port = listener.local_addr().unwrap().port();
8928 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8929 let url = format!("http://127.0.0.1:{port}/r");
8930 let mut client = Client::new();
8933 client.entropy = Arc::new(FixedDraw(20_099));
8934 let start = std::time::Instant::now();
8935 let response =
8936 with_deadline(Duration::from_secs(5), client.get(&url).retries(2).send())
8937 .await
8938 .unwrap();
8939 assert_eq!(response.status().as_u16(), 200);
8940 assert_eq!(origin_hits.load(Ordering::SeqCst), 2);
8941 assert_eq!(target_hits.load(Ordering::SeqCst), 2);
8942 assert!(
8943 start.elapsed() >= Duration::from_millis(200),
8944 "the hop's retry is the chain's second: {:?}",
8945 start.elapsed()
8946 );
8947 }
8948
8949 #[tokio::test]
8950 async fn a_post_redirected_to_get_under_a_deadline_gets_no_retry() {
8951 use axum::response::IntoResponse;
8952 let target_hits = Arc::new(AtomicU32::new(0));
8953 let target = Arc::clone(&target_hits);
8954 let app = axum::Router::new()
8955 .route(
8956 "/r",
8957 axum::routing::post(|| async { axum::response::Redirect::to("/t") }),
8958 )
8959 .route(
8960 "/t",
8961 axum::routing::get(move || async move {
8962 if target.fetch_add(1, Ordering::SeqCst) == 0 {
8963 axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response()
8964 } else {
8965 "done".into_response()
8966 }
8967 }),
8968 );
8969 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
8970 let port = listener.local_addr().unwrap().port();
8971 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
8972 let url = format!("http://127.0.0.1:{port}/r");
8973 let response = with_deadline(
8974 Duration::from_secs(3),
8975 Client::new().post(&url).retries(3).send(),
8976 )
8977 .await
8978 .unwrap();
8979 assert_eq!(response.status().as_u16(), 503, "as without a deadline");
8980 assert_eq!(target_hits.load(Ordering::SeqCst), 1);
8981 }
8982
8983 #[tokio::test]
8984 async fn a_backoff_past_the_hop_deadline_returns_the_response() {
8985 let (url, hits) = counting(Some(502), &[]).await;
8986 let policy = RetryPolicy {
8987 max_retries: 20,
8988 ..RetryPolicy::default()
8989 };
8990 let gate = RetryGate::with_deadline(None, None, None, true);
8991 let client = reqwest::Client::new();
8994 let _ = client.get(&url).send().await;
8995 let result = send_one(
8996 &client,
8997 &Method::GET,
8998 &url,
8999 &HeaderMap::new(),
9000 None,
9001 &policy,
9002 &FixedDraw(400),
9008 false,
9009 Some(crate::time::ambient_instant() + Duration::from_secs(1)),
9010 false,
9011 &gate,
9012 false,
9013 None,
9014 None,
9015 )
9016 .await;
9017 assert_eq!(
9018 result.map(|response| response.status().as_u16()).ok(),
9019 Some(502),
9020 "the last response, not a deadline error"
9021 );
9022 assert!(hits.load(Ordering::SeqCst) >= 3, "no retry ran");
9024 }
9025
9026 #[tokio::test]
9027 async fn a_transport_retry_past_the_hop_deadline_is_not_charged() {
9028 let app = axum::Router::new().route(
9029 "/hang",
9030 axum::routing::get(|| async {
9031 std::future::pending::<()>().await;
9032 ""
9033 }),
9034 );
9035 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9036 let port = listener.local_addr().unwrap().port();
9037 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
9038 let host = format!("127.0.0.1:{port}");
9039 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
9040 let gate =
9041 RetryGate::with_deadline(None, Some(Arc::clone(&budgets)), Some(&host), true);
9042 let result = send_one(
9044 &reqwest::Client::new(),
9045 &Method::GET,
9046 &format!("http://{host}/hang"),
9047 &HeaderMap::new(),
9048 None,
9049 &RetryPolicy::default(),
9050 &crate::entropy::SeededEntropy::new(1),
9051 false,
9052 Some(crate::time::ambient_instant() + Duration::from_millis(200)),
9053 false,
9054 &gate,
9055 false,
9056 None,
9057 None,
9058 )
9059 .await;
9060 assert!(result.is_err(), "{result:?}");
9061 let full = f64::from(RetryBudgetConfig::default().capacity);
9062 let available = budgets.for_host(&host).available();
9063 assert!(
9064 (full - available).abs() < f64::EPSILON,
9065 "no tokens for a retry that cannot run: {available}"
9066 );
9067 }
9068
9069 #[tokio::test]
9070 async fn a_retry_needs_a_minimum_attempt_before_the_hop_deadline() {
9071 let (url, hits) = counting(Some(502), &[]).await;
9072 let gate = RetryGate::with_deadline(None, None, None, true);
9073 let result = send_one(
9076 &reqwest::Client::new(),
9077 &Method::GET,
9078 &url,
9079 &HeaderMap::new(),
9080 None,
9081 &RetryPolicy::default(),
9082 &FixedDraw(100),
9083 false,
9084 Some(crate::time::ambient_instant() + Duration::from_millis(105)),
9085 false,
9086 &gate,
9087 false,
9088 None,
9089 None,
9090 )
9091 .await;
9092 assert_eq!(
9093 result.map(|response| response.status().as_u16()).ok(),
9094 Some(502),
9095 "the response, not a retry that cannot run"
9096 );
9097 assert_eq!(hits.load(Ordering::SeqCst), 1);
9098 }
9099
9100 #[tokio::test]
9101 async fn an_origin_failure_after_a_redirect_charges_the_origin() {
9102 let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9104 let dead_port = closed.local_addr().unwrap().port();
9105 drop(closed);
9106 let target = format!("http://127.0.0.1:{dead_port}/t");
9107 let hits = Arc::new(AtomicU32::new(0));
9108 let counter = Arc::clone(&hits);
9109 let app = axum::Router::new().route(
9110 "/r",
9111 axum::routing::get(move || {
9112 let target = target.clone();
9113 let first = counter.fetch_add(1, Ordering::SeqCst) == 0;
9114 async move {
9115 if !first {
9116 std::future::pending::<()>().await;
9117 }
9118 axum::response::Redirect::temporary(&target)
9119 }
9120 }),
9121 );
9122 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9123 let origin_port = listener.local_addr().unwrap().port();
9124 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
9125
9126 let client = Client::with_timeout(Duration::from_millis(300));
9127 let budgets = client.retry.budgets.clone().unwrap();
9128 let result = client
9129 .get(format!("http://127.0.0.1:{origin_port}/r"))
9130 .retries(2)
9131 .send()
9132 .await;
9133 assert!(result.is_err(), "{result:?}");
9134 let full = f64::from(RetryBudgetConfig::default().capacity);
9135 let cost = f64::from(RetryBudgetConfig::default().transient_cost);
9136 let origin = budgets.for_host(&format!("127.0.0.1:{origin_port}"));
9137 let dead = budgets.for_host(&format!("127.0.0.1:{dead_port}"));
9138 assert!(
9139 (full - dead.available() - cost).abs() < f64::EPSILON,
9140 "the dead target pays its one retry: {}",
9141 dead.available()
9142 );
9143 assert!(
9144 (full - origin.available() - cost).abs() < f64::EPSILON,
9145 "the origin pays the retry after its own timeout: {}",
9146 origin.available()
9147 );
9148 }
9149
9150 #[tokio::test]
9151 async fn a_redirect_host_is_refilled_when_the_final_body_fails() {
9152 let target_app = axum::Router::new().route(
9154 "/t",
9155 axum::routing::get(|| async {
9156 axum::body::Body::from_stream(futures::stream::once(async {
9157 Err::<Bytes, _>(std::io::Error::other("broken body"))
9158 }))
9159 }),
9160 );
9161 let target_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9162 let target_port = target_listener.local_addr().unwrap().port();
9163 tokio::spawn(async move { axum::serve(target_listener, target_app).await.unwrap() });
9164 let target = format!("http://127.0.0.1:{target_port}/t");
9165 let origin_app = axum::Router::new().route(
9166 "/r",
9167 axum::routing::get(move || {
9168 let target = target.clone();
9169 async move { axum::response::Redirect::temporary(&target) }
9170 }),
9171 );
9172 let origin_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9173 let origin_port = origin_listener.local_addr().unwrap().port();
9174 tokio::spawn(async move { axum::serve(origin_listener, origin_app).await.unwrap() });
9175
9176 let client = Client::new();
9177 let budgets = client.retry.budgets.clone().unwrap();
9178 let drained = budgets.for_host(&format!("127.0.0.1:{target_port}"));
9179 while drained.try_acquire(RetryKind::Transient) {}
9180 let before = drained.available();
9181 let result = client
9182 .get(format!("http://127.0.0.1:{origin_port}/r"))
9183 .send()
9184 .await;
9185 assert!(result.is_err(), "{result:?}");
9186 assert!(drained.available() > before, "the target got its refill");
9187 }
9188
9189 #[tokio::test]
9190 async fn ten_redirects_are_followed_with_and_without_a_deadline() {
9191 let app = axum::Router::new().route(
9193 "/r/{n}",
9194 axum::routing::get(
9195 |axum::extract::Path(n): axum::extract::Path<u32>| async move {
9196 use axum::response::IntoResponse;
9197 if n == 0 {
9198 "done".into_response()
9199 } else {
9200 axum::response::Redirect::temporary(&format!("/r/{}", n - 1))
9201 .into_response()
9202 }
9203 },
9204 ),
9205 );
9206 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9207 let port = listener.local_addr().unwrap().port();
9208 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
9209 let ten = format!("http://127.0.0.1:{port}/r/10");
9210 let eleven = format!("http://127.0.0.1:{port}/r/11");
9211
9212 let plain = Client::new().get(&ten).send().await.unwrap();
9213 assert_eq!(plain.status().as_u16(), 200, "as reqwest's own limit");
9214 let timed = with_deadline(Duration::from_secs(3), Client::new().get(&ten).send())
9215 .await
9216 .unwrap();
9217 assert_eq!(timed.status().as_u16(), 200);
9218
9219 assert!(Client::new().get(&eleven).send().await.is_err());
9220 assert!(
9221 with_deadline(Duration::from_secs(3), Client::new().get(&eleven).send())
9222 .await
9223 .is_err()
9224 );
9225 }
9226
9227 #[tokio::test]
9228 async fn a_caller_deadline_header_is_capped_at_the_time_left() {
9229 let seen = Arc::new(Mutex::new(Vec::new()));
9230 let record = Arc::clone(&seen);
9231 let app = axum::Router::new().route(
9232 "/x",
9233 axum::routing::get(move |headers: HeaderMap| async move {
9234 let values: Vec<String> = headers
9235 .get_all(DEADLINE_HEADER)
9236 .iter()
9237 .map(|value| value.to_str().unwrap().to_owned())
9238 .collect();
9239 record.lock().unwrap().push(values);
9240 }),
9241 );
9242 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9243 let port = listener.local_addr().unwrap().port();
9244 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
9245 let url = format!("http://127.0.0.1:{port}/x");
9246
9247 let send = Client::new()
9249 .get(&url)
9250 .header(DEADLINE_HEADER, "5000")
9251 .send();
9252 with_deadline(Duration::from_secs(1), send).await.unwrap();
9253 let send = Client::new()
9255 .get(&url)
9256 .header(DEADLINE_HEADER, "200")
9257 .send();
9258 with_deadline(Duration::from_secs(3), send).await.unwrap();
9259 Client::new()
9261 .get(&url)
9262 .header(DEADLINE_HEADER, "5000")
9263 .send()
9264 .await
9265 .unwrap();
9266
9267 let seen = seen.lock().unwrap().clone();
9268 assert_eq!(seen.len(), 3);
9269 assert_eq!(seen[0].len(), 1, "one header: {seen:?}");
9270 assert!(seen[0][0].parse::<u64>().unwrap() <= 1_000, "{seen:?}");
9271 assert_eq!(seen[1], vec!["200".to_owned()]);
9272 assert_eq!(seen[2], vec!["5000".to_owned()]);
9273 }
9274
9275 #[tokio::test]
9276 async fn plain_path_follows_a_redirect_without_a_deadline() {
9277 let (url, seen) = redirecting(Duration::ZERO).await;
9278 let response = Client::new().get(&url).send().await.unwrap();
9279 assert_eq!(response.status().as_u16(), 200);
9280 assert_eq!(seen.lock().unwrap().len(), 2);
9281 }
9282
9283 #[tokio::test]
9284 #[allow(clippy::await_holding_lock)]
9285 async fn plain_path_stops_a_hanging_call_at_the_deadline() {
9286 let _lock = lock();
9287 let (url, hits) = counting(None, &[]).await;
9288 let result =
9289 with_deadline(Duration::from_millis(300), Client::new().get(&url).send()).await;
9290 assert!(
9291 matches!(result, Err(ClientError::DeadlineExceeded)),
9292 "{result:?}"
9293 );
9294 assert_eq!(
9295 hits.load(Ordering::SeqCst),
9296 1,
9297 "no retry after the deadline"
9298 );
9299 }
9300
9301 #[tokio::test]
9302 #[allow(clippy::await_holding_lock)]
9303 async fn plain_path_does_not_start_a_retry_that_cannot_fit() {
9304 let _lock = lock();
9305 let (url, hits) = counting(Some(503), &[]).await;
9306 let response = with_deadline(
9309 Duration::from_millis(700),
9310 Client::new().get(&url).retries(50).send(),
9311 )
9312 .await
9313 .unwrap();
9314 assert_eq!(response.status().as_u16(), 503);
9315 let made = hits.load(Ordering::SeqCst);
9316 assert!(
9317 (2..51).contains(&made),
9318 "retries stop at the deadline: {made}"
9319 );
9320 }
9321
9322 #[tokio::test]
9323 #[allow(clippy::await_holding_lock)]
9324 async fn plain_path_skips_a_retry_after_wait_past_the_deadline() {
9325 let _lock = lock();
9326 let (url, hits) = counting(Some(429), &[("retry-after", "10")]).await;
9327 let response = with_deadline(Duration::from_secs(2), Client::new().get(&url).send())
9328 .await
9329 .unwrap();
9330 assert_eq!(
9331 response.status().as_u16(),
9332 429,
9333 "the 429 comes back at once"
9334 );
9335 assert_eq!(hits.load(Ordering::SeqCst), 1);
9336 }
9337
9338 #[tokio::test]
9339 #[allow(clippy::await_holding_lock)]
9340 async fn plain_path_obeys_the_retry_budget() {
9341 let _lock = lock();
9342 let (url, hits) = counting(Some(503), &[]).await;
9343 let client = one_retry_client();
9344 let first = client.get(&url).send().await.unwrap();
9345 assert_eq!(first.status().as_u16(), 503);
9346 assert_eq!(hits.load(Ordering::SeqCst), 2, "one retry from the budget");
9347 let second = client.get(&url).send().await.unwrap();
9348 assert_eq!(second.status().as_u16(), 503);
9349 assert_eq!(
9350 hits.load(Ordering::SeqCst),
9351 3,
9352 "the empty budget blocks retries"
9353 );
9354 }
9355
9356 #[tokio::test]
9357 async fn custom_path_obeys_the_deadline_and_the_budget() {
9358 let (url, hits) = counting(Some(503), &[]).await;
9359 let response = with_deadline(
9360 Duration::from_millis(700),
9361 Client::new().get(&url).retries(50).no_redirect().send(),
9362 )
9363 .await
9364 .unwrap();
9365 assert_eq!(response.status().as_u16(), 503);
9366 let made = hits.load(Ordering::SeqCst);
9367 assert!(
9368 (2..51).contains(&made),
9369 "retries stop at the deadline: {made}"
9370 );
9371
9372 let client = one_retry_client();
9373 client.get(&url).no_redirect().send().await.unwrap();
9374 assert_eq!(
9375 hits.load(Ordering::SeqCst),
9376 made + 2,
9377 "one retry from the budget"
9378 );
9379 client.get(&url).no_redirect().send().await.unwrap();
9380 assert_eq!(
9381 hits.load(Ordering::SeqCst),
9382 made + 3,
9383 "the empty budget blocks retries"
9384 );
9385 }
9386
9387 #[tokio::test]
9388 async fn custom_path_reports_a_deadline_timeout_as_deadline_exceeded() {
9389 let (url, hits) = counting(None, &[]).await;
9390 let result = with_deadline(
9391 Duration::from_millis(300),
9392 Client::new().get(&url).no_redirect().send(),
9393 )
9394 .await;
9395 assert!(
9396 matches!(result, Err(ClientError::DeadlineExceeded)),
9397 "{result:?}"
9398 );
9399 assert_eq!(
9400 hits.load(Ordering::SeqCst),
9401 1,
9402 "no retry after the deadline"
9403 );
9404 }
9405
9406 async fn stalled_body() -> String {
9409 let app = axum::Router::new().route(
9410 "/x",
9411 axum::routing::get(|| async {
9412 let stream = futures::stream::pending::<Result<bytes::Bytes, std::io::Error>>();
9413 axum::response::Response::new(axum::body::Body::from_stream(stream))
9414 }),
9415 );
9416 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
9417 let addr = listener.local_addr().unwrap();
9418 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
9419 format!("http://127.0.0.1:{}/x", addr.port())
9420 }
9421
9422 #[tokio::test]
9423 #[allow(clippy::await_holding_lock)]
9424 async fn a_body_stalled_past_the_deadline_is_deadline_exceeded() {
9425 let _lock = lock();
9426 let url = stalled_body().await;
9427 let plain =
9428 with_deadline(Duration::from_millis(300), Client::new().get(&url).send()).await;
9429 assert!(
9430 matches!(plain, Err(ClientError::DeadlineExceeded)),
9431 "{plain:?}"
9432 );
9433 let custom = with_deadline(
9434 Duration::from_millis(300),
9435 Client::new().get(&url).no_redirect().send(),
9436 )
9437 .await;
9438 assert!(
9439 matches!(custom, Err(ClientError::DeadlineExceeded)),
9440 "{custom:?}"
9441 );
9442 }
9443
9444 #[tokio::test]
9445 async fn ssrf_safe_path_fails_at_once_when_the_deadline_has_passed() {
9446 let result = Deadline::at(tokio::time::Instant::now())
9447 .scope(Client::new().get_ssrf_safe("https://example.com/").send())
9448 .await;
9449 assert!(
9450 matches!(result, Err(ClientError::DeadlineExceeded)),
9451 "{result:?}"
9452 );
9453 }
9454
9455 #[test]
9456 fn a_redirect_hop_uses_the_budget_of_its_own_host() {
9457 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
9458 let origin = RetryGate::with_deadline(
9459 None,
9460 Some(Arc::clone(&budgets)),
9461 Some("origin:443"),
9462 true,
9463 );
9464 let hop = origin.for_hop("https://target/x", &mut std::collections::HashSet::new());
9465 assert!(Arc::ptr_eq(
9466 hop.budget.as_ref().unwrap(),
9467 &budgets.for_host("target:443")
9468 ));
9469 assert!(!Arc::ptr_eq(
9470 hop.budget.as_ref().unwrap(),
9471 origin.budget.as_ref().unwrap()
9472 ));
9473 }
9474
9475 #[test]
9476 fn a_redirect_chain_refills_each_host_once() {
9477 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
9478 let origin_budget = budgets.for_host("a:443");
9479 let target_budget = budgets.for_host("b:443");
9480 while origin_budget.try_acquire(RetryKind::Transient) {}
9481 while target_budget.try_acquire(RetryKind::Transient) {}
9482 let origin =
9483 RetryGate::with_deadline(None, Some(Arc::clone(&budgets)), Some("a:443"), true);
9484 let mut refilled = std::collections::HashSet::from(["a:443".to_owned()]);
9485 origin.check().unwrap();
9487 let after_first = origin_budget.available();
9488 let before_b = target_budget.available();
9489 origin
9490 .for_hop("https://b/1", &mut refilled)
9491 .check()
9492 .unwrap();
9493 let after_b = target_budget.available();
9494 assert!(after_b > before_b, "the first visit to b refills it");
9495 origin
9496 .for_hop("https://a/2", &mut refilled)
9497 .check()
9498 .unwrap();
9499 origin
9500 .for_hop("https://b/3", &mut refilled)
9501 .check()
9502 .unwrap();
9503 assert!(
9504 (origin_budget.available() - after_first).abs() < f64::EPSILON,
9505 "the origin is not credited again"
9506 );
9507 assert!(
9508 (target_budget.available() - after_b).abs() < f64::EPSILON,
9509 "the target is credited once"
9510 );
9511 }
9512
9513 #[test]
9514 fn rekey_moves_to_the_budget_of_the_answering_host() {
9515 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
9516 let mut gate = RetryGate::with_deadline(
9517 None,
9518 Some(Arc::clone(&budgets)),
9519 Some("origin:443"),
9520 true,
9521 );
9522 let origin = Arc::clone(gate.budget.as_ref().unwrap());
9523 gate.rekey("https://origin/x");
9524 assert!(
9525 Arc::ptr_eq(gate.budget.as_ref().unwrap(), &origin),
9526 "same host"
9527 );
9528 gate.rekey("https://target/x");
9529 assert!(Arc::ptr_eq(
9530 gate.budget.as_ref().unwrap(),
9531 &budgets.for_host("target:443")
9532 ));
9533 }
9534
9535 #[test]
9536 fn a_redirected_answer_refills_the_destination_budget() {
9537 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
9538 let target = budgets.for_host("target:443");
9539 while target.try_acquire(RetryKind::Transient) {}
9540 let gate = RetryGate::with_deadline(
9541 None,
9542 Some(Arc::clone(&budgets)),
9543 Some("origin:443"),
9544 true,
9545 );
9546 let origin = Arc::clone(gate.budget.as_ref().unwrap());
9547 let (before, origin_before) = (target.available(), origin.available());
9548
9549 gate.record_destination("https://origin/x");
9550 assert!(
9551 (target.available() - before).abs() < f64::EPSILON,
9552 "same host"
9553 );
9554
9555 gate.record_destination("https://target/x");
9556 assert!(target.available() > before, "the destination is refilled");
9557 assert!(
9558 (origin.available() - origin_before).abs() < f64::EPSILON,
9559 "the origin budget is unchanged"
9560 );
9561 assert!(
9562 Arc::ptr_eq(gate.budget.as_ref().unwrap(), &origin),
9563 "no rekey"
9564 );
9565 }
9566
9567 #[tokio::test(start_paused = true)]
9568 async fn an_expired_deadline_reclassifies_a_hop_error() {
9569 let gate = Deadline::at(tokio::time::Instant::now())
9570 .scope(async { RetryGate::start(None, None, true) })
9571 .await;
9572 let error = gate.classify(ClientError::InvalidUrl("dns stalled".into()));
9573 assert!(matches!(error, ClientError::DeadlineExceeded));
9574 let open = RetryGate::start(None, None, true);
9575 let error = open.classify(ClientError::InvalidUrl("dns".into()));
9576 assert!(matches!(error, ClientError::InvalidUrl(_)));
9577 }
9578
9579 #[tokio::test]
9580 async fn a_rejected_ssrf_safe_host_is_not_refilled() {
9581 let client = Client::new();
9582 let budget = client
9583 .retry
9584 .budgets
9585 .clone()
9586 .unwrap()
9587 .for_host("127.0.0.1:80");
9588 while budget.try_acquire(RetryKind::Transient) {}
9589 let empty = budget.available();
9590 let result = client.get_ssrf_safe("http://127.0.0.1/").send().await;
9592 assert!(result.is_err(), "{result:?}");
9593 assert!(
9594 (budget.available() - empty).abs() < f64::EPSILON,
9595 "no refill without an attempt: {}",
9596 budget.available()
9597 );
9598 }
9599
9600 #[tokio::test(start_paused = true)]
9601 async fn a_followed_redirect_host_is_refilled_after_the_deadline() {
9602 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
9603 let target = budgets.for_host("t:80");
9604 while target.try_acquire(RetryKind::Transient) {}
9605 let empty = target.available();
9606 let gate = RetryGate::with_deadline(
9607 Some(Deadline::after(Duration::from_millis(1))),
9608 Some(budgets),
9609 Some("o:80"),
9610 true,
9611 );
9612 tokio::time::advance(Duration::from_millis(5)).await;
9614 let mut seen = std::collections::HashSet::from(["o:80".to_owned()]);
9615 gate.record_destinations(&["http://t/".to_owned()], &mut seen);
9616 assert!(
9617 target.available() > empty,
9618 "the target's attempt started, so it is refilled"
9619 );
9620 }
9621
9622 #[tokio::test(start_paused = true)]
9623 async fn a_request_past_its_deadline_does_not_refill_the_budget() {
9624 let budgets = Arc::new(RetryBudgets::new(&RetryBudgetConfig::default()));
9625 let budget = budgets.for_host("h:80");
9626 while budget.try_acquire(RetryKind::Transient) {}
9627 let empty = budget.available();
9628 let late = RetryGate::with_deadline(
9629 Some(Deadline::at(tokio::time::Instant::now())),
9630 Some(Arc::clone(&budgets)),
9631 Some("h:80"),
9632 true,
9633 );
9634 assert!(late.check().is_err());
9635 assert!(
9636 (budget.available() - empty).abs() < f64::EPSILON,
9637 "no refill"
9638 );
9639
9640 let short = RetryGate::with_deadline(
9642 Some(Deadline::after(Duration::from_millis(1))),
9643 Some(Arc::clone(&budgets)),
9644 Some("h:80"),
9645 true,
9646 );
9647 tokio::time::advance(Duration::from_millis(5)).await;
9648 assert!(short.check().is_err());
9649 assert!(
9650 (budget.available() - empty).abs() < f64::EPSILON,
9651 "no refill without an attempt"
9652 );
9653
9654 let live = RetryGate::with_deadline(None, Some(budgets), Some("h:80"), true);
9655 assert!(
9656 (budget.available() - empty).abs() < f64::EPSILON,
9657 "not when built"
9658 );
9659 live.check().unwrap();
9660 let once = budget.available();
9661 assert!(once > empty, "a live request refills at its first attempt");
9662 live.check().unwrap();
9663 assert!((budget.available() - once).abs() < f64::EPSILON, "once");
9664 }
9665
9666 #[test]
9667 fn budget_keys_keep_the_port() {
9668 assert_eq!(url_host("http://svc:8001/a").as_deref(), Some("svc:8001"));
9669 assert_eq!(url_host("https://svc/a").as_deref(), Some("svc:443"));
9670 assert_eq!(url_host("http://svc/a").as_deref(), Some("svc:80"));
9671 assert_eq!(url_host("/relative"), None);
9672 }
9673
9674 #[test]
9675 fn a_retry_ending_in_a_client_or_server_error_keeps_its_tokens_spent() {
9676 let budget = Arc::new(RetryBudget::new(&RetryBudgetConfig::default()));
9677 let gate = RetryGate {
9678 deadline: None,
9679 budgets: None,
9680 budget: Some(Arc::clone(&budget)),
9681 host: None,
9682 send_header: true,
9683 refill_pending: AtomicBool::new(false),
9684 pending_retry: std::sync::Mutex::new(None),
9685 caller_deadline: std::sync::OnceLock::new(),
9686 signer: None,
9687 };
9688 assert!(gate.allow(RetryKind::Transient, Duration::ZERO));
9689 gate.check().unwrap();
9691 gate.finish(Some(RetryKind::Transient), 500);
9692 assert!((budget.available() - 486.0).abs() < f64::EPSILON);
9693 gate.finish(Some(RetryKind::Transient), 200);
9694 assert!((budget.available() - 500.0).abs() < f64::EPSILON);
9695 }
9696 }
9697}