From 0837e942abff268e1ef238d4706cbf0452b7cb73 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 31 Aug 2026 01:00:29 +0000 Subject: [PATCH] test(client,adapters): retry policy + streaming terminal arms + credential/header-parameter arms (CLI-03, COV-10, COV-12) - new tests/retry_policy_wire.rs: POST+500 = 1 upstream hit (method gate), GET 500/503/200 = 3 hits (retry-to-success), budget exhaustion stops retries under a 50-attempt cap (wall-time + hit-count bounds) - forward_stream terminal arms on the wire: oversized SSE line -> one INTERNAL terminal envelope; mid-stream transport abort (staged via a notify gate so the abort is genuinely mid-stream) -> terminal envelope after the delivered frame; pending event EOF-flush; dead port through both forward() and forward_stream() - Once-path decode arms: malformed application/json 200 -> INTERNAL decode envelope; application/octet-stream 200 -> byte-array envelope - COV-12 credential arms: ApiKey and Basic malformed values fail loudly without echoing secret material; declared header-param invalid name/value rejections - new tests/client_config_reload.rs: config() reflects a reloaded config (FWD-12 atomicity half) --- src/adapters/forward.rs | 382 ++++++++++++++++++++++++++++++++++ tests/client_config_reload.rs | 33 +++ tests/retry_policy_wire.rs | 164 +++++++++++++++ 3 files changed, 579 insertions(+) create mode 100644 tests/client_config_reload.rs create mode 100644 tests/retry_policy_wire.rs diff --git a/src/adapters/forward.rs b/src/adapters/forward.rs index 5bc6c33..97d9cb8 100644 --- a/src/adapters/forward.rs +++ b/src/adapters/forward.rs @@ -2826,6 +2826,388 @@ mod tests { ); } + /// FWD-07 Once-path decode arm: a `200` with `application/json` and + /// a body that does not parse surfaces a single `INTERNAL` decode + /// error envelope — never a partial or fabricated success. + #[tokio::test] + async fn malformed_json_body_200_decodes_to_an_internal_error_envelope() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response(200, "application/json", b"{not json".to_vec()) + })) + .await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Err(err) => { + assert_eq!(err.code, "INTERNAL", "message was: {}", err.message); + assert!( + err.message.contains("failed to decode response body"), + "message was: {}", + err.message + ); + } + other => panic!("expected INTERNAL decode envelope, got {other:?}"), + } + } + + /// FWD-07 Once-path binary arm: a `200` with + /// `application/octet-stream` surfaces the body as a JSON byte + /// array (bounded by the response cap like every other read path). + #[tokio::test] + async fn binary_octet_stream_200_surfaces_as_a_byte_array_envelope() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response( + 200, + "application/octet-stream", + vec![0x00, 0xFF, 0x10, 0x42], + ) + })) + .await; + let envelope = call_forward(&base, noop_context()).await; + match envelope.result { + Ok(value) => assert_eq!(value, json!([0, 255, 16, 66])), + other => panic!("expected a byte-array envelope, got {other:?}"), + } + } + + /// COV-10 (a): an upstream SSE line longer than the 1 MiB line cap + /// with no newline terminates `forward_stream` with a single + /// `INTERNAL` terminal envelope, then the stream ends. + #[tokio::test] + async fn oversized_upstream_sse_line_terminates_with_one_internal_envelope() { + let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n"; + let base = spawn_sse_responder_with_writer(head, move |mut sock| async move { + use tokio::io::AsyncWriteExt; + let line = vec![b'a'; SSE_EVENT_BUFFER_CAP + 1]; + let _ = sock.write_all(b"data: ").await; + let _ = sock.write_all(&line).await; + let _ = sock.write_all(b"\n\n").await; + let _ = sock.flush().await; + }) + .await; + let stream = forward_stream( + &minimal_client(), + &base, + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({"type": "object"}), + &[], + json!({}), + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!( + envelopes.len(), + 1, + "the parse-overflow terminal envelope is the only output" + ); + match &envelopes[0].result { + Err(err) => { + assert_eq!(err.code, "INTERNAL", "message was: {}", err.message); + assert!( + err.message.contains("SSE parse error"), + "message was: {}", + err.message + ); + } + other => panic!("expected one INTERNAL terminal envelope, got {other:?}"), + } + } + + /// COV-10 (b): the responder emits one complete `data:` frame, lets + /// the client consume it, then kills the connection before the + /// declared body length is met — a genuine mid-stream transport + /// abort, not a graceful EOF. The delivered event passes, a single + /// `INTERNAL` terminal error envelope follows, the stream ends. + #[tokio::test] + async fn aborted_socket_mid_stream_emits_a_terminal_error_envelope() { + let head = + "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: 512\r\n\r\n"; + let gate = TestArc::new(tokio::sync::Notify::new()); + let notify = TestArc::clone(&gate); + let base = spawn_sse_responder_with_writer(head, move |mut sock| async move { + use tokio::io::AsyncWriteExt; + let _ = sock.write_all(b"data: {\"n\":1}\n\n").await; + let _ = sock.flush().await; + notify.notified().await; + drop(sock); + }) + .await; + let stream = forward_stream( + &minimal_client(), + &base, + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({"type": "object"}), + &[], + json!({}), + noop_context(), + ); + tokio::pin!(stream); + let first = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("the complete frame must deliver within the test budget") + .expect("stream must deliver the frame before the abort"); + assert_eq!(first.result.clone().unwrap(), json!({"n": 1})); + gate.notify_one(); + let terminal = tokio::time::timeout(Duration::from_secs(5), stream.next()) + .await + .expect("the terminal envelope must follow within the test budget") + .expect("the stream must end with the terminal envelope"); + match terminal.result { + Err(err) => { + assert_eq!(err.code, "INTERNAL", "message was: {}", err.message); + assert!( + err.message.contains("SSE stream error"), + "message was: {}", + err.message + ); + } + other => panic!("expected the terminal error envelope, got {other:?}"), + } + let ended = tokio::time::timeout(Duration::from_secs(1), stream.next()).await; + assert!( + matches!(ended, Ok(None) | Err(_)), + "the stream ends after the terminal envelope" + ); + } + + /// COV-10 (c): the responder ends the stream with a pending event + /// (data lines received, no trailing blank line) — the WHATWG + /// EOF-flush dispatches that pending event as a final success + /// envelope at EOF. + #[tokio::test] + async fn pending_event_flushes_at_eof_without_a_trailing_blank_line() { + let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n"; + let base = spawn_sse_responder_with_writer(head, move |mut sock| async move { + use tokio::io::AsyncWriteExt; + let _ = sock + .write_all(b"data: {\"n\":1}\n\ndata: {\"final\":true}\n") + .await; + let _ = sock.flush().await; + let _ = sock.shutdown().await; + }) + .await; + let stream = forward_stream( + &minimal_client(), + &base, + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({"type": "object"}), + &[], + json!({}), + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!(envelopes.len(), 2); + assert_eq!(envelopes[0].result.clone().unwrap(), json!({"n": 1})); + assert_eq!( + envelopes[1].result.clone().unwrap(), + json!({"final": true}), + "EOF-flush dispatches the pending event" + ); + } + + /// COV-10 (d), Once-path transport arm: a connect-refused upstream + /// surfaces a single `INTERNAL` error envelope through `forward`. + #[tokio::test] + async fn dead_port_forward_yields_an_internal_envelope() { + let envelope = forward( + &minimal_client(), + "http://127.0.0.1:9", + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({"type": "object"}), + &[], + json!({}), + noop_context(), + ) + .await; + match envelope.result { + Err(err) => { + assert_eq!(err.code, "INTERNAL", "message was: {}", err.message); + assert!( + err.message.contains("HTTP request failed"), + "message was: {}", + err.message + ); + } + other => panic!("expected INTERNAL transport envelope, got {other:?}"), + } + } + + /// COV-10 (d), streaming-path transport arm: the same dead upstream + /// through `forward_stream` surfaces a single terminal `INTERNAL` + /// error envelope (the streaming build/send-failure terminal arm). + #[tokio::test] + async fn dead_port_forward_stream_yields_a_terminal_internal_envelope() { + let stream = forward_stream( + &minimal_client(), + "http://127.0.0.1:9", + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &serde_json::json!({"type": "object"}), + &[], + json!({}), + noop_context(), + ); + let envelopes = collect_stream(stream).await; + assert_eq!( + envelopes.len(), + 1, + "exactly one terminal envelope, then the stream ends" + ); + match &envelopes[0].result { + Err(err) => { + assert_eq!(err.code, "INTERNAL", "message was: {}", err.message); + assert!( + err.message.contains("HTTP request failed"), + "message was: {}", + err.message + ); + } + other => panic!("expected terminal INTERNAL envelope, got {other:?}"), + } + } + + /// COV-12, loud-credential arm (ApiKey): an ApiKey credential value + /// carrying invalid HTTP header bytes fails loudly and never echoes + /// the credential material (FWD-08 family). + #[tokio::test] + async fn api_key_with_invalid_value_fails_loudly_without_echoing_secrets() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response(200, "application/json", b"{}".to_vec()) + })) + .await; + let ctx = ctx_with_capability("svc", "key\u{0003}-secret-marker".to_string()); + let envelope = call_forward_authed( + &base, + ctx, + &Some(HttpAuthScheme::ApiKey { + header_name: "x-api-key".to_string(), + }), + ) + .await; + match envelope.result { + Err(err) => { + assert!( + err.message + .contains("refusing to send the request unauthenticated"), + "message was: {}", + err.message + ); + assert!( + !err.message.contains("secret-marker") && !err.message.contains("key\u{0003}"), + "error must not echo credential material: {}", + err.message + ); + } + other => panic!("expected loud credential error, got {other:?}"), + } + } + + /// COV-12, loud-credential arm (Basic): a Basic credential pair + /// with invalid HTTP header bytes fails loudly like Bearer/ApiKey + /// (FWD-08 family). + #[tokio::test] + async fn basic_credential_with_invalid_value_fails_loudly_without_echoing_secrets() { + let base = spawn_responder(TestArc::new(|_parts| { + http_response(200, "application/json", b"{}".to_vec()) + })) + .await; + let ctx = ctx_with_capability("svc", "basic\u{0001}-secret-marker".to_string()); + let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Basic)).await; + match envelope.result { + Err(err) => { + assert!( + err.message + .contains("refusing to send the request unauthenticated"), + "message was: {}", + err.message + ); + assert!( + !err.message.contains("secret-marker") + && !err.message.contains("basic\u{0001}"), + "error must not echo credential material: {}", + err.message + ); + } + other => panic!("expected loud credential error, got {other:?}"), + } + } + + /// COV-12: a declared header parameter with an invalid *value* + /// (control bytes) is rejected loudly at build time rather than + /// silently dropped from the request (FWD-08 family). + #[test] + fn declared_header_param_with_invalid_value_is_rejected() { + let ctx = noop_context(); + let schema = json!({ + "type": "object", + "properties": {"X-Trace": {"type": "string", "wire": "header"}}, + }); + let err = build_request( + "https://api.example.com", + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &schema, + &json!({"X-Trace": "bad\u{0000}value"}), + &ctx, + ) + .expect_err("invalid header-param value must fail loudly"); + assert!( + err.message.contains("X-Trace") && err.message.contains("invalid value"), + "message was: {}", + err.message + ); + } + + /// COV-12: a declared header parameter whose name is not a valid + /// HTTP header name is rejected loudly at build time (FWD-08 + /// family, the header-param sibling of the default-header arm). + #[test] + fn declared_header_param_with_invalid_name_is_rejected() { + let ctx = noop_context(); + let schema = json!({ + "type": "object", + "properties": {"bad header": {"type": "string", "wire": "header"}}, + }); + let err = build_request( + "https://api.example.com", + "/x", + "GET", + &None, + &TestHashMap::new(), + "svc", + &schema, + &json!({"bad header": "v"}), + &ctx, + ) + .expect_err("invalid header-param name must fail loudly"); + assert!( + err.message.contains("valid HTTP header name"), + "message was: {}", + err.message + ); + } + /// CLI-02 (a): a same-host 302 is followed, and the credential /// header is forwarded to the followed hop — the FWD-03 policy only /// scrubs headers across *cross-host* hops, so a same-origin hop diff --git a/tests/client_config_reload.rs b/tests/client_config_reload.rs new file mode 100644 index 0000000..5db0f61 --- /dev/null +++ b/tests/client_config_reload.rs @@ -0,0 +1,33 @@ +//! COV-12/FWD-12: `SharedHttpClient::config()` reflects the config the +//! running clients were built from, and a `reload` swaps the config +//! together with both clients in one atomic store — after a reload, +//! `config()` shows the new generation and the old one is unreachable +//! for every late reader (FWD-12's atomicity assertion). + +use std::time::Duration; + +use alkhttp::client::{HttpClientConfig, SharedHttpClient}; + +#[test] +fn config_accessor_tracks_a_reloaded_config() { + let initial = HttpClientConfig { + retry_after_ceiling: Duration::from_secs(11), + ..HttpClientConfig::default() + }; + let http = SharedHttpClient::new(initial).expect("initial client builds"); + assert_eq!( + http.config().retry_after_ceiling, + Duration::from_secs(11), + "config() starts as the config the client was built from" + ); + let reloaded = HttpClientConfig { + retry_after_ceiling: Duration::from_secs(22), + ..HttpClientConfig::default() + }; + futures::executor::block_on(http.reload(reloaded)).expect("reload succeeds"); + assert_eq!( + http.config().retry_after_ceiling, + Duration::from_secs(22), + "config() reflects the reloaded generation, not the initial one (FWD-12)" + ); +} diff --git a/tests/retry_policy_wire.rs b/tests/retry_policy_wire.rs new file mode 100644 index 0000000..a2a5bb2 --- /dev/null +++ b/tests/retry_policy_wire.rs @@ -0,0 +1,164 @@ +//! Retry-policy wire coverage (review-002 CLI-03, FWD-04): counting +//! responders against the real `SharedHttpClient` middleware stack. +//! +//! Pinned here: +//! +//! - the method gate — a POST gets exactly one upstream hit even when +//! the upstream answers 500 (non-idempotent requests never re-send); +//! - the idempotent retry path — a GET that 500s twice then succeeds +//! results in exactly three upstream hits; +//! - the wall-clock budget — with a generous attempt cap but a small +//! `max_total_retry_duration`, an always-500 upstream stops the +//! retry loop on budget exhaustion, bounding both the wall time and +//! the hit count. + +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use alkhttp::client::{HttpClientConfig, RetryConfig, SharedHttpClient}; +use std::sync::Mutex; + +/// A raw-TCP HTTP/1.1 responder that counts accepted connections and +/// answers each request from a per-attempt script: every entry is one +/// response head; the last entry repeats when the script runs out. +struct ScriptedResponder { + addr: std::net::SocketAddr, + hits: Arc, +} + +impl ScriptedResponder { + async fn spawn(script: Vec<&'static str>) -> Self { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("responder binds"); + let addr = listener.local_addr().expect("responder address"); + let hits = Arc::new(AtomicU32::new(0)); + let script = Arc::new(Mutex::new(script)); + let hits_loop = Arc::clone(&hits); + let script_loop = Arc::clone(&script); + tokio::spawn(async move { + loop { + let Ok((mut sock, _)) = listener.accept().await else { + return; + }; + hits_loop.fetch_add(1, Ordering::SeqCst); + let head = { + let mut queue = script_loop.lock().unwrap_or_else(|e| e.into_inner()); + if queue.len() <= 1 { + queue + .first() + .copied() + .unwrap_or("HTTP/1.1 500 Internal Server Error") + } else { + queue.remove(0) + } + }; + let body = ""; + let response = format!( + "{head}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ); + let _ = tokio::io::AsyncWriteExt::write_all(&mut sock, response.as_bytes()).await; + tokio::io::AsyncWriteExt::shutdown(&mut sock).await.ok(); + } + }); + Self { addr, hits } + } + + fn hits(&self) -> u32 { + self.hits.load(Ordering::SeqCst) + } +} + +fn retrying_config(max_total_retry_duration: Duration) -> HttpClientConfig { + HttpClientConfig { + request_timeout: Some(Duration::from_secs(5)), + connect_timeout: Some(Duration::from_secs(2)), + read_timeout: Some(Duration::from_secs(2)), + retry: RetryConfig { + max_retries: 50, + initial_backoff: Duration::from_millis(10), + max_retry_interval: Duration::from_millis(50), + }, + max_total_retry_duration, + ..HttpClientConfig::default() + } +} + +#[tokio::test] +async fn post_500_receives_exactly_one_upstream_hit() { + let responder = ScriptedResponder::spawn(vec!["HTTP/1.1 500 Internal Server Error"]).await; + let http = + SharedHttpClient::new(retrying_config(Duration::from_secs(10))).expect("client builds"); + let response = http + .client() + .post(format!("http://{}/mutate", responder.addr)) + .send() + .await + .expect("500 is a delivered response, not a transport error"); + assert_eq!(response.status(), 500); + assert_eq!( + responder.hits(), + 1, + "the method gate must bypass the retry middleware for POST: a non-idempotent request is never re-sent" + ); +} + +#[tokio::test] +async fn get_500_twice_then_success_is_exactly_three_hits() { + let responder = ScriptedResponder::spawn(vec![ + "HTTP/1.1 500 Internal Server Error", + "HTTP/1.1 503 Service Unavailable", + "HTTP/1.1 200 OK", + ]) + .await; + let http = + SharedHttpClient::new(retrying_config(Duration::from_secs(10))).expect("client builds"); + let response = http + .client() + .get(format!("http://{}/flaky", responder.addr)) + .send() + .await + .expect("retry-to-success delivers the final response"); + assert_eq!(response.status(), 200); + assert_eq!( + responder.hits(), + 3, + "two failed attempts retried, third attempt succeeded" + ); +} + +#[tokio::test] +async fn budget_exhaustion_stops_retries_despite_a_generous_attempt_cap() { + let responder = ScriptedResponder::spawn(vec!["HTTP/1.1 500 Internal Server Error"]).await; + let http = + SharedHttpClient::new(retrying_config(Duration::from_millis(400))).expect("client builds"); + let started = std::time::Instant::now(); + let response = http + .client() + .get(format!("http://{}/stuck", responder.addr)) + .send() + .await + .expect("500 is a delivered response, not a transport error"); + assert_eq!(response.status(), 500); + let hits = responder.hits(); + assert_eq!( + response.status(), + 500, + "the surfaced status is the final upstream 500" + ); + assert!( + hits >= 2, + "at least one retry ran before the budget closed, hits: {hits}" + ); + assert!( + hits < 50, + "budget exhaustion must stop retries long before the 50-attempt cap, hits: {hits}" + ); + assert!( + started.elapsed() < Duration::from_secs(3), + "wall time is bounded by max_total_retry_duration + one attempt, took {:?} over {hits} hits", + started.elapsed() + ); +}