fix: do not re-record requests served from the replay snapshot

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<RecordedExchange>`) 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<RecordedExchange>`) for a filesystem-free replay/fixture backend.
 
 `run()` binds every listener, starts a shared recorder and command processor, and returns a `ClusterHandle` exposing:
 

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<ResponseSource>,
     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,

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() {

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);
+}

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 {

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();
 

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()