diff --git a/Cargo.lock b/Cargo.lock index 00b0187..f7233d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1207,7 +1207,7 @@ dependencies = [ [[package]] name = "partly-proxy-echo" -version = "0.3.0" +version = "0.4.0" dependencies = [ "base64", "bytes", @@ -1224,7 +1224,7 @@ dependencies = [ [[package]] name = "partly-proxy-lib" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "base64", @@ -1264,7 +1264,7 @@ dependencies = [ [[package]] name = "partly-proxy-storage-jsonl" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-stream", "async-trait", @@ -1280,7 +1280,7 @@ dependencies = [ [[package]] name = "partly-proxy-storage-sqlite" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-stream", "async-trait", @@ -1295,7 +1295,7 @@ dependencies = [ [[package]] name = "partly-proxy-types" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "base64", diff --git a/Cargo.toml b/Cargo.toml index 5a5280c..45d88b5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/*"] [workspace.package] -version = "0.3.0" +version = "0.4.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.3.0", path = "crates/partly-proxy-types" } -partly-proxy-storage-jsonl = { version = "0.3.0", path = "crates/partly-proxy-storage-jsonl" } -partly-proxy-storage-sqlite = { version = "0.3.0", path = "crates/partly-proxy-storage-sqlite" } +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-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 ffa75bb..83daecf 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 in-memory ring. Durable persistence — NDJSON file, SQLite database, or anything else implementing `SnapshotStorage` — is configured separately by passing a `SharedStorage` to `Recorder::with_storage` or `ProxyClusterBuilder::storage(...)`. Mixing the two concerns into a single `persist_path` field would couple the recording cap to one specific backend; the split keeps both axes independent. +`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. ### 3.4 `UpstreamTlsConfig` @@ -109,15 +109,18 @@ Everything is built through `ProxyClusterBuilder`: let cluster = ProxyClusterBuilder::new() .recording(RecordingConfig { /* … */ }) .add_middleware(GlobalAuthMiddleware) // applies to all upstreams - .add_upstream("api", api_config) // no middleware, no replay + .add_upstream("api", api_config) // no middleware, no snapshots .add_upstream_with_middleware("billing", b_cfg, b_mw) // per-upstream middleware - .add_upstream_with("legacy", l_cfg, l_mw, Some(replay_source)) + // 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))) // For deterministic playback against a snapshot file (no upstream dial): - .add_upstream_with_mode("frozen", f_cfg, f_mw, Some(snapshot), Mode::Replay) + .add_upstream_with_mode("frozen", f_cfg, f_mw, Some(Snapshots::from_storage(frozen_store, strategy)), 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. + `run()` binds every listener, starts a shared recorder and command processor, and returns a `ClusterHandle` exposing: - `addr(name) -> Option` — the bound address for a named upstream. @@ -132,13 +135,15 @@ Assertions are not exposed as a Rust API. They are driven exclusively through th | Shared across cluster | Per upstream | |----------------------|--------------| -| Recorder (single buffer + persist file) | Forwarder and connection pool | +| Recorder (single in-memory ring buffer) | Forwarder and connection pool | | 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 replay source | +| | Optional `Snapshots` medium (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. + --- ## 5. Request Lifecycle @@ -313,7 +318,7 @@ impl ProxyMiddleware for StripAuth { #### When and where it runs -- **On record** (lifecycle stage 9): the proxy clones the request/response into a working `ProxyRequest` / `ProxyResponse`, runs `redact_request_for_snapshot` and `redact_response_for_snapshot` across the chain in registration order, **then** computes the body hash, serialises, and writes to the in-memory ring and `persist_path`. The original request/response returned to the client is untouched — the live caller still sees its `Authorization` header. +- **On record** (lifecycle stage 9): the proxy clones the request/response into a working `ProxyRequest` / `ProxyResponse`, runs `redact_request_for_snapshot` and `redact_response_for_snapshot` across the chain in registration order, **then** computes the body hash, serialises, and writes to the in-memory ring and the upstream's durable medium (if one is attached). The original request/response returned to the client is untouched — the live caller still sees its `Authorization` header. - **On replay lookup** (lifecycle stage 7): before the live `ProxyRequest` is used to compute the lookup key, the proxy runs `redact_request_for_snapshot` across the chain on a working copy. The key is computed from the redacted copy. Because the snapshot on disk was written with the same redaction applied, the hashes agree and the lookup hits. The original request continues into the chain unmodified. Both calls are infallible by design — they are pure rewrites, not policy decisions. If a middleware needs to fail-stop on a missing field, it should do so in `handle`, not here. @@ -391,6 +396,8 @@ let replay = ReplaySource::new(exchanges, MatchStrategy::MethodUriAndBodyHash); let replay = ReplaySource::from_jsonl(path, MatchStrategy::MethodUriAndBodyHash)?; ``` +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. + ### 8.1 Match strategies | Strategy | Key | Notes | @@ -449,19 +456,23 @@ Bodies serialise as base64 in JSON; the NDJSON format is round-trippable into a 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). -- `persist_path` 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. +- 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. +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. + ### 9.2 Recorder API The shared `Recorder` is cheaply cloneable and exposes async methods: -- `record(exchange)` — insert (also appends to disk if `persist_path` is set). Before insertion, the exchange is passed through every middleware's `redact_request_for_snapshot` / `redact_response_for_snapshot` (see §6.4), so secrets are stripped before bytes leave the recorder. The body hash stored on the recorded request is computed *after* redaction, which is what makes hash-based replay lookups continue to work. +- `record(exchange)` — insert into the in-memory ring, and append to the upstream's durable medium first if one is registered for `exchange.upstream`. Before insertion, the exchange is passed through every middleware's `redact_request_for_snapshot` / `redact_response_for_snapshot` (see §6.4), so secrets are stripped before bytes leave the recorder. The body hash stored on the recorded request is computed *after* redaction, which is what makes hash-based replay lookups continue to work. - `exchanges()` — clone the full buffer. - `len()`, `clear()`. - `any_matching(pred)`, `count_matching(pred)`, `find_matching(pred)` — predicate-based scans. +- `storage_for(upstream)` — borrow the durable backend registered for a named upstream, if any. +- `flush()` — fence every registered durable backend (called on shutdown). -When the in-memory buffer is full, the oldest exchange is evicted. +The durable append happens *before* the exchange lands in the in-memory ring, so a storage error stops the exchange from becoming visible to predicate scans. When the in-memory buffer is full, the oldest exchange is evicted. ### 9.3 Provenance @@ -646,9 +657,11 @@ The crate does not ship a hosting binary — wiring `ProxyClusterBuilder` into a ### 20.1 Record once, replay forever -1. Configure one upstream with recording enabled and a `persist_path`. -2. Drive the system under test against the proxy. Real upstream traffic accumulates in NDJSON. -3. In future test runs, load the NDJSON via `ReplaySource::from_jsonl` and add it to the upstream. The real upstream is no longer needed. +1. Register the upstream in `Mode::Record` with a `Snapshots::from_storage(jsonl, strategy)` medium 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. + +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. ### 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 09b2113..fe9bb7f 100644 --- a/crates/partly-proxy-lib/benches/common/mod.rs +++ b/crates/partly-proxy-lib/benches/common/mod.rs @@ -20,7 +20,8 @@ use hyper_util::{ }; use partly_proxy_echo as echo; use partly_proxy_lib::{ - ClusterHandle, ProxyClusterBuilder, ProxyConfig, RecordingConfig, SharedStorage, UpstreamTarget, + ClusterHandle, MatchStrategy, ProxyClusterBuilder, ProxyConfig, RecordingConfig, SharedStorage, + Snapshots, UpstreamTarget, }; use tempfile::TempDir; use tokio::task::JoinHandle; @@ -134,16 +135,17 @@ pub async fn spawn_proxy(recording: Recording) -> ProxyHandle { max_in_memory: 10_000, }; - let mut builder = ProxyClusterBuilder::new().recording(cfg).add_upstream( + let snapshots = storage + .map(|storage| Snapshots::from_storage(storage, MatchStrategy::MethodUriAndBodyHash)); + let builder = ProxyClusterBuilder::new().recording(cfg).add_upstream_with( "upstream", ProxyConfig::http( "127.0.0.1:0".parse().unwrap(), UpstreamTarget::new(format!("http://{echo_addr}")), ), + Vec::new(), + snapshots, ); - if let Some(storage) = storage { - builder = builder.storage(storage); - } let cluster = builder.run().await.expect("cluster build"); let proxy_addr = cluster.addr("upstream").expect("bound addr"); diff --git a/crates/partly-proxy-lib/examples/host.rs b/crates/partly-proxy-lib/examples/host.rs index 24429f3..b6b8497 100644 --- a/crates/partly-proxy-lib/examples/host.rs +++ b/crates/partly-proxy-lib/examples/host.rs @@ -22,7 +22,8 @@ use std::{net::SocketAddr, sync::Arc}; use partly_proxy_lib::{ - ProxyClusterBuilder, ProxyConfig, RecordingConfig, Result, SharedStorage, UpstreamTarget, + MatchStrategy, ProxyClusterBuilder, ProxyConfig, RecordingConfig, Result, SharedStorage, + Snapshots, UpstreamTarget, }; #[tokio::main] @@ -47,25 +48,32 @@ async fn main() -> Result<()> { .expect("PARTLY_PROXY_TCP_CONTROL_BIND must be a valid SocketAddr") }); - let storage: Option = match std::env::var("PARTLY_PROXY_RECORDING_PATH").ok() { - Some(path) => Some(Arc::new( - partly_proxy_lib::jsonl::JsonlStorage::open(path).await?, - )), + // 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() { + Some(path) => { + let storage: SharedStorage = + Arc::new(partly_proxy_lib::jsonl::JsonlStorage::open(path).await?); + Some(Snapshots::from_storage( + storage, + MatchStrategy::MethodUriAndBodyHash, + )) + } None => None, }; let mut builder = ProxyClusterBuilder::new() .recording(RecordingConfig::in_memory(10_000)) - .add_upstream( + .add_upstream_with( "upstream", ProxyConfig::http(proxy_bind, UpstreamTarget::new(upstream_url)), + Vec::new(), + snapshots, ); if let Some(addr) = tcp_control_bind { builder = builder.tcp_control_plane(addr); } - if let Some(storage) = storage { - builder = builder.storage(storage); - } let cluster = builder.run().await?; tracing::info!( diff --git a/crates/partly-proxy-lib/src/builder.rs b/crates/partly-proxy-lib/src/builder.rs index 9a75995..741eec4 100644 --- a/crates/partly-proxy-lib/src/builder.rs +++ b/crates/partly-proxy-lib/src/builder.rs @@ -8,7 +8,7 @@ //! deterministic. use std::{ - collections::{BTreeMap, HashSet}, + collections::{BTreeMap, HashMap, HashSet}, net::SocketAddr, sync::Arc, }; @@ -26,7 +26,7 @@ use crate::{ middleware::{ProxyMiddleware, SharedMiddleware}, proxy_io::{ProxyRequest, ProxyResponse}, recorder::Recorder, - replay::ReplaySource, + replay::Snapshots, upstream::UpstreamRegistry, }; @@ -50,7 +50,6 @@ pub struct ProxyClusterBuilder { upstreams: Vec, global_middleware: Vec, tcp_control_addr: Option, - storage: Option, replay_miss_handler: ReplayMissHandler, } @@ -62,7 +61,6 @@ impl Default for ProxyClusterBuilder { upstreams: Vec::new(), global_middleware: Vec::new(), tcp_control_addr: None, - storage: None, replay_miss_handler: default_replay_miss_handler(), } } @@ -78,7 +76,6 @@ impl std::fmt::Debug for ProxyClusterBuilder { ) .field("global_middleware", &self.global_middleware.len()) .field("tcp_control_addr", &self.tcp_control_addr) - .field("storage", &self.storage.is_some()) .finish_non_exhaustive() } } @@ -88,7 +85,9 @@ pub(crate) struct UpstreamSpec { pub name: String, pub config: ProxyConfig, pub middleware: Vec, - pub replay: Option, + /// Per-upstream snapshot medium — loaded for replay and (in `Record`) + /// appended to as the recording sink. Resolved at `run()`. + pub snapshots: Option, pub mode: Mode, pub replay_miss_handler: ReplayMissHandler, } @@ -98,7 +97,7 @@ impl std::fmt::Debug for UpstreamSpec { f.debug_struct("UpstreamSpec") .field("name", &self.name) .field("middleware", &self.middleware.len()) - .field("replay", &self.replay.is_some()) + .field("snapshots", &self.snapshots.is_some()) .field("mode", &self.mode) .finish_non_exhaustive() } @@ -153,7 +152,7 @@ impl ProxyClusterBuilder { name: name.into(), config, middleware: Vec::new(), - replay: None, + snapshots: None, mode: self.default_mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -173,7 +172,7 @@ impl ProxyClusterBuilder { name: name.into(), config, middleware, - replay: None, + snapshots: None, mode: self.default_mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -181,9 +180,15 @@ impl ProxyClusterBuilder { } /// Register an upstream with both per-upstream middleware and an - /// optional replay source. Uses the builder's current + /// optional [`Snapshots`] medium. Uses the builder's current /// [`default_mode`](Self::default_mode). /// + /// The `snapshots` medium 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. + /// /// See `SPECIFICATION.md` §8.3: in `Record` mode, stubs take priority /// over replay, which takes priority over the upstream forward. To /// replay snapshots without ever forwarding to the upstream, call @@ -193,13 +198,13 @@ impl ProxyClusterBuilder { name: impl Into, config: ProxyConfig, middleware: Vec, - replay: Option, + snapshots: Option, ) -> Self { self.upstreams.push(UpstreamSpec { name: name.into(), config, middleware, - replay, + snapshots, mode: self.default_mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -213,20 +218,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 - /// recorder. + /// upstream's [`Snapshots`] medium. pub fn add_upstream_with_mode( mut self, name: impl Into, config: ProxyConfig, middleware: Vec, - replay: Option, + snapshots: Option, mode: Mode, ) -> Self { self.upstreams.push(UpstreamSpec { name: name.into(), config, middleware, - replay, + snapshots, mode, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -251,7 +256,7 @@ impl ProxyClusterBuilder { name: name.into(), config, middleware, - replay: None, + snapshots: None, mode: Mode::Replay, replay_miss_handler: Arc::clone(&self.replay_miss_handler), }); @@ -299,19 +304,6 @@ impl ProxyClusterBuilder { self } - /// Override the recorder's storage backend. - /// - /// When set, `run()` builds the recorder via - /// [`Recorder::with_storage`](crate::Recorder::with_storage) and the - /// provided `SharedStorage` is used for every recorded exchange. When - /// unset, the recorder falls back to opening the default backend from - /// `RecordingConfig::persist_path` (NDJSON when the `storage-jsonl` - /// feature is on, in-memory only otherwise). - pub fn storage(mut self, storage: SharedStorage) -> Self { - self.storage = Some(storage); - self - } - /// Bind every listener and start its accept loop. /// /// Returns a [`ClusterHandle`](crate::ClusterHandle) once all listeners @@ -329,20 +321,39 @@ impl ProxyClusterBuilder { } } - let recorder = match self.storage.clone() { - Some(storage) => Recorder::with_storage(self.recording.clone(), Some(storage)), - None => Recorder::new(self.recording.clone()), - }; + // 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. + 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); + } + Some(replay) + } + None => None, + }; + resolved.push((spec, replay)); + } + + let recorder = Recorder::with_routes(self.recording.clone(), routes); let (shutdown_tx, shutdown_rx) = watch::channel::>(None); let mut upstreams = BTreeMap::new(); let mut registry = UpstreamRegistry::default(); let global_middleware = self.global_middleware; - for spec in self.upstreams { + for (spec, replay) in resolved { let name = spec.name.clone(); match listener::spawn_listener( spec, + replay, global_middleware.clone(), recorder.clone(), shutdown_rx.clone(), diff --git a/crates/partly-proxy-lib/src/cluster.rs b/crates/partly-proxy-lib/src/cluster.rs index a2fd7c2..573d729 100644 --- a/crates/partly-proxy-lib/src/cluster.rs +++ b/crates/partly-proxy-lib/src/cluster.rs @@ -93,8 +93,9 @@ impl ClusterHandle { &self.recording } - /// Shared recorder — cheap to clone. Holds the in-memory ring and - /// optionally appends to the configured NDJSON file. See + /// 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. 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 013ba0d..10aa4cf 100644 --- a/crates/partly-proxy-lib/src/config.rs +++ b/crates/partly-proxy-lib/src/config.rs @@ -95,10 +95,9 @@ 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 -/// separately via -/// [`Recorder::with_storage`](crate::Recorder::with_storage) or -/// [`ProxyClusterBuilder::storage`](crate::ProxyClusterBuilder::storage). +/// [`SnapshotStorage`](crate::SnapshotStorage) — is configured per +/// upstream by attaching a [`Snapshots`](crate::Snapshots) medium via +/// [`ProxyClusterBuilder::add_upstream_with`](crate::ProxyClusterBuilder::add_upstream_with). #[derive(Debug, Clone)] pub struct RecordingConfig { /// Whether exchanges are recorded at all. diff --git a/crates/partly-proxy-lib/src/lib.rs b/crates/partly-proxy-lib/src/lib.rs index 0dbf085..fd8970d 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}; +pub use replay::{MatchStrategy, ReplaySource, Snapshots}; pub use stub::{RequestMatcher, StubEntry, StubStore, StubbedResponse}; pub use wire::{StubFields, WireCommand, WireFilter, WireResponse}; diff --git a/crates/partly-proxy-lib/src/listener.rs b/crates/partly-proxy-lib/src/listener.rs index 7206ed2..72d35a6 100644 --- a/crates/partly-proxy-lib/src/listener.rs +++ b/crates/partly-proxy-lib/src/listener.rs @@ -36,6 +36,7 @@ use crate::{ middleware::{self, SharedMiddleware, Terminal, TerminalFuture}, proxy_io::{ProxyRequest, ProxyResponse}, recorder::Recorder, + replay::ReplaySource, tls::build_tls_acceptor, upstream::UpstreamRuntime, }; @@ -51,6 +52,7 @@ pub(crate) struct RunningListener { /// Bind the listener for one upstream spec and spawn its accept loop. pub(crate) async fn spawn_listener( spec: UpstreamSpec, + replay: Option, global_middleware: Vec, recorder: Recorder, shutdown: watch::Receiver>, @@ -75,7 +77,7 @@ pub(crate) async fn spawn_listener( forwarder, recorder, middleware, - spec.replay, + replay, spec.mode, spec.replay_miss_handler, )); diff --git a/crates/partly-proxy-lib/src/recorder.rs b/crates/partly-proxy-lib/src/recorder.rs index a2ba7d0..4bd8c4a 100644 --- a/crates/partly-proxy-lib/src/recorder.rs +++ b/crates/partly-proxy-lib/src/recorder.rs @@ -1,16 +1,22 @@ //! Shared traffic recorder — see `SPECIFICATION.md` §9.2. //! -//! The recorder owns an in-memory ring buffer of `RecordedExchange`s and, -//! optionally, a pluggable [`SnapshotStorage`](crate::SnapshotStorage) -//! medium for durable persistence. It is cheaply cloneable (`Arc`-backed); -//! every listener task and the future control plane share one instance -//! per cluster. +//! The recorder owns a single cluster-wide in-memory ring buffer of +//! `RecordedExchange`s — that ring backs the cluster-wide assertion/query +//! API (`QueryTraffic`, `AssertSeen`, `AssertCount`). Durable persistence, +//! by contrast, is configured *per upstream*: the recorder holds a routing +//! map from upstream name to its [`SnapshotStorage`](crate::SnapshotStorage) +//! medium, and each exchange is appended to the medium registered for its +//! own upstream (if any). It is cheaply cloneable (`Arc`-backed); every +//! listener task and the control plane share one instance per cluster. //! //! Redaction (`redact_request_for_snapshot` / `redact_response_for_snapshot`, //! §6.4) happens in the lifecycle code *before* this recorder is called — //! the recorder hashes and stores whatever it is handed. -use std::{collections::VecDeque, sync::Arc}; +use std::{ + collections::{HashMap, VecDeque}, + sync::Arc, +}; use partly_proxy_types::{RecordedExchange, Result, SharedStorage}; use tokio::sync::{Notify, RwLock}; @@ -28,7 +34,7 @@ impl std::fmt::Debug for Recorder { f.debug_struct("Recorder") .field("enabled", &self.inner.config.enabled) .field("max_in_memory", &self.inner.config.max_in_memory) - .field("has_storage", &self.inner.storage.is_some()) + .field("storage_routes", &self.inner.storage.len()) .finish_non_exhaustive() } } @@ -36,8 +42,10 @@ impl std::fmt::Debug for Recorder { struct RecorderInner { config: RecordingConfig, state: RwLock, - /// Pluggable durable medium. `None` ⇒ in-memory only. - storage: Option, + /// Per-upstream durable media, keyed by upstream name. An exchange is + /// appended to the medium registered for its `upstream`; upstreams with + /// no entry (and exchanges with no name) are kept in memory only. + storage: HashMap, /// Fired (via `notify_waiters`) every time a new exchange is recorded. /// The wait-for assertion loop registers a waiter before each predicate /// check, so notifications that arrive between checks are not lost. @@ -49,20 +57,21 @@ struct RecorderState { } impl Recorder { - /// Build an in-memory-only recorder. Persistence — NDJSON, - /// `SQLite`, object store, or anything else implementing - /// [`SnapshotStorage`](crate::SnapshotStorage) — is configured by - /// constructing the backend yourself and threading it through - /// [`Recorder::with_storage`] or - /// [`ProxyClusterBuilder::storage`](crate::ProxyClusterBuilder::storage). + /// 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 + /// [`add_upstream_with`](crate::ProxyClusterBuilder::add_upstream_with); + /// the builder threads the resulting routes through + /// [`Recorder::with_routes`]. pub fn new(config: RecordingConfig) -> Self { - Self::with_storage(config, None) + Self::with_routes(config, HashMap::new()) } - /// Build a recorder from an already-constructed storage backend. - /// Synchronous — no I/O. The caller owns the file/connection - /// lifecycle that produced `storage`. - pub fn with_storage(config: RecordingConfig, storage: Option) -> Self { + /// Build a recorder from a map of upstream name to its durable storage + /// backend. Synchronous — no I/O. The caller owns the file/connection + /// lifecycle that produced each medium. + pub fn with_routes(config: RecordingConfig, storage: HashMap) -> Self { let initial_capacity = config.max_in_memory.min(1024); Self { inner: Arc::new(RecorderInner { @@ -94,16 +103,17 @@ impl Recorder { self.inner.config.max_in_memory } - /// View the underlying storage backend, if any. - pub fn storage(&self) -> Option<&SharedStorage> { - self.inner.storage.as_ref() + /// View the durable storage backend registered for `upstream`, if any. + pub fn storage_for(&self, upstream: &str) -> Option<&SharedStorage> { + self.inner.storage.get(upstream) } /// Insert an exchange. If the buffer is at capacity, the oldest entry is - /// evicted first (FIFO). When the recorder has a storage backend, the - /// exchange is appended to durable storage *before* it lands in memory — - /// that way a storage error stops the exchange from becoming visible - /// to predicate scans. + /// evicted first (FIFO). When the exchange's `upstream` has a registered + /// storage backend, the exchange is appended to that medium *before* it + /// lands in memory — that way a storage error stops the exchange from + /// becoming visible to predicate scans. Exchanges whose upstream has no + /// registered medium (or no name) are kept in memory only. /// /// When recording is disabled, this is a no-op. pub async fn record(&self, exchange: RecordedExchange) -> Result<()> { @@ -117,7 +127,11 @@ impl Recorder { // sides observe insertions in the same sequence. let mut state = self.inner.state.write().await; - if let Some(storage) = &self.inner.storage { + if let Some(storage) = exchange + .upstream + .as_deref() + .and_then(|name| self.inner.storage.get(name)) + { storage.append(&exchange).await?; } @@ -141,12 +155,12 @@ impl Recorder { Ok(()) } - /// Make every previously appended exchange durable. Delegates to - /// `SnapshotStorage::flush`. Called explicitly by callers and by - /// `ClusterHandle::shutdown` to fence batching backends (object - /// store) before tear-down. + /// Make every previously appended exchange durable across all + /// registered media. Delegates to `SnapshotStorage::flush`. Called + /// explicitly by callers and by `ClusterHandle::shutdown` to fence + /// batching backends (object store) before tear-down. pub async fn flush(&self) -> Result<()> { - if let Some(storage) = &self.inner.storage { + for storage in self.inner.storage.values() { storage.flush().await?; } Ok(()) @@ -316,7 +330,10 @@ mod tests { let path = dir.path().join("trace.ndjson"); let storage: SharedStorage = Arc::new(crate::jsonl::JsonlStorage::open(&path).await.unwrap()); - let recorder = Recorder::with_storage(RecordingConfig::in_memory(10), Some(storage)); + // `make_exchange` stamps the upstream name "api"; route that name to + // the JSONL medium so the exchanges land on disk. + let routes = HashMap::from([("api".to_owned(), storage)]); + let recorder = Recorder::with_routes(RecordingConfig::in_memory(10), routes); recorder .record(make_exchange("/a", b"hello")) .await diff --git a/crates/partly-proxy-lib/src/replay.rs b/crates/partly-proxy-lib/src/replay.rs index 064311e..cda1a35 100644 --- a/crates/partly-proxy-lib/src/replay.rs +++ b/crates/partly-proxy-lib/src/replay.rs @@ -28,7 +28,8 @@ 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, SnapshotStorage, hash::sha256_hex, + ExchangeOutcome, RecordedExchange, RecordedRequest, Result, SharedStorage, SnapshotStorage, + hash::sha256_hex, }; use crate::{ @@ -69,6 +70,76 @@ impl std::fmt::Debug for MatchStrategy { } } +/// 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 { + strategy: MatchStrategy, + 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, strategy: MatchStrategy) -> 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 { + Self { + strategy, + 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(), self.strategy).await?; + Ok((replay, Some(storage))) + } + SnapshotsSource::InMemory(exchanges) => { + Ok((ReplaySource::new(exchanges, self.strategy), 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("strategy", &self.strategy) + .field("source", &kind) + .finish() + } +} + /// Key used by `MethodUriAndBodyHash`: (method, path+query, body sha-256 hex). type IndexKey = (String, String, String); @@ -478,9 +549,12 @@ mod tests { .await .expect("open jsonl"), ); - let recorder = crate::recorder::Recorder::with_storage( + // 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)]); + let recorder = crate::recorder::Recorder::with_routes( crate::config::RecordingConfig::in_memory(100), - Some(storage), + routes, ); for n in 0..3 { let req = RecordedRequest::from_parts( diff --git a/crates/partly-proxy-lib/tests/record.rs b/crates/partly-proxy-lib/tests/record.rs index 1cc613a..28cb13a 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, ProxyClusterBuilder, ProxyConfig, RecordedExchange, - RecordingConfig, UpstreamTarget, + ClusterHandle, ExchangeOutcome, MatchStrategy, ProxyClusterBuilder, ProxyConfig, + RecordedExchange, RecordingConfig, Snapshots, UpstreamTarget, }; use tokio::task::JoinHandle; @@ -146,8 +146,15 @@ async fn ndjson_persist_file_is_replayable() { ); let cluster = ProxyClusterBuilder::new() .recording(RecordingConfig::in_memory(100)) - .storage(storage) - .add_upstream("upstream", cfg) + .add_upstream_with( + "upstream", + cfg, + Vec::new(), + Some(Snapshots::from_storage( + storage, + MatchStrategy::MethodUriAndBodyHash, + )), + ) .run() .await .unwrap(); @@ -180,8 +187,8 @@ 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 -/// `ProxyClusterBuilder::storage(...)` plumbing. +/// keeps the exchanges in memory. Used to verify the per-upstream +/// `Snapshots::from_storage(...)` plumbing. #[derive(Debug, Default)] struct TrackingStorage { appended: tokio::sync::Mutex>, @@ -215,7 +222,7 @@ impl partly_proxy_lib::SnapshotStorage for TrackingStorage { } #[tokio::test] -async fn custom_storage_via_builder_storage_setter() { +async fn custom_storage_via_per_upstream_snapshots() { let (echo_addr, _t) = spawn_echo().await; let storage = std::sync::Arc::new(TrackingStorage::default()); let cfg = ProxyConfig::http( @@ -226,8 +233,15 @@ async fn custom_storage_via_builder_storage_setter() { ); let cluster = ProxyClusterBuilder::new() .recording(RecordingConfig::in_memory(100)) - .storage(storage.clone()) - .add_upstream("upstream", cfg) + .add_upstream_with( + "upstream", + cfg, + Vec::new(), + Some(Snapshots::from_storage( + storage.clone(), + MatchStrategy::MethodUriAndBodyHash, + )), + ) .run() .await .unwrap(); diff --git a/crates/partly-proxy-lib/tests/replay.rs b/crates/partly-proxy-lib/tests/replay.rs index b3f5abf..ab5cdc4 100644 --- a/crates/partly-proxy-lib/tests/replay.rs +++ b/crates/partly-proxy-lib/tests/replay.rs @@ -13,8 +13,8 @@ use partly_proxy_echo as echo; use partly_proxy_lib::{ Command, ExchangeOutcome, MatchStrategy, Mode, Next, ProxyClusterBuilder, ProxyConfig, ProxyMiddleware, ProxyRequest, ProxyResponse, RecordedExchange, RecordedRequest, - RecordedResponse, RecordingConfig, ReplaySource, RequestContext, RequestMatcher, - ResponseSource, Result as ProxyResult, SharedMiddleware, StubbedResponse, UpstreamTarget, + RecordedResponse, RecordingConfig, RequestContext, RequestMatcher, ResponseSource, + Result as ProxyResult, SharedMiddleware, Snapshots, StubbedResponse, UpstreamTarget, }; use tokio::task::JoinHandle; @@ -80,7 +80,7 @@ async fn replay_hit_serves_recorded_response_without_touching_upstream() { a }; - let replay = ReplaySource::new( + let replay = Snapshots::in_memory( vec![make_recorded( Method::GET, "/health", @@ -125,7 +125,7 @@ async fn replay_mode_miss_returns_503_without_touching_upstream() { drop(l); a }; - let replay = ReplaySource::new( + let replay = Snapshots::in_memory( vec![make_recorded(Method::GET, "/health", b"", 200, b"replayed")], MatchStrategy::MethodUriAndBodyHash, ); @@ -174,7 +174,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 = ReplaySource::new( + let replay = Snapshots::in_memory( vec![make_recorded(Method::GET, "/health", b"", 200, b"replayed")], MatchStrategy::MethodUriAndBodyHash, ); @@ -219,7 +219,7 @@ async fn stub_takes_priority_over_replay() { drop(l); a }; - let replay = ReplaySource::new( + let replay = Snapshots::in_memory( vec![make_recorded(Method::GET, "/x", b"", 200, b"from-replay")], MatchStrategy::MethodUriAndBodyHash, ); @@ -307,7 +307,7 @@ async fn replay_lookup_uses_redact_request_for_snapshot() { a }; let snapshot = make_recorded(Method::GET, "/secure", b"", 200, b"ok"); - let replay = ReplaySource::new(vec![snapshot], MatchStrategy::MethodUriAndBodyHash); + let replay = Snapshots::in_memory(vec![snapshot], MatchStrategy::MethodUriAndBodyHash); let cluster = ProxyClusterBuilder::new() .add_upstream_with( "api", @@ -343,7 +343,7 @@ async fn replay_records_served_exchanges_when_recording_enabled() { drop(l); a }; - let replay = ReplaySource::new( + let replay = Snapshots::in_memory( vec![make_recorded(Method::GET, "/x", b"", 200, b"replay-body")], MatchStrategy::MethodUriAndBodyHash, ); @@ -471,7 +471,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 = ReplaySource::new( + let replay = Snapshots::in_memory( vec![make_recorded(Method::GET, "/x", b"", 200, b"replayed")], MatchStrategy::MethodUriAndBodyHash, ); @@ -502,7 +502,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 = ReplaySource::new( + let replay = Snapshots::in_memory( vec![make_recorded(Method::GET, "/x", b"", 200, b"replayed")], MatchStrategy::MethodUriAndBodyHash, );