diff --git a/Cargo.toml b/Cargo.toml index b25c627a..22649cde 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -84,6 +84,12 @@ tower = { version = "0.5", features = ["util"] } # disk mid-link more than once, and the failure reads as the linker crashing # rather than as a full disk. # +# The saving arrives on the next `cargo clean` and not before. Cargo leaves +# artefacts it has superseded, so a target directory that has been built in for +# a while holds both shapes and briefly grows. On the worktree this was written +# in, one that had reached 31 GB rebuilt from clean — the whole workspace, +# every test binary included — into 3.4 GB. +# # Set `debug = true` here, temporarily, if you do need a stepping debugger. [profile.dev] debug = "line-tables-only" diff --git a/crates/vibescrobble-index/src/bin/vibescrobble-replay.rs b/crates/vibescrobble-index/src/bin/vibescrobble-replay.rs new file mode 100644 index 00000000..9274c029 --- /dev/null +++ b/crates/vibescrobble-index/src/bin/vibescrobble-replay.rs @@ -0,0 +1,189 @@ +//! Record a firehose, and play it back with nothing else running. +//! +//! ```text +//! vibescrobble-replay record --pds http://localhost:3000 --out session.jsonl +//! vibescrobble-replay serve --file session.jsonl --listen 127.0.0.1:3010 +//! vibescrobble-index --pds http://127.0.0.1:3010 +//! ``` +//! +//! Why it exists, and what it is not, is in +//! [`vibescrobble_index::replay`](../vibescrobble_index/replay/index.html). + +#![forbid(unsafe_code)] + +use std::io::Write; +use std::path::PathBuf; + +use tokio::sync::mpsc; +use vibescrobble_index::follow::{read_frames, Frame}; +use vibescrobble_index::replay::{router, Pace, Recorded, Recording}; +use vibescrobble_index::ServerRef; + +/// What `--help` prints. +const USAGE: &str = "\ +vibescrobble-replay — record a firehose, and play it back + + vibescrobble-replay record --out [--pds ] [--frames ] + vibescrobble-replay serve --file [--listen ] [--rate ] + +Options + --pds Server to record from (default http://localhost:3000) + --out Where to write the recording + --frames Stop after this many frames (default: until interrupted) + --file Recording to play back + --listen Address to serve it on (default 127.0.0.1:3010) + --rate Frames per second, or 0 for as fast as it will go + (default 5) + +A recording is one JSON object per line, so trimming it to the frames that +reproduce something is a text editor and nothing else. +"; + +/// Where to record from when nothing says otherwise. +const DEFAULT_PDS: &str = "http://localhost:3000"; + +/// Where to serve a playback when nothing says otherwise. +/// +/// Not a port any other component uses, so a replay can run beside a real +/// stack rather than instead of it. +const DEFAULT_LISTEN: &str = "127.0.0.1:3010"; + +/// Frames per second a playback hands over when nothing says otherwise. +/// +/// Slow enough that a canvas watching it looks like something is happening +/// rather than flashing once and stopping. +const DEFAULT_RATE: f64 = 5.0; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let args: Vec = std::env::args().skip(1).collect(); + if args.is_empty() || args.iter().any(|arg| arg == "--help" || arg == "-h") { + print!("{USAGE}"); + return Ok(()); + } + + vibescrobble_index::init_tracing(); + + match args[0].as_str() { + "record" => record(&args[1..]).await, + "serve" => serve(&args[1..]).await, + other => { + eprintln!("vibescrobble-replay: no such command: {other}\n\n{USAGE}"); + std::process::exit(2); + } + } +} + +/// Subscribes to a firehose and writes what it sees. +/// +/// Appends as it goes rather than buffering to the end, so that a recording +/// interrupted with Ctrl-C is a recording rather than nothing. That is the +/// ordinary way to end one: a developer records until the thing they are +/// chasing happens. +async fn record(args: &[String]) -> Result<(), Box> { + let mut pds = DEFAULT_PDS.to_owned(); + let mut out: Option = None; + let mut limit: Option = None; + + let mut rest = args.iter(); + while let Some(arg) = rest.next() { + match arg.as_str() { + "--pds" => pds = rest.next().ok_or("--pds wants a url")?.clone(), + "--out" => out = Some(PathBuf::from(rest.next().ok_or("--out wants a path")?)), + "--frames" => { + limit = Some(rest.next().ok_or("--frames wants a count")?.parse()?); + } + other => return Err(format!("unexpected argument: {other}").into()), + } + } + let out = out.ok_or("record needs --out")?; + + let server = ServerRef::parse(&pds)?; + let file = std::fs::File::create(&out)?; + let mut file = std::io::BufWriter::new(file); + + let (sender, mut frames) = mpsc::channel::(vibescrobble_index::follow::FRAME_BUFFER); + let url = format!("{}/firehose", server.base_url()); + tracing::info!(%url, path = %out.display(), "recording"); + + let http = reqwest::Client::new(); + let reading = tokio::spawn(async move { + if let Err(err) = read_frames(&http, &url, &sender).await { + tracing::warn!(error = %err, "the stream ended"); + } + }); + + let mut written = 0usize; + while let Some(frame) = frames.recv().await { + let recorded = Recorded::from(frame); + writeln!(file, "{}", serde_json::to_string(&recorded)?)?; + // Flushed per frame: the ordinary end of a recording is Ctrl-C, and a + // buffer that had not reached the disk would take the interesting part + // with it. + file.flush()?; + written += 1; + if limit.is_some_and(|limit| written >= limit) { + break; + } + } + reading.abort(); + tracing::info!(frames = written, path = %out.display(), "recorded"); + Ok(()) +} + +/// Serves a recording as a firehose. +async fn serve(args: &[String]) -> Result<(), Box> { + let mut file: Option = None; + let mut listen: Option = None; + let mut rate = DEFAULT_RATE; + + let mut rest = args.iter(); + while let Some(arg) = rest.next() { + match arg.as_str() { + "--file" => file = Some(PathBuf::from(rest.next().ok_or("--file wants a path")?)), + "--listen" => listen = Some(rest.next().ok_or("--listen wants an address")?.clone()), + "--rate" => rate = rest.next().ok_or("--rate wants a number")?.parse()?, + other => return Err(format!("unexpected argument: {other}").into()), + } + } + let file = file.ok_or("serve needs --file")?; + + let recording = Recording::parse(&std::fs::read_to_string(&file)?)?; + let frames = recording.frames.len(); + let accounts = recording.accounts().len(); + // Where the recording came from, unless told otherwise: an index checks + // that a server's zone sits under the host it was pointed at, so a + // playback on any other port is refused however correct its frames are. + let listen = listen + .or_else(|| recording.listen_hint()) + .unwrap_or_else(|| DEFAULT_LISTEN.to_owned()); + let pace = Pace { + per_second: (rate > 0.0).then_some(rate), + }; + + let listener = tokio::net::TcpListener::bind(&listen).await?; + let addr = listener.local_addr()?; + println!( + "replaying {} — {frames} frame(s), {accounts} account(s)", + file.display() + ); + println!(" firehose http://{addr}/firehose"); + println!( + " follow it vibescrobble-index --pds http://localhost:{}", + addr.port() + ); + if let Some(zone) = recording.zone() { + println!(" minted under {zone}"); + } + println!( + " pace {}", + match pace.per_second { + Some(rate) => format!("{rate} frame(s) per second"), + None => "as fast as the socket takes them".to_owned(), + } + ); + println!(); + + axum::serve(listener, router(recording, pace)).await?; + Ok(()) +} diff --git a/crates/vibescrobble-index/src/lib.rs b/crates/vibescrobble-index/src/lib.rs index ac25cddb..a0f30515 100644 --- a/crates/vibescrobble-index/src/lib.rs +++ b/crates/vibescrobble-index/src/lib.rs @@ -108,6 +108,7 @@ pub mod follow; pub mod handle; pub mod http; pub mod profile; +pub mod replay; pub mod report; pub mod resolve; pub mod server; diff --git a/crates/vibescrobble-index/src/replay.rs b/crates/vibescrobble-index/src/replay.rs new file mode 100644 index 00000000..32dc2425 --- /dev/null +++ b/crates/vibescrobble-index/src/replay.rs @@ -0,0 +1,488 @@ +//! A recorded firehose, and a server that plays it back. +//! +//! # What this is for +//! +//! Working on the index, the query service or the canvas needs a personal data +//! server and a swarm in front of it: three terminals and a population, before +//! anything downstream has a byte to look at. The canvas already ships a mock +//! for exactly that reason. This is the same idea one layer down — a recording +//! of a real stream, played back with nothing else running. +//! +//! What it buys over keeping a `--data` directory is that a recording is a +//! *sequence*, not a state. The same frames arrive in the same order every +//! time, as fast or as slowly as you ask, so a read-side bug is reproduced +//! rather than re-provoked, and a recording can be trimmed by hand to the +//! three frames that cause it. +//! +//! # The format is one frame per line +//! +//! JSON Lines, each carrying the server-sent event's name and its data. Not a +//! clever container: the point is that a person can open a recording, delete +//! the frames that are not the bug, and hand the file to somebody else. +//! +//! # What it does not pretend to be +//! +//! A personal data server. It answers the firehose and the account listing an +//! index asks for on every pass, and nothing else — no provisioning, no +//! repositories, no blobs. An index following it fills in what it can and +//! retries what it cannot, which is what it does against a real server that is +//! part way up. + +use std::sync::Arc; + +use axum::extract::{Query, State}; +use axum::response::sse::Sse; +use axum::response::IntoResponse; +use axum::routing::get; +use axum::Router; +use serde::{Deserialize, Serialize}; + +use crate::follow::Frame; + +/// One frame as a recording holds it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Recorded { + /// The `event:` name the frame carried, if any. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event: Option, + /// Everything from the frame's `data:` lines. + pub data: String, +} + +impl From for Recorded { + fn from(frame: Frame) -> Self { + Self { + event: frame.event, + data: frame.data, + } + } +} + +impl Recorded { + /// The sequence number this frame carries, if it is a commit. + /// + /// Read out of the data rather than stored beside it, so that a recording + /// somebody has edited by hand cannot disagree with itself. + pub fn seq(&self) -> Option { + serde_json::from_str::(&self.data) + .ok()? + .get("seq")? + .as_u64() + } + + /// The DID this frame is about, if it names one. + pub fn did(&self) -> Option { + serde_json::from_str::(&self.data) + .ok()? + .get("did")? + .as_str() + .map(str::to_owned) + } +} + +/// A recording, in the order it was captured. +#[derive(Debug, Clone, Default)] +pub struct Recording { + /// Every frame, in order. + pub frames: Vec, +} + +impl Recording { + /// Parses a recording, skipping blank lines and refusing bad ones by line. + /// + /// A line that will not parse is an error naming the line, because a + /// recording is a file people edit and the useful answer to a typo is + /// which line it is on. + pub fn parse(text: &str) -> Result { + let mut frames = Vec::new(); + for (number, line) in text.lines().enumerate() { + if line.trim().is_empty() { + continue; + } + let frame = serde_json::from_str(line).map_err(|source| RecordingError::Line { + line: number + 1, + source, + })?; + frames.push(frame); + } + Ok(Self { frames }) + } + + /// Renders a recording back to the format [`Recording::parse`] reads. + pub fn render(&self) -> String { + let mut out = String::new(); + for frame in &self.frames { + out.push_str(&serde_json::to_string(frame).expect("a frame serializes")); + out.push('\n'); + } + out + } + + /// Every account the recording mentions, in the order first seen. + /// + /// An index asks a server what accounts it has on every survey pass. A + /// recording knows only the accounts that wrote something, which is fewer + /// than a real server would list and is the honest answer: nothing here + /// saw the others. + pub fn accounts(&self) -> Vec { + let mut seen = Vec::new(); + for frame in &self.frames { + if let Some(did) = frame.did() { + if !seen.contains(&did) { + seen.push(did); + } + } + } + seen + } + + /// The zone the recorded accounts were minted under, if it can be told. + /// + /// An index refuses to survey a server whose `/health` does not name a + /// zone — that is how it tells a vibescrobble server from anything else + /// answering on a port — so a replay has to name one or it is a firehose + /// nobody will read past. + /// + /// It is derived rather than recorded: a `did:web` is the zone with a + /// label in front of it, so dropping the leftmost label of the first + /// account in the recording gives the zone that minted it. Derived, so a + /// recording somebody trimmed by hand cannot end up claiming a zone none + /// of its frames are from. + pub fn zone(&self) -> Option { + let (zone, _port) = self.zone_and_port()?; + Some(zone) + } + + /// The zone and, where the recording came from a development stack, its + /// port. + /// + /// They are separated because a server declares its zone *without* a port — + /// `agents.localhost`, not `agents.localhost:3400` — while the account + /// hostnames it mints carry one. A replay that declared the port would be + /// refused by the containment check for a reason that reads like a trust + /// failure and is a formatting difference. + fn zone_and_port(&self) -> Option<(String, Option)> { + let did = self.accounts().into_iter().next()?; + let host = did.strip_prefix("did:web:")?.replace("%3A", ":"); + let (_label, rest) = host.split_once('.')?; + Some(match rest.rsplit_once(':') { + Some((zone, port)) => (zone.to_owned(), port.parse().ok()), + None => (rest.to_owned(), None), + }) + .filter(|(zone, _)| !zone.is_empty()) + } + + /// The address a playback should listen on to be believed. + /// + /// An index does not take a server's word for the zone it mints under: it + /// checks that the zone sits under the host it was pointed at, which is + /// the containment rule the whole trust model rests on. A replay is + /// serving somebody else's recording, so it satisfies that rule only by + /// listening where the recording came from. + /// + /// Loopback and the recorded port, then — `agents.localhost:3400` becomes + /// `127.0.0.1:3400`, and an index pointed at `http://localhost:3400` + /// surveys it as it would the server that was recorded. `None` when the + /// recording names no port, which is a recording from a deployment rather + /// than from a development stack. + pub fn listen_hint(&self) -> Option { + let (_zone, port) = self.zone_and_port()?; + Some(format!("127.0.0.1:{}", port?)) + } + + /// The frames after `cursor`, or all of them when there is none. + /// + /// A cursor is `:` as the firehose defines it, and only the + /// sequence number is used: a recording is one instance by construction. + pub fn after(&self, cursor: Option<&str>) -> &[Recorded] { + let Some(seq) = cursor + .and_then(|cursor| cursor.rsplit(':').next()) + .and_then(|seq| seq.parse::().ok()) + else { + return &self.frames; + }; + let start = self + .frames + .iter() + .position(|frame| frame.seq().is_some_and(|found| found > seq)) + .unwrap_or(self.frames.len()); + &self.frames[start..] + } +} + +/// Why a recording could not be read. +#[derive(Debug, thiserror::Error)] +pub enum RecordingError { + /// One line is not a frame. + #[error("line {line} is not a recorded frame: {source}")] + Line { + /// Which line, counting from one. + line: usize, + /// What serde said. + source: serde_json::Error, + }, +} + +/// How fast a replay hands frames over. +#[derive(Debug, Clone, Copy)] +pub struct Pace { + /// Frames per second, or `None` for as fast as the socket takes them. + pub per_second: Option, +} + +impl Pace { + /// The gap between two frames, if there is one. + pub fn gap(self) -> Option { + let rate = self.per_second?; + (rate > 0.0).then(|| std::time::Duration::from_secs_f64(1.0 / rate)) + } +} + +/// What the replay server is serving. +#[derive(Clone)] +struct Playback { + recording: Arc, + pace: Pace, +} + +/// A router that serves `recording` as a firehose. +pub fn router(recording: Recording, pace: Pace) -> Router { + let state = Playback { + recording: Arc::new(recording), + pace, + }; + Router::new() + .route("/firehose", get(firehose)) + .route("/health", get(health)) + .route("/xrpc/zone.quernstone.listAgents", get(list_agents)) + .with_state(state) +} + +/// The recording, as a server-sent event stream. +async fn firehose( + State(playback): State, + Query(params): Query>, +) -> impl IntoResponse { + let cursor = params.get("cursor").cloned(); + let frames: Vec = playback.recording.after(cursor.as_deref()).to_vec(); + let gap = playback.pace.gap(); + + let stream = async_stream::stream(frames, gap); + Sse::new(stream) +} + +/// What this is and how much of it there is. +async fn health(State(playback): State) -> impl IntoResponse { + axum::Json(serde_json::json!({ + "status": "ok", + // The field an index looks for before it will read anything else here. + "zone": playback.recording.zone(), + // And the ones that say this is not the server it is pretending to be. + "replay": true, + "frames": playback.recording.frames.len(), + "accounts": playback.recording.accounts().len(), + })) +} + +/// The accounts the recording mentions, shaped as the real listing is. +async fn list_agents(State(playback): State) -> impl IntoResponse { + let agents: Vec = playback + .recording + .accounts() + .into_iter() + .map(|did| serde_json::json!({ "did": did })) + .collect(); + axum::Json(serde_json::json!({ "agents": agents })) +} + +/// The stream half, kept apart because it is the only part that is a `Stream`. +mod async_stream { + use std::convert::Infallible; + use std::future::Future; + use std::pin::Pin; + use std::task::{Context, Poll}; + use std::time::Duration; + + use axum::response::sse::Event; + use futures_core::Stream; + + use super::Recorded; + + /// A stream that hands over `frames`, waiting `gap` between them. + /// + /// Hand-rolled for the reason [`crate::follow::read_frames`] is: the crate + /// that would provide this also provides a great deal that nothing here + /// uses, and the state machine is one index and one sleep. + pub(super) fn stream( + frames: Vec, + gap: Option, + ) -> impl Stream> + Send + 'static { + Playing { + frames, + at: 0, + gap, + waiting: None, + } + } + + struct Playing { + frames: Vec, + at: usize, + gap: Option, + waiting: Option>>, + } + + impl Stream for Playing { + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); + if let Some(sleep) = this.waiting.as_mut() { + match sleep.as_mut().poll(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(()) => this.waiting = None, + } + } + let Some(frame) = this.frames.get(this.at) else { + // The end of a recording is the end of the stream, and a + // subscriber that reconnects gets it again from its cursor — + // which is what a real server does when it has nothing more. + return Poll::Ready(None); + }; + this.at += 1; + let mut event = Event::default().data(frame.data.clone()); + if let Some(name) = &frame.event { + event = event.event(name.clone()); + } + if let Some(gap) = this.gap { + this.waiting = Some(Box::pin(tokio::time::sleep(gap))); + } + Poll::Ready(Some(Ok(event))) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn recording() -> Recording { + Recording::parse( + r#"{"event":"info","data":"{\"instance\":\"abc\",\"oldest\":1}"} +{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:a\"}"} +{"event":"commit","data":"{\"seq\":2,\"did\":\"did:web:b\"}"} +{"event":"commit","data":"{\"seq\":3,\"did\":\"did:web:a\"}"} +"#, + ) + .expect("the fixture parses") + } + + #[test] + fn a_recording_round_trips_through_its_own_format() { + let first = recording(); + let again = Recording::parse(&first.render()).expect("re-read"); + assert_eq!(first.frames, again.frames); + } + + #[test] + fn a_blank_line_is_not_a_frame() { + let parsed = Recording::parse("\n\n{\"data\":\"{}\"}\n\n").expect("parses"); + assert_eq!(parsed.frames.len(), 1); + } + + #[test] + fn a_line_that_will_not_parse_says_which_line() { + let err = Recording::parse("{\"data\":\"{}\"}\nnot json\n").expect_err("refused"); + assert!(matches!(err, RecordingError::Line { line: 2, .. }), "{err}"); + } + + #[test] + fn a_cursor_hands_back_only_what_came_after_it() { + let recording = recording(); + let after = recording.after(Some("abc:1")); + assert_eq!(after.len(), 2); + assert_eq!(after[0].seq(), Some(2)); + } + + #[test] + fn no_cursor_is_the_whole_recording() { + assert_eq!(recording().after(None).len(), 4); + } + + #[test] + fn a_cursor_past_the_end_hands_back_nothing() { + assert!(recording().after(Some("abc:99")).is_empty()); + } + + #[test] + fn a_cursor_that_is_not_one_is_ignored_rather_than_refused() { + // A recording is a development tool and a mistyped cursor should show + // you the stream, not an error page. + assert_eq!(recording().after(Some("nonsense")).len(), 4); + } + + #[test] + fn the_zone_comes_off_the_accounts_rather_than_being_declared() { + let recording = Recording::parse( + r#"{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:marmot.agents.localhost%3A3400\"}"} +"#, + ) + .expect("parses"); + // Without the port: that is how a server declares its own zone, and + // an index compares the two. + assert_eq!(recording.zone().as_deref(), Some("agents.localhost")); + } + + #[test] + fn a_playback_listens_where_the_recording_came_from() { + // Not a convenience: an index checks that a server's zone sits under + // the host it was pointed at, so a playback anywhere else is refused + // however correct its frames are. + let recording = Recording::parse( + r#"{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:marmot.agents.localhost%3A3400\"}"} +"#, + ) + .expect("parses"); + assert_eq!(recording.listen_hint().as_deref(), Some("127.0.0.1:3400")); + } + + #[test] + fn a_recording_with_no_port_suggests_no_address() { + let recording = Recording::parse( + r#"{"event":"commit","data":"{\"seq\":1,\"did\":\"did:web:kestrel.agents.example\"}"} +"#, + ) + .expect("parses"); + assert_eq!(recording.zone().as_deref(), Some("agents.example")); + assert_eq!(recording.listen_hint(), None); + } + + #[test] + fn a_recording_of_nothing_claims_no_zone() { + // Better than guessing one: an index that is told a zone reads on, and + // reading on into an empty recording is not an improvement. + assert_eq!(Recording::default().zone(), None); + } + + #[test] + fn the_accounts_are_the_ones_that_said_something() { + assert_eq!(recording().accounts(), vec!["did:web:a", "did:web:b"]); + } + + #[test] + fn a_pace_of_none_waits_for_nothing() { + assert!(Pace { per_second: None }.gap().is_none()); + assert!(Pace { + per_second: Some(0.0) + } + .gap() + .is_none()); + assert_eq!( + Pace { + per_second: Some(4.0) + } + .gap(), + Some(std::time::Duration::from_millis(250)) + ); + } +} diff --git a/docs/running-locally.md b/docs/running-locally.md index ced02f0f..9c6a4e1c 100644 --- a/docs/running-locally.md +++ b/docs/running-locally.md @@ -310,6 +310,30 @@ The groups on their own: curl -s localhost:3003/work # or 3002, the index that computed them ``` +## Working on the read side without the write side + +The index, the query service and the canvas need a personal data server and a +swarm in front of them: three terminals and a population before anything +downstream has a byte to look at. A recording removes both. + +```sh +./target/debug/vibescrobble-replay record --out session.jsonl # while the stack runs +./target/debug/vibescrobble-replay serve --file session.jsonl # later, alone +./scripts/dev-index.sh # pointed at the replay +``` + +`serve` listens on the port the recording came from, because an index checks +that a server's zone sits under the host it was reached at and a playback +anywhere else is refused however correct its frames are. It answers the +firehose and the account listing, which is what an index reads on every pass, +and nothing else. + +A recording is one JSON object per line, so trimming it to the frames that +reproduce something is a text editor. What it buys over keeping a `--data` +directory is that a recording is a *sequence* rather than a state: the same +frames in the same order every time, as fast or as slowly as `--rate` says, so +a read-side bug is reproduced rather than re-provoked. + ## Asking the query service things The index serves the whole view and a stream of what changes, and nothing diff --git a/plan/local-dev.md b/plan/local-dev.md index 639202a9..f22b9c1c 100644 --- a/plan/local-dev.md +++ b/plan/local-dev.md @@ -38,15 +38,37 @@ script claims. So the compile is not where the time goes. What the build costs is *disk*, and that is what fills a machine: see the dev profile under Done. -- [ ] **Capture and replay a firehose.** Working on the index, the query - service or the canvas needs a personal data server and a swarm, and the - canvas already has a mock for exactly that reason. A recording of a real - stream, replayed into an index with nothing else running, would extend - that one layer down and make a read-side bug reproducible rather than - re-provoked. ## Done +- [x] **A test that failed about once in twelve suite runs, found and fixed.** + `a_proof_missing_a_node_does_not_verify` asserted that a record's path + through the Merkle search tree was more than one node deep. A record key + is a TID and a tree's shape comes from the hashes of its keys, so that + depth varies run to run: it held nineteen times in twenty and failed the + whole suite the other time. Nothing in the test needed the depth. It was + measured at two failures in twenty-five runs before, and none in thirty + after. + + It took three sightings to catch, because the first two were seen through + a `cargo test` whose output had been sent to `/dev/null` — which is its + own lesson about checking exit statuses without keeping what produced + them. +- [x] **Capture and replay a firehose.** `vibescrobble-replay record` writes a + stream to a file, one JSON object per line so that trimming it to the + frames that reproduce something is a text editor; `serve` plays it back. + Driven from a recording alone, with no personal data server and no swarm + running, an index and a query service reached fifteen agents, twenty-nine + scrobbles and two work groups. + + Two things had to be right and were found by running it rather than by + reading. The playback declares the zone *without* a port, because that is + how a server declares its own and an index compares the two — with the + port it was refused for what reads like a trust failure and is a + formatting difference. And it listens on the port the recording came + from by default, because the containment check compares the declared zone + against the host it was reached at, so a playback anywhere else is + refused however correct its frames are. - [x] **Watch modes, and the one component that refuses.** `--watch` on any `dev-*.sh` reruns the whole script — rebuild, that half's tests, banner and all — whenever anything under `crates/` moves. It reruns the script @@ -95,6 +117,12 @@ that is what fills a machine: see the dev profile under Done. 43 MB of binary, 59.6 s to 47.2 s of build. The workspace gains more than that share, because the facade crate's test binaries link every crate in it and there is one per suite. + + The saving arrives on the next `cargo clean`. Cargo keeps artefacts it + has superseded, so a directory that has been built in for a while holds + both shapes and briefly grows; the worktree this was written in had + reached 31 GB and rebuilt from clean, whole workspace and every test + binary, into 3.4 GB. - [x] **Simulate failure, not only the happy path.** A server that goes away mid-stream is now a test: `crates/vibescrobble/tests/firehose.rs` cuts a subscriber's connection, keeps writing records it cannot see, and