From bc0d76f3455481d389ad128019914057bbd247f0 Mon Sep 17 00:00:00 2001 From: Josiah Bull Date: Tue, 16 Jun 2026 16:05:43 +1200 Subject: [PATCH 1/2] refactor!: remove MatchStrategy, always match by method+uri+body-hash MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Custom match strategy was never needed — replay only ever uses MethodUriAndBodyHash. Remove the MatchStrategy enum and CustomMatcher type alias, drop the strategy parameter from ReplaySource::{new, from_jsonl,from_storage} and Snapshots::{from_storage,in_memory}, and collapse the lookup to the single hash-indexed path (no more per-call match on strategy, no linear-scan branch). Breaking change: bump 0.4.0 -> 0.5.0 (pre-1.0 minor = breaking boundary). Spec §2/§5/§8.1/§8.1.1 updated to describe the fixed match key instead of pluggable strategies. --- Cargo.lock | 10 +- Cargo.toml | 8 +- SPECIFICATION.md | 31 +-- crates/partly-proxy-lib/benches/common/mod.rs | 7 +- crates/partly-proxy-lib/examples/host.rs | 9 +- crates/partly-proxy-lib/src/lib.rs | 2 +- crates/partly-proxy-lib/src/replay.rs | 229 +++++------------- crates/partly-proxy-lib/tests/record.rs | 14 +- crates/partly-proxy-lib/tests/replay.rs | 93 ++++--- 9 files changed, 152 insertions(+), 251 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f7233d9..e8991a1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1207,7 +1207,7 @@ dependencies = [ [[package]] name = "partly-proxy-echo" -version = "0.4.0" +version = "0.5.0" dependencies = [ "base64", "bytes", @@ -1224,7 +1224,7 @@ dependencies = [ [[package]] name = "partly-proxy-lib" -version = "0.4.0" +version = "0.5.0" dependencies = [ "async-trait", "base64", @@ -1264,7 +1264,7 @@ dependencies = [ [[package]] name = "partly-proxy-storage-jsonl" -version = "0.4.0" +version = "0.5.0" dependencies = [ "async-stream", "async-trait", @@ -1280,7 +1280,7 @@ dependencies = [ [[package]] name = "partly-proxy-storage-sqlite" -version = "0.4.0" +version = "0.5.0" dependencies = [ "async-stream", "async-trait", @@ -1295,7 +1295,7 @@ dependencies = [ [[package]] name = "partly-proxy-types" -version = "0.4.0" +version = "0.5.0" dependencies = [ "async-trait", "base64", diff --git a/Cargo.toml b/Cargo.toml index 45d88b5..1661ee9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/*"] [workspace.package] -version = "0.4.0" +version = "0.5.0" edition = "2024" license = "MIT OR Apache-2.0" repository = "https://github.com/thepartly/partly-proxy" @@ -17,9 +17,9 @@ rust-version = "1.85" # - `cargo publish` / `cargo release` use `version` from the registry. # Keep these versions in lock-step with `workspace.package.version` above; # `release.toml` (shared-version = true) enforces that on release. -partly-proxy-types = { version = "0.4.0", path = "crates/partly-proxy-types" } -partly-proxy-storage-jsonl = { version = "0.4.0", path = "crates/partly-proxy-storage-jsonl" } -partly-proxy-storage-sqlite = { version = "0.4.0", path = "crates/partly-proxy-storage-sqlite" } +partly-proxy-types = { version = "0.5.0", path = "crates/partly-proxy-types" } +partly-proxy-storage-jsonl = { version = "0.5.0", path = "crates/partly-proxy-storage-jsonl" } +partly-proxy-storage-sqlite = { version = "0.5.0", path = "crates/partly-proxy-storage-sqlite" } # `partly-proxy-echo` is `publish = false`; only consumed as a dev-dep # inside the workspace. Path-only is fine since dev-deps without a # version are stripped from the published manifest. diff --git a/SPECIFICATION.md b/SPECIFICATION.md index 83daecf..600de70 100644 --- a/SPECIFICATION.md +++ b/SPECIFICATION.md @@ -25,7 +25,7 @@ The design centres on a single request lifecycle through which all behaviours | Capability | |------------| | Record live upstream traffic — in-memory ring buffer, optional NDJSON disk persistence | -| Replay from recorded snapshots — indexed by method+path+body-hash; custom matcher supported | +| Replay from recorded snapshots — indexed by method+path+body-hash | | Body-aware replay matching — SHA-256 body hash | | Inject runtime stubs — with optional fire-count limit and artificial delay | | Pause/resume traffic — globally or per upstream | @@ -156,7 +156,7 @@ Each incoming request flows through the following ordered stages. Earlier stages 4. **Body collection** — request body is buffered into bytes for hash-based matching. 5. **Middleware chain** — global middleware then per-upstream middleware, composed via `Next`. Each middleware decides whether to call `next.run(req, ctx).await`. The innermost call falls through to the terminal stages below. 6. **Terminal: Stub scan** — first matching active stub wins. Honours its optional artificial `delay`. Decrements its `times` counter; removes the stub when exhausted. -7. **Terminal: Replay lookup** — if a replay source is configured, the proxy makes a working copy of the request, runs `redact_request_for_snapshot` across the middleware chain on that copy, then looks the copy up by the chosen match strategy. A hit returns the recorded response. The original request is unchanged. +7. **Terminal: Replay lookup** — if a replay source is configured, the proxy makes a working copy of the request, runs `redact_request_for_snapshot` across the middleware chain on that copy, then looks the copy up by its (method, path+query, body hash) key. A hit returns the recorded response. The original request is unchanged. 8. **Terminal: Miss handling** — if no stub and no replay hit, the next step is governed by the upstream's [`Mode`](#83-mode-interactions): - **`Mode::Record`** — forward to the upstream. A failure here surfaces as `Err(ProxyError::Upstream*)` back through the middleware chain, where any middleware can catch and recover. If no middleware catches, the proxy returns `502 Bad Gateway`. - **`Mode::Replay`** — never touch the upstream. The proxy returns `503 Service Unavailable` with body `{}` and `Content-Type: application/json`. @@ -388,32 +388,35 @@ A stub matches a request when **all** of the following hold (any unset field is ## 8. Replay -A `ReplaySource` is an immutable snapshot of recorded exchanges with a chosen match strategy: +A `ReplaySource` is an immutable snapshot of recorded exchanges, indexed for O(1) lookup: ```rust -let replay = ReplaySource::new(exchanges, MatchStrategy::MethodUriAndBodyHash); +let replay = ReplaySource::new(exchanges); // or -let replay = ReplaySource::from_jsonl(path, MatchStrategy::MethodUriAndBodyHash)?; +let replay = ReplaySource::from_jsonl(path)?; ``` -Upstreams do not take a `ReplaySource` directly. Instead they take a `Snapshots` medium (§3.3, §4): `Snapshots::from_storage(store, strategy)` loads the source from a durable backend at `run()` *and* registers that backend as the upstream's recording sink, while `Snapshots::in_memory(exchanges, strategy)` wraps an in-memory list for replay-only use. The loaded source is consulted on the hot path exactly as described below. +Upstreams do not take a `ReplaySource` directly. Instead they take a `Snapshots` medium (§3.3, §4): `Snapshots::from_storage(store)` loads the source from a durable backend at `run()` *and* registers that backend as the upstream's recording sink, while `Snapshots::in_memory(exchanges)` wraps an in-memory list for replay-only use. The loaded source is consulted on the hot path exactly as described below. -### 8.1 Match strategies +### 8.1 Match key -| Strategy | Key | Notes | -|----------|-----|-------| -| `MethodUriAndBodyHash` (default) | (method, origin-form URI [path + query string], SHA-256 hex of body) | Distinguishes identical endpoints called with different query parameters or payloads. Query-string bytes are compared verbatim; reordered parameters are treated as a different request. | -| `Custom(closure)` | arbitrary | `Fn(&RecordedRequest, &Request) -> bool` — falls back to linear scan | +Matching is fixed: every request is keyed on **(method, origin-form URI [path + query string], SHA-256 hex of body)**. There is no pluggable match strategy — this single key is the only matching scheme. -`MethodUriAndBodyHash` builds an index at construction time for O(1) lookup. `Custom` is the only other supported strategy; coarser keys (method-only, method+path) and normalised variants (query-parameter canonicalisation, method+URI ignoring query) are intentionally not provided — callers who want those semantics express them as a `Custom` closure. +| Property | Value | +|----------|-------| +| Key | (method, origin-form URI [path + query string], SHA-256 hex of body) | +| Lookup | O(1) — an index is built once at construction | +| Query string | Compared verbatim; reordered parameters are treated as a different request | +| Body | Compared by SHA-256, so identical endpoints called with different payloads stay distinct | + +Coarser keys (method-only, method+path), normalised variants (query-parameter canonicalisation, method+URI ignoring query), and arbitrary user predicates are intentionally not provided — the matching key is deliberately the same on the record and replay sides so hashes always agree. ### 8.1.1 Scale target Replay must remain usable with snapshot files containing **10,000 to 100,000 exchanges** — these are realistic sizes for a recorded end-to-end suite, not a worst case to be discouraged. Concretely: - `ReplaySource::from_jsonl(...)` parses a 100k-line file in a single pass; it does not hold the whole file in a `String` and must stream line-by-line (e.g. `BufReader::lines`) to keep peak memory bounded by the largest single exchange, not the file size. -- Index construction for `MethodUriAndBodyHash` is O(n) in the number of exchanges; lookup remains O(1) per request regardless of snapshot size. Hash-map capacity should be preallocated from the exchange count to avoid repeated rehashing during load. -- `Custom` matchers fall back to a linear scan, which is O(n) per request. With a 100k-exchange snapshot this is the slow path; use it sparingly or pre-filter via the upstream/path before invoking the custom predicate. +- Index construction is O(n) in the number of exchanges; lookup remains O(1) per request regardless of snapshot size. Hash-map capacity should be preallocated from the exchange count to avoid repeated rehashing during load. - Memory budget at 100k exchanges with typical JSON payloads (~1–4 KiB body each) is on the order of hundreds of MiB. The proxy keeps decoded `Bytes` bodies in the source verbatim — there is no per-exchange duplication into the recorder unless `Replay + recording` is enabled. ### 8.2 Reusability diff --git a/crates/partly-proxy-lib/benches/common/mod.rs b/crates/partly-proxy-lib/benches/common/mod.rs index fe9bb7f..a43e891 100644 --- a/crates/partly-proxy-lib/benches/common/mod.rs +++ b/crates/partly-proxy-lib/benches/common/mod.rs @@ -20,8 +20,8 @@ use hyper_util::{ }; use partly_proxy_echo as echo; use partly_proxy_lib::{ - ClusterHandle, MatchStrategy, ProxyClusterBuilder, ProxyConfig, RecordingConfig, SharedStorage, - Snapshots, UpstreamTarget, + ClusterHandle, ProxyClusterBuilder, ProxyConfig, RecordingConfig, SharedStorage, Snapshots, + UpstreamTarget, }; use tempfile::TempDir; use tokio::task::JoinHandle; @@ -135,8 +135,7 @@ pub async fn spawn_proxy(recording: Recording) -> ProxyHandle { max_in_memory: 10_000, }; - let snapshots = storage - .map(|storage| Snapshots::from_storage(storage, MatchStrategy::MethodUriAndBodyHash)); + let snapshots = storage.map(Snapshots::from_storage); let builder = ProxyClusterBuilder::new().recording(cfg).add_upstream_with( "upstream", ProxyConfig::http( diff --git a/crates/partly-proxy-lib/examples/host.rs b/crates/partly-proxy-lib/examples/host.rs index b6b8497..db37ea4 100644 --- a/crates/partly-proxy-lib/examples/host.rs +++ b/crates/partly-proxy-lib/examples/host.rs @@ -22,8 +22,8 @@ use std::{net::SocketAddr, sync::Arc}; use partly_proxy_lib::{ - MatchStrategy, ProxyClusterBuilder, ProxyConfig, RecordingConfig, Result, SharedStorage, - Snapshots, UpstreamTarget, + ProxyClusterBuilder, ProxyConfig, RecordingConfig, Result, SharedStorage, Snapshots, + UpstreamTarget, }; #[tokio::main] @@ -55,10 +55,7 @@ async fn main() -> Result<()> { Some(path) => { let storage: SharedStorage = Arc::new(partly_proxy_lib::jsonl::JsonlStorage::open(path).await?); - Some(Snapshots::from_storage( - storage, - MatchStrategy::MethodUriAndBodyHash, - )) + Some(Snapshots::from_storage(storage)) } None => None, }; diff --git a/crates/partly-proxy-lib/src/lib.rs b/crates/partly-proxy-lib/src/lib.rs index fd8970d..b6cda51 100644 --- a/crates/partly-proxy-lib/src/lib.rs +++ b/crates/partly-proxy-lib/src/lib.rs @@ -46,6 +46,6 @@ pub use partly_proxy_types::{ }; pub use proxy_io::{ProxyRequest, ProxyResponse}; pub use recorder::Recorder; -pub use replay::{MatchStrategy, ReplaySource, Snapshots}; +pub use replay::{ReplaySource, Snapshots}; pub use stub::{RequestMatcher, StubEntry, StubStore, StubbedResponse}; pub use wire::{StubFields, WireCommand, WireFilter, WireResponse}; diff --git a/crates/partly-proxy-lib/src/replay.rs b/crates/partly-proxy-lib/src/replay.rs index cda1a35..7743362 100644 --- a/crates/partly-proxy-lib/src/replay.rs +++ b/crates/partly-proxy-lib/src/replay.rs @@ -1,13 +1,8 @@ //! Replay source — see `SPECIFICATION.md` §8. //! -//! A `ReplaySource` is an immutable bundle of recorded exchanges plus a -//! match strategy. The two supported strategies (per §8.1) are: -//! -//! - `MethodUriAndBodyHash` — O(1) hash-indexed lookup, built once at -//! construction. Keys on method, origin-form URI (path + query string), -//! and the body SHA-256. The default. -//! - `Custom(closure)` — linear scan with a user predicate. Use sparingly -//! on large snapshots (§8.1.1). +//! A `ReplaySource` is an immutable bundle of recorded exchanges indexed +//! for O(1) lookup. The lookup key is `(method, origin-form URI (path + +//! query string), body SHA-256)`, built once at construction (§8.1). //! //! Lookups go through every middleware's `redact_request_for_snapshot` //! before the lookup key is computed (§8.2.1), so a request that carried @@ -28,8 +23,7 @@ use std::{collections::HashMap, sync::Arc}; #[cfg(any(test, feature = "storage-jsonl"))] use partly_proxy_types::ProxyError; use partly_proxy_types::{ - ExchangeOutcome, RecordedExchange, RecordedRequest, Result, SharedStorage, SnapshotStorage, - hash::sha256_hex, + ExchangeOutcome, RecordedExchange, Result, SharedStorage, SnapshotStorage, hash::sha256_hex, }; use crate::{ @@ -37,39 +31,6 @@ use crate::{ proxy_io::{ProxyRequest, ProxyResponse}, }; -/// Predicate type used by [`MatchStrategy::Custom`]. Defined as a type alias -/// so the trait-object type doesn't trip clippy's `type_complexity` lint. -pub type CustomMatcher = Arc bool + Send + Sync>; - -/// Match strategy for a [`ReplaySource`]. -#[derive(Clone, Default)] -pub enum MatchStrategy { - /// `(method, uri.path_and_query(), sha256_hex(body))`. Hash-indexed; - /// O(1) lookup. - /// - /// Includes the query string so APIs that pass their data in query - /// parameters with an empty body (e.g. `GET /vehicle?plate=ABC123`) - /// match per-request rather than collapsing to the first snapshot at - /// that path. Query-string bytes are compared verbatim — parameter - /// reordering is treated as a different request; callers needing - /// canonicalisation should use [`MatchStrategy::Custom`]. - #[default] - MethodUriAndBodyHash, - /// User-supplied predicate; falls back to a linear scan over every - /// exchange. The closure is invoked with the on-disk - /// [`RecordedRequest`] and the live (already-redacted) [`ProxyRequest`]. - Custom(CustomMatcher), -} - -impl std::fmt::Debug for MatchStrategy { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::MethodUriAndBodyHash => f.write_str("MethodUriAndBodyHash"), - Self::Custom(_) => f.write_str("Custom()"), - } - } -} - /// Per-upstream snapshot medium handed to /// [`add_upstream_with`](crate::ProxyClusterBuilder::add_upstream_with). /// @@ -80,7 +41,6 @@ impl std::fmt::Debug for MatchStrategy { /// appended back to the same medium. There is no separate cluster-wide /// storage knob — recording is configured per upstream, here. pub struct Snapshots { - strategy: MatchStrategy, source: SnapshotsSource, } @@ -95,18 +55,16 @@ enum SnapshotsSource { impl Snapshots { /// Use a durable [`SharedStorage`] medium (e.g. a JSONL file) as both /// the replay source and the recording sink for this upstream. - pub fn from_storage(storage: SharedStorage, strategy: MatchStrategy) -> Self { + pub fn from_storage(storage: SharedStorage) -> Self { Self { - strategy, source: SnapshotsSource::Storage(storage), } } /// Use an in-memory list of exchanges as a replay-only source. Nothing /// recorded at runtime is written back — the medium is read-only. - pub fn in_memory(exchanges: Vec, strategy: MatchStrategy) -> Self { + pub fn in_memory(exchanges: Vec) -> Self { Self { - strategy, source: SnapshotsSource::InMemory(exchanges), } } @@ -117,12 +75,10 @@ impl Snapshots { pub(crate) async fn resolve(self) -> Result<(ReplaySource, Option)> { match self.source { SnapshotsSource::Storage(storage) => { - let replay = ReplaySource::from_storage(storage.as_ref(), self.strategy).await?; + let replay = ReplaySource::from_storage(storage.as_ref()).await?; Ok((replay, Some(storage))) } - SnapshotsSource::InMemory(exchanges) => { - Ok((ReplaySource::new(exchanges, self.strategy), None)) - } + SnapshotsSource::InMemory(exchanges) => Ok((ReplaySource::new(exchanges), None)), } } } @@ -133,14 +89,11 @@ impl std::fmt::Debug for Snapshots { SnapshotsSource::Storage(_) => "Storage", SnapshotsSource::InMemory(_) => "InMemory", }; - f.debug_struct("Snapshots") - .field("strategy", &self.strategy) - .field("source", &kind) - .finish() + f.debug_struct("Snapshots").field("source", &kind).finish() } } -/// Key used by `MethodUriAndBodyHash`: (method, path+query, body sha-256 hex). +/// Lookup key: (method, path+query, body sha-256 hex). type IndexKey = (String, String, String); /// Cheap-to-clone replay source. Behind an `Arc`, so several listeners can @@ -151,18 +104,16 @@ pub struct ReplaySource { } struct ReplaySourceInner { - strategy: MatchStrategy, exchanges: Vec, - /// Populated only for `MethodUriAndBodyHash`. Maps the lookup key to an - /// index into `exchanges`. The *first* exchange written for a given key - /// wins on collision — that way replay is deterministic across reloads. + /// Maps the lookup key to an index into `exchanges`. The *first* + /// exchange written for a given key wins on collision — that way + /// replay is deterministic across reloads. index: HashMap, } impl std::fmt::Debug for ReplaySource { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ReplaySource") - .field("strategy", &self.inner.strategy) .field("exchanges", &self.inner.exchanges.len()) .field("index", &self.inner.index.len()) .finish_non_exhaustive() @@ -171,14 +122,10 @@ impl std::fmt::Debug for ReplaySource { impl ReplaySource { /// Build a replay source from an in-memory list of exchanges. - pub fn new(exchanges: Vec, strategy: MatchStrategy) -> Self { - let index = build_index(&exchanges, &strategy); + pub fn new(exchanges: Vec) -> Self { + let index = build_index(&exchanges); Self { - inner: Arc::new(ReplaySourceInner { - strategy, - exchanges, - index, - }), + inner: Arc::new(ReplaySourceInner { exchanges, index }), } } @@ -196,11 +143,11 @@ impl ReplaySource { /// feature is off, callers should use [`ReplaySource::from_storage`] /// with whichever backend they prefer. #[cfg(feature = "storage-jsonl")] - pub fn from_jsonl(path: impl AsRef, strategy: MatchStrategy) -> Result { + pub fn from_jsonl(path: impl AsRef) -> Result { let file = match std::fs::File::open(&path) { Ok(f) => f, Err(e) if e.kind() == std::io::ErrorKind::NotFound => { - return Ok(Self::new(Vec::new(), strategy)); + return Ok(Self::new(Vec::new())); } Err(e) => return Err(ProxyError::Recording(e)), }; @@ -217,7 +164,7 @@ impl ReplaySource { let exchange = partly_proxy_storage_jsonl::parse_ndjson_line(&line, lineno)?; exchanges.push(exchange); } - Ok(Self::new(exchanges, strategy)) + Ok(Self::new(exchanges)) } /// Drain a `SnapshotStorage`'s `load()` stream into a replay source. @@ -226,18 +173,15 @@ impl ReplaySource { /// over any storage backend. Peak memory during construction is /// bounded by the largest single exchange — the stream is consumed /// one item at a time, then the assembled `Vec` feeds the existing - /// `build_index` for O(1) `MethodUriAndBodyHash` lookups. - pub async fn from_storage( - storage: &dyn SnapshotStorage, - strategy: MatchStrategy, - ) -> Result { + /// `build_index` for O(1) lookups. + pub async fn from_storage(storage: &dyn SnapshotStorage) -> Result { use futures::StreamExt; let mut stream = storage.load(); let mut exchanges = Vec::new(); while let Some(item) = stream.next().await { exchanges.push(item?); } - Ok(Self::new(exchanges, strategy)) + Ok(Self::new(exchanges)) } /// Number of exchanges in the source. @@ -250,9 +194,9 @@ impl ReplaySource { self.inner.exchanges.is_empty() } - /// Look up a response for `req`. Returns `None` on miss, on a hit with an - /// `Error` outcome (errors are intentionally not replayed — use stubs for - /// that), or when the match strategy refuses the request. + /// Look up a response for `req`. Returns `None` on miss or on a hit with + /// an `Error` outcome (errors are intentionally not replayed — use stubs + /// for that). /// /// `chain` is the effective middleware list — its /// `redact_request_for_snapshot` hooks fire on a working copy of `req` @@ -260,26 +204,16 @@ impl ReplaySource { pub fn lookup(&self, req: &ProxyRequest, chain: &[SharedMiddleware]) -> Option { let mut redacted = req.clone(); middleware::redact_request(chain, &mut redacted); - let matched = match &self.inner.strategy { - MatchStrategy::MethodUriAndBodyHash => { - let key = ( - redacted.method.as_str().to_owned(), - path_and_query_of_uri(&redacted.uri), - sha256_hex(&redacted.body), - ); - self.inner - .index - .get(&key) - .and_then(|&i| self.inner.exchanges.get(i)) - } - MatchStrategy::Custom(f) => self - .inner - .exchanges - .iter() - .find(|e| f(&e.request, &redacted)), - }; - - let exchange = matched?; + let key = ( + redacted.method.as_str().to_owned(), + path_and_query_of_uri(&redacted.uri), + sha256_hex(&redacted.body), + ); + let exchange = self + .inner + .index + .get(&key) + .and_then(|&i| self.inner.exchanges.get(i))?; match &exchange.outcome { ExchangeOutcome::Response(r) => Some(ProxyResponse { status: r.status(), @@ -292,13 +226,7 @@ impl ReplaySource { } } -fn build_index( - exchanges: &[RecordedExchange], - strategy: &MatchStrategy, -) -> HashMap { - if !matches!(strategy, MatchStrategy::MethodUriAndBodyHash) { - return HashMap::new(); - } +fn build_index(exchanges: &[RecordedExchange]) -> HashMap { let mut index = HashMap::with_capacity(exchanges.len()); for (i, e) in exchanges.iter().enumerate() { if !matches!(e.outcome, ExchangeOutcome::Response(_)) { @@ -393,14 +321,11 @@ mod tests { } #[test] - fn hash_strategy_finds_exact_match() { - let src = ReplaySource::new( - vec![ - make_exchange(Method::GET, "/health", b"", 200), - make_exchange(Method::POST, "/orders", b"{\"n\":1}", 201), - ], - MatchStrategy::MethodUriAndBodyHash, - ); + fn lookup_finds_exact_match() { + let src = ReplaySource::new(vec![ + make_exchange(Method::GET, "/health", b"", 200), + make_exchange(Method::POST, "/orders", b"{\"n\":1}", 201), + ]); let resp = src.lookup(&live(Method::GET, "/health", b""), &[]).unwrap(); assert_eq!(resp.status, StatusCode::OK); assert_eq!( @@ -411,14 +336,11 @@ mod tests { } #[test] - fn hash_strategy_distinguishes_by_body() { - let src = ReplaySource::new( - vec![ - make_exchange(Method::POST, "/orders", b"{\"n\":1}", 201), - make_exchange(Method::POST, "/orders", b"{\"n\":2}", 202), - ], - MatchStrategy::MethodUriAndBodyHash, - ); + fn lookup_distinguishes_by_body() { + let src = ReplaySource::new(vec![ + make_exchange(Method::POST, "/orders", b"{\"n\":1}", 201), + make_exchange(Method::POST, "/orders", b"{\"n\":2}", 202), + ]); let r1 = src .lookup(&live(Method::POST, "/orders", b"{\"n\":1}"), &[]) .unwrap(); @@ -430,11 +352,8 @@ mod tests { } #[test] - fn hash_strategy_misses_when_method_or_path_or_body_differs() { - let src = ReplaySource::new( - vec![make_exchange(Method::GET, "/health", b"", 200)], - MatchStrategy::MethodUriAndBodyHash, - ); + fn lookup_misses_when_method_or_path_or_body_differs() { + let src = ReplaySource::new(vec![make_exchange(Method::GET, "/health", b"", 200)]); assert!( src.lookup(&live(Method::POST, "/health", b""), &[]) .is_none() @@ -447,19 +366,15 @@ mod tests { } #[test] - fn hash_strategy_distinguishes_by_query_string() { + fn lookup_distinguishes_by_query_string() { // Regression: query-string-driven APIs (data in the query, empty // body) used to collapse to the first snapshot at a given path - // because the lookup key dropped the query. The default strategy - // now includes path+query, so each distinct query string is its - // own entry. - let src = ReplaySource::new( - vec![ - make_exchange(Method::GET, "/vehicle?plate=ABC123", b"", 200), - make_exchange(Method::GET, "/vehicle?plate=XYZ999", b"", 201), - ], - MatchStrategy::MethodUriAndBodyHash, - ); + // because the lookup key dropped the query. The key now includes + // path+query, so each distinct query string is its own entry. + let src = ReplaySource::new(vec![ + make_exchange(Method::GET, "/vehicle?plate=ABC123", b"", 200), + make_exchange(Method::GET, "/vehicle?plate=XYZ999", b"", 201), + ]); let abc = src .lookup(&live(Method::GET, "/vehicle?plate=ABC123", b""), &[]) .unwrap(); @@ -481,7 +396,7 @@ mod tests { } #[test] - fn hash_strategy_tolerates_absolute_form_recorded_uri() { + fn lookup_tolerates_absolute_form_recorded_uri() { // Live requests arrive in origin-form; recorder may store an // absolute-form URI (`http://host/path?query`). The index must // strip scheme+authority but keep the query. @@ -491,7 +406,7 @@ mod tests { b"", 200, ); - let src = ReplaySource::new(vec![recorded], MatchStrategy::MethodUriAndBodyHash); + let src = ReplaySource::new(vec![recorded]); let hit = src .lookup(&live(Method::GET, "/vehicle?plate=ABC123", b""), &[]) .unwrap(); @@ -514,28 +429,10 @@ mod tests { }, Duration::from_millis(1), ); - let src = ReplaySource::new(vec![ex], MatchStrategy::MethodUriAndBodyHash); + let src = ReplaySource::new(vec![ex]); assert!(src.lookup(&live(Method::GET, "/oops", b""), &[]).is_none()); } - #[test] - fn custom_strategy_runs_predicate() { - let src = ReplaySource::new( - vec![ - make_exchange(Method::GET, "/a", b"", 200), - make_exchange(Method::GET, "/b", b"", 201), - make_exchange(Method::GET, "/c", b"", 202), - ], - MatchStrategy::Custom(Arc::new(|recorded, live| { - recorded.uri.contains("/b") && live.method == Method::GET - })), - ); - let resp = src - .lookup(&live(Method::GET, "/anything", b""), &[]) - .unwrap(); - assert_eq!(resp.status, StatusCode::CREATED); - } - #[cfg(feature = "storage-jsonl")] #[tokio::test] async fn from_jsonl_round_trips() { @@ -579,7 +476,7 @@ mod tests { .unwrap(); } - let src = ReplaySource::from_jsonl(&path, MatchStrategy::MethodUriAndBodyHash).unwrap(); + let src = ReplaySource::from_jsonl(&path).unwrap(); assert_eq!(src.len(), 3); let resp = src.lookup(&live(Method::GET, "/n/1", b""), &[]).unwrap(); assert_eq!(resp.body, Bytes::from_static(b"body-1")); @@ -619,9 +516,7 @@ mod tests { let storage = MockStorage { exchanges: exchanges.clone(), }; - let src = ReplaySource::from_storage(&storage, MatchStrategy::MethodUriAndBodyHash) - .await - .unwrap(); + let src = ReplaySource::from_storage(&storage).await.unwrap(); assert_eq!(src.len(), 2); let resp = src .lookup(&live(Method::POST, "/b", b"{\"n\":1}"), &[]) @@ -649,9 +544,7 @@ mod tests { ))])) } } - let err = ReplaySource::from_storage(&BadStorage, MatchStrategy::MethodUriAndBodyHash) - .await - .unwrap_err(); + let err = ReplaySource::from_storage(&BadStorage).await.unwrap_err(); assert!(matches!(err, ProxyError::Recording(_))); } } diff --git a/crates/partly-proxy-lib/tests/record.rs b/crates/partly-proxy-lib/tests/record.rs index 28cb13a..7a5d677 100644 --- a/crates/partly-proxy-lib/tests/record.rs +++ b/crates/partly-proxy-lib/tests/record.rs @@ -4,8 +4,8 @@ use std::{net::SocketAddr, time::Duration}; use partly_proxy_echo as echo; use partly_proxy_lib::{ - ClusterHandle, ExchangeOutcome, MatchStrategy, ProxyClusterBuilder, ProxyConfig, - RecordedExchange, RecordingConfig, Snapshots, UpstreamTarget, + ClusterHandle, ExchangeOutcome, ProxyClusterBuilder, ProxyConfig, RecordedExchange, + RecordingConfig, Snapshots, UpstreamTarget, }; use tokio::task::JoinHandle; @@ -150,10 +150,7 @@ async fn ndjson_persist_file_is_replayable() { "upstream", cfg, Vec::new(), - Some(Snapshots::from_storage( - storage, - MatchStrategy::MethodUriAndBodyHash, - )), + Some(Snapshots::from_storage(storage)), ) .run() .await @@ -237,10 +234,7 @@ async fn custom_storage_via_per_upstream_snapshots() { "upstream", cfg, Vec::new(), - Some(Snapshots::from_storage( - storage.clone(), - MatchStrategy::MethodUriAndBodyHash, - )), + Some(Snapshots::from_storage(storage.clone())), ) .run() .await diff --git a/crates/partly-proxy-lib/tests/replay.rs b/crates/partly-proxy-lib/tests/replay.rs index ab5cdc4..10138a9 100644 --- a/crates/partly-proxy-lib/tests/replay.rs +++ b/crates/partly-proxy-lib/tests/replay.rs @@ -11,10 +11,10 @@ use bytes::Bytes; use http::{HeaderMap, Method, StatusCode}; use partly_proxy_echo as echo; use partly_proxy_lib::{ - Command, ExchangeOutcome, MatchStrategy, Mode, Next, ProxyClusterBuilder, ProxyConfig, - ProxyMiddleware, ProxyRequest, ProxyResponse, RecordedExchange, RecordedRequest, - RecordedResponse, RecordingConfig, RequestContext, RequestMatcher, ResponseSource, - Result as ProxyResult, SharedMiddleware, Snapshots, StubbedResponse, UpstreamTarget, + Command, ExchangeOutcome, Mode, Next, ProxyClusterBuilder, ProxyConfig, ProxyMiddleware, + ProxyRequest, ProxyResponse, RecordedExchange, RecordedRequest, RecordedResponse, + RecordingConfig, RequestContext, RequestMatcher, ResponseSource, Result as ProxyResult, + SharedMiddleware, Snapshots, StubbedResponse, UpstreamTarget, }; use tokio::task::JoinHandle; @@ -80,16 +80,13 @@ async fn replay_hit_serves_recorded_response_without_touching_upstream() { a }; - let replay = Snapshots::in_memory( - vec![make_recorded( - Method::GET, - "/health", - b"", - 200, - b"{\"ok\":true}", - )], - MatchStrategy::MethodUriAndBodyHash, - ); + let replay = Snapshots::in_memory(vec![make_recorded( + Method::GET, + "/health", + b"", + 200, + b"{\"ok\":true}", + )]); let cluster = ProxyClusterBuilder::new() .add_upstream_with_mode( "api", @@ -125,10 +122,13 @@ async fn replay_mode_miss_returns_503_without_touching_upstream() { drop(l); a }; - let replay = Snapshots::in_memory( - vec![make_recorded(Method::GET, "/health", b"", 200, b"replayed")], - MatchStrategy::MethodUriAndBodyHash, - ); + let replay = Snapshots::in_memory(vec![make_recorded( + Method::GET, + "/health", + b"", + 200, + b"replayed", + )]); let cluster = ProxyClusterBuilder::new() .add_upstream_with_mode( "api", @@ -174,10 +174,13 @@ async fn record_mode_miss_falls_through_to_upstream() { // SPECIFICATION.md §8.3: in Mode::Record a replay miss falls through to // the upstream so the new exchange can be recorded. let (echo_addr, _t) = spawn_echo().await; - let replay = Snapshots::in_memory( - vec![make_recorded(Method::GET, "/health", b"", 200, b"replayed")], - MatchStrategy::MethodUriAndBodyHash, - ); + let replay = Snapshots::in_memory(vec![make_recorded( + Method::GET, + "/health", + b"", + 200, + b"replayed", + )]); let cluster = ProxyClusterBuilder::new() .add_upstream_with_mode( "api", @@ -219,10 +222,13 @@ async fn stub_takes_priority_over_replay() { drop(l); a }; - let replay = Snapshots::in_memory( - vec![make_recorded(Method::GET, "/x", b"", 200, b"from-replay")], - MatchStrategy::MethodUriAndBodyHash, - ); + let replay = Snapshots::in_memory(vec![make_recorded( + Method::GET, + "/x", + b"", + 200, + b"from-replay", + )]); let cluster = ProxyClusterBuilder::new() .add_upstream_with( "api", @@ -307,7 +313,7 @@ async fn replay_lookup_uses_redact_request_for_snapshot() { a }; let snapshot = make_recorded(Method::GET, "/secure", b"", 200, b"ok"); - let replay = Snapshots::in_memory(vec![snapshot], MatchStrategy::MethodUriAndBodyHash); + let replay = Snapshots::in_memory(vec![snapshot]); let cluster = ProxyClusterBuilder::new() .add_upstream_with( "api", @@ -343,10 +349,13 @@ async fn replay_records_served_exchanges_when_recording_enabled() { drop(l); a }; - let replay = Snapshots::in_memory( - vec![make_recorded(Method::GET, "/x", b"", 200, b"replay-body")], - MatchStrategy::MethodUriAndBodyHash, - ); + let replay = Snapshots::in_memory(vec![make_recorded( + Method::GET, + "/x", + b"", + 200, + b"replay-body", + )]); let cluster = ProxyClusterBuilder::new() .recording(RecordingConfig::in_memory(10)) .add_upstream_with( @@ -471,10 +480,13 @@ async fn response_source_stub_marks_ctx() { #[tokio::test] async fn response_source_snapshot_marks_ctx() { let captured = Arc::new(Mutex::new(None)); - let replay = Snapshots::in_memory( - vec![make_recorded(Method::GET, "/x", b"", 200, b"replayed")], - MatchStrategy::MethodUriAndBodyHash, - ); + let replay = Snapshots::in_memory(vec![make_recorded( + Method::GET, + "/x", + b"", + 200, + b"replayed", + )]); let cluster = ProxyClusterBuilder::new() .add_upstream_with_mode( "api", @@ -502,10 +514,13 @@ async fn response_source_snapshot_marks_ctx() { #[tokio::test] async fn response_source_replay_miss_marks_ctx() { let captured = Arc::new(Mutex::new(None)); - let replay = Snapshots::in_memory( - vec![make_recorded(Method::GET, "/x", b"", 200, b"replayed")], - MatchStrategy::MethodUriAndBodyHash, - ); + let replay = Snapshots::in_memory(vec![make_recorded( + Method::GET, + "/x", + b"", + 200, + b"replayed", + )]); let cluster = ProxyClusterBuilder::new() .add_upstream_with_mode( "api", From 729f34bdca02dcaca5971da3b840385ce2a331b0 Mon Sep 17 00:00:00 2001 From: Josiah Bull Date: Tue, 16 Jun 2026 16:46:05 +1200 Subject: [PATCH 2/2] refactor!: remove ReplaySource/Snapshots from the public API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replay is now configured purely by attaching a SnapshotStorage backend to an upstream — there is no caller-facing replay type. - `ReplaySource` and `Snapshots` are removed from the public API (`ReplaySource` is now `pub(crate)`, built internally from the attached backend's `load()` stream at `run()`; `Snapshots` is deleted). - `add_upstream_with` / `add_upstream_with_mode` take `Option` directly instead of `Option`. - Add `InMemoryStorage`, a real in-memory `SnapshotStorage` backend in partly-proxy-types (promotes the former storage.rs doc example), for filesystem-free replay/fixtures — replaces `Snapshots::in_memory`. - Drop the now-unused `ReplaySource::from_jsonl`; the round-trip test loads via `from_storage` instead. No version bump: 0.5.0 already covers these breaking changes vs main's 0.4.0 baseline. Spec §3.3/§4/§4.1/§8/§9.1/§20.1 updated to the storage-backend attach model (also clears leftover `strategy` args from the §4/§20.1 examples). --- SPECIFICATION.md | 46 +++-- crates/partly-proxy-lib/benches/common/mod.rs | 5 +- crates/partly-proxy-lib/examples/host.rs | 7 +- crates/partly-proxy-lib/src/builder.rs | 58 +++--- crates/partly-proxy-lib/src/cluster.rs | 5 +- crates/partly-proxy-lib/src/config.rs | 8 +- crates/partly-proxy-lib/src/lib.rs | 5 +- crates/partly-proxy-lib/src/recorder.rs | 2 +- crates/partly-proxy-lib/src/replay.rs | 172 ++++-------------- crates/partly-proxy-lib/tests/record.rs | 18 +- crates/partly-proxy-lib/tests/replay.rs | 28 +-- crates/partly-proxy-types/src/lib.rs | 2 +- crates/partly-proxy-types/src/storage.rs | 90 +++++---- 13 files changed, 183 insertions(+), 263 deletions(-) diff --git a/SPECIFICATION.md b/SPECIFICATION.md index 600de70..41262cd 100644 --- a/SPECIFICATION.md +++ b/SPECIFICATION.md @@ -79,7 +79,7 @@ Scheme is auto-detected from `base_url` — HTTP and HTTPS upstreams use the sam | `enabled: bool` | `true` | Whether exchanges are recorded | | `max_in_memory: usize` | `10_000` | Cap for the in-memory ring buffer (FIFO eviction) | -`RecordingConfig` controls only the cluster-wide in-memory ring (the `enabled` flag and the `max_in_memory` cap that backs the assertion/query API). Durable persistence — NDJSON file, SQLite database, or anything else implementing `SnapshotStorage` — is configured **per upstream** by attaching a `Snapshots` medium when the upstream is registered (`add_upstream_with` / `add_upstream_with_mode`); see §9.1. Keeping the recording cap separate from the storage backend keeps both axes independent, and making storage per-upstream lets each upstream record to (and replay from) its own file. +`RecordingConfig` controls only the cluster-wide in-memory ring (the `enabled` flag and the `max_in_memory` cap that backs the assertion/query API). Durable persistence — NDJSON file, SQLite database, in-memory (`InMemoryStorage`), or anything else implementing `SnapshotStorage` — is configured **per upstream** by attaching a storage backend when the upstream is registered (`add_upstream_with` / `add_upstream_with_mode`); see §9.1. Keeping the recording cap separate from the storage backend keeps both axes independent, and making storage per-upstream lets each upstream record to (and replay from) its own backend. ### 3.4 `UpstreamTlsConfig` @@ -106,20 +106,24 @@ One certificate per listener; no SNI multiplexing. Everything is built through `ProxyClusterBuilder`: ```rust +// Each upstream attaches its own SnapshotStorage backend (Arc-wrapped). +let legacy_store: SharedStorage = Arc::new(JsonlStorage::open("legacy.ndjson").await?); +let frozen_store: SharedStorage = Arc::new(JsonlStorage::open("frozen.ndjson").await?); + let cluster = ProxyClusterBuilder::new() .recording(RecordingConfig { /* … */ }) .add_middleware(GlobalAuthMiddleware) // applies to all upstreams - .add_upstream("api", api_config) // no middleware, no snapshots + .add_upstream("api", api_config) // no middleware, no storage .add_upstream_with_middleware("billing", b_cfg, b_mw) // per-upstream middleware - // A per-upstream snapshot medium: loaded for replay AND appended to while recording. - .add_upstream_with("legacy", l_cfg, l_mw, Some(Snapshots::from_storage(legacy_store, strategy))) + // A per-upstream storage backend: loaded for replay AND appended to while recording. + .add_upstream_with("legacy", l_cfg, l_mw, Some(legacy_store)) // For deterministic playback against a snapshot file (no upstream dial): - .add_upstream_with_mode("frozen", f_cfg, f_mw, Some(Snapshots::from_storage(frozen_store, strategy)), Mode::Replay) + .add_upstream_with_mode("frozen", f_cfg, f_mw, Some(frozen_store), Mode::Replay) .run() .await?; ``` -The `Snapshots` medium handed to an upstream drives both directions of the record/replay round-trip: at `run()` its existing contents are loaded and indexed into a replay source, and in `Mode::Record` every served exchange for that upstream is appended back to the same medium. There is no cluster-wide storage setter — give each upstream its own medium to keep recordings separate. Use `Snapshots::in_memory(exchanges, strategy)` for a replay-only source that is never written back. +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. `run()` binds every listener, starts a shared recorder and command processor, and returns a `ClusterHandle` exposing: @@ -139,7 +143,7 @@ Assertions are not exposed as a Rust API. They are driven exclusively through th | Command channel and processor | Middleware chain (global middleware + that upstream's middleware, in that order) | | Global middleware | Active stubs | | | Pause flag and resume signal | -| | Optional `Snapshots` medium (replay source + durable recording sink) | +| | Optional `SnapshotStorage` backend (replay source + durable recording sink) | | | Optional inbound TLS acceptor | The recorder's in-memory ring is cluster-wide (it backs the assertion/query API, which filters by upstream). Durable storage, by contrast, is per upstream: the recorder routes each exchange to the medium registered for its upstream name, so every upstream persists to its own file. @@ -388,15 +392,19 @@ A stub matches a request when **all** of the following hold (any unset field is ## 8. Replay -A `ReplaySource` is an immutable snapshot of recorded exchanges, indexed for O(1) lookup: +Replay is configured by attaching a `SnapshotStorage` backend to an upstream (§3.3, §4) — there is no caller-facing replay type. At `run()` the cluster drains the backend's `load()` stream into an internal, immutable replay index (keyed for O(1) lookup, §8.1) and registers the same backend as the upstream's recording sink, so one backend drives both directions of the round-trip: ```rust -let replay = ReplaySource::new(exchanges); -// or -let replay = ReplaySource::from_jsonl(path)?; +// durable: records to and replays from the same NDJSON file +let store: SharedStorage = Arc::new(JsonlStorage::open(path).await?); +.add_upstream_with("api", cfg, mw, Some(store)) + +// filesystem-free replay/fixture backend +let store: SharedStorage = Arc::new(InMemoryStorage::from(exchanges)); +.add_upstream_with("api", cfg, mw, Some(store)) ``` -Upstreams do not take a `ReplaySource` directly. Instead they take a `Snapshots` medium (§3.3, §4): `Snapshots::from_storage(store)` loads the source from a durable backend at `run()` *and* registers that backend as the upstream's recording sink, while `Snapshots::in_memory(exchanges)` wraps an in-memory list for replay-only use. The loaded source is consulted on the hot path exactly as described below. +The loaded index is consulted on the hot path exactly as described below. ### 8.1 Match key @@ -415,7 +423,7 @@ Coarser keys (method-only, method+path), normalised variants (query-parameter ca Replay must remain usable with snapshot files containing **10,000 to 100,000 exchanges** — these are realistic sizes for a recorded end-to-end suite, not a worst case to be discouraged. Concretely: -- `ReplaySource::from_jsonl(...)` parses a 100k-line file in a single pass; it does not hold the whole file in a `String` and must stream line-by-line (e.g. `BufReader::lines`) to keep peak memory bounded by the largest single exchange, not the file size. +- Loading a backend at `run()` drains its `load()` stream in a single pass; a file-backed `SnapshotStorage` (e.g. NDJSON) must stream line-by-line rather than hold the whole file in a `String`, keeping peak memory bounded by the largest single exchange, not the file size. - Index construction is O(n) in the number of exchanges; lookup remains O(1) per request regardless of snapshot size. Hash-map capacity should be preallocated from the exchange count to avoid repeated rehashing during load. - Memory budget at 100k exchanges with typical JSON payloads (~1–4 KiB body each) is on the order of hundreds of MiB. The proxy keeps decoded `Bytes` bodies in the source verbatim — there is no per-exchange duplication into the recorder unless `Replay + recording` is enabled. @@ -454,15 +462,15 @@ The builder defaults to `Mode::Record` (`add_upstream`, `add_upstream_with_middl - `RecordedResponse`: status, headers, body bytes. - `RecordedExchange`: unique id, optional `upstream` name (set in cluster mode), timestamp, duration, request, **either** a response or an error string, and a string-keyed `labels` map for caller-supplied metadata. -Bodies serialise as base64 in JSON; the NDJSON format is round-trippable into a `ReplaySource`. +Bodies serialise as base64 in JSON; the NDJSON format round-trips — a file recorded in one run loads back as the replay index in the next. A single recording session can produce **10,000 to 100,000 exchanges** in one NDJSON file — long-running end-to-end suites realistically generate this volume — and the format must remain usable at that scale. Concretely: - The on-disk format is strictly one exchange per line, append-only. Loading a 100k-exchange file is a single streaming pass (no whole-file parse, no JSON-array wrapper). - Storage writes are append-only and per-exchange — a long suite never rewrites earlier lines, so the file grows linearly and is safe to truncate or `tail -f` mid-run. -- Round-tripping a 100k-line NDJSON file into a `ReplaySource` is supported and exercised; see §8.1.1 for the loader's complexity properties. +- Round-tripping a 100k-line NDJSON file back into the replay index is supported and exercised; see §8.1.1 for the loader's complexity properties. -Durable storage is attached per upstream via a `Snapshots` medium (§3.3, §4); the recorder holds a map from upstream name to its `SnapshotStorage` backend and appends each exchange to the medium registered for its `upstream`. Exchanges whose upstream has no attached medium (or no name) are kept in the in-memory ring only. +Durable storage is attached per upstream as a `SnapshotStorage` backend (§3.3, §4); the recorder holds a map from upstream name to its backend and appends each exchange to the backend registered for its `upstream`. Exchanges whose upstream has no attached backend (or no name) are kept in the in-memory ring only. ### 9.2 Recorder API @@ -660,11 +668,11 @@ The crate does not ship a hosting binary — wiring `ProxyClusterBuilder` into a ### 20.1 Record once, replay forever -1. Register the upstream in `Mode::Record` with a `Snapshots::from_storage(jsonl, strategy)` medium pointing at a fresh NDJSON file. +1. Register the upstream in `Mode::Record`, attaching a JSONL `SnapshotStorage` backend (`JsonlStorage::open`) pointing at a fresh NDJSON file. 2. Drive the system under test against the proxy. Real upstream traffic accumulates in that file. -3. In future test runs, register the same upstream in `Mode::Replay` with a `Snapshots::from_storage` medium pointing at the same file. The medium is loaded into the replay source at `run()`; the real upstream is no longer needed. +3. In future test runs, register the same upstream in `Mode::Replay` with a backend pointing at the same file. The backend is loaded into the replay index at `run()`; the real upstream is no longer needed. -Because the medium is the same in both directions, a re-run in `Mode::Record` replays any request already in the file rather than re-recording it (the snapshot acts as a deduplicating cache, §8.3) and only forwards genuinely new requests. Delete the file to force a clean re-capture. +Because the backend is the same in both directions, a re-run in `Mode::Record` replays any request already in the file rather than re-recording it (the snapshot acts as a deduplicating cache, §8.3) and only forwards genuinely new requests. Delete the file to force a clean re-capture. ### 20.2 Ad-hoc mock for a single test diff --git a/crates/partly-proxy-lib/benches/common/mod.rs b/crates/partly-proxy-lib/benches/common/mod.rs index a43e891..96b521a 100644 --- a/crates/partly-proxy-lib/benches/common/mod.rs +++ b/crates/partly-proxy-lib/benches/common/mod.rs @@ -20,8 +20,7 @@ use hyper_util::{ }; use partly_proxy_echo as echo; use partly_proxy_lib::{ - ClusterHandle, ProxyClusterBuilder, ProxyConfig, RecordingConfig, SharedStorage, Snapshots, - UpstreamTarget, + ClusterHandle, ProxyClusterBuilder, ProxyConfig, RecordingConfig, SharedStorage, UpstreamTarget, }; use tempfile::TempDir; use tokio::task::JoinHandle; @@ -135,7 +134,7 @@ pub async fn spawn_proxy(recording: Recording) -> ProxyHandle { max_in_memory: 10_000, }; - let snapshots = storage.map(Snapshots::from_storage); + let snapshots = storage; let builder = ProxyClusterBuilder::new().recording(cfg).add_upstream_with( "upstream", ProxyConfig::http( diff --git a/crates/partly-proxy-lib/examples/host.rs b/crates/partly-proxy-lib/examples/host.rs index db37ea4..dbb215d 100644 --- a/crates/partly-proxy-lib/examples/host.rs +++ b/crates/partly-proxy-lib/examples/host.rs @@ -22,8 +22,7 @@ use std::{net::SocketAddr, sync::Arc}; use partly_proxy_lib::{ - ProxyClusterBuilder, ProxyConfig, RecordingConfig, Result, SharedStorage, Snapshots, - UpstreamTarget, + ProxyClusterBuilder, ProxyConfig, RecordingConfig, Result, SharedStorage, UpstreamTarget, }; #[tokio::main] @@ -51,11 +50,11 @@ async fn main() -> Result<()> { // Storage is configured per upstream: the same medium is loaded for // replay and appended to while recording. Here the single "upstream" // gets its own JSONL file when PARTLY_PROXY_RECORDING_PATH is set. - let snapshots: Option = match std::env::var("PARTLY_PROXY_RECORDING_PATH").ok() { + let snapshots: Option = match std::env::var("PARTLY_PROXY_RECORDING_PATH").ok() { Some(path) => { let storage: SharedStorage = Arc::new(partly_proxy_lib::jsonl::JsonlStorage::open(path).await?); - Some(Snapshots::from_storage(storage)) + Some(storage) } None => None, }; diff --git a/crates/partly-proxy-lib/src/builder.rs b/crates/partly-proxy-lib/src/builder.rs index 741eec4..e4d66f2 100644 --- a/crates/partly-proxy-lib/src/builder.rs +++ b/crates/partly-proxy-lib/src/builder.rs @@ -26,7 +26,7 @@ use crate::{ middleware::{ProxyMiddleware, SharedMiddleware}, proxy_io::{ProxyRequest, ProxyResponse}, recorder::Recorder, - replay::Snapshots, + replay::ReplaySource, upstream::UpstreamRegistry, }; @@ -85,9 +85,9 @@ pub(crate) struct UpstreamSpec { pub name: String, pub config: ProxyConfig, pub middleware: Vec, - /// Per-upstream snapshot medium — loaded for replay and (in `Record`) + /// Per-upstream storage backend — loaded for replay and (in `Record`) /// appended to as the recording sink. Resolved at `run()`. - pub snapshots: Option, + pub storage: Option, pub mode: Mode, pub replay_miss_handler: ReplayMissHandler, } @@ -97,7 +97,7 @@ impl std::fmt::Debug for UpstreamSpec { f.debug_struct("UpstreamSpec") .field("name", &self.name) .field("middleware", &self.middleware.len()) - .field("snapshots", &self.snapshots.is_some()) + .field("storage", &self.storage.is_some()) .field("mode", &self.mode) .finish_non_exhaustive() } @@ -152,7 +152,7 @@ impl ProxyClusterBuilder { name: name.into(), config, middleware: Vec::new(), - snapshots: None, + storage: None, mode: self.default_mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -172,7 +172,7 @@ impl ProxyClusterBuilder { name: name.into(), config, middleware, - snapshots: None, + storage: None, mode: self.default_mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -180,14 +180,17 @@ impl ProxyClusterBuilder { } /// Register an upstream with both per-upstream middleware and an - /// optional [`Snapshots`] medium. Uses the builder's current - /// [`default_mode`](Self::default_mode). + /// optional [`SnapshotStorage`](crate::SnapshotStorage) backend. Uses the + /// builder's current [`default_mode`](Self::default_mode). /// - /// The `snapshots` medium is the single per-upstream storage knob: at + /// The `storage` backend is the single per-upstream storage knob: at /// [`run()`](Self::run) its existing contents are loaded into the replay /// source, and in [`Mode::Record`] every new exchange for this upstream - /// is appended back to it. Give each upstream its own medium (e.g. its - /// own JSONL file) to keep recordings separate. + /// is appended back to it. Construct a backend + /// (e.g. `JsonlStorage::open(path)` or + /// [`InMemoryStorage`](crate::InMemoryStorage)), wrap it in an `Arc`, and + /// pass it here. Give each upstream its own backend to keep recordings + /// separate. /// /// See `SPECIFICATION.md` §8.3: in `Record` mode, stubs take priority /// over replay, which takes priority over the upstream forward. To @@ -198,13 +201,13 @@ impl ProxyClusterBuilder { name: impl Into, config: ProxyConfig, middleware: Vec, - snapshots: Option, + storage: Option, ) -> Self { self.upstreams.push(UpstreamSpec { name: name.into(), config, middleware, - snapshots, + storage, mode: self.default_mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -218,20 +221,20 @@ impl ProxyClusterBuilder { /// missing snapshot yields the replay-miss response (default `503 {}`). /// In [`Mode::Record`] the terminal falls through to the upstream on /// miss and (when recording is enabled) appends the exchange to the - /// upstream's [`Snapshots`] medium. + /// upstream's `storage` backend. pub fn add_upstream_with_mode( mut self, name: impl Into, config: ProxyConfig, middleware: Vec, - snapshots: Option, + storage: Option, mode: Mode, ) -> Self { self.upstreams.push(UpstreamSpec { name: name.into(), config, middleware, - snapshots, + storage, mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -256,7 +259,7 @@ impl ProxyClusterBuilder { name: name.into(), config, middleware, - snapshots: None, + storage: None, mode: Mode::Replay, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -321,20 +324,19 @@ impl ProxyClusterBuilder { } } - // Resolve each upstream's snapshot medium up front: load its - // contents into a replay source for the hot path, and collect the - // durable media into a per-upstream routing map for the recorder. - // Loading is async (it streams the backend), so it happens here in - // `run()` rather than in the synchronous `add_upstream_*` builders. + // Resolve each upstream's storage backend up front: load its + // contents into a replay source for the hot path, and register the + // backend in a per-upstream routing map so the recorder appends new + // exchanges back to it. Loading is async (it streams the backend), + // so it happens here in `run()` rather than in the synchronous + // `add_upstream_*` builders. let mut routes: HashMap = HashMap::new(); let mut resolved = Vec::with_capacity(self.upstreams.len()); for mut spec in self.upstreams { - let replay = match spec.snapshots.take() { - Some(snapshots) => { - let (replay, storage) = snapshots.resolve().await?; - if let Some(storage) = storage { - routes.insert(spec.name.clone(), storage); - } + let replay = match spec.storage.take() { + Some(storage) => { + let replay = ReplaySource::from_storage(storage.as_ref()).await?; + routes.insert(spec.name.clone(), storage); Some(replay) } None => None, diff --git a/crates/partly-proxy-lib/src/cluster.rs b/crates/partly-proxy-lib/src/cluster.rs index 573d729..06c7d70 100644 --- a/crates/partly-proxy-lib/src/cluster.rs +++ b/crates/partly-proxy-lib/src/cluster.rs @@ -94,9 +94,8 @@ impl ClusterHandle { } /// Shared recorder — cheap to clone. Holds the cluster-wide in-memory - /// ring and routes each exchange to its upstream's durable medium (if - /// one was attached via a [`Snapshots`](crate::Snapshots)). See - /// `SPECIFICATION.md` §9. + /// ring and routes each exchange to its upstream's durable storage + /// backend (if one was attached). See `SPECIFICATION.md` §9. pub fn recorder(&self) -> &Recorder { &self.recorder } diff --git a/crates/partly-proxy-lib/src/config.rs b/crates/partly-proxy-lib/src/config.rs index 10aa4cf..54681f2 100644 --- a/crates/partly-proxy-lib/src/config.rs +++ b/crates/partly-proxy-lib/src/config.rs @@ -96,7 +96,7 @@ impl Default for UpstreamTarget { /// Controls the recorder's in-memory ring buffer only. Persistence — /// NDJSON file, `SQLite` database, or anything else implementing /// [`SnapshotStorage`](crate::SnapshotStorage) — is configured per -/// upstream by attaching a [`Snapshots`](crate::Snapshots) medium via +/// upstream by attaching a storage backend via /// [`ProxyClusterBuilder::add_upstream_with`](crate::ProxyClusterBuilder::add_upstream_with). #[derive(Debug, Clone)] pub struct RecordingConfig { @@ -139,9 +139,9 @@ impl RecordingConfig { /// and no replay hit: /// /// - [`Mode::Record`] forwards to the upstream and records the exchange. -/// When a [`ReplaySource`](crate::ReplaySource) is also configured, replay -/// hits short-circuit before the forward (so previously-seen requests -/// don't re-hit the upstream). +/// When a storage backend is also attached, replay hits short-circuit +/// before the forward (so previously-seen requests don't re-hit the +/// upstream). /// - [`Mode::Replay`] never touches the upstream. A miss yields a `503` with /// an empty-JSON-object body (`{}`). #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] diff --git a/crates/partly-proxy-lib/src/lib.rs b/crates/partly-proxy-lib/src/lib.rs index b6cda51..e1189b6 100644 --- a/crates/partly-proxy-lib/src/lib.rs +++ b/crates/partly-proxy-lib/src/lib.rs @@ -41,11 +41,10 @@ pub use partly_proxy_storage_jsonl as jsonl; #[cfg(feature = "storage-sqlite")] pub use partly_proxy_storage_sqlite as sqlite; pub use partly_proxy_types::{ - ExchangeOutcome, ProxyError, RecordedExchange, RecordedRequest, RecordedResponse, Result, - SharedStorage, SnapshotStorage, + ExchangeOutcome, InMemoryStorage, ProxyError, RecordedExchange, RecordedRequest, + RecordedResponse, Result, SharedStorage, SnapshotStorage, }; pub use proxy_io::{ProxyRequest, ProxyResponse}; pub use recorder::Recorder; -pub use replay::{ReplaySource, Snapshots}; pub use stub::{RequestMatcher, StubEntry, StubStore, StubbedResponse}; pub use wire::{StubFields, WireCommand, WireFilter, WireResponse}; diff --git a/crates/partly-proxy-lib/src/recorder.rs b/crates/partly-proxy-lib/src/recorder.rs index 4bd8c4a..b7b22e9 100644 --- a/crates/partly-proxy-lib/src/recorder.rs +++ b/crates/partly-proxy-lib/src/recorder.rs @@ -60,7 +60,7 @@ impl Recorder { /// Build an in-memory-only recorder with no durable media. Persistence /// — NDJSON, `SQLite`, object store, or anything else implementing /// [`SnapshotStorage`](crate::SnapshotStorage) — is configured per - /// upstream by attaching a [`Snapshots`](crate::Snapshots) medium via + /// upstream by attaching a storage backend via /// [`add_upstream_with`](crate::ProxyClusterBuilder::add_upstream_with); /// the builder threads the resulting routes through /// [`Recorder::with_routes`]. diff --git a/crates/partly-proxy-lib/src/replay.rs b/crates/partly-proxy-lib/src/replay.rs index 7743362..e79d011 100644 --- a/crates/partly-proxy-lib/src/replay.rs +++ b/crates/partly-proxy-lib/src/replay.rs @@ -1,29 +1,28 @@ //! Replay source — see `SPECIFICATION.md` §8. //! -//! A `ReplaySource` is an immutable bundle of recorded exchanges indexed -//! for O(1) lookup. The lookup key is `(method, origin-form URI (path + -//! query string), body SHA-256)`, built once at construction (§8.1). +//! `ReplaySource` is a crate-internal, immutable bundle of recorded +//! exchanges indexed for O(1) lookup. The lookup key is `(method, +//! origin-form URI (path + query string), body SHA-256)`, built once at +//! construction (§8.1). +//! +//! It is not part of the public API: callers attach a +//! [`SnapshotStorage`](crate::SnapshotStorage) backend to an upstream +//! (e.g. `JsonlStorage` or [`InMemoryStorage`](crate::InMemoryStorage)), +//! and the cluster builds the `ReplaySource` from that backend's `load()` +//! stream at `run()`. //! //! Lookups go through every middleware's `redact_request_for_snapshot` //! before the lookup key is computed (§8.2.1), so a request that carried //! a live `Authorization` header still matches a snapshot recorded with //! that header stripped. -// Only `from_jsonl` uses these — gate them so `--no-default-features` -// builds don't trip an unused-import warning. -#[cfg(feature = "storage-jsonl")] -use std::io::{BufRead, BufReader}; -#[cfg(feature = "storage-jsonl")] -use std::path::Path; use std::{collections::HashMap, sync::Arc}; -// `ProxyError` is only constructed by the JSONL loader path (and the -// from_storage tests). Gate the import accordingly to keep -// --no-default-features warning-free. -#[cfg(any(test, feature = "storage-jsonl"))] +// `ProxyError` is only constructed in the storage-error test below. +#[cfg(test)] use partly_proxy_types::ProxyError; use partly_proxy_types::{ - ExchangeOutcome, RecordedExchange, Result, SharedStorage, SnapshotStorage, hash::sha256_hex, + ExchangeOutcome, RecordedExchange, Result, SnapshotStorage, hash::sha256_hex, }; use crate::{ @@ -31,75 +30,14 @@ use crate::{ proxy_io::{ProxyRequest, ProxyResponse}, }; -/// Per-upstream snapshot medium handed to -/// [`add_upstream_with`](crate::ProxyClusterBuilder::add_upstream_with). -/// -/// A single `Snapshots` drives both ends of the record/replay round-trip. -/// At cluster [`run()`](crate::ProxyClusterBuilder::run) its existing -/// contents are loaded and indexed into a [`ReplaySource`]; in -/// [`Mode::Record`](crate::Mode) every new exchange for that upstream is -/// appended back to the same medium. There is no separate cluster-wide -/// storage knob — recording is configured per upstream, here. -pub struct Snapshots { - source: SnapshotsSource, -} - -enum SnapshotsSource { - /// Durable medium — loaded for replay, appended to while recording. - Storage(SharedStorage), - /// In-memory exchanges — replay only, never recorded back. Handy for - /// tests and fixtures that don't want to touch the filesystem. - InMemory(Vec), -} - -impl Snapshots { - /// Use a durable [`SharedStorage`] medium (e.g. a JSONL file) as both - /// the replay source and the recording sink for this upstream. - pub fn from_storage(storage: SharedStorage) -> Self { - Self { - source: SnapshotsSource::Storage(storage), - } - } - - /// Use an in-memory list of exchanges as a replay-only source. Nothing - /// recorded at runtime is written back — the medium is read-only. - pub fn in_memory(exchanges: Vec) -> Self { - Self { - source: SnapshotsSource::InMemory(exchanges), - } - } - - /// Resolve into the replay source consulted on the hot path and, for a - /// durable medium, the storage handle to register as the upstream's - /// recording sink. Called once per upstream at cluster `run()`. - pub(crate) async fn resolve(self) -> Result<(ReplaySource, Option)> { - match self.source { - SnapshotsSource::Storage(storage) => { - let replay = ReplaySource::from_storage(storage.as_ref()).await?; - Ok((replay, Some(storage))) - } - SnapshotsSource::InMemory(exchanges) => Ok((ReplaySource::new(exchanges), None)), - } - } -} - -impl std::fmt::Debug for Snapshots { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let kind = match &self.source { - SnapshotsSource::Storage(_) => "Storage", - SnapshotsSource::InMemory(_) => "InMemory", - }; - f.debug_struct("Snapshots").field("source", &kind).finish() - } -} - /// Lookup key: (method, path+query, body sha-256 hex). type IndexKey = (String, String, String); /// Cheap-to-clone replay source. Behind an `Arc`, so several listeners can -/// share one source. +/// share one source. Crate-internal — built from an attached +/// [`SnapshotStorage`] at cluster `run()`, never constructed by callers. #[derive(Clone)] -pub struct ReplaySource { +pub(crate) struct ReplaySource { inner: Arc, } @@ -122,59 +60,20 @@ impl std::fmt::Debug for ReplaySource { impl ReplaySource { /// Build a replay source from an in-memory list of exchanges. - pub fn new(exchanges: Vec) -> Self { + pub(crate) fn new(exchanges: Vec) -> Self { let index = build_index(&exchanges); Self { inner: Arc::new(ReplaySourceInner { exchanges, index }), } } - /// Stream an NDJSON file line-by-line into a replay source. - /// - /// If the file does not exist an empty [`ReplaySource`] is returned — - /// this covers the common case where a snapshots file has not been - /// created yet (e.g. first run in record mode). - /// - /// The loader reads one exchange per line and never materialises the - /// whole file as a single string (per §8.1.1's 100k-exchange scale - /// target). Each malformed line yields a `ProxyError::Recording`. - /// - /// Gated on the `storage-jsonl` Cargo feature (on by default). When the - /// feature is off, callers should use [`ReplaySource::from_storage`] - /// with whichever backend they prefer. - #[cfg(feature = "storage-jsonl")] - pub fn from_jsonl(path: impl AsRef) -> Result { - let file = match std::fs::File::open(&path) { - Ok(f) => f, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => { - return Ok(Self::new(Vec::new())); - } - Err(e) => return Err(ProxyError::Recording(e)), - }; - let reader = BufReader::new(file); - let mut exchanges = Vec::new(); - for (lineno, line) in reader.lines().enumerate() { - let line = line.map_err(ProxyError::Recording)?; - if line.trim().is_empty() { - continue; - } - // Single source of truth for the parse: the JSONL backend - // crate exports the same helper its own `load()` uses. No - // inline copy in this file. - let exchange = partly_proxy_storage_jsonl::parse_ndjson_line(&line, lineno)?; - exchanges.push(exchange); - } - Ok(Self::new(exchanges)) - } - /// Drain a `SnapshotStorage`'s `load()` stream into a replay source. /// - /// The async counterpart to [`from_jsonl`](Self::from_jsonl), generic - /// over any storage backend. Peak memory during construction is + /// Generic over any storage backend. Peak memory during construction is /// bounded by the largest single exchange — the stream is consumed - /// one item at a time, then the assembled `Vec` feeds the existing - /// `build_index` for O(1) lookups. - pub async fn from_storage(storage: &dyn SnapshotStorage) -> Result { + /// one item at a time, then the assembled `Vec` feeds `build_index` + /// for O(1) lookups. + pub(crate) async fn from_storage(storage: &dyn SnapshotStorage) -> Result { use futures::StreamExt; let mut stream = storage.load(); let mut exchanges = Vec::new(); @@ -184,16 +83,12 @@ impl ReplaySource { Ok(Self::new(exchanges)) } - /// Number of exchanges in the source. - pub fn len(&self) -> usize { + /// Number of exchanges in the source. Test-only introspection. + #[cfg(test)] + pub(crate) fn len(&self) -> usize { self.inner.exchanges.len() } - /// Whether the source is empty. - pub fn is_empty(&self) -> bool { - self.inner.exchanges.is_empty() - } - /// Look up a response for `req`. Returns `None` on miss or on a hit with /// an `Error` outcome (errors are intentionally not replayed — use stubs /// for that). @@ -201,7 +96,11 @@ impl ReplaySource { /// `chain` is the effective middleware list — its /// `redact_request_for_snapshot` hooks fire on a working copy of `req` /// before the lookup key is computed. - pub fn lookup(&self, req: &ProxyRequest, chain: &[SharedMiddleware]) -> Option { + pub(crate) fn lookup( + &self, + req: &ProxyRequest, + chain: &[SharedMiddleware], + ) -> Option { let mut redacted = req.clone(); middleware::redact_request(chain, &mut redacted); let key = ( @@ -435,10 +334,10 @@ mod tests { #[cfg(feature = "storage-jsonl")] #[tokio::test] - async fn from_jsonl_round_trips() { + async fn jsonl_storage_round_trips() { // Build a recorder backed by an explicit JsonlStorage, drive - // some exchanges through it, then load that NDJSON file via - // ReplaySource. + // some exchanges through it, then load that NDJSON file back into a + // ReplaySource via `from_storage`. let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("trace.ndjson"); let storage: partly_proxy_types::SharedStorage = Arc::new( @@ -447,8 +346,9 @@ mod tests { .expect("open jsonl"), ); // Route the "api" upstream (stamped on each exchange below) to the - // JSONL medium so records land on disk. - let routes = std::collections::HashMap::from([("api".to_owned(), storage)]); + // JSONL medium so records land on disk. Keep an `Arc` clone to load + // from afterwards. + let routes = std::collections::HashMap::from([("api".to_owned(), storage.clone())]); let recorder = crate::recorder::Recorder::with_routes( crate::config::RecordingConfig::in_memory(100), routes, @@ -476,7 +376,7 @@ mod tests { .unwrap(); } - let src = ReplaySource::from_jsonl(&path).unwrap(); + let src = ReplaySource::from_storage(storage.as_ref()).await.unwrap(); assert_eq!(src.len(), 3); let resp = src.lookup(&live(Method::GET, "/n/1", b""), &[]).unwrap(); assert_eq!(resp.body, Bytes::from_static(b"body-1")); diff --git a/crates/partly-proxy-lib/tests/record.rs b/crates/partly-proxy-lib/tests/record.rs index 7a5d677..ad4417e 100644 --- a/crates/partly-proxy-lib/tests/record.rs +++ b/crates/partly-proxy-lib/tests/record.rs @@ -5,7 +5,7 @@ use std::{net::SocketAddr, time::Duration}; use partly_proxy_echo as echo; use partly_proxy_lib::{ ClusterHandle, ExchangeOutcome, ProxyClusterBuilder, ProxyConfig, RecordedExchange, - RecordingConfig, Snapshots, UpstreamTarget, + RecordingConfig, UpstreamTarget, }; use tokio::task::JoinHandle; @@ -146,12 +146,7 @@ async fn ndjson_persist_file_is_replayable() { ); let cluster = ProxyClusterBuilder::new() .recording(RecordingConfig::in_memory(100)) - .add_upstream_with( - "upstream", - cfg, - Vec::new(), - Some(Snapshots::from_storage(storage)), - ) + .add_upstream_with("upstream", cfg, Vec::new(), Some(storage)) .run() .await .unwrap(); @@ -185,7 +180,7 @@ async fn ndjson_persist_file_is_replayable() { /// Storage backend that counts every `append` and `flush` it sees and /// keeps the exchanges in memory. Used to verify the per-upstream -/// `Snapshots::from_storage(...)` plumbing. +/// `add_upstream_with(storage)` plumbing. #[derive(Debug, Default)] struct TrackingStorage { appended: tokio::sync::Mutex>, @@ -230,12 +225,7 @@ async fn custom_storage_via_per_upstream_snapshots() { ); let cluster = ProxyClusterBuilder::new() .recording(RecordingConfig::in_memory(100)) - .add_upstream_with( - "upstream", - cfg, - Vec::new(), - Some(Snapshots::from_storage(storage.clone())), - ) + .add_upstream_with("upstream", cfg, Vec::new(), Some(storage.clone())) .run() .await .unwrap(); diff --git a/crates/partly-proxy-lib/tests/replay.rs b/crates/partly-proxy-lib/tests/replay.rs index 10138a9..10af650 100644 --- a/crates/partly-proxy-lib/tests/replay.rs +++ b/crates/partly-proxy-lib/tests/replay.rs @@ -11,10 +11,10 @@ use bytes::Bytes; use http::{HeaderMap, Method, StatusCode}; use partly_proxy_echo as echo; use partly_proxy_lib::{ - Command, ExchangeOutcome, Mode, Next, ProxyClusterBuilder, ProxyConfig, ProxyMiddleware, - ProxyRequest, ProxyResponse, RecordedExchange, RecordedRequest, RecordedResponse, - RecordingConfig, RequestContext, RequestMatcher, ResponseSource, Result as ProxyResult, - SharedMiddleware, Snapshots, StubbedResponse, UpstreamTarget, + Command, ExchangeOutcome, InMemoryStorage, Mode, Next, ProxyClusterBuilder, ProxyConfig, + ProxyMiddleware, ProxyRequest, ProxyResponse, RecordedExchange, RecordedRequest, + RecordedResponse, RecordingConfig, RequestContext, RequestMatcher, ResponseSource, + Result as ProxyResult, SharedMiddleware, SharedStorage, StubbedResponse, UpstreamTarget, }; use tokio::task::JoinHandle; @@ -43,6 +43,10 @@ fn cfg(url: String) -> ProxyConfig { ) } +fn in_memory_store(exchanges: Vec) -> SharedStorage { + Arc::new(InMemoryStorage::from(exchanges)) +} + fn make_recorded( method: Method, path: &str, @@ -80,7 +84,7 @@ async fn replay_hit_serves_recorded_response_without_touching_upstream() { a }; - let replay = Snapshots::in_memory(vec![make_recorded( + let replay = in_memory_store(vec![make_recorded( Method::GET, "/health", b"", @@ -122,7 +126,7 @@ async fn replay_mode_miss_returns_503_without_touching_upstream() { drop(l); a }; - let replay = Snapshots::in_memory(vec![make_recorded( + let replay = in_memory_store(vec![make_recorded( Method::GET, "/health", b"", @@ -174,7 +178,7 @@ async fn record_mode_miss_falls_through_to_upstream() { // SPECIFICATION.md §8.3: in Mode::Record a replay miss falls through to // the upstream so the new exchange can be recorded. let (echo_addr, _t) = spawn_echo().await; - let replay = Snapshots::in_memory(vec![make_recorded( + let replay = in_memory_store(vec![make_recorded( Method::GET, "/health", b"", @@ -222,7 +226,7 @@ async fn stub_takes_priority_over_replay() { drop(l); a }; - let replay = Snapshots::in_memory(vec![make_recorded( + let replay = in_memory_store(vec![make_recorded( Method::GET, "/x", b"", @@ -313,7 +317,7 @@ async fn replay_lookup_uses_redact_request_for_snapshot() { a }; let snapshot = make_recorded(Method::GET, "/secure", b"", 200, b"ok"); - let replay = Snapshots::in_memory(vec![snapshot]); + let replay = in_memory_store(vec![snapshot]); let cluster = ProxyClusterBuilder::new() .add_upstream_with( "api", @@ -349,7 +353,7 @@ async fn replay_records_served_exchanges_when_recording_enabled() { drop(l); a }; - let replay = Snapshots::in_memory(vec![make_recorded( + let replay = in_memory_store(vec![make_recorded( Method::GET, "/x", b"", @@ -480,7 +484,7 @@ async fn response_source_stub_marks_ctx() { #[tokio::test] async fn response_source_snapshot_marks_ctx() { let captured = Arc::new(Mutex::new(None)); - let replay = Snapshots::in_memory(vec![make_recorded( + let replay = in_memory_store(vec![make_recorded( Method::GET, "/x", b"", @@ -514,7 +518,7 @@ async fn response_source_snapshot_marks_ctx() { #[tokio::test] async fn response_source_replay_miss_marks_ctx() { let captured = Arc::new(Mutex::new(None)); - let replay = Snapshots::in_memory(vec![make_recorded( + let replay = in_memory_store(vec![make_recorded( Method::GET, "/x", b"", diff --git a/crates/partly-proxy-types/src/lib.rs b/crates/partly-proxy-types/src/lib.rs index 8bb6260..131d040 100644 --- a/crates/partly-proxy-types/src/lib.rs +++ b/crates/partly-proxy-types/src/lib.rs @@ -23,4 +23,4 @@ pub mod testing; pub use error::{BoxError, ProxyError, Result}; pub use recorded::{ExchangeOutcome, RecordedExchange, RecordedRequest, RecordedResponse}; -pub use storage::{SharedStorage, SnapshotStorage}; +pub use storage::{InMemoryStorage, SharedStorage, SnapshotStorage}; diff --git a/crates/partly-proxy-types/src/storage.rs b/crates/partly-proxy-types/src/storage.rs index 959b2c1..f1ddaf7 100644 --- a/crates/partly-proxy-types/src/storage.rs +++ b/crates/partly-proxy-types/src/storage.rs @@ -2,46 +2,14 @@ //! //! Any crate can implement [`SnapshotStorage`]; the trait surface uses //! only types from this crate, so backends don't need `partly-proxy-lib`. -//! First-party backends: `partly-proxy-storage-jsonl`, -//! `partly-proxy-storage-sqlite`. -//! -//! # Example -//! -//! ``` -//! use std::sync::Mutex; -//! -//! use async_trait::async_trait; -//! use partly_proxy_types::storage::{BoxStream, SnapshotStorage}; -//! use partly_proxy_types::{RecordedExchange, Result}; -//! -//! #[derive(Debug, Default)] -//! pub struct InMemoryStorage { -//! exchanges: Mutex>, -//! } -//! -//! #[async_trait] -//! impl SnapshotStorage for InMemoryStorage { -//! async fn append(&self, exchange: &RecordedExchange) -> Result<()> { -//! self.exchanges.lock().unwrap().push(exchange.clone()); -//! Ok(()) -//! } -//! -//! async fn flush(&self) -> Result<()> { -//! Ok(()) -//! } -//! -//! fn load(&self) -> BoxStream<'_, Result> { -//! let snapshot = self.exchanges.lock().unwrap().clone(); -//! Box::pin(futures::stream::iter(snapshot.into_iter().map(Ok))) -//! } -//! } -//! ``` +//! First-party backends: [`InMemoryStorage`] (in this crate), +//! `partly-proxy-storage-jsonl`, `partly-proxy-storage-sqlite`. //! //! Backends can opt into the shared conformance battery by enabling the //! `testing` Cargo feature and calling //! [`testing::run_conformance`](crate::testing::run_conformance). -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use async_trait::async_trait; // Re-exported so implementers don't have to add `futures` themselves. @@ -72,3 +40,55 @@ pub trait SnapshotStorage: Send + Sync + std::fmt::Debug { /// Cheap-to-clone handle on a [`SnapshotStorage`]. pub type SharedStorage = Arc; + +/// In-memory [`SnapshotStorage`] backed by a `Mutex>`. +/// +/// The zero-dependency backend for replay-only fixtures and tests that +/// don't want to touch the filesystem: seed it from a list of exchanges, +/// wrap it in an `Arc`, and attach it to an upstream. `load()` replays the +/// seeded exchanges; `append()` keeps any newly recorded ones in the same +/// vec (so it round-trips like a file would). +/// +/// ``` +/// use std::sync::Arc; +/// +/// use partly_proxy_types::{InMemoryStorage, SharedStorage}; +/// +/// let store: SharedStorage = Arc::new(InMemoryStorage::from(vec![/* exchanges */])); +/// ``` +#[derive(Debug, Default)] +pub struct InMemoryStorage { + exchanges: Mutex>, +} + +impl InMemoryStorage { + /// An empty store. + pub fn new() -> Self { + Self::default() + } +} + +impl From> for InMemoryStorage { + fn from(exchanges: Vec) -> Self { + Self { + exchanges: Mutex::new(exchanges), + } + } +} + +#[async_trait] +impl SnapshotStorage for InMemoryStorage { + async fn append(&self, exchange: &RecordedExchange) -> Result<()> { + self.exchanges.lock().unwrap().push(exchange.clone()); + Ok(()) + } + + async fn flush(&self) -> Result<()> { + Ok(()) + } + + fn load(&self) -> ExchangeStream<'_> { + let snapshot = self.exchanges.lock().unwrap().clone(); + Box::pin(futures::stream::iter(snapshot.into_iter().map(Ok))) + } +}