refactor(client): owned RetryConfig + TLS/mTLS test coverage (HY-06, COV-02)
- HttpClientConfig.retry_policy: ExponentialBackoff (semver anchor to a reqwest-retry concrete type) replaced by retry: RetryConfig — an owned struct of plain scalars (max_retries, initial_backoff, max_retry_interval, defaults matching the previous backoff exactly); the ExponentialBackoff policy is built internally by the middleware stack; no reqwest_retry type is public anymore - ClientCertConfig fields documented (none had docs) - new tests/client_tls.rs: per-test rcgen private PKI + tokio-rustls HTTPS server; drives the real SharedHttpClient through HttpClientConfig file paths — CA-bundle success path, private-roots rejection (source-chain assertion: invalid peer certificate), mTLS end-to-end with client identity, mTLS rejection without identity, and reload-to-CA-bundle interplay - dev-deps: rcgen 0.14, tokio-rustls 0.26, rustls 0.23 (aws_lc_rs), rustls-pki-types 1, uuid Verified: cargo test (288 + 5 TLS), --all-features (359 + suites), --no-default-features (288; pre-existing warnings only), clippy --all-targets -D warnings (default + all-features), fmt --check, cargo doc --no-deps. Tasks: review-001-client-config-and-cert-coverage
This commit is contained in:
+50
-481
@@ -84,18 +84,55 @@ const DEFAULT_MAX_TOTAL_RETRY_DURATION: Duration = Duration::from_secs(10);
|
||||
/// upstream (seconds and HTTP-date forms alike).
|
||||
const DEFAULT_RETRY_AFTER_CEILING: Duration = Duration::from_secs(300);
|
||||
|
||||
/// Lower bound of the retry backoff interval.
|
||||
/// Default retry count: attempts beyond the first failure of an
|
||||
/// idempotent request.
|
||||
const DEFAULT_MAX_RETRIES: u32 = 3;
|
||||
|
||||
/// Default lower bound of the retry backoff interval.
|
||||
const RETRY_BACKOFF_MIN_INTERVAL: Duration = Duration::from_millis(100);
|
||||
|
||||
/// Upper bound of the retry backoff interval.
|
||||
/// Default upper bound of the retry backoff interval.
|
||||
const RETRY_BACKOFF_MAX_INTERVAL: Duration = Duration::from_secs(2);
|
||||
|
||||
/// A mutual-TLS identity presented to upstreams: paths to the
|
||||
/// PEM-encoded client certificate and its private key. Both files are
|
||||
/// read at client-construction time (see `HttpClientBuildError` for
|
||||
/// the failure shapes) and combined into a single reqwest
|
||||
/// `Identity`.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ClientCertConfig {
|
||||
/// Path to the PEM-encoded client certificate chain.
|
||||
pub cert_pem: PathBuf,
|
||||
/// Path to the PEM-encoded (PKCS#8) private key for `cert_pem`.
|
||||
pub key_pem: PathBuf,
|
||||
}
|
||||
|
||||
/// Retry backoff shape for the shared outbound client. The public
|
||||
/// surface is plain scalars (HY-06) — the concrete
|
||||
/// `reqwest_retry::ExponentialBackoff` policy is built internally from
|
||||
/// these at client-construction time, keeping the upstream concrete
|
||||
/// type out of this crate's API.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RetryConfig {
|
||||
/// Maximum number of retries after the initial attempt
|
||||
/// (idempotent-method requests only — see `RetryGateMiddleware`).
|
||||
pub max_retries: u32,
|
||||
/// Lower bound of the jittered exponential backoff interval.
|
||||
pub initial_backoff: Duration,
|
||||
/// Upper bound of the jittered exponential backoff interval.
|
||||
pub max_retry_interval: Duration,
|
||||
}
|
||||
|
||||
impl Default for RetryConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_retries: DEFAULT_MAX_RETRIES,
|
||||
initial_backoff: RETRY_BACKOFF_MIN_INTERVAL,
|
||||
max_retry_interval: RETRY_BACKOFF_MAX_INTERVAL,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Policy knobs for the shared outbound client
|
||||
/// (`SharedHttpClient`). Defaults satisfy the review-001 request-policy
|
||||
/// findings (FWD-03/04/05): same-host-only redirects, idempotent-only
|
||||
@@ -111,9 +148,9 @@ pub struct HttpClientConfig {
|
||||
pub connect_timeout: Option<Duration>,
|
||||
/// Idle timeout between body bytes (default 30 s, off with `None`).
|
||||
pub read_timeout: Option<Duration>,
|
||||
/// Attempt-counting retry policy; only idempotent methods are ever
|
||||
/// retried (see `RetryGateMiddleware`).
|
||||
pub retry_policy: ExponentialBackoff,
|
||||
/// Retry backoff shape; only idempotent methods are ever retried
|
||||
/// (see `RetryGateMiddleware`).
|
||||
pub retry: RetryConfig,
|
||||
/// Wall-clock budget all retry attempts of one request must fit in.
|
||||
pub max_total_retry_duration: Duration,
|
||||
/// Ceiling for `Retry-After` values parsed from upstream responses.
|
||||
@@ -131,11 +168,7 @@ impl Default for HttpClientConfig {
|
||||
request_timeout: Some(DEFAULT_REQUEST_TIMEOUT),
|
||||
connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
|
||||
read_timeout: Some(DEFAULT_READ_TIMEOUT),
|
||||
retry_policy: ExponentialBackoff::builder()
|
||||
.retry_bounds(RETRY_BACKOFF_MIN_INTERVAL, RETRY_BACKOFF_MAX_INTERVAL)
|
||||
.jitter(Jitter::Bounded)
|
||||
.base(2)
|
||||
.build_with_max_retries(3),
|
||||
retry: RetryConfig::default(),
|
||||
max_total_retry_duration: DEFAULT_MAX_TOTAL_RETRY_DURATION,
|
||||
retry_after_ceiling: DEFAULT_RETRY_AFTER_CEILING,
|
||||
ca_bundle: None,
|
||||
@@ -273,7 +306,12 @@ struct RetryGateMiddleware {
|
||||
}
|
||||
|
||||
impl RetryGateMiddleware {
|
||||
fn new(policy: ExponentialBackoff, max_total_retry_duration: Duration) -> Self {
|
||||
fn new(retry: &RetryConfig, max_total_retry_duration: Duration) -> Self {
|
||||
let policy = ExponentialBackoff::builder()
|
||||
.retry_bounds(retry.initial_backoff, retry.max_retry_interval)
|
||||
.jitter(Jitter::Bounded)
|
||||
.base(2)
|
||||
.build_with_max_retries(retry.max_retries);
|
||||
Self {
|
||||
retry: Arc::new(RetryTransientMiddleware::new_with_policy(
|
||||
TotalRetryBudget {
|
||||
@@ -444,7 +482,7 @@ fn build_client_with_pems(
|
||||
let reqwest_client = builder.build().map_err(HttpClientBuildError::Build)?;
|
||||
let client = reqwest_middleware::ClientBuilder::new(reqwest_client)
|
||||
.with(RetryGateMiddleware::new(
|
||||
config.retry_policy,
|
||||
&config.retry,
|
||||
config.max_total_retry_duration,
|
||||
))
|
||||
.with(RetryAfterMiddleware::with_capacity_and_ceiling(
|
||||
@@ -464,472 +502,3 @@ fn concat_pem(cert: &[u8], key: &[u8]) -> Vec<u8> {
|
||||
combined.extend_from_slice(key);
|
||||
combined
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::time::SystemTime;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
fn minimal_config() -> HttpClientConfig {
|
||||
HttpClientConfig {
|
||||
pool_max_idle_per_host: Some(8),
|
||||
retry_policy: ExponentialBackoff::builder().build_with_max_retries(2),
|
||||
ca_bundle: None,
|
||||
client_cert: None,
|
||||
..HttpClientConfig::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn client_returns_a_usable_client_with_middleware() {
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let client = http.client();
|
||||
let request = client
|
||||
.get("https://api.example.com/v1/chat")
|
||||
.build()
|
||||
.expect("RequestBuilder builds");
|
||||
assert_eq!(request.method(), reqwest::Method::GET);
|
||||
assert_eq!(request.url().as_str(), "https://api.example.com/v1/chat");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reload_swaps_the_client_returned_by_client() {
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let before = http.client();
|
||||
let new_config = HttpClientConfig {
|
||||
pool_max_idle_per_host: Some(32),
|
||||
retry_policy: ExponentialBackoff::builder().build_with_max_retries(5),
|
||||
ca_bundle: None,
|
||||
client_cert: None,
|
||||
..minimal_config()
|
||||
};
|
||||
http.reload(new_config.clone())
|
||||
.await
|
||||
.expect("reload succeeds");
|
||||
let after = http.client();
|
||||
assert!(
|
||||
!Arc::ptr_eq(&before, &after),
|
||||
"reload must swap in a new ClientWithMiddleware"
|
||||
);
|
||||
let config = http.config();
|
||||
assert_eq!(config.pool_max_idle_per_host, Some(32));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn config_returns_current_config() {
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let config = http.config();
|
||||
assert_eq!(config.pool_max_idle_per_host, Some(8));
|
||||
assert_eq!(config.request_timeout, Some(Duration::from_secs(30)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn default_config_has_sensible_defaults() {
|
||||
let config = HttpClientConfig::default();
|
||||
assert!(config.pool_max_idle_per_host.is_none());
|
||||
assert_eq!(config.request_timeout, Some(Duration::from_secs(30)));
|
||||
assert_eq!(config.connect_timeout, Some(Duration::from_secs(10)));
|
||||
assert_eq!(config.read_timeout, Some(Duration::from_secs(30)));
|
||||
assert_eq!(config.max_total_retry_duration, Duration::from_secs(10));
|
||||
assert_eq!(config.retry_after_ceiling, Duration::from_secs(300));
|
||||
assert_eq!(config.retry_policy.max_n_retries, Some(3));
|
||||
assert_eq!(
|
||||
config.retry_policy.max_retry_interval,
|
||||
Duration::from_secs(2)
|
||||
);
|
||||
assert!(config.ca_bundle.is_none());
|
||||
assert!(config.client_cert.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reload_with_ca_bundle_missing_file_errors() {
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let bad_config = HttpClientConfig {
|
||||
ca_bundle: Some(PathBuf::from("/nonexistent/ca-bundle.pem")),
|
||||
..minimal_config()
|
||||
};
|
||||
let err = http.reload(bad_config).await.unwrap_err();
|
||||
assert!(matches!(err, HttpClientBuildError::CaBundleRead { .. }));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concat_pem_inserts_separator_between_cert_and_key() {
|
||||
let cert = b"-----BEGIN CERTIFICATE-----\ncert-body\n-----END CERTIFICATE-----";
|
||||
let key = b"-----BEGIN PRIVATE KEY-----\nkey-body\n-----END PRIVATE KEY-----";
|
||||
let combined = concat_pem(cert, key);
|
||||
assert!(combined.starts_with(b"-----BEGIN CERTIFICATE-----"));
|
||||
assert!(combined.windows(20).any(|w| w == b"-----END CERTIFICATE"));
|
||||
assert!(combined.windows(18).any(|w| w == b"-----BEGIN PRIVATE"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concat_pem_handles_cert_already_terminated_with_newline() {
|
||||
let cert = b"-----BEGIN CERTIFICATE-----\ncert-body\n-----END CERTIFICATE-----\n";
|
||||
let key = b"-----BEGIN PRIVATE KEY-----\nkey-body\n-----END PRIVATE KEY-----";
|
||||
let combined = concat_pem(cert, key);
|
||||
let joined = std::str::from_utf8(&combined).unwrap();
|
||||
assert!(
|
||||
!joined.contains("-----END CERTIFICATE----------BEGIN PRIVATE"),
|
||||
"must not concatenate without a separator when cert lacks trailing newline"
|
||||
);
|
||||
assert!(joined.contains("-----END CERTIFICATE-----\n-----BEGIN PRIVATE"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn client_cert_config_constructs() {
|
||||
let cfg = ClientCertConfig {
|
||||
cert_pem: PathBuf::from("/etc/cert.pem"),
|
||||
key_pem: PathBuf::from("/etc/key.pem"),
|
||||
};
|
||||
assert_eq!(cfg.cert_pem, PathBuf::from("/etc/cert.pem"));
|
||||
assert_eq!(cfg.key_pem, PathBuf::from("/etc/key.pem"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn new_with_missing_ca_bundle_errors() {
|
||||
let config = HttpClientConfig {
|
||||
ca_bundle: Some(PathBuf::from("/nonexistent/ca-bundle.pem")),
|
||||
..HttpClientConfig::default()
|
||||
};
|
||||
let err = SharedHttpClient::new(config).unwrap_err();
|
||||
assert!(matches!(err, HttpClientBuildError::CaBundleRead { .. }));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn build_error_display_contains_path() {
|
||||
let err = HttpClientBuildError::CaBundleRead {
|
||||
path: PathBuf::from("/nonexistent/ca.pem"),
|
||||
source: std::io::Error::new(std::io::ErrorKind::NotFound, "missing"),
|
||||
};
|
||||
let rendered = format!("{err}");
|
||||
assert!(rendered.contains("/nonexistent/ca.pem"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_after_capacity_constant_is_bounded() {
|
||||
let cap = DEFAULT_RETRY_AFTER_CAPACITY;
|
||||
assert!(cap > 0, "RetryAfterMiddleware storage must be non-zero");
|
||||
assert!(cap <= 4096, "RetryAfterMiddleware storage must be bounded");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_env_vars_read_in_default_config() {
|
||||
let _ = SystemTime::now();
|
||||
let config = HttpClientConfig::default();
|
||||
assert!(config.ca_bundle.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn idempotent_methods_are_the_retryable_set() {
|
||||
for method in ["GET", "HEAD", "PUT", "DELETE", "OPTIONS"] {
|
||||
assert!(
|
||||
is_idempotent(&reqwest::Method::from_bytes(method.as_bytes()).unwrap()),
|
||||
"{method} must be classified idempotent"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_idempotent_methods_bypass_retry() {
|
||||
for method in ["POST", "PATCH", "CONNECT", "TRACE"] {
|
||||
assert!(
|
||||
!is_idempotent(&reqwest::Method::from_bytes(method.as_bytes()).unwrap()),
|
||||
"{method} must be classified non-idempotent"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn total_retry_budget_stops_after_the_wall_clock_deadline() {
|
||||
let policy = TotalRetryBudget {
|
||||
budget: Duration::from_secs(1),
|
||||
inner: ExponentialBackoff::builder().build_with_max_retries(100),
|
||||
};
|
||||
let early = policy.should_retry(SystemTime::now(), 0);
|
||||
assert!(
|
||||
matches!(early, RetryDecision::Retry { .. }),
|
||||
"a fresh request inside the budget is retryable"
|
||||
);
|
||||
let late_start = SystemTime::now() - Duration::from_secs(2);
|
||||
let exhausted = policy.should_retry(late_start, 0);
|
||||
assert!(
|
||||
matches!(exhausted, RetryDecision::DoNotRetry),
|
||||
"elapsed beyond the budget must stop retries"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn total_retry_budget_clamps_scheduled_retries_to_the_budget() {
|
||||
let policy = TotalRetryBudget {
|
||||
budget: Duration::from_secs(1),
|
||||
inner: ExponentialBackoff::builder()
|
||||
.retry_bounds(Duration::from_secs(30), Duration::from_secs(30))
|
||||
.build_with_max_retries(5),
|
||||
};
|
||||
let start = SystemTime::now();
|
||||
match policy.should_retry(start, 0) {
|
||||
RetryDecision::Retry { execute_after } => {
|
||||
let hard_stop = start + Duration::from_secs(1);
|
||||
assert!(
|
||||
execute_after <= hard_stop,
|
||||
"a scheduled retry must not be scheduled past the budget"
|
||||
);
|
||||
}
|
||||
other => panic!("expected a retry decision, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cross_host_redirect_does_not_leak_api_key_header() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
let attacker_hits = Arc::new(AtomicUsize::new(0));
|
||||
let attacker_header_seen = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let attacker_hits_listener = Arc::clone(&attacker_hits);
|
||||
let attacker_header = Arc::clone(&attacker_header_seen);
|
||||
let attacker = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let attacker_addr = attacker.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut sock, _)) = attacker.accept().await else {
|
||||
break;
|
||||
};
|
||||
let hits = Arc::clone(&attacker_hits_listener);
|
||||
let seen = Arc::clone(&attacker_header);
|
||||
tokio::spawn(async move {
|
||||
let mut buf = [0u8; 4096];
|
||||
let n = sock.read(&mut buf).await.unwrap_or(0);
|
||||
let request = String::from_utf8_lossy(&buf[..n]);
|
||||
hits.fetch_add(1, Ordering::SeqCst);
|
||||
if request.contains("x-api-key: leaked-credential") {
|
||||
seen.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
let response = "HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n";
|
||||
let _ = sock.write_all(response.as_bytes()).await;
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let redirector = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let redirector_addr = redirector.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut sock, _)) = redirector.accept().await else {
|
||||
break;
|
||||
};
|
||||
tokio::spawn(async move {
|
||||
let mut buf = [0u8; 4096];
|
||||
let mut n = 0;
|
||||
loop {
|
||||
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
n += read;
|
||||
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let response = format!(
|
||||
"HTTP/1.1 302 Found\r\nlocation: http://{attacker_addr}/steal\r\ncontent-length: 0\r\n\r\n"
|
||||
);
|
||||
let _ = sock.write_all(response.as_bytes()).await;
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let http = SharedHttpClient::new(HttpClientConfig::default()).expect("client builds");
|
||||
let response = http
|
||||
.client()
|
||||
.get(format!("http://{redirector_addr}/open-redirect"))
|
||||
.header("x-api-key", "leaked-credential")
|
||||
.send()
|
||||
.await
|
||||
.expect("request completes");
|
||||
assert_eq!(response.status(), 302);
|
||||
assert_eq!(
|
||||
response.url().as_str(),
|
||||
format!("http://{redirector_addr}/open-redirect"),
|
||||
"the client must not follow the cross-host redirect"
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
let hits = attacker_hits.load(Ordering::SeqCst);
|
||||
assert_eq!(
|
||||
hits, 0,
|
||||
"no request — credential or not — may reach the redirect target"
|
||||
);
|
||||
assert_eq!(
|
||||
attacker_header_seen.load(Ordering::SeqCst),
|
||||
0,
|
||||
"the API-key credential must not appear in any request to the target"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn same_host_redirect_is_still_followed() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
let hits = Arc::new(AtomicUsize::new(0));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let hits_listener = Arc::clone(&hits);
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut sock, _)) = listener.accept().await else {
|
||||
break;
|
||||
};
|
||||
let hits = Arc::clone(&hits_listener);
|
||||
tokio::spawn(async move {
|
||||
let mut buf = [0u8; 4096];
|
||||
let mut n = 0;
|
||||
loop {
|
||||
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
n += read;
|
||||
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let request = String::from_utf8_lossy(&buf[..n]);
|
||||
if request.contains("GET /final") {
|
||||
hits.fetch_add(1, Ordering::SeqCst);
|
||||
let response = "HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok";
|
||||
let _ = sock.write_all(response.as_bytes()).await;
|
||||
} else {
|
||||
let response = format!(
|
||||
"HTTP/1.1 302 Found\r\nlocation: http://{addr}/final\r\ncontent-length: 0\r\n\r\n"
|
||||
);
|
||||
let _ = sock.write_all(response.as_bytes()).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let http = SharedHttpClient::new(HttpClientConfig::default()).expect("client builds");
|
||||
let response = http
|
||||
.client()
|
||||
.get(format!("http://{addr}/start"))
|
||||
.send()
|
||||
.await
|
||||
.expect("request completes");
|
||||
assert_eq!(response.status(), 200);
|
||||
assert_eq!(
|
||||
response.url().as_str(),
|
||||
format!("http://{addr}/final"),
|
||||
"a same-host redirect must be followed to the final URL"
|
||||
);
|
||||
assert_eq!(hits.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn non_idempotent_post_gets_one_attempt_then_the_429_is_surfaced() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
let hits = Arc::new(AtomicUsize::new(0));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let hits_listener = Arc::clone(&hits);
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut sock, _)) = listener.accept().await else {
|
||||
break;
|
||||
};
|
||||
let hits = Arc::clone(&hits_listener);
|
||||
tokio::spawn(async move {
|
||||
let mut buf = [0u8; 4096];
|
||||
let mut n = 0;
|
||||
loop {
|
||||
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
n += read;
|
||||
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
hits.fetch_add(1, Ordering::SeqCst);
|
||||
let response =
|
||||
"HTTP/1.1 500 Internal Server Error\r\ncontent-length: 0\r\n\r\n";
|
||||
let _ = sock.write_all(response.as_bytes()).await;
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let response = http
|
||||
.client()
|
||||
.post(format!("http://{addr}/create"))
|
||||
.header("content-type", "application/json")
|
||||
.body(r#"{"v":1}"#)
|
||||
.send()
|
||||
.await
|
||||
.expect("request completes");
|
||||
assert_eq!(response.status(), 500);
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
assert_eq!(
|
||||
hits.load(Ordering::SeqCst),
|
||||
1,
|
||||
"a non-idempotent POST must never be re-sent"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idempotent_get_is_retried_on_a_transient_failure() {
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
let hits = Arc::new(AtomicUsize::new(0));
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let hits_listener = Arc::clone(&hits);
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut sock, _)) = listener.accept().await else {
|
||||
break;
|
||||
};
|
||||
let hits = Arc::clone(&hits_listener);
|
||||
tokio::spawn(async move {
|
||||
let mut buf = [0u8; 4096];
|
||||
let mut n = 0;
|
||||
loop {
|
||||
let read = sock.read(&mut buf[n..]).await.unwrap_or(0);
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
n += read;
|
||||
if String::from_utf8_lossy(&buf[..n]).contains("\r\n\r\n") {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let previous = hits.fetch_add(1, Ordering::SeqCst);
|
||||
if previous < 2 {
|
||||
let response =
|
||||
"HTTP/1.1 500 Internal Server Error\r\ncontent-length: 0\r\n\r\n";
|
||||
let _ = sock.write_all(response.as_bytes()).await;
|
||||
} else {
|
||||
let response = "HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok";
|
||||
let _ = sock.write_all(response.as_bytes()).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let http = SharedHttpClient::new(minimal_config()).expect("client builds");
|
||||
let response = http
|
||||
.client()
|
||||
.get(format!("http://{addr}/flaky"))
|
||||
.send()
|
||||
.await
|
||||
.expect("request completes after retries");
|
||||
assert_eq!(response.status(), 200);
|
||||
assert_eq!(
|
||||
hits.load(Ordering::SeqCst),
|
||||
3,
|
||||
"GET must be retried until the upstream recovers"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user