diff --git a/SPECIFICATION.md b/SPECIFICATION.md index 41262cd..98a1855 100644 --- a/SPECIFICATION.md +++ b/SPECIFICATION.md @@ -123,7 +123,7 @@ let cluster = ProxyClusterBuilder::new() .await?; ``` -The storage backend handed to an upstream drives both directions of the record/replay round-trip: at `run()` its existing contents (via `SnapshotStorage::load`) are loaded and indexed into an internal replay source, and in `Mode::Record` every served exchange for that upstream is appended back to the same backend. There is no cluster-wide storage setter — give each upstream its own backend to keep recordings separate. Use `InMemoryStorage` (seeded from a `Vec`) for a filesystem-free replay/fixture backend. +The storage backend handed to an upstream drives both directions of the record/replay round-trip: at `run()` its existing contents (via `SnapshotStorage::load`) are loaded and indexed into an internal replay source, and in `Mode::Record` every *newly forwarded* exchange for that upstream is appended back to the same backend. Requests that the replay source already answers are served from it and are **not** appended again — the backend is a deduplicating cache, not an append-on-every-request log (see §8.3 and §20.1). There is no cluster-wide storage setter — give each upstream its own backend to keep recordings separate. Use `InMemoryStorage` (seeded from a `Vec`) for a filesystem-free replay/fixture backend. `run()` binds every listener, starts a shared recorder and command processor, and returns a `ClusterHandle` exposing: diff --git a/crates/partly-proxy-lib/src/listener.rs b/crates/partly-proxy-lib/src/listener.rs index 72d35a6..466815a 100644 --- a/crates/partly-proxy-lib/src/listener.rs +++ b/crates/partly-proxy-lib/src/listener.rs @@ -316,6 +316,7 @@ async fn handle_request( &runtime_for_block, &original_request, &resp, + ctx.response_source(), started.elapsed(), ) .await; @@ -419,11 +420,18 @@ async fn record_success_exchange( runtime: &UpstreamRuntime, original_request: &ProxyRequest, final_response: &ProxyResponse, + source: Option, duration: std::time::Duration, ) { if !runtime.recorder.is_enabled() { return; } + // Deduplicating cache (SPECIFICATION.md §8.3/§20.1): a response served from + // the replay snapshot is already on record — recording it again would + // multiply entries for an already-seen request. + if source == Some(ResponseSource::Snapshot) { + return; + } let (recorded_req, recorded_resp) = build_recorded(runtime, original_request, final_response); persist_exchange( runtime, diff --git a/crates/partly-proxy-lib/tests/assertions.rs b/crates/partly-proxy-lib/tests/assertions.rs index 0bf3c75..7d0ccd3 100644 --- a/crates/partly-proxy-lib/tests/assertions.rs +++ b/crates/partly-proxy-lib/tests/assertions.rs @@ -1,42 +1,14 @@ //! Wait-for semantics of `AssertSeen` and `AssertCount` (see //! `SPECIFICATION.md` §14.1). -use std::{ - net::SocketAddr, - time::{Duration, Instant}, -}; +use std::time::{Duration, Instant}; -use partly_proxy_echo as echo; use partly_proxy_lib::{ - Command, CommandResponse, ProxyClusterBuilder, ProxyConfig, RecordingConfig, TrafficFilter, - UpstreamTarget, + Command, CommandResponse, ProxyClusterBuilder, RecordingConfig, TrafficFilter, }; -use tokio::task::JoinHandle; - -async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { - let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); - let task = tokio::spawn(async move { - let _ = echo::serve(listener).await; - }); - (addr, task) -} -fn http_client() -> reqwest::Client { - reqwest::Client::builder() - .no_proxy() - .timeout(Duration::from_secs(5)) - .build() - .unwrap() -} - -fn cfg(url: String) -> ProxyConfig { - ProxyConfig::http( - "127.0.0.1:0".parse().unwrap(), - UpstreamTarget::new(url) - .with_connect_timeout(Duration::from_secs(1)) - .with_request_timeout(Duration::from_secs(5)), - ) -} +mod common; +use common::{cfg, http_client, spawn_echo}; #[tokio::test] async fn assert_seen_blocks_until_traffic_arrives() { diff --git a/crates/partly-proxy-lib/tests/common/mod.rs b/crates/partly-proxy-lib/tests/common/mod.rs new file mode 100644 index 0000000..326f7db --- /dev/null +++ b/crates/partly-proxy-lib/tests/common/mod.rs @@ -0,0 +1,108 @@ +//! Shared helpers for the integration test binaries. +//! +//! Lives under `tests/common/` (a subdirectory, not a top-level `tests/*.rs`) +//! so Cargo treats it as a module to include via `mod common;` rather than as +//! its own test binary. Each including binary only exercises a subset of these, +//! so unused-helper warnings are expected and silenced. +#![allow(dead_code)] + +use std::{net::SocketAddr, path::Path, time::Duration}; + +use bytes::Bytes; +use http::{HeaderMap, Method}; +use partly_proxy_echo as echo; +use partly_proxy_lib::{ + ExchangeOutcome, ProxyConfig, RecordedExchange, RecordedRequest, RecordedResponse, + SnapshotStorage, UpstreamTarget, jsonl::JsonlStorage, +}; +use tokio::{io::AsyncBufReadExt, task::JoinHandle}; + +/// Bind an in-process echo upstream on an ephemeral port and serve it on a +/// background task. Returns the bound address and the task handle. +pub async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { + let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); + let task = tokio::spawn(async move { + let _ = echo::serve(listener).await; + }); + (addr, task) +} + +/// A reqwest client that ignores any ambient proxy env vars and times out +/// after 5s — the standard client for driving the proxy from a test. +pub fn http_client() -> reqwest::Client { + reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(5)) + .build() + .expect("reqwest client builds") +} + +/// A `ProxyConfig` bound to an ephemeral port, forwarding to `url` with short +/// (1s connect / 5s request) test timeouts. +pub fn cfg(url: String) -> ProxyConfig { + ProxyConfig::http( + "127.0.0.1:0".parse().unwrap(), + UpstreamTarget::new(url) + .with_connect_timeout(Duration::from_secs(1)) + .with_request_timeout(Duration::from_secs(5)), + ) +} + +/// An address that nothing is listening on — bind an ephemeral port, capture +/// it, then drop the listener. Any forward to it fails, which lets a test +/// prove a response came from a stub/replay rather than the upstream. +pub fn unreachable_addr() -> SocketAddr { + let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let a = l.local_addr().unwrap(); + drop(l); + a +} + +/// Count non-blank NDJSON lines at `path`, streaming line-by-line so it stays +/// cheap on large snapshot files. A missing file counts as zero. +pub async fn ndjson_line_count(path: &Path) -> usize { + let Ok(file) = tokio::fs::File::open(path).await else { + return 0; + }; + let mut lines = tokio::io::BufReader::new(file).lines(); + let mut count = 0; + while let Some(line) = lines.next_line().await.expect("read NDJSON line") { + if !line.trim().is_empty() { + count += 1; + } + } + count +} + +/// Write a single recorded exchange into a fresh NDJSON file at `path`. +pub async fn seed_snapshot( + path: &Path, + method: Method, + uri: &str, + req_body: &[u8], + resp_body: &[u8], +) { + let req = RecordedRequest::from_parts( + &method, + &uri.parse().unwrap(), + &HeaderMap::new(), + Bytes::copy_from_slice(req_body), + ); + let resp = RecordedResponse { + status: 200, + headers: Vec::new(), + body: Bytes::copy_from_slice(resp_body), + }; + let storage = JsonlStorage::open(path).await.unwrap(); + storage + .append(&RecordedExchange::new( + Some("upstream".to_owned()), + req, + ExchangeOutcome::Response(resp), + Duration::from_millis(1), + )) + .await + .unwrap(); + storage.flush().await.unwrap(); + drop(storage); +} diff --git a/crates/partly-proxy-lib/tests/control_plane_tcp.rs b/crates/partly-proxy-lib/tests/control_plane_tcp.rs index 0fd3339..cf9756d 100644 --- a/crates/partly-proxy-lib/tests/control_plane_tcp.rs +++ b/crates/partly-proxy-lib/tests/control_plane_tcp.rs @@ -2,29 +2,14 @@ use std::{net::SocketAddr, time::Duration}; -use partly_proxy_echo as echo; use partly_proxy_lib::{ProxyClusterBuilder, ProxyConfig, RecordingConfig, UpstreamTarget}; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, net::TcpStream, - task::JoinHandle, }; -async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { - let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); - let task = tokio::spawn(async move { - let _ = echo::serve(listener).await; - }); - (addr, task) -} - -fn http_client() -> reqwest::Client { - reqwest::Client::builder() - .no_proxy() - .timeout(Duration::from_secs(5)) - .build() - .unwrap() -} +mod common; +use common::{http_client, spawn_echo}; /// Send one JSON line, return the next response line. async fn rt(addr: SocketAddr, line: &str) -> serde_json::Value { diff --git a/crates/partly-proxy-lib/tests/forward.rs b/crates/partly-proxy-lib/tests/forward.rs index 24fc675..da73163 100644 --- a/crates/partly-proxy-lib/tests/forward.rs +++ b/crates/partly-proxy-lib/tests/forward.rs @@ -6,44 +6,22 @@ //! reqwest client. We never reach the public internet — the upstream is //! always in the same tokio runtime. -use std::{net::SocketAddr, time::Duration}; +use std::time::Duration; -use partly_proxy_echo as echo; use partly_proxy_lib::{ClusterHandle, ProxyClusterBuilder, ProxyConfig, UpstreamTarget}; -use tokio::task::JoinHandle; - -/// Spawn the echo upstream on 127.0.0.1:0 and return (addr, task). -async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { - let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); - let task = tokio::spawn(async move { - let _ = echo::serve(listener).await; - }); - (addr, task) -} + +mod common; +use common::{cfg, http_client, spawn_echo, unreachable_addr}; /// Spawn the proxy in front of an upstream URL and return the cluster handle. async fn spawn_proxy(upstream_url: String) -> ClusterHandle { - let cfg = ProxyConfig::http( - "127.0.0.1:0".parse().unwrap(), - UpstreamTarget::new(upstream_url) - .with_connect_timeout(Duration::from_secs(1)) - .with_request_timeout(Duration::from_secs(5)), - ); ProxyClusterBuilder::new() - .add_upstream("upstream", cfg) + .add_upstream("upstream", cfg(upstream_url)) .run() .await .expect("cluster builds") } -fn http_client() -> reqwest::Client { - reqwest::Client::builder() - .no_proxy() - .timeout(Duration::from_secs(5)) - .build() - .expect("reqwest client builds") -} - #[tokio::test] async fn forwards_get_to_upstream_and_returns_body() { let (echo_addr, _echo_task) = spawn_echo().await; @@ -119,14 +97,7 @@ async fn upstream_status_is_proxied_verbatim() { #[tokio::test] async fn unreachable_upstream_yields_502() { - // Bind a listener, capture its addr, then drop the listener so the port - // is free for the test window. The upstream URL will refuse connections. - let unreachable = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let a = l.local_addr().unwrap(); - drop(l); - a - }; + let unreachable = unreachable_addr(); let cluster = spawn_proxy(format!("http://{unreachable}")).await; let proxy_addr = cluster.addr("upstream").unwrap(); diff --git a/crates/partly-proxy-lib/tests/middleware.rs b/crates/partly-proxy-lib/tests/middleware.rs index 1e3d489..bdfff3c 100644 --- a/crates/partly-proxy-lib/tests/middleware.rs +++ b/crates/partly-proxy-lib/tests/middleware.rs @@ -2,7 +2,6 @@ //! rewrites, short-circuits, recovery, and snapshot-boundary redaction. use std::{ - net::SocketAddr, sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -13,38 +12,13 @@ use std::{ use async_trait::async_trait; use bytes::Bytes; use http::StatusCode; -use partly_proxy_echo as echo; use partly_proxy_lib::{ - ClusterHandle, Next, ProxyClusterBuilder, ProxyConfig, ProxyMiddleware, ProxyRequest, - ProxyResponse, RecordingConfig, RequestContext, Result as ProxyResult, SharedMiddleware, - UpstreamTarget, + ClusterHandle, Next, ProxyClusterBuilder, ProxyMiddleware, ProxyRequest, ProxyResponse, + RecordingConfig, RequestContext, Result as ProxyResult, SharedMiddleware, }; -use tokio::task::JoinHandle; -async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { - let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); - let task = tokio::spawn(async move { - let _ = echo::serve(listener).await; - }); - (addr, task) -} - -fn http_client() -> reqwest::Client { - reqwest::Client::builder() - .no_proxy() - .timeout(Duration::from_secs(5)) - .build() - .expect("reqwest client builds") -} - -fn upstream_cfg(url: String) -> ProxyConfig { - ProxyConfig::http( - "127.0.0.1:0".parse().unwrap(), - UpstreamTarget::new(url) - .with_connect_timeout(Duration::from_secs(1)) - .with_request_timeout(Duration::from_secs(5)), - ) -} +mod common; +use common::{cfg, http_client, spawn_echo}; struct ShortCircuit200; @@ -71,7 +45,7 @@ async fn short_circuit_middleware_skips_forwarding() { .recording(RecordingConfig::in_memory(10)) .add_upstream_with_middleware( "api", - upstream_cfg(format!("http://{echo_addr}")), + cfg(format!("http://{echo_addr}")), vec![Arc::new(ShortCircuit200) as SharedMiddleware], ) .run() @@ -122,7 +96,7 @@ async fn response_body_rewrite_lands_on_the_wire() { let cluster = ProxyClusterBuilder::new() .add_upstream_with_middleware( "api", - upstream_cfg(format!("http://{echo_addr}")), + cfg(format!("http://{echo_addr}")), vec![Arc::new(PrefixBody { prefix: b"PREFIX:" }) as SharedMiddleware], ) .run() @@ -169,7 +143,7 @@ async fn request_body_rewrite_reaches_upstream() { let cluster = ProxyClusterBuilder::new() .add_upstream_with_middleware( "api", - upstream_cfg(format!("http://{echo_addr}")), + cfg(format!("http://{echo_addr}")), vec![Arc::new(RewriteRequestBody { new_body: b"REWRITTEN", }) as SharedMiddleware], @@ -225,7 +199,7 @@ async fn middleware_can_recover_from_upstream_failure() { let cluster = ProxyClusterBuilder::new() .add_upstream_with_middleware( "api", - upstream_cfg(format!("http://{unreachable}")), + cfg(format!("http://{unreachable}")), vec![Arc::new(Recover) as SharedMiddleware], ) .run() @@ -280,7 +254,7 @@ async fn snapshot_redaction_strips_secrets_from_recorder_only() { .recording(RecordingConfig::in_memory(10)) .add_upstream_with_middleware( "api", - upstream_cfg(format!("http://{echo_addr}")), + cfg(format!("http://{echo_addr}")), vec![Arc::new(StripAuth) as SharedMiddleware], ) .run() @@ -367,7 +341,7 @@ async fn global_middleware_runs_before_per_upstream() { .add_middleware(global) .add_upstream_with_middleware( "api", - upstream_cfg(format!("http://{echo_addr}")), + cfg(format!("http://{echo_addr}")), vec![Arc::new(local) as SharedMiddleware], ) .run() diff --git a/crates/partly-proxy-lib/tests/record.rs b/crates/partly-proxy-lib/tests/record.rs index ad4417e..e6791b7 100644 --- a/crates/partly-proxy-lib/tests/record.rs +++ b/crates/partly-proxy-lib/tests/record.rs @@ -1,45 +1,25 @@ //! End-to-end recording through a real listener + forwarder. -use std::{net::SocketAddr, time::Duration}; +use std::{sync::Arc, time::Duration}; -use partly_proxy_echo as echo; +use http::Method; use partly_proxy_lib::{ ClusterHandle, ExchangeOutcome, ProxyClusterBuilder, ProxyConfig, RecordedExchange, - RecordingConfig, UpstreamTarget, + RecordingConfig, SharedStorage, UpstreamTarget, jsonl::JsonlStorage, }; -use tokio::task::JoinHandle; -async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { - let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); - let task = tokio::spawn(async move { - let _ = echo::serve(listener).await; - }); - (addr, task) -} +mod common; +use common::{cfg, http_client, ndjson_line_count, seed_snapshot, spawn_echo, unreachable_addr}; async fn spawn_proxy(upstream_url: String, recording: RecordingConfig) -> ClusterHandle { - let cfg = ProxyConfig::http( - "127.0.0.1:0".parse().unwrap(), - UpstreamTarget::new(upstream_url) - .with_connect_timeout(Duration::from_secs(1)) - .with_request_timeout(Duration::from_secs(5)), - ); ProxyClusterBuilder::new() .recording(recording) - .add_upstream("upstream", cfg) + .add_upstream("upstream", cfg(upstream_url)) .run() .await .expect("cluster builds") } -fn http_client() -> reqwest::Client { - reqwest::Client::builder() - .no_proxy() - .timeout(Duration::from_secs(5)) - .build() - .expect("reqwest client builds") -} - #[tokio::test] async fn successful_exchange_is_recorded_in_memory() { let (echo_addr, _echo_task) = spawn_echo().await; @@ -92,12 +72,7 @@ async fn successful_exchange_is_recorded_in_memory() { #[tokio::test] async fn unreachable_upstream_records_error_outcome() { - let unreachable = { - let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let a = l.local_addr().unwrap(); - drop(l); - a - }; + let unreachable = unreachable_addr(); let cluster = spawn_proxy( format!("http://{unreachable}"), RecordingConfig::in_memory(10), @@ -282,6 +257,60 @@ async fn disabled_recording_keeps_buffer_empty() { cluster.shutdown().await.unwrap(); } +/// A request already present in the snapshot file is served from the snapshot +/// (proved here by pointing the upstream at an unreachable address: a forward +/// would 502), but the recorder appended it a second time, so the file grew +/// from one line to two. Per SPECIFICATION.md §8.3/§20.1 the snapshot is a +/// deduplicating cache, so it must stay at one line. +#[tokio::test] +async fn record_mode_does_not_re_record_request_already_in_snapshot() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("trace.ndjson"); + seed_snapshot(&path, Method::POST, "/existing", b"hello", b"FROM-SNAPSHOT").await; + assert_eq!( + ndjson_line_count(&path).await, + 1, + "seed should write one line" + ); + + let storage: SharedStorage = Arc::new(JsonlStorage::open(&path).await.unwrap()); + let cfg = ProxyConfig::http( + "127.0.0.1:0".parse().unwrap(), + UpstreamTarget::new(format!("http://{}", unreachable_addr())) + .with_connect_timeout(Duration::from_millis(500)) + .with_request_timeout(Duration::from_secs(2)), + ); + let cluster = ProxyClusterBuilder::new() + .recording(RecordingConfig::in_memory(100)) + .add_upstream_with("upstream", cfg, Vec::new(), Some(storage)) + .run() + .await + .unwrap(); + let proxy = cluster.addr("upstream").unwrap(); + + let resp = http_client() + .post(format!("http://{proxy}/existing")) + .body("hello") + .send() + .await + .unwrap(); + // 200 + the snapshot body proves the request was replayed, not forwarded + // (the unreachable upstream would have produced a 502). + assert_eq!(resp.status(), 200, "request in snapshot must be replayed"); + assert_eq!(resp.text().await.unwrap(), "FROM-SNAPSHOT"); + + // Give the recorder ample time to (wrongly) append a duplicate. + tokio::time::sleep(Duration::from_millis(300)).await; + cluster.shutdown().await.unwrap(); + + assert_eq!( + ndjson_line_count(&path).await, + 1, + "a request already present in the snapshot must NOT be re-recorded \ + (SPECIFICATION.md §8.3/§20.1: the snapshot is a deduplicating cache)" + ); +} + /// Poll the recorder until it reaches at least `target` exchanges or times /// out. The lifecycle records exchanges *after* the response has been sent, /// so the client-side `await` returning is not sufficient by itself. diff --git a/crates/partly-proxy-lib/tests/replay.rs b/crates/partly-proxy-lib/tests/replay.rs index 10af650..799bea9 100644 --- a/crates/partly-proxy-lib/tests/replay.rs +++ b/crates/partly-proxy-lib/tests/replay.rs @@ -1,7 +1,6 @@ //! Replay layered with middleware, stubs and the live forwarder. use std::{ - net::SocketAddr, sync::{Arc, Mutex}, time::Duration, }; @@ -9,39 +8,15 @@ use std::{ use async_trait::async_trait; use bytes::Bytes; use http::{HeaderMap, Method, StatusCode}; -use partly_proxy_echo as echo; use partly_proxy_lib::{ - Command, ExchangeOutcome, InMemoryStorage, Mode, Next, ProxyClusterBuilder, ProxyConfig, - ProxyMiddleware, ProxyRequest, ProxyResponse, RecordedExchange, RecordedRequest, - RecordedResponse, RecordingConfig, RequestContext, RequestMatcher, ResponseSource, - Result as ProxyResult, SharedMiddleware, SharedStorage, StubbedResponse, UpstreamTarget, + Command, ExchangeOutcome, InMemoryStorage, Mode, Next, ProxyClusterBuilder, ProxyMiddleware, + ProxyRequest, ProxyResponse, RecordedExchange, RecordedRequest, RecordedResponse, + RecordingConfig, RequestContext, RequestMatcher, ResponseSource, Result as ProxyResult, + SharedMiddleware, SharedStorage, StubbedResponse, }; -use tokio::task::JoinHandle; - -async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { - let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); - let task = tokio::spawn(async move { - let _ = echo::serve(listener).await; - }); - (addr, task) -} - -fn http_client() -> reqwest::Client { - reqwest::Client::builder() - .no_proxy() - .timeout(Duration::from_secs(5)) - .build() - .unwrap() -} -fn cfg(url: String) -> ProxyConfig { - ProxyConfig::http( - "127.0.0.1:0".parse().unwrap(), - UpstreamTarget::new(url) - .with_connect_timeout(Duration::from_secs(1)) - .with_request_timeout(Duration::from_secs(5)), - ) -} +mod common; +use common::{cfg, http_client, spawn_echo, unreachable_addr}; fn in_memory_store(exchanges: Vec) -> SharedStorage { Arc::new(InMemoryStorage::from(exchanges)) @@ -343,10 +318,13 @@ async fn replay_lookup_uses_redact_request_for_snapshot() { } #[tokio::test] -async fn replay_records_served_exchanges_when_recording_enabled() { - // §8.3 says "Every served exchange — whether the response came from a - // middleware short-circuit, a stub, or replay — is recorded under the - // upstream name". +async fn record_mode_does_not_re_record_a_replay_hit() { + // SPECIFICATION.md §8.3/§20.1: in `Mode::Record` the snapshot is a + // deduplicating cache — a request already present is replayed "rather than + // re-recording it". Serving a replay hit must therefore NOT append another + // copy to the recorder (which would multiply entries for an already-seen + // request). The upstream is unreachable, so a 200 + the snapshot body also + // proves the request was replayed, not forwarded. let unreachable = { let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let a = l.local_addr().unwrap(); @@ -373,33 +351,25 @@ async fn replay_records_served_exchanges_when_recording_enabled() { .unwrap(); let proxy = cluster.addr("api").unwrap(); - let _ = http_client() + let resp = http_client() .get(format!("http://{proxy}/x")) .send() .await - .unwrap() - .text() - .await .unwrap(); + assert_eq!( + resp.status(), + 200, + "replay hit, not a forward to the dead upstream" + ); + assert_eq!(resp.text().await.unwrap(), "replay-body"); - // Wait until the recorder has caught up. - let deadline = std::time::Instant::now() + Duration::from_secs(2); - loop { - if cluster.recorder().len().await >= 1 { - break; - } - if std::time::Instant::now() >= deadline { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - let exchanges = cluster.recorder().exchanges().await; - assert_eq!(exchanges.len(), 1); - let resp = exchanges[0] - .outcome - .as_response() - .expect("response outcome"); - assert_eq!(resp.body, Bytes::from_static(b"replay-body")); + // Give the recorder ample time to (wrongly) append the replayed exchange. + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + cluster.recorder().len().await, + 0, + "a replayed request is already on record and must not be recorded again" + ); cluster.shutdown().await.unwrap(); } @@ -437,13 +407,6 @@ impl ProxyMiddleware for ShortCircuit { } } -fn unreachable_addr() -> SocketAddr { - let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let a = l.local_addr().unwrap(); - drop(l); - a -} - #[tokio::test] async fn response_source_stub_marks_ctx() { let captured = Arc::new(Mutex::new(None)); diff --git a/crates/partly-proxy-lib/tests/stubs.rs b/crates/partly-proxy-lib/tests/stubs.rs index 03832d4..988f46a 100644 --- a/crates/partly-proxy-lib/tests/stubs.rs +++ b/crates/partly-proxy-lib/tests/stubs.rs @@ -1,41 +1,17 @@ //! Stubs, pause/resume, and the in-process command plane driving the //! live listener. -use std::{net::SocketAddr, time::Duration}; +use std::time::Duration; use bytes::Bytes; use http::StatusCode; -use partly_proxy_echo as echo; use partly_proxy_lib::{ - Command, CommandResponse, ProxyClusterBuilder, ProxyConfig, RecordingConfig, RequestMatcher, - StubbedResponse, TrafficFilter, UpstreamTarget, + Command, CommandResponse, ProxyClusterBuilder, RecordingConfig, RequestMatcher, + StubbedResponse, TrafficFilter, }; -use tokio::task::JoinHandle; -async fn spawn_echo() -> (SocketAddr, JoinHandle<()>) { - let (addr, listener) = echo::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); - let task = tokio::spawn(async move { - let _ = echo::serve(listener).await; - }); - (addr, task) -} - -fn http_client() -> reqwest::Client { - reqwest::Client::builder() - .no_proxy() - .timeout(Duration::from_secs(5)) - .build() - .unwrap() -} - -fn cfg(url: String) -> ProxyConfig { - ProxyConfig::http( - "127.0.0.1:0".parse().unwrap(), - UpstreamTarget::new(url) - .with_connect_timeout(Duration::from_secs(1)) - .with_request_timeout(Duration::from_secs(5)), - ) -} +mod common; +use common::{cfg, http_client, spawn_echo}; #[tokio::test] async fn stub_overrides_upstream_response() {