diff --git a/Cargo.lock b/Cargo.lock index 4afc51b3..5eaff92c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -883,6 +883,7 @@ dependencies = [ "clap", "coppa-audio", "ctrlc", + "hound", "manta-decode", "manta-engine", "manta-input", diff --git a/README.md b/README.md index 286978ab..ada3fc03 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,28 @@ cargo build --release -p manta-cli manta gen v1 --out /tmp/v1 manta decode /tmp/v1/v1.wav +# See a real spot on the DX cluster telnet port -- no radio required. +# --realtime paces the 120 s recording in wall-clock time (instead of +# draining it in ~3 s) so a client has time to connect; the spot itself +# appears about 12 s in. +cat > server.toml <<'TOML' +[server] +station_callsign = "N0CALL" +# Loopback only: without this, bind_addr defaults to 0.0.0.0 and this demo +# would publish the telnet, JSON/WebSocket and metrics ports on every +# interface -- see docs/RUNBOOKS/network-exposure.md before dropping it. +bind_addr = "127.0.0.1" +TOML +manta listen --source /tmp/v1/v1.wav --server-config server.toml --realtime +# then, in another terminal, any time in the next two minutes: +telnet localhost 7300 +# at the "login:" prompt, type any callsign and press Enter -- this is a +# real login, not optional, and the server only starts streaming spots to +# you once it lands +# then, at the "de N0CALL-# >" prompt: type sh/dx to see the spot, or just +# wait -- if you logged in within the first ~12 s, the spot arrives on its +# own once decoded, with no further command needed + # Copy live CW from a public KiwiSDR on 40 m manta listen --kiwi-host kiwi.example.org --kiwi-freq 7030000 @@ -111,7 +133,7 @@ manta listen --json --kiwi-host kiwi.example.org --kiwi-freq 7030000 | Source | How | Status | | --- | --- | --- | -| IQ / audio WAV file | `decode`, `listen --source` | Working | +| IQ / audio WAV file | `decode`, `listen --source` | Working -- `listen --source` takes a 2-channel IQ WAV (what `decode`/`gen` use, sidecar-aware; any channelizer rate other than 48 kHz, or 48 kHz with a `.json` sidecar) or a 48 kHz mono/stereo rig-audio WAV (48 kHz stereo with no sidecar downmixes like mono); add `--realtime`/`--loop` to pace or repeat file replay (`--loop` requires `--realtime` when `--server-config` is given, so looped spots aren't published with runaway future timestamps). A "channelizer rate" is any `fs` where `fs / 93.75` is a power of two -- 12/24/48/96/192/384 kHz and so on, not 44.1 or 100 kHz | | Sound card (rig audio passband) | `listen --device` | Working, 48 kHz input only | | KiwiSDR over the network | `listen --kiwi-host` | Working | | RTL-SDR, Airspy, SDRplay, HackRF, and anything else SoapySDR drives | `listen --soapy-driver`, feature `soapy` | Working, needs hardware soak | @@ -122,10 +144,11 @@ enforced by criterion benches. ## Outputs -- Decoded text or JSON Lines on stdout today. +- Decoded text or JSON Lines on stdout. - RBN-format `DX de` spots over the DX cluster telnet protocol (port 7300) - and a JSON Lines / WebSocket stream (port 7301): in progress, see - [ROADMAP.md](ROADMAP.md) milestone M3. + and a JSON Lines / WebSocket stream (port 7301), started with + `listen --server-config`; see the Quickstart above for a hardware-free + demo, and [ROADMAP.md](ROADMAP.md) milestone M3 for acceptance status. The decode path is deterministic: the same file in produces byte-identical spot logs out. That is a hard requirement, and CI enforces it with golden diff --git a/ROADMAP.md b/ROADMAP.md index b0c8e3c9..9a8e5f8b 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -99,7 +99,10 @@ metrics endpoint, spot JSON Schema contributed to `dispensa`. **Accept when:** - A stock DX cluster client (e.g. `telnet`, N1MM) connects, logs in, and - receives well-formed RBN-format spots. + receives well-formed RBN-format spots. **Met and automated, no hardware + needed**: `crates/manta-cli/tests/telnet_e2e.rs` drives this end-to-end + from `manta gen` output through `listen --source --realtime + --server-config` to a real telnet/JSON-Lines client (MAN-121). - **Parity benchmark**: on ≥ 2 h of recorded contest-weekend IQ, manta achieves ≥ 80 % recall of RBN's spots for the same slice with ≤ 5 % false (bogus-call) spots. Numbers published in the repo, whatever they are. diff --git a/crates/manta-cli/Cargo.toml b/crates/manta-cli/Cargo.toml index 5fed79d5..60d3c8af 100644 --- a/crates/manta-cli/Cargo.toml +++ b/crates/manta-cli/Cargo.toml @@ -34,6 +34,7 @@ tracing-subscriber = { workspace = true } [dev-dependencies] coppa-audio = { workspace = true } +hound = { workspace = true } serde_json = { workspace = true } manta-testkit = { workspace = true } tempfile = { workspace = true } diff --git a/crates/manta-cli/src/main.rs b/crates/manta-cli/src/main.rs index 0b1ab0e8..385e93ab 100644 --- a/crates/manta-cli/src/main.rs +++ b/crates/manta-cli/src/main.rs @@ -67,8 +67,12 @@ enum Command { /// Input device name substring (default input device if omitted). #[arg(long, conflicts_with = "source")] device: Option, - /// Replay a WAV file instead of a live device (paced by its own - /// sample rate via AudioIqSource; used for demos and testing). + /// Replay a WAV file instead of a live device: either a 2-channel + /// IQ WAV (what `manta gen`/`decode` use) at a channelizer rate -- + /// fs/93.75 must be a power of two, i.e. 12/24/48/96/192/384 kHz + /// and so on, NOT arbitrary rates like 44.1 or 100 kHz -- or a + /// 48 kHz mono rig-audio WAV. Drains as fast as it can be read + /// unless --realtime is given. #[arg(long, conflicts_with = "device")] source: Option, /// KiwiSDR receiver hostname. Requires --kiwi-freq. @@ -177,6 +181,30 @@ enum Command { /// this machine's copy of the file happens to say." #[arg(long, value_parser = parse_replay_epoch)] replay_epoch: Option, + /// Replay the --source file at wall-clock realtime instead of as + /// fast as it can be read, so a telnet/JSON client has time to + /// connect and observe a spot (MAN-121). Off by default: unpaced + /// replay is far faster and is what the test suite and `soak` rely + /// on. Output is byte-identical either way -- pacing only decides + /// when samples are delivered, never which. + #[arg(long, requires = "source")] + realtime: bool, + /// Restart the --source file at end-of-file instead of exiting, for + /// a demo left running (MAN-121). Note: the spot dedupe window + /// suppresses a repeat spot for the same callsign and frequency for + /// 10 minutes of recording time, so a short looped file yields + /// roughly one spot per 10 minutes, not one per pass. Combine with + /// --realtime for a live-paced demo. + /// + /// REQUIRES --realtime when --server-config is also given: an + /// unpaced loop never ends and advances its sample clock ~30-40x + /// faster than wall time, so `SpotBus` would keep publishing spots + /// stamped ever further into the future to real telnet/JSON + /// clients (round-14 review). Enforced below, in the Listen arm, + /// rather than by clap, so a plain (non-networked) unpaced loop -- + /// which publishes to nobody -- keeps working. + #[arg(long = "loop", requires = "source")] + loop_replay: bool, }, /// Run the listen pipeline for a fixed duration, checking for panics /// and unbounded memory growth (ROADMAP M1 accept criterion). @@ -381,7 +409,11 @@ fn open_source( fn open_audio_source(device: Option, source: Option) -> Result> { Ok(match source { - Some(path) => Box::new(manta_input::AudioIqSource::from_wav_file(&path)?), + // Dispatches on the WAV's own channel count (MAN-121): a 2-channel + // IQ WAV -- what `manta gen`/`decode` already use -- decodes + // directly, at its own native rate; anything else is still treated + // as a real rig-audio passband via AudioIqSource, unchanged. + Some(path) => manta_input::open_replay_wav(&path)?, None => Box::new(manta_input::AudioIqSource::from_device(device.as_deref())?), }) } @@ -532,9 +564,7 @@ fn session_nonce_for_replay_path(path: &std::path::Path) -> Result { let mut file = std::fs::File::open(path) .with_context(|| format!("opening {} to derive its replay identity", path.display()))?; - const OFFSET_BASIS: u64 = 0xcbf29ce484222325; - const PRIME: u64 = 0x0000_0100_0000_01b3; - let mut hash = OFFSET_BASIS; + let mut hash = FNV_OFFSET_BASIS; let mut buf = [0u8; 64 * 1024]; loop { let n = file @@ -545,12 +575,47 @@ fn session_nonce_for_replay_path(path: &std::path::Path) -> Result { } for &byte in &buf[..n] { hash ^= byte as u64; - hash = hash.wrapping_mul(PRIME); + hash = hash.wrapping_mul(FNV_PRIME); } } Ok(hash as u128) } +/// FNV-1a-64's published constants, shared by the content hash above and +/// the RF mix-in below so the whole replay identity is one continuous +/// FNV-1a stream over (recording bytes || effective RF frequency). +const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325; +const FNV_PRIME: u64 = 0x0000_0100_0000_01b3; + +/// The FULL replay session identity: the recording's content hash mixed +/// with the RF center frequency the session will actually publish spots +/// at. +/// +/// The WAV bytes alone are not the identity of a replay observation +/// (round-14 review). The RF frequency comes from OUTSIDE the WAV -- the +/// `.json` sidecar, or a `--dial-freq-hz` override -- so two copies +/// of one baseband recording tagged with different `center_freq_hz` +/// values, or one file replayed twice at two different dial frequencies, +/// hashed identically. Every other spot-`id` input (station, track id, +/// sample timestamp, callsign) is identical across those runs too, so the +/// ids collided outright while describing genuinely different RF +/// observations -- and `SpotMessage`'s own docs note that cqdx keys on +/// that id and may overwrite or drop a collision. +/// +/// Takes the EFFECTIVE frequency -- read off the fully-wrapped source, so +/// a `--dial-freq-hz` override is what gets mixed in, exactly as it is +/// what gets published -- and mixes its IEEE-754 bits, which keeps the +/// value deterministic across builds and machines just like the byte +/// stream it continues. +fn session_nonce_for_replay(path: &std::path::Path, center_freq_hz: f64) -> Result { + let mut hash = session_nonce_for_replay_path(path)? as u64; + for &byte in ¢er_freq_hz.to_bits().to_le_bytes() { + hash ^= byte as u64; + hash = hash.wrapping_mul(FNV_PRIME); + } + Ok(hash as u128) +} + /// Clap value parser for `--dial-freq-hz`: rejects non-finite (NaN/infinity) /// and non-positive values at CLI-parse time, before they're baked into /// `FixedCenterFreqSource` and silently propagate into malformed RBN/JSON @@ -1004,6 +1069,8 @@ fn main() -> Result<()> { server_config, dial_freq_hz, replay_epoch, + realtime, + loop_replay, } => { let is_file_replay = source.is_some(); // Captured before `open_source` consumes `source` below -- @@ -1017,7 +1084,21 @@ fn main() -> Result<()> { let has_hpsdr_source = hpsdr_host.is_some(); #[cfg(not(feature = "hpsdr"))] let has_hpsdr_source = false; - let has_rf_aware_source = kiwi_host.is_some() || has_soapy_source || has_hpsdr_source; + // A 2-channel IQ WAV whose `.json` sidecar declares a + // POSITIVE center frequency DOES report a real RF frequency -- + // probe cheaply (never fails, even for a missing path) so this + // stays ahead of all file I/O and a bad-flag error still beats + // a bad-file error. Deliberately the RF probe, not + // `replay_wav_has_iq_sidecar`'s format probe: a sidecar + // declaring `0.0` makes the file IQ but supplies no dial + // frequency, so the gate must still fire for it (MAN-121, + // round-2 review). + let file_declares_rf = source + .as_deref() + .and_then(manta_input::replay_wav_center_freq_hz) + .is_some(); + let has_rf_aware_source = + kiwi_host.is_some() || has_soapy_source || has_hpsdr_source || file_declares_rf; let source_name = if kiwi_host.is_some() { "kiwi" } else if has_soapy_source { @@ -1033,12 +1114,37 @@ fn main() -> Result<()> { if server_config.is_some() && !has_rf_aware_source && dial_freq_hz.is_none() { bail!( "--dial-freq-hz is required with --server-config when using a plain \ - audio device or --source WAV file -- neither reports a real RF \ - frequency (KiwiSDR/SoapySDR already know theirs from \ + audio device, a rig-audio --source WAV file (mono, or 2-channel at \ + 48 kHz with no .json sidecar), or an IQ WAV whose sidecar \ + declares center_freq_hz = 0 (baseband/unknown) -- none of these report \ + a real RF frequency (a 2-channel IQ WAV whose sidecar declares a \ + positive center_freq_hz does, as do KiwiSDR/SoapySDR via \ --kiwi-freq/--soapy-freq)" ); } + // An unpaced loop is unbounded in BOTH directions: it never + // reaches EOF, and its sample clock runs ~30-40x faster than + // wall time. `Dedupe` releases a repeat spot every 600 s of + // that simulated time, and `SpotBus::unix_ts_for` adds + // `sample_ts` to the fixed replay epoch -- so a networked + // unpaced loop publishes spots timestamped progressively + // further into the future, forever, to real clients that have + // no way to tell them from current observations. Paced + // (`--realtime`) looping keeps simulated and wall time + // together, so it stays truthful. Checked here, alongside the + // --dial-freq-hz gate and ahead of all file I/O, so this stays + // a flag error rather than a file error (round-14 review). + if loop_replay && server_config.is_some() && !realtime { + bail!( + "--loop with --server-config also requires --realtime -- an unpaced loop \ + advances its sample clock ~30-40x faster than wall time and never ends, so \ + telnet/JSON clients would receive spots timestamped progressively further \ + into the future. Add --realtime for a live-paced looping demo, or drop \ + --server-config to loop without publishing to clients" + ); + } + let kiwi = KiwiOpts { host: kiwi_host, port: kiwi_port, @@ -1057,6 +1163,16 @@ fn main() -> Result<()> { let hpsdr_source: Option> = None; let src = match hpsdr_source { Some(src) => src, + // `--loop` requires --source (clap `requires = "source"`), + // so `replay_path` is guaranteed Some here. Bypasses + // `open_source` entirely -- `LoopingWavSource` opens the + // file itself (and reopens it at EOF), so there's nothing + // for `open_source` to hand back. + None if loop_replay => Box::new(manta_input::LoopingWavSource::new( + replay_path + .clone() + .expect("--loop requires --source (clap-enforced)"), + )?) as Box, None => { #[cfg(feature = "soapy")] { @@ -1085,6 +1201,17 @@ fn main() -> Result<()> { }), None => src, }; + // Pacing wraps the OUTSIDE of everything above -- loop first + // (so pacing measures the continuous looped stream, not a + // clock that restarts each pass), then the dial-freq override, + // then realtime pacing (MAN-121 Decision 5). A pure sleep + // wrapper: never touches which samples are delivered, only + // when, so --realtime output is byte-identical to unpaced. + let src: Box = if realtime { + Box::new(manta_input::PacedSource::new(src)) + } else { + src + }; // Kept alive for the process lifetime: dropping it would stop // the spawned server tasks. `None` when --server-config wasn't @@ -1121,7 +1248,17 @@ fn main() -> Result<()> { // either). let epoch = resolve_epoch(replay_path.as_deref(), replay_epoch)?; let session_nonce: u128 = match &replay_path { - Some(replay_path) => session_nonce_for_replay_path(replay_path)?, + // The EFFECTIVE center frequency, read off the + // fully-wrapped `src` (so a --dial-freq-hz + // override counts), is part of the replay + // identity -- it comes from the sidecar or the + // flag, never from the WAV bytes, so hashing the + // bytes alone collided two different RF + // observations onto one spot id (round-14 + // review). + Some(replay_path) => { + session_nonce_for_replay(replay_path, src.center_freq_hz())? + } // Live session: `epoch` above is already SystemTime::now(). None => epoch .duration_since(std::time::SystemTime::UNIX_EPOCH) @@ -1569,6 +1706,39 @@ mod tests { ); } + // Round-14 review: the RF frequency a replay session publishes at comes + // from the sidecar or --dial-freq-hz, never from the WAV bytes, so the + // content hash alone let two genuinely different RF observations of the + // same recording collide on every JSON spot id. + #[test] + fn session_nonce_for_replay_separates_the_same_recording_at_different_frequencies() { + let f = write_temp_file(b"one recording, two dial frequencies"); + let a = session_nonce_for_replay(f.path(), 14_027_000.0).unwrap(); + let b = session_nonce_for_replay(f.path(), 7_027_000.0).unwrap(); + assert_ne!( + a, b, + "the same recording replayed at two RF frequencies must not share a session nonce" + ); + } + + #[test] + fn session_nonce_for_replay_is_deterministic_for_the_same_recording_and_frequency() { + let f = write_temp_file(b"one recording, one dial frequency"); + assert_eq!( + session_nonce_for_replay(f.path(), 14_027_000.0).unwrap(), + session_nonce_for_replay(f.path(), 14_027_000.0).unwrap() + ); + } + + #[test] + fn session_nonce_for_replay_still_separates_different_recordings_at_one_frequency() { + let a = session_nonce_for_replay(write_temp_file(b"contest weekend").path(), 14_027_000.0) + .unwrap(); + let b = session_nonce_for_replay(write_temp_file(b"quiet weeknight").path(), 14_027_000.0) + .unwrap(); + assert_ne!(a, b); + } + // MAN-32/MAN-42: start_spot_server spawns one RBN uplink task per // configured [[rbn_uplink]] target, only for those that are enabled. diff --git a/crates/manta-cli/tests/cli.rs b/crates/manta-cli/tests/cli.rs index 278d6031..0a698492 100644 --- a/crates/manta-cli/tests/cli.rs +++ b/crates/manta-cli/tests/cli.rs @@ -107,6 +107,270 @@ fn server_config_without_dial_freq_for_audio_source_is_a_clean_error() { assert!(stderr.contains("--dial-freq-hz"), "stderr: {stderr}"); } +/// MAN-121 Scenario 1: the README's own `manta gen v1 --out /tmp/v1` output +/// is 2-channel IQ WAV at 96 kHz, which `listen --source` used to hard-reject +/// with "AudioIqSource requires 48000 Hz, got 96000" -- it should decode +/// instead, exactly as `manta decode` already does. +#[test] +fn listen_source_accepts_the_iq_wav_that_gen_writes() { + let dir = tempfile::tempdir().unwrap(); + let spec = manta_testkit::vectors::VectorSpec { + duration_s: 15.0, + ..manta_testkit::vectors::v1() + }; + manta_testkit::vectors::write_fixture_set(&spec, dir.path()).unwrap(); + + let out = manta() + .args(["listen", "--source"]) + .arg(dir.path().join("v1.wav")) + .arg("--json") + .output() + .unwrap(); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!(!stderr.contains("requires 48000 Hz"), "stderr: {stderr}"); + assert!(out.status.success(), "stderr: {stderr}"); + let stdout = String::from_utf8_lossy(&out.stdout); + assert!( + stdout.lines().any(|l| l.contains("\"spot\"")), + "expected at least one spot line, got stdout: {stdout}" + ); +} + +/// MAN-121: a 2-channel IQ WAV with a sidecar already knows its own RF +/// center frequency (`WavIqSource` reads it), so `--server-config` must not +/// demand `--dial-freq-hz` for it the way it does for a plain audio source. +#[test] +fn listen_server_config_needs_no_dial_freq_for_a_sidecar_backed_iq_wav() { + let dir = tempfile::tempdir().unwrap(); + let spec = manta_testkit::vectors::VectorSpec { + duration_s: 15.0, + ..manta_testkit::vectors::v1() + }; + manta_testkit::vectors::write_fixture_set(&spec, dir.path()).unwrap(); + + let toml_path = dir.path().join("server.toml"); + std::fs::write( + &toml_path, + "[server]\nstation_callsign = \"W5AU\"\nbind_addr = \"127.0.0.1\"\ntelnet_port = 0\njson_port = 0\nmetrics_port = 0\n", + ) + .unwrap(); + + let out = manta() + .args(["listen", "--source"]) + .arg(dir.path().join("v1.wav")) + .arg("--server-config") + .arg(&toml_path) + .arg("--json") + .output() + .unwrap(); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!(!stderr.contains("--dial-freq-hz"), "stderr: {stderr}"); + assert!(out.status.success(), "stderr: {stderr}"); +} + +/// A plain mono 48 kHz audio WAV still has no real RF reference, so the +/// existing gate must still fire for it -- only sidecar-backed IQ WAVs are +/// exempted. +#[test] +fn listen_server_config_still_requires_dial_freq_for_a_mono_wav() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("rig.wav"); + let spec = hound::WavSpec { + channels: 1, + sample_rate: 48_000, + bits_per_sample: 32, + sample_format: hound::SampleFormat::Float, + }; + let mut w = hound::WavWriter::create(&wav, spec).unwrap(); + for _ in 0..480 { + w.write_sample(0.0f32).unwrap(); + } + w.finalize().unwrap(); + + let toml_path = dir.path().join("server.toml"); + std::fs::write( + &toml_path, + "[server]\nstation_callsign = \"W5AU\"\nbind_addr = \"127.0.0.1\"\ntelnet_port = 0\njson_port = 0\nmetrics_port = 0\n", + ) + .unwrap(); + + let out = manta() + .args(["listen", "--source"]) + .arg(&wav) + .arg("--server-config") + .arg(&toml_path) + .output() + .unwrap(); + assert!(!out.status.success()); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!(stderr.contains("--dial-freq-hz"), "stderr: {stderr}"); +} + +/// MAN-121 Decision 5: `--realtime` only decides WHEN samples are +/// delivered, never WHICH -- output must be byte-identical either way, the +/// determinism guarantee the broad review demands of any pacing change. +#[test] +fn realtime_replay_is_byte_identical_to_unpaced_replay() { + let dir = tempfile::tempdir().unwrap(); + let spec = manta_testkit::vectors::VectorSpec { + duration_s: 15.0, + ..manta_testkit::vectors::v1() + }; + manta_testkit::vectors::write_fixture_set(&spec, dir.path()).unwrap(); + let wav = dir.path().join("v1.wav"); + + let unpaced = manta() + .args(["listen", "--source"]) + .arg(&wav) + .arg("--json") + .output() + .unwrap(); + assert!(unpaced.status.success()); + + let paced = manta() + .args(["listen", "--source"]) + .arg(&wav) + .arg("--json") + .arg("--realtime") + .output() + .unwrap(); + assert!( + paced.status.success(), + "stderr: {}", + String::from_utf8_lossy(&paced.stderr) + ); + + assert_eq!(unpaced.stdout, paced.stdout); +} + +#[test] +fn realtime_requires_source() { + let out = manta().args(["listen", "--realtime"]).output().unwrap(); + assert_eq!( + out.status.code(), + Some(2), + "stderr: {}", + String::from_utf8_lossy(&out.stderr) + ); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!(stderr.contains("--source "), "stderr: {stderr}"); +} + +/// MAN-121 Decision 6: `--loop` restarts the file at EOF, so a demo can be +/// left running past a single pass's worth of events. +#[test] +fn loop_replay_keeps_producing_events_past_the_end_of_the_file() { + let dir = tempfile::tempdir().unwrap(); + let spec = manta_testkit::vectors::VectorSpec { + duration_s: 5.0, + ..manta_testkit::vectors::v1() + }; + manta_testkit::vectors::write_fixture_set(&spec, dir.path()).unwrap(); + let wav = dir.path().join("v1.wav"); + + // One unpaced pass, for a baseline event count. + let single_pass = manta() + .args(["listen", "--source"]) + .arg(&wav) + .arg("--json") + .output() + .unwrap(); + assert!(single_pass.status.success()); + let single_pass_lines = String::from_utf8_lossy(&single_pass.stdout).lines().count(); + + // Looped: kill it after a wall-clock budget comfortably exceeding one + // unpaced pass, so it must have wrapped at least once. + let mut child = manta() + .args(["listen", "--source"]) + .arg(&wav) + .arg("--json") + .arg("--loop") + .stdout(std::process::Stdio::piped()) + .spawn() + .unwrap(); + std::thread::sleep(std::time::Duration::from_secs(3)); + let _ = child.kill(); + let out = child.wait_with_output().unwrap(); + let looped_lines = String::from_utf8_lossy(&out.stdout).lines().count(); + + assert!( + looped_lines > single_pass_lines, + "expected --loop to produce more events than a single pass \ + ({single_pass_lines}), got {looped_lines}" + ); +} + +#[test] +fn loop_requires_source() { + let out = manta().args(["listen", "--loop"]).output().unwrap(); + assert_eq!( + out.status.code(), + Some(2), + "stderr: {}", + String::from_utf8_lossy(&out.stderr) + ); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!(stderr.contains("--source "), "stderr: {stderr}"); +} + +/// Round-14 review: an unpaced `--loop` never ends and advances its sample +/// clock ~30-40x faster than wall time, so a networked one would publish +/// spots timestamped ever further into the future to real clients. Like the +/// --dial-freq-hz gate, it is a flag error checked ahead of all file I/O, +/// so nonexistent paths still provoke exactly this message. +#[test] +fn loop_with_server_config_but_no_realtime_is_a_clean_error() { + let out = manta() + .args([ + "listen", + "--source", + "/nonexistent.wav", + "--server-config", + "/nonexistent.toml", + "--dial-freq-hz", + "14027000", + "--loop", + ]) + .output() + .unwrap(); + assert!( + !out.status.success(), + "expected a clean failure for a networked unpaced loop" + ); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!(stderr.contains("--realtime"), "stderr: {stderr}"); +} + +/// The same combination WITH --realtime passes the flag gate -- it must +/// fail on the missing file instead, proving the gate above is about +/// pacing and not about `--loop` plus `--server-config` as such. +#[test] +fn loop_with_server_config_and_realtime_passes_the_flag_gate() { + let out = manta() + .args([ + "listen", + "--source", + "/nonexistent.wav", + "--server-config", + "/nonexistent.toml", + "--dial-freq-hz", + "14027000", + "--loop", + "--realtime", + ]) + .output() + .unwrap(); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!( + !stderr.contains("also requires --realtime"), + "the pacing gate must not fire when --realtime is given: {stderr}" + ); + assert!( + stderr.contains("/nonexistent.wav"), + "expected the missing-file error instead: {stderr}" + ); +} + #[test] fn dial_freq_hz_rejects_non_finite_and_non_positive_values() { for bad in ["nan", "inf", "-inf", "0", "-14027000"] { diff --git a/crates/manta-cli/tests/telnet_e2e.rs b/crates/manta-cli/tests/telnet_e2e.rs new file mode 100644 index 00000000..0e29b4ec --- /dev/null +++ b/crates/manta-cli/tests/telnet_e2e.rs @@ -0,0 +1,269 @@ +//! ROADMAP M3 acceptance gate, automated: "a stock DX cluster client +//! connects, logs in, and receives well-formed RBN-format spots" -- driven +//! entirely from `manta gen`-shaped output, with no SDR and no hand-built +//! recording (MAN-121). + +use std::io::{BufRead, BufReader, Write}; +use std::net::{TcpListener, TcpStream}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +fn manta() -> Command { + Command::new(env!("CARGO_BIN_EXE_manta")) +} + +/// Ephemeral ports HELD by their own listeners until the child is about to +/// start. There is no way to learn a port the CLI assigned itself -- +/// `start_spot_server` binds from the TOML, not from an OS-assigned `:0` +/// the caller could read back -- so the ports must be chosen before the +/// child starts. +/// +/// Reserving all of them at once and releasing them together is what makes +/// the three DISTINCT (round-14 review): binding and immediately releasing +/// one port at a time let the OS hand the very same port back for the next +/// call, and `cargo test --workspace` runs these two tests in parallel with +/// each other and with the rest of the suite. A duplicate port makes one of +/// `start_spot_server`'s three required binds fail, which the connect +/// retry loop below cannot repair -- it just spends the whole 40 s deadline +/// and reports a timeout, i.e. a flake in a required acceptance test. +/// +/// The residual bind-then-release window (release -> child bind) is +/// unavoidable without a `:0`-and-report-back mode in the CLI, but it is +/// now a few milliseconds wide and, when it does lose, fails loudly via +/// `connect_with_retry`'s child-exit check rather than silently. +struct ReservedPorts(Vec); + +impl ReservedPorts { + fn reserve(n: usize) -> Self { + ReservedPorts( + (0..n) + .map(|_| TcpListener::bind("127.0.0.1:0").unwrap()) + .collect(), + ) + } + + fn ports(&self) -> Vec { + self.0 + .iter() + .map(|l| l.local_addr().unwrap().port()) + .collect() + } + + /// Hand the ports back to the OS, immediately before spawning the child + /// that binds them. + fn release(self) { + drop(self.0); + } +} + +struct KillOnDrop(Child); + +impl Drop for KillOnDrop { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn spawn_listen(wav: &std::path::Path, server_toml: &std::path::Path) -> KillOnDrop { + let child = manta() + .args(["listen", "--source"]) + .arg(wav) + .args(["--server-config"]) + .arg(server_toml) + .args(["--realtime"]) + // Deliberately NO --dial-freq-hz -- the fixture's own .json + // sidecar supplies the real RF frequency (MAN-121 Decision 2). + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn manta listen"); + KillOnDrop(child) +} + +fn connect_with_retry(child: &mut Child, port: u16, deadline: Instant) -> TcpStream { + loop { + match TcpStream::connect(("127.0.0.1", port)) { + Ok(s) => return s, + Err(e) => { + // A bind failure in the child (a port lost between release + // and spawn, a bad config) is terminal: retrying cannot + // repair it, and without this check the test burns its + // whole deadline before reporting a bare connect timeout + // that says nothing about the real cause (round-14 + // review). + if let Some(status) = child.try_wait().unwrap() { + panic!( + "manta exited with {status} before 127.0.0.1:{port} accepted a \ + connection (last connect error: {e}) -- it most likely failed to bind" + ); + } + if Instant::now() >= deadline { + panic!("could not connect to 127.0.0.1:{port} within deadline: {e}"); + } + std::thread::sleep(Duration::from_millis(50)); + } + } + } +} + +/// Reads lines until one contains `needle`, bounded by `deadline`. Never an +/// upper bound the CALLER should tune tightly -- pacing is wall-clock, so +/// this deadline exists only to fail the test cleanly instead of hanging +/// forever, not to assert a specific latency. +fn wait_for_line_containing( + reader: &mut BufReader, + needle: &str, + deadline: Instant, +) -> String { + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + panic!("timed out waiting for a line containing {needle:?}"); + } + reader.get_ref().set_read_timeout(Some(remaining)).unwrap(); + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) => panic!("connection closed before a line containing {needle:?} arrived"), + Ok(_) => { + if line.contains(needle) { + return line; + } + } + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => + { + panic!("timed out waiting for a line containing {needle:?}"); + } + Err(e) => panic!("read error waiting for {needle:?}: {e}"), + } + } +} + +fn write_server_toml( + dir: &std::path::Path, + telnet_port: u16, + json_port: u16, + metrics_port: u16, +) -> std::path::PathBuf { + let path = dir.join("server.toml"); + std::fs::write( + &path, + format!( + "[server]\n\ + station_callsign = \"W5AU\"\n\ + bind_addr = \"127.0.0.1\"\n\ + telnet_port = {telnet_port}\n\ + json_port = {json_port}\n\ + metrics_port = {metrics_port}\n" + ), + ) + .unwrap(); + path +} + +// 30s: the first spot reaches the wire ~12s into paced replay (2s +// calibration + SPEC 2.1's ~2.05s warmup/confirm floor + the vector's own +// lead-in), leaving comfortable margin for a CI runner without making the +// test needlessly slow. +const FIXTURE_DURATION_S: f64 = 30.0; +const SPOT_WAIT_DEADLINE_S: u64 = 40; + +#[test] +fn a_stock_telnet_client_receives_a_well_formed_rbn_spot_from_gen_output() { + let dir = tempfile::tempdir().unwrap(); + let spec = manta_testkit::vectors::VectorSpec { + duration_s: FIXTURE_DURATION_S, + ..manta_testkit::vectors::v1() + }; + let manifest = manta_testkit::vectors::write_fixture_set(&spec, dir.path()).unwrap(); + let wav = dir.path().join("v1.wav"); + + let reserved = ReservedPorts::reserve(3); + let ports = reserved.ports(); + let (telnet_port, json_port, metrics_port) = (ports[0], ports[1], ports[2]); + let server_toml = write_server_toml(dir.path(), telnet_port, json_port, metrics_port); + reserved.release(); + + let mut child = spawn_listen(&wav, &server_toml); + + let deadline = Instant::now() + Duration::from_secs(SPOT_WAIT_DEADLINE_S); + let stream = connect_with_retry(&mut child.0, telnet_port, deadline); + let mut reader = BufReader::new(stream); + + wait_for_line_containing(&mut reader, "login", deadline); + reader.get_ref().write_all(b"W5AU\r\n").unwrap(); + + let spot_line = wait_for_line_containing(&mut reader, "DX de", deadline); + + assert!( + child.0.try_wait().unwrap().is_none(), + "the server process must still be running when the spot arrives" + ); + + assert!(spot_line.contains("W1AW"), "spot line: {spot_line}"); + // The vector's own expected_freq_hz (from the sidecar-declared RF + // frequency, not an audio-tone offset) -- e.g. "14012.4". + let expected_khz = manifest.expected_freq_hz / 1000.0; + let khz_str = format!("{expected_khz:.0}"); + assert!( + spot_line.contains(&khz_str), + "expected a frequency near {expected_khz:.1} kHz, got: {spot_line}" + ); +} + +#[test] +fn a_json_lines_client_receives_the_same_spot() { + let dir = tempfile::tempdir().unwrap(); + let spec = manta_testkit::vectors::VectorSpec { + duration_s: FIXTURE_DURATION_S, + ..manta_testkit::vectors::v1() + }; + let manifest = manta_testkit::vectors::write_fixture_set(&spec, dir.path()).unwrap(); + let wav = dir.path().join("v1.wav"); + + let reserved = ReservedPorts::reserve(3); + let ports = reserved.ports(); + let (telnet_port, json_port, metrics_port) = (ports[0], ports[1], ports[2]); + let server_toml = write_server_toml(dir.path(), telnet_port, json_port, metrics_port); + reserved.release(); + + let mut child = spawn_listen(&wav, &server_toml); + + let deadline = Instant::now() + Duration::from_secs(SPOT_WAIT_DEADLINE_S); + let stream = connect_with_retry(&mut child.0, json_port, deadline); + stream + .set_read_timeout(Some(Duration::from_secs(SPOT_WAIT_DEADLINE_S))) + .unwrap(); + let mut reader = BufReader::new(stream); + + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + panic!("timed out waiting for a spot on the JSON Lines port"); + } + reader.get_ref().set_read_timeout(Some(remaining)).unwrap(); + let mut line = String::new(); + let n = reader.read_line(&mut line).unwrap_or(0); + if n == 0 { + panic!("JSON stream closed before a spot arrived"); + } + let Ok(value) = serde_json::from_str::(&line) else { + continue; // e.g. a keepalive/non-spot control line + }; + let Some(dx_call) = value.get("dxCall").and_then(|v| v.as_str()) else { + continue; + }; + assert_eq!(dx_call, "W1AW", "unexpected dxCall in: {value}"); + let freq = value["frequency"] + .as_f64() + .expect("frequency field should be numeric"); + assert!( + (freq - manifest.expected_freq_hz).abs() < 500.0, + "expected frequency near {}, got {freq}", + manifest.expected_freq_hz + ); + break; + } +} diff --git a/crates/manta-dsp/src/channelizer.rs b/crates/manta-dsp/src/channelizer.rs index 5dabf856..08056a30 100644 --- a/crates/manta-dsp/src/channelizer.rs +++ b/crates/manta-dsp/src/channelizer.rs @@ -65,16 +65,30 @@ pub struct Channelizer { } impl Channelizer { + /// Whether `fs` is a rate this channelizer supports (`fs/93.75` a + /// power of two) -- exactly the predicate `new` enforces, exposed so a + /// caller can reject an unsupported rate BEFORE anything is sized from + /// it. `manta_engine::listen` allocates a `CALIBRATION_SECONDS * fs` + /// buffer, so a source reporting a wild rate (a malformed WAV header + /// claiming ~4 GS/s, say) would try to allocate tens of GiB and abort + /// the process before `new` ever got to return this error (MAN-121 + /// review). Non-finite and negative rates answer `false`: `nf.round() + /// as usize` saturates them to 0, which is not a power of two. + pub fn supports_rate(fs: f64) -> bool { + let nf = fs / CHANNEL_SPACING_HZ; + let n = nf.round() as usize; + (nf - n as f64).abs() <= 1e-9 && n.is_power_of_two() + } + /// A channelizer for a supported table rate (`fs/93.75` a power of /// two). SPEC §1.1. pub fn new(fs: f64, center_freq_hz: f64) -> Result { - let nf = fs / CHANNEL_SPACING_HZ; - let n = nf.round() as usize; - if (nf - n as f64).abs() > 1e-9 || !n.is_power_of_two() { + if !Self::supports_rate(fs) { return Err(format!( "unsupported sample rate {fs}: fs/93.75 must be a power of two" )); } + let n = (fs / CHANNEL_SPACING_HZ).round() as usize; Ok(Channelizer { n, hop: n / 4, diff --git a/crates/manta-engine/src/listen.rs b/crates/manta-engine/src/listen.rs index bb23c20b..b088dbc9 100644 --- a/crates/manta-engine/src/listen.rs +++ b/crates/manta-engine/src/listen.rs @@ -39,6 +39,15 @@ pub fn listen( let fs = src.sample_rate(); let center_freq_hz = src.center_freq_hz(); + // Built BEFORE the calibration buffer is sized, not after: `calib_n` + // is derived from `fs`, so a source reporting an unsupported rate -- + // in the extreme, a malformed WAV header claiming a rate near + // `u32::MAX` -- used to allocate tens of GiB and abort the process + // before this rate check could return its error (MAN-121 review). + // Same fail-fast rationale as the config validation above. + let mut ch = manta_dsp::channelizer::Channelizer::new(fs, center_freq_hz) + .map_err(|e| anyhow::anyhow!(e))?; + let calib_n = (fs * CALIBRATION_SECONDS).round() as usize; let mut calib = vec![Complex32::new(0.0, 0.0); calib_n]; let mut filled = 0; @@ -49,8 +58,6 @@ pub fn listen( } filled += n; } - let mut ch = manta_dsp::channelizer::Channelizer::new(fs, center_freq_hz) - .map_err(|e| anyhow::anyhow!(e))?; let hop = ch.hop() as u64; let mut tm = crate::track::TrackManager::new( ch.n_channels(), diff --git a/crates/manta-input/src/lib.rs b/crates/manta-input/src/lib.rs index ef4e817c..e69c7fe4 100644 --- a/crates/manta-input/src/lib.rs +++ b/crates/manta-input/src/lib.rs @@ -8,6 +8,12 @@ pub use audio::{AudioIqSource, TARGET_RATE_HZ}; pub mod kiwi; pub use kiwi::KiwiIqSource; +pub mod pace; +pub use pace::PacedSource; + +pub mod replay; +pub use replay::LoopingWavSource; + #[cfg(feature = "soapy")] pub mod soapy; #[cfg(feature = "soapy")] @@ -123,6 +129,154 @@ impl IqSource for WavIqSource { } } +/// Upper bound on the sample rate `open_replay_wav` will adopt from an IQ +/// replay WAV's own header, in S/s. +/// +/// `Channelizer::supports_rate` constrains the rate's SHAPE but not its +/// SIZE, and everything sized from a replay rate -- `manta_engine::listen`'s +/// `CALIBRATION_SECONDS * fs` complex buffer, the channelizer's `fs/93.75` +/// -point FFT and prototype filter -- is allocated from a number a +/// few-hundred-byte file can claim freely. 10 MS/s is deliberately the same +/// ceiling `manta-cli`'s `--hpsdr-rate` already applies to a LIVE source, so +/// replay is bounded no more tightly than real hardware: it admits every +/// rate any supported receiver produces (the largest well-shaped rate under +/// it is 6.144 MS/s, a ~98 MiB calibration buffer) while turning the +/// 5.9 GiB-and-up headers into the documented startup error instead of an +/// OOM abort. +pub const MAX_REPLAY_RATE_HZ: f64 = 10_000_000.0; + +/// Open a replay WAV as whichever `IqSource` its layout implies: 2 channels +/// = complex IQ (what `manta gen` writes and `manta decode` reads, at its +/// own native rate, via `WavIqSource`); anything else = a real rig-audio +/// passband, which `AudioIqSource` converts to analytic form via Hilbert +/// transform and still requires at exactly `TARGET_RATE_HZ`. +/// +/// `manta_engine::listen()` is rate-agnostic -- `Channelizer::new` accepts +/// any `fs` where `fs / 93.75` is a power of two, and 96000 / 93.75 = 1024 +/// -- so this is a reader choice, not a resample. See MAN-121. +/// +/// Channel count alone is only unambiguous away from `TARGET_RATE_HZ` +/// (48 kHz): `AudioIqSource` never accepted anything but 48 kHz, so a +/// 2-channel file at any other rate could never have been a rig-audio +/// capture before this dispatch existed, and routing it to `WavIqSource` +/// regresses nothing. At exactly 48 kHz a 2-channel file is genuinely +/// ambiguous -- it's both a legal `WavIqSource` rate and `AudioIqSource`'s +/// only rate, and a stereo soundcard recording of a receiver's passband +/// (a real, common rig-audio capture) is indistinguishable from IQ by +/// channel count alone. A `.json` sidecar (what `gen`/`decode`'s IQ +/// files always carry) breaks the tie in favor of IQ; with no sidecar, a +/// 48 kHz 2-channel file keeps its pre-MAN-121 `AudioIqSource` downmix +/// path rather than being silently misread as `Complex32::new(I, Q)`. +/// +/// The IQ branch additionally rejects any rate the channelizer can't +/// accept (`fs/93.75` not a power of two) *and* any rate above +/// `MAX_REPLAY_RATE_HZ`, because it is the branch that adopts the FILE's +/// own rate -- see the guard's own comment for why that has to happen +/// here rather than in `manta_engine::listen`. +/// +/// The tie-break asks `replay_wav_has_iq_sidecar` -- "is there a parseable +/// IQ sidecar?" -- and deliberately NOT `replay_wav_center_freq_hz`, which +/// answers the separate question "does that sidecar declare an RF +/// frequency good enough to satisfy the CLI's `--dial-freq-hz` gate?" +/// (round-2 review). Those are different questions: a baseband or +/// unknown-dial capture whose sidecar says `center_freq_hz: 0.0` is still +/// unambiguously IQ in FORMAT, and conflating the two silently downmixed +/// such a 48 kHz file through `AudioIqSource`'s Hilbert path, discarding +/// Q, even when the caller supplied `--dial-freq-hz`. +pub fn open_replay_wav(path: &Path) -> Result> { + let spec = hound::WavReader::open(path) + .with_context(|| format!("open WAV {}", path.display()))? + .spec(); + let is_iq = spec.channels == 2 + && (spec.sample_rate != TARGET_RATE_HZ || replay_wav_has_iq_sidecar(path)); + if is_iq { + // The IQ path is the one that adopts the FILE's own rate, so it is + // also the one that must bound it. `manta_engine::listen` sizes a + // `CALIBRATION_SECONDS * fs` complex buffer from `sample_rate()` + // and only THEN calls `Channelizer::new` to validate the rate, so + // a malformed 2-channel header claiming a rate near `u32::MAX` + // would ask for tens of GiB and abort the process instead of + // reporting the documented unsupported-rate error. Reject it here, + // before the source is ever handed to the engine (MAN-121 review). + // `AudioIqSource` needs no equivalent guard -- it already accepts + // exactly `TARGET_RATE_HZ` and nothing else. + // + // Two independent things have to be true, and `supports_rate` + // only answers the first: the rate's SHAPE (`fs/93.75` a power of + // two) and its MAGNITUDE. `supports_rate` has no upper bound, so + // a malformed header claiming e.g. 393_216_000 (= 93.75 * 2^22, + // a perfectly well-shaped rate) passes it and still asks + // `manta_engine::listen` for a ~5.9 GiB calibration buffer -- and + // `Channelizer::new` for a 2^22-point FFT and prototype filter -- + // from a file that may be a few hundred bytes long. Bound the + // magnitude too (MAN-121 round-15 review). + let fs = spec.sample_rate as f64; + if !manta_dsp::channelizer::Channelizer::supports_rate(fs) { + bail!( + "unsupported sample rate {fs} in {}: an IQ replay WAV must use a channelizer \ + rate (fs/93.75 a power of two -- 12/24/48/96/192/384 kHz and so on)", + path.display() + ); + } + if fs > MAX_REPLAY_RATE_HZ { + bail!( + "unsupported sample rate {fs} in {}: an IQ replay WAV must be at most \ + {MAX_REPLAY_RATE_HZ} S/s -- a higher rate is a malformed or unsupported \ + header, and sizing the decoder's startup buffers from it would exhaust \ + memory before the rate could be reported", + path.display() + ); + } + Ok(Box::new(WavIqSource::open(path)?)) + } else { + Ok(Box::new(AudioIqSource::from_wav_file(path)?)) + } +} + +/// Whatever `center_freq_hz` a replay WAV's sidecar declares: `Some` for 2 +/// channels plus a parseable `.json` carrying a finite +/// `center_freq_hz` -- including `0.0`, which means "baseband/unknown", not +/// "not IQ" -- and `None` for anything else. +/// +/// Deliberately swallows every error, including a nonexistent path, so a +/// CLI flag-validation gate can run BEFORE any real file I/O and still +/// report a missing flag rather than a missing file -- an ordering +/// `crates/manta-cli/tests/cli.rs`'s +/// `server_config_without_dial_freq_for_audio_source_is_a_clean_error` +/// asserts and documents in its own comment. Both public wrappers below +/// inherit that property. +fn replay_wav_sidecar_center_freq_hz(path: &Path) -> Option { + let spec = hound::WavReader::open(path).ok()?.spec(); + if spec.channels != 2 { + return None; + } + let text = std::fs::read_to_string(path.with_extension("json")).ok()?; + let sc: Sidecar = serde_json::from_str(&text).ok()?; + sc.center_freq_hz.is_finite().then_some(sc.center_freq_hz) +} + +/// FORMAT question: does this replay WAV carry a valid IQ sidecar (2 +/// channels plus a parseable `.json` with a finite `center_freq_hz`)? +/// +/// This is the 48 kHz tie-break `open_replay_wav` uses, and it is +/// deliberately independent of what that frequency actually *is*: a +/// sidecar declaring `center_freq_hz: 0.0` (baseband, or a dial frequency +/// the operator will supply with `--dial-freq-hz` instead) still marks the +/// file as IQ, so its Q channel is read rather than thrown away. +pub fn replay_wav_has_iq_sidecar(path: &Path) -> bool { + replay_wav_sidecar_center_freq_hz(path).is_some() +} + +/// RF question: the real RF center frequency a replay WAV *declares* -- a +/// valid IQ sidecar whose `center_freq_hz` is finite and positive -- or +/// `None` for anything else, including an IQ file that declares `0.0`. +/// +/// Only this answers "may `--server-config` go without `--dial-freq-hz`?"; +/// use `replay_wav_has_iq_sidecar` to decide how to READ the file. +pub fn replay_wav_center_freq_hz(path: &Path) -> Option { + replay_wav_sidecar_center_freq_hz(path).filter(|hz| *hz > 0.0) +} + /// Drain an IqSource to a Vec (file-mode helper). ARCHITECTURE §3. pub fn read_all(src: &mut dyn IqSource) -> Result> { let mut all = Vec::new(); @@ -237,4 +391,314 @@ mod tests { assert_eq!(src.read(&mut buf).unwrap(), 100); assert_eq!(src.read(&mut buf).unwrap(), 0); // EOF } + + fn write_mono_f32_wav(path: &std::path::Path, samples: &[f32], fs: u32) { + let spec = hound::WavSpec { + channels: 1, + sample_rate: fs, + bits_per_sample: 32, + sample_format: hound::SampleFormat::Float, + }; + let mut w = hound::WavWriter::create(path, spec).unwrap(); + for s in samples { + w.write_sample(*s).unwrap(); + } + w.finalize().unwrap(); + } + + // MAN-121: `open_replay_wav` dispatches on channel count so `listen + // --source` can accept the same 2-channel IQ WAV `manta gen`/`decode` + // already use, not just AudioIqSource's mono rig-audio format. + #[test] + fn open_replay_wav_reads_a_stereo_iq_file_at_its_native_rate() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("v1.wav"); + write_f32_wav(&wav, &samples(), 96_000); + std::fs::write( + dir.path().join("v1.json"), + r#"{"center_freq_hz": 14000000.0}"#, + ) + .unwrap(); + + let src = open_replay_wav(&wav).unwrap(); + assert_eq!(src.sample_rate(), 96_000.0); + assert_eq!(src.center_freq_hz(), 14_000_000.0); + } + + #[test] + fn open_replay_wav_reads_a_mono_48k_file_as_an_audio_source() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("rig.wav"); + write_mono_f32_wav(&wav, &vec![0.0f32; 480], 48_000); + + let src = open_replay_wav(&wav).unwrap(); + assert_eq!(src.sample_rate(), 48_000.0); + assert_eq!(src.center_freq_hz(), 0.0); + } + + // MAN-121 remediation: a 2-channel 48 kHz WAV with no sidecar is the + // exact collision with AudioIqSource's only supported rate -- a stereo + // soundcard recording of rig audio looks identical to IQ by channel + // count alone. Without a sidecar it must keep the pre-MAN-121 + // AudioIqSource downmix-and-Hilbert path, not be reinterpreted as + // Complex32::new(I, Q). Ch0 is constant zero and ch1 carries a large, + // distinctive value: WavIqSource would read the pair verbatim + // (0.0, 0.9); AudioIqSource downmixes to ch0 alone (constant zero) and + // the Hilbert transform of an all-zero signal is all-zero, so the two + // paths are unambiguous from the output alone. + #[test] + fn open_replay_wav_treats_a_sidecarless_48k_stereo_file_as_rig_audio_not_iq() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("rig.wav"); + let samples: Vec = (0..300).map(|_| Complex32::new(0.0, 0.9)).collect(); + write_f32_wav(&wav, &samples, 48_000); + + let mut src = open_replay_wav(&wav).unwrap(); + assert_eq!(src.sample_rate(), 48_000.0); + assert_eq!(src.center_freq_hz(), 0.0); + let all = read_all(&mut *src).unwrap(); + assert!( + all.iter().all(|s| s.re == 0.0 && s.im == 0.0), + "expected the AudioIqSource downmix+Hilbert path (all-zero output for \ + all-zero ch0), got non-zero samples -- the file was read as raw IQ instead" + ); + } + + // Companion to the above: a sidecar is exactly the signal that should + // still win at 48 kHz, since `gen`/`decode`'s own IQ files may legally + // be 48 kHz (48000 / 93.75 = 512, a valid channelizer rate). + #[test] + fn open_replay_wav_still_reads_a_sidecar_backed_48k_stereo_file_as_iq() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("v1.wav"); + write_f32_wav(&wav, &samples(), 48_000); + std::fs::write( + dir.path().join("v1.json"), + r#"{"center_freq_hz": 14000000.0}"#, + ) + .unwrap(); + + let mut src = open_replay_wav(&wav).unwrap(); + assert_eq!(src.sample_rate(), 48_000.0); + assert_eq!(src.center_freq_hz(), 14_000_000.0); + let all = read_all(&mut *src).unwrap(); + assert_eq!(all, samples()); + } + + // MAN-121 review: an unsupported IQ rate must be rejected HERE, not + // deep inside `manta_engine::listen`, which sizes a two-second complex + // calibration buffer from `sample_rate()` before `Channelizer::new` + // ever validates it -- a header claiming a rate near `u32::MAX` would + // request tens of GiB and abort the process instead of reporting this + // error. 100 kHz is the smallest realistic case of the same class: + // 100000 / 93.75 = 1066.67, not a power of two. + #[test] + fn open_replay_wav_rejects_an_iq_file_at_an_unsupported_channelizer_rate() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("odd.wav"); + write_f32_wav(&wav, &samples(), 100_000); + + // `Box` isn't `Debug`, so `unwrap_err` is unavailable. + let err = match open_replay_wav(&wav) { + Ok(_) => panic!("expected an unsupported-rate error, got a source"), + Err(e) => e.to_string(), + }; + assert!( + err.contains("unsupported sample rate"), + "expected an unsupported-rate error, got: {err}" + ); + } + + // The pathological case the guard above exists for, stated exactly: a + // 2-channel header claiming 200 MS/s. `listen` would have asked for + // 2 * 2e8 complex samples (~3.2 GiB) before validating the rate. (Not + // literally `u32::MAX`, which `hound`'s own writer can't even encode + // -- its byte-rate field overflows -- but the same class of header.) + #[test] + fn open_replay_wav_rejects_an_absurd_iq_sample_rate_before_any_allocation() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("huge.wav"); + write_f32_wav(&wav, &samples(), 200_000_000); + + assert!(open_replay_wav(&wav).is_err()); + } + + // The half of that class the shape check alone does NOT catch + // (MAN-121 round-15 review): 393_216_000 = 93.75 * 2^22 is a + // perfectly WELL-SHAPED channelizer rate, so `supports_rate` says + // yes, yet `manta_engine::listen` would size a ~5.9 GiB calibration + // buffer (and a 2^22-point FFT) from this ~8 KiB file. Only the + // magnitude bound rejects it. + #[test] + fn open_replay_wav_rejects_a_well_shaped_but_absurdly_large_iq_rate() { + let fs = 393_216_000; + assert!( + manta_dsp::channelizer::Channelizer::supports_rate(fs as f64), + "this test is only meaningful for a rate the SHAPE check accepts" + ); + + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("wellshaped-huge.wav"); + write_f32_wav(&wav, &samples(), fs); + + let err = match open_replay_wav(&wav) { + Ok(_) => panic!("expected an unsupported-rate error, got a source"), + Err(e) => e.to_string(), + }; + assert!( + err.contains("unsupported sample rate") && err.contains("at most"), + "expected the magnitude-bound error, got: {err}" + ); + } + + // The bound admits every rate a real receiver produces -- the + // ceiling's own largest well-shaped rate must still open. + #[test] + fn open_replay_wav_accepts_the_largest_realistic_iq_rate() { + let fs = 6_144_000; + assert!(fs as f64 <= MAX_REPLAY_RATE_HZ); + + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("fast.wav"); + write_f32_wav(&wav, &samples(), fs); + + let src = open_replay_wav(&wav).unwrap(); + assert_eq!(src.sample_rate(), fs as f64); + } + + // Round-2 review: format detection and RF-frequency validation are + // separate questions. A 48 kHz 2-channel IQ file whose sidecar declares + // `center_freq_hz: 0.0` (baseband, or a dial frequency the operator + // passes with --dial-freq-hz) is still IQ in FORMAT -- routing it to + // AudioIqSource would discard Q and synthesize it back via Hilbert. + // Same ch0=0 / ch1=0.9 discriminator as the sidecarless test above: + // WavIqSource returns the pair verbatim, AudioIqSource returns zeros. + #[test] + fn open_replay_wav_reads_a_zero_freq_sidecar_48k_stereo_file_as_iq() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("baseband.wav"); + let samples: Vec = (0..300).map(|_| Complex32::new(0.0, 0.9)).collect(); + write_f32_wav(&wav, &samples, 48_000); + std::fs::write( + dir.path().join("baseband.json"), + r#"{"center_freq_hz": 0.0}"#, + ) + .unwrap(); + + let mut src = open_replay_wav(&wav).unwrap(); + assert_eq!(src.sample_rate(), 48_000.0); + let all = read_all(&mut *src).unwrap(); + assert_eq!( + all, samples, + "a zero-frequency sidecar still declares IQ -- Q must not be \ + discarded and re-synthesized by the AudioIqSource path" + ); + } + + // The other half of the same split: declaring `0.0` marks the file as + // IQ but does NOT report a real RF frequency, so the CLI's + // --dial-freq-hz gate must still fire for it. + #[test] + fn a_zero_freq_sidecar_declares_iq_format_but_no_rf_frequency() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("baseband.wav"); + write_f32_wav(&wav, &samples(), 48_000); + std::fs::write( + dir.path().join("baseband.json"), + r#"{"center_freq_hz": 0.0}"#, + ) + .unwrap(); + + assert!(replay_wav_has_iq_sidecar(&wav)); + assert_eq!(replay_wav_center_freq_hz(&wav), None); + } + + #[test] + fn replay_wav_has_iq_sidecar_is_false_without_a_parseable_sidecar() { + let dir = tempfile::tempdir().unwrap(); + + let no_sidecar = dir.path().join("nosidecar.wav"); + write_f32_wav(&no_sidecar, &samples(), 48_000); + assert!(!replay_wav_has_iq_sidecar(&no_sidecar)); + + let bad_sidecar = dir.path().join("badsidecar.wav"); + write_f32_wav(&bad_sidecar, &samples(), 48_000); + std::fs::write(dir.path().join("badsidecar.json"), "not json").unwrap(); + assert!(!replay_wav_has_iq_sidecar(&bad_sidecar)); + + let mono = dir.path().join("rig.wav"); + write_mono_f32_wav(&mono, &vec![0.0f32; 480], 48_000); + std::fs::write( + dir.path().join("rig.json"), + r#"{"center_freq_hz": 7030000.0}"#, + ) + .unwrap(); + assert!(!replay_wav_has_iq_sidecar(&mono)); + + // Never fails, for the same reason replay_wav_center_freq_hz never + // does: the CLI gate probes before any file I/O is meant to fail. + assert!(!replay_wav_has_iq_sidecar(std::path::Path::new( + "/nonexistent.wav" + ))); + } + + #[test] + fn open_replay_wav_still_rejects_a_mono_file_at_the_wrong_rate() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("rig.wav"); + write_mono_f32_wav(&wav, &vec![0.0f32; 441], 44_100); + + match open_replay_wav(&wav) { + Ok(_) => panic!("expected an error for a 44100 Hz mono file"), + Err(err) => assert!( + format!("{err}").contains("48000"), + "expected the AudioIqSource rate error, got: {err}" + ), + } + } + + #[test] + fn replay_wav_center_freq_hz_reports_a_sidecar_backed_iq_file() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("v1.wav"); + write_f32_wav(&wav, &samples(), 96_000); + std::fs::write( + dir.path().join("v1.json"), + r#"{"center_freq_hz": 14000000.0}"#, + ) + .unwrap(); + + assert_eq!(replay_wav_center_freq_hz(&wav), Some(14_000_000.0)); + } + + #[test] + fn replay_wav_center_freq_hz_is_none_for_mono_missing_and_malformed_inputs() { + let dir = tempfile::tempdir().unwrap(); + + // Mono WAV, no sidecar possible (a real audio source, not IQ). + let mono = dir.path().join("rig.wav"); + write_mono_f32_wav(&mono, &vec![0.0f32; 480], 48_000); + assert_eq!(replay_wav_center_freq_hz(&mono), None); + + // 2ch WAV with no sidecar at all. + let no_sidecar = dir.path().join("nosidecar.wav"); + write_f32_wav(&no_sidecar, &samples(), 96_000); + assert_eq!(replay_wav_center_freq_hz(&no_sidecar), None); + + // 2ch WAV with an unparseable sidecar. + let bad_sidecar = dir.path().join("badsidecar.wav"); + write_f32_wav(&bad_sidecar, &samples(), 96_000); + std::fs::write(dir.path().join("badsidecar.json"), "not json").unwrap(); + assert_eq!(replay_wav_center_freq_hz(&bad_sidecar), None); + + // Nonexistent path -- MUST NOT panic or error, only return None, + // since a CLI flag-validation gate calls this before any file I/O + // is meant to fail (crates/manta-cli/tests/cli.rs's + // server_config_without_dial_freq_for_audio_source_is_a_clean_error + // depends on this exact ordering). + assert_eq!( + replay_wav_center_freq_hz(std::path::Path::new("/nonexistent.wav")), + None + ); + } } diff --git a/crates/manta-input/src/pace.rs b/crates/manta-input/src/pace.rs new file mode 100644 index 00000000..99c46d1d --- /dev/null +++ b/crates/manta-input/src/pace.rs @@ -0,0 +1,349 @@ +//! Wall-clock pacing for file replay (MAN-121). A pure sleep wrapper: it +//! delivers exactly the samples its inner source delivers, in the same +//! order, and only decides *when*. The decode path is untouched, which is +//! what keeps `--realtime` output byte-identical to unpaced output. + +use crate::IqSource; +use anyhow::Result; +use num_complex::Complex32; +use std::sync::atomic::AtomicBool; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +/// Paces `inner` to wall-clock realtime at its own sample rate. +/// +/// Drift-free by construction: sleeps are computed from CUMULATIVE +/// delivered samples (including the chunk being returned) against a single +/// start instant (taken on the first read), not per-chunk, so a chunk +/// that arrives late is absorbed rather than compounded. If the consumer +/// falls behind the recording, `due <= elapsed` and this never sleeps at +/// all -- pacing degrades to unpaced instead of ever stalling the +/// pipeline. +pub struct PacedSource { + inner: Box, + fs: f64, + delivered: u64, + /// The recording clock's origin, started LAZILY on the first `read()` + /// rather than at construction. `manta-cli` builds the paced source + /// before it resolves the replay epoch, hashes the whole recording for + /// the session nonce, and binds the telnet/JSON listeners -- all of + /// which happen between `new()` and the first read. Anchoring the + /// clock at construction credited that setup time against the + /// recording's own timeline, so the first buffer's pacing debt was + /// already partly spent before the servers were even listening and a + /// client had that much less of the window to connect in. + start: Option, +} + +impl PacedSource { + pub fn new(inner: Box) -> Self { + let fs = inner.sample_rate(); + PacedSource { + inner, + fs, + delivered: 0, + start: None, + } + } + + /// Wall-clock since the recording clock started, or zero before the + /// first `read()` has started it. + fn elapsed(&self) -> Duration { + self.start.map_or(Duration::ZERO, |start| start.elapsed()) + } +} + +/// How long `read()` must still wait before it may RETURN a buffer that +/// brings the cumulative delivered count to `delivered`, given `elapsed` +/// wall-clock since the paced source started. +/// +/// The count is the one INCLUDING the buffer about to be returned, not the +/// one before it (round-2 review): pacing on the previous count would hand +/// every buffer to the consumer a full chunk before its own recording +/// interval had elapsed, and the very first chunk -- which for +/// `manta_engine::listen` is the entire two-second calibration buffer -- +/// would be delivered instantly, so a spot inside it could be published +/// before any client had time to connect. +/// +/// `None` means "return now": the consumer is already at or past the +/// recording's own clock, so pacing degrades to unpaced rather than ever +/// stalling the pipeline. Extracted as a pure function so the pacing +/// decision can be asserted exactly, instead of inferred from a wall-clock +/// upper bound on a whole `read()`. Such bounds measure the machine as +/// much as the code and are flake-prone on a loaded CI runner -- the +/// `--features hpsdr` job in particular runs this module's tests inside a +/// far heavier `manta-input` test binary (every UDP-loopback HPSDR test +/// too) than the default job does. +fn pacing_delay(delivered: u64, fs: f64, elapsed: Duration) -> Option { + let due = Duration::from_secs_f64(delivered as f64 / fs); + if due > elapsed { + Some(due - elapsed) + } else { + None + } +} + +impl IqSource for PacedSource { + fn sample_rate(&self) -> f64 { + self.inner.sample_rate() + } + + fn center_freq_hz(&self) -> f64 { + self.inner.center_freq_hz() + } + + fn confirmed_live_handle(&self) -> Option> { + // Do not swallow the inner source's own liveness signal (MAN-55). + self.inner.confirmed_live_handle() + } + + fn read(&mut self, buf: &mut [Complex32]) -> Result { + // Start the recording clock here, on the first read, and BEFORE + // the inner read runs -- the time the inner source itself spends + // producing this buffer is part of the recording's own interval, + // not something to sleep on top of. + self.start.get_or_insert_with(Instant::now); + let n = self.inner.read(buf)?; + self.delivered += n as u64; + // Sleep AFTER the read, against the count that INCLUDES this + // buffer: the samples this call is about to hand back must have + // had their own recording interval elapse first. Sleeping before + // the read instead paced against the PREVIOUS call's samples, so + // `listen`'s first read -- the whole two-second calibration + // buffer -- returned at once and anything decoded from it could + // reach the servers before a client could connect. + if let Some(delay) = pacing_delay(self.delivered, self.fs, self.elapsed()) { + std::thread::sleep(delay); + } + Ok(n) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct VecSource { + samples: Vec, + cursor: usize, + fs: f64, + } + + impl IqSource for VecSource { + fn sample_rate(&self) -> f64 { + self.fs + } + fn center_freq_hz(&self) -> f64 { + 0.0 + } + fn read(&mut self, buf: &mut [Complex32]) -> Result { + let n = buf.len().min(self.samples.len() - self.cursor); + buf[..n].copy_from_slice(&self.samples[self.cursor..self.cursor + n]); + self.cursor += n; + Ok(n) + } + } + + fn drain(src: &mut dyn IqSource, chunk: usize) -> Vec { + let mut all = Vec::new(); + let mut buf = vec![Complex32::new(0.0, 0.0); chunk]; + loop { + let n = src.read(&mut buf).unwrap(); + if n == 0 { + return all; + } + all.extend_from_slice(&buf[..n]); + } + } + + #[test] + fn paced_source_delivers_the_same_samples_in_the_same_order() { + // A high fake fs keeps this test fast -- pacing math is the same + // regardless of rate. + let samples: Vec = (0..500) + .map(|i| Complex32::new(i as f32, -(i as f32))) + .collect(); + let src = VecSource { + samples: samples.clone(), + cursor: 0, + fs: 1_000_000.0, + }; + let mut paced = PacedSource::new(Box::new(src)); + let drained = drain(&mut paced, 64); + assert_eq!(drained, samples); + } + + #[test] + fn paced_source_takes_at_least_the_recording_duration() { + let samples = vec![Complex32::new(0.0, 0.0); 2400]; + let src = VecSource { + samples, + cursor: 0, + fs: 8000.0, + }; + let mut paced = PacedSource::new(Box::new(src)); + let start = Instant::now(); + drain(&mut paced, 256); + // 2400 samples at 8000 S/s = 0.3s. Generous lower bound only -- + // never assert an upper bound, that is CI-flaky. + assert!( + start.elapsed() >= Duration::from_millis(250), + "elapsed {:?} was too short for a 0.3s recording", + start.elapsed() + ); + } + + #[test] + fn paced_source_does_not_sleep_when_the_consumer_is_already_behind() { + struct SlowSource { + inner: VecSource, + } + impl IqSource for SlowSource { + fn sample_rate(&self) -> f64 { + self.inner.sample_rate() + } + fn center_freq_hz(&self) -> f64 { + 0.0 + } + fn read(&mut self, buf: &mut [Complex32]) -> Result { + std::thread::sleep(Duration::from_millis(200)); + self.inner.read(buf) + } + } + // 800 samples at 8000 S/s = 0.1s of "recording" pacing, but the + // single read() call already takes 0.2s -- pacing must add ~nothing + // on top of that one call. + let samples = vec![Complex32::new(0.0, 0.0); 800]; + let slow = SlowSource { + inner: VecSource { + samples, + cursor: 0, + fs: 8000.0, + }, + }; + let mut paced = PacedSource::new(Box::new(slow)); + let mut buf = vec![Complex32::new(0.0, 0.0); 800]; + assert_eq!(paced.read(&mut buf).unwrap(), 800); + // Asserted on the pacing DECISION, not on a wall-clock upper bound + // for the whole call: `elapsed` is already >= the inner source's + // own 200ms by the time `read()` returns, while the recording is + // only worth 800/8000 = 100ms, so the next read's delay is + // guaranteed to be None on any machine at any load. A `<280ms` + // bound on the whole call asserted the same thing but could be + // broken by an oversubscribed CI runner overshooting the 200ms + // sleep rather than by pacing compounding. + assert_eq!( + pacing_delay(paced.delivered, paced.fs, paced.elapsed()), + None, + "pacing must ask for no delay once the consumer has fallen behind" + ); + // The pure decision itself, pinned exactly: 100ms of recording + // delivered, 200ms of wall-clock spent -- no delay, and no + // compounding of the two. + assert_eq!(pacing_delay(800, 8000.0, Duration::from_millis(200)), None); + assert_eq!( + pacing_delay(800, 8000.0, Duration::from_millis(40)), + Some(Duration::from_millis(60)), + "when the consumer is AHEAD, the delay is the remaining debt only" + ); + } + + // Round-2 review: the FIRST read must be paced too. `listen`'s first + // read asks for the whole two-second calibration buffer, and pacing + // against the count BEFORE that buffer made it return instantly -- + // every sample of the first chunk was handed to the decoder at + // startup, so a spot inside it could reach the servers before a client + // could connect. Asserted on the single first call, not on a drain. + #[test] + fn paced_source_paces_the_very_first_read() { + let samples = vec![Complex32::new(0.0, 0.0); 800]; + let src = VecSource { + samples, + cursor: 0, + fs: 8000.0, + }; + let mut paced = PacedSource::new(Box::new(src)); + let mut buf = vec![Complex32::new(0.0, 0.0); 800]; + let start = Instant::now(); + assert_eq!(paced.read(&mut buf).unwrap(), 800); + // 800 samples at 8000 S/s = 0.1s of recording. Generous lower + // bound only -- never an upper bound, that is CI-flaky. + assert!( + start.elapsed() >= Duration::from_millis(90), + "the first read returned after {:?}, before its own 0.1s of \ + recording had elapsed", + start.elapsed() + ); + } + + // The recording clock must start at the FIRST READ, not at + // construction. `manta-cli` builds the paced source, then resolves the + // replay epoch, hashes the whole recording for the session nonce, and + // binds the telnet/JSON listeners before `listen()` ever reads a + // sample. Charging that setup time to the recording meant the first + // buffer -- `listen`'s whole two-second calibration read -- came due + // that much sooner, eating into the window a client has to connect + // after the servers are actually up. + #[test] + fn paced_source_clock_starts_at_the_first_read_not_at_construction() { + let samples = vec![Complex32::new(0.0, 0.0); 800]; + let src = VecSource { + samples, + cursor: 0, + fs: 8000.0, + }; + let mut paced = PacedSource::new(Box::new(src)); + // Stand in for the CLI's own between-construction-and-first-read + // setup: epoch resolution, whole-file hashing, server bind. + std::thread::sleep(Duration::from_millis(150)); + assert_eq!( + paced.elapsed(), + Duration::ZERO, + "the recording clock must not run before the first read" + ); + let mut buf = vec![Complex32::new(0.0, 0.0); 800]; + let start = Instant::now(); + assert_eq!(paced.read(&mut buf).unwrap(), 800); + // 800 samples at 8000 S/s = 0.1s of recording, owed in FULL from + // this point -- not reduced by the 150ms of setup above. Generous + // lower bound only; never an upper bound, that is CI-flaky. + assert!( + start.elapsed() >= Duration::from_millis(90), + "the first read returned after {:?}, so the 150ms of setup \ + before it was credited against the recording's own clock", + start.elapsed() + ); + } + + #[test] + fn paced_source_passes_eof_through() { + let samples = vec![Complex32::new(0.0, 0.0); 5]; + let src = VecSource { + samples, + cursor: 0, + fs: 8000.0, + }; + let mut paced = PacedSource::new(Box::new(src)); + let mut buf = vec![Complex32::new(0.0, 0.0); 5]; + assert_eq!(paced.read(&mut buf).unwrap(), 5); + assert_eq!(paced.read(&mut buf).unwrap(), 0); + // EOF must not advance the delivered counter, or every further + // read at EOF would accrue a larger and larger sleep debt against + // a source that has nothing left to give. Asserted on the counter + // and on the pacing decision rather than on a wall-clock upper + // bound for the EOF read, which a loaded CI runner can break for + // reasons that have nothing to do with pacing. + assert_eq!(paced.delivered, 5, "EOF must not advance `delivered`"); + // The whole debt a 5-sample read at 8 kS/s can ever ask for is + // 5/8000 s, and it is already spent by the time EOF is reached. + assert_eq!( + pacing_delay(5, 8000.0, Duration::ZERO), + Some(Duration::from_secs_f64(5.0 / 8000.0)) + ); + assert_eq!( + pacing_delay(paced.delivered, paced.fs, paced.elapsed()), + None, + "the 625us debt is long spent -- EOF reads must not sleep" + ); + } +} diff --git a/crates/manta-input/src/replay.rs b/crates/manta-input/src/replay.rs new file mode 100644 index 00000000..d9828697 --- /dev/null +++ b/crates/manta-input/src/replay.rs @@ -0,0 +1,95 @@ +//! Looping file replay (`--loop`, MAN-121). + +use crate::{open_replay_wav, IqSource}; +use anyhow::Result; +use num_complex::Complex32; +use std::path::PathBuf; + +/// Reopens the file at EOF so replay never ends. +/// +/// Reopening (rather than rewinding) is what lets this work for BOTH +/// replay flavours -- `WavIqSource` holds a cursor with no public rewind, +/// and `AudioIqSource` wraps an opaque `coppa_audio::AudioSource` with no +/// rewind of its own either. +/// +/// The wrap point is a hard discontinuity in the sample stream (last +/// sample straight to first), which the channelizer sees as a click. +/// Harmless for a demo; this is not a substitute for a genuinely long +/// recording. +pub struct LoopingWavSource { + inner: Box, + path: PathBuf, +} + +impl LoopingWavSource { + pub fn new(path: PathBuf) -> Result { + let inner = open_replay_wav(&path)?; + Ok(LoopingWavSource { inner, path }) + } +} + +impl IqSource for LoopingWavSource { + fn sample_rate(&self) -> f64 { + self.inner.sample_rate() + } + + fn center_freq_hz(&self) -> f64 { + self.inner.center_freq_hz() + } + + fn read(&mut self, buf: &mut [Complex32]) -> Result { + let n = self.inner.read(buf)?; + if n > 0 { + return Ok(n); + } + self.inner = open_replay_wav(&self.path)?; + self.inner.read(buf) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn write_f32_wav(path: &std::path::Path, samples: &[Complex32], fs: u32) { + let spec = hound::WavSpec { + channels: 2, + sample_rate: fs, + bits_per_sample: 32, + sample_format: hound::SampleFormat::Float, + }; + let mut w = hound::WavWriter::create(path, spec).unwrap(); + for s in samples { + w.write_sample(s.re).unwrap(); + w.write_sample(s.im).unwrap(); + } + w.finalize().unwrap(); + } + + #[test] + fn loops_past_end_of_file_reproducing_the_same_samples() { + let dir = tempfile::tempdir().unwrap(); + let wav = dir.path().join("short.wav"); + let samples: Vec = (0..10) + .map(|i| Complex32::new(i as f32, -(i as f32))) + .collect(); + // 96 kHz, not an arbitrary rate: `open_replay_wav` now rejects an + // IQ WAV whose rate the channelizer can't accept (MAN-121 review), + // and 48 kHz would take the sidecarless-stereo tie-break down the + // AudioIqSource path instead. This test is about loop-past-EOF + // semantics, not rate. + write_f32_wav(&wav, &samples, 96_000); + + let mut src = LoopingWavSource::new(wav).unwrap(); + let mut buf = vec![Complex32::new(0.0, 0.0); 10]; + + // First pass. + assert_eq!(src.read(&mut buf).unwrap(), 10); + assert_eq!(buf, samples); + + // EOF triggers a reopen; the second pass reproduces the same data + // rather than returning 0 and stopping. + assert_eq!(src.read(&mut buf).unwrap(), 10); + assert_eq!(buf, samples); + } +} diff --git a/docs/DECISIONS/2026-09-07-man121-hardware-free-replay.md b/docs/DECISIONS/2026-09-07-man121-hardware-free-replay.md new file mode 100644 index 00000000..28833c3c --- /dev/null +++ b/docs/DECISIONS/2026-09-07-man121-hardware-free-replay.md @@ -0,0 +1,206 @@ +# MAN-121: a hardware-free path to a real spot on the telnet port + +MAN-121 (2026-09-05 broad review, hit list row O-01) found the README's own +quickstart couldn't reach the server it documents: `manta gen v1 --out +/tmp/v1` writes 2-channel 96 kHz IQ, but `manta listen --source` only +accepted 48 kHz mono audio (`AudioIqSource`), hard-rejecting with +`"AudioIqSource requires 48000 Hz, got 96000"`. Separately, file replay had +no pacing at all -- a 120 s recording drained in ~3.5 s, so the process +(and its telnet/JSON servers) exited before any client could plausibly +connect and observe a spot. Together, ROADMAP.md's M3 acceptance gate ("a +stock DX cluster client connects and receives well-formed spots") was not +verifiable by anyone without SDR hardware. + +## Decision 1 — dispatch on WAV channel count and rate/sidecar, not a new flag or vector family + +`manta_input::open_replay_wav(path)` reads the WAV header: 2 channels at any +rate other than `TARGET_RATE_HZ` (48 kHz), or 2 channels at 48 kHz with a +parseable `.json` sidecar, means complex IQ (`WavIqSource`, +sidecar-aware -- what `gen`/`decode` already use); anything else is treated +as a real rig-audio passband (`AudioIqSource`, still 48 kHz-only, unchanged). + +Rejected alternatives: +- **Resampling `AudioIqSource`'s input to 48 kHz.** Not just unnecessary but + wrong: `manta_engine::listen()` is already rate-agnostic + (`Channelizer::new` accepts any `fs` where `fs / 93.75` is a power of two, + and `96000 / 93.75 = 1024`). A resample step would be lossy, slower, and + would put a non-trivial DSP stage in the determinism-critical path for no + benefit. This was a reader-choice bug, not a rate-conversion gap. +- **A `gen --audio-48k` variant.** Every vector (V1-V10) hardcodes + `fs: 96_000.0` and is a frozen golden vector; a new 48 kHz family would + still leave `gen v1`'s actual output unusable by `listen`, which is + exactly what the ticket's first scenario complains about. +- **Relaxing `AudioIqSource`'s 48 kHz requirement.** It's correct for its + scenario and pinned in `docs/DECISIONS/2026-07-17-m1-implementation-pins.md` + pin 3 (`coppa-audio`'s resampler is unreachable at M1). + +**Round-1 review correction:** channel count *alone* is not the +discriminator this section originally claimed. `AudioIqSource` never +accepts anything but 48 kHz, so a 2-channel file at any other rate could +never have been valid rig audio before this dispatch existed -- +unambiguous. But at exactly 48 kHz, a 2-channel file is genuinely +ambiguous: it's both a legal `WavIqSource` rate (`48000 / 93.75 = 512`, a +valid channelizer rate) and `AudioIqSource`'s only rate, and a stereo +soundcard recording of a receiver's passband is indistinguishable from IQ +by channel count alone. Dispatching on channel count regardless of rate +silently misread that recording as `Complex32::new(I, Q)` -- wrong output, +no error, no override. The fix keeps channel count as the discriminator +away from 48 kHz, and at 48 kHz breaks the tie with the same `.json` +sidecar Decision 2 already uses for RF-awareness: present means IQ, absent +means the pre-existing `AudioIqSource` downmix path. `coppa_audio::WavSource` +(which `AudioIqSource` wraps) downmixes any multi-channel WAV by *striding* +(`step_by(channels)`), discarding channel 1 -- correct for rig audio, wrong +for IQ, which is exactly why the 48 kHz case needs the sidecar tie-break +rather than channel count alone. + +## Decision 2 — keep the `--dial-freq-hz` gate where it is; make it format-aware with a never-failing probe + +Fixing Decision 1 alone still leaves `listen --source +--server-config ...` failing with `"--dial-freq-hz is required..."`, since +that gate keyed on *source kind* (file vs. KiwiSDR/SoapySDR), not on +whether the source actually knows its RF frequency. A `gen` IQ WAV *does* +know it -- its `.json` sidecar carries `center_freq_hz`, which +`WavIqSource::open` already reads. + +`manta_input::replay_wav_center_freq_hz(path)` returns the RF center a +replay WAV *declares* (2 channels + a parseable sidecar with a finite, +positive `center_freq_hz`), and returns `None` on **any** error, including +a nonexistent path. The gate folds this into `has_rf_aware_source` and +stays exactly where it was in the CLI handler, ahead of all real file I/O. + +**Round-2 review correction:** Decision 1's format tie-break must *not* +reuse this RF probe. The first cut did, so a 48 kHz 2-channel IQ file whose +sidecar declares `center_freq_hz: 0.0` (baseband, or an unknown dial the +operator supplies with `--dial-freq-hz`) failed the positive-frequency test +and was classified as rig audio -- its Q channel discarded and re-synthesized +through the Hilbert path, contradicting Decision 1's own "parseable sidecar" +wording. The two questions are now separate functions over one private +helper: `replay_wav_has_iq_sidecar` (FORMAT -- finite `center_freq_hz`, +`0.0` included) drives `open_replay_wav`, and `replay_wav_center_freq_hz` +(RF -- finite and positive) drives this gate alone. Both still swallow every +error, so the ordering invariant below is untouched. + +This preserves an existing, test-enforced invariant: +`crates/manta-cli/tests/cli.rs::server_config_without_dial_freq_for_audio_source_is_a_clean_error` +provokes the gate with `/nonexistent.wav` and asserts the error names the +*flag*, not a file-open failure -- i.e. flag validation must run before any +file is touched. A probe that could itself fail on a bad path would invert +that ordering; swallowing every error (rather than propagating one) is +what keeps it intact. + +## Decision 3 — `--realtime` is opt-in, not the default + +The ticket's own Gherkin permits this explicitly: "paced in real time by +default (**or an explicit `--realtime`/`--loop` flag is available**)." +Taking the flag branch: + +1. The broad review's protect-list states pacing "must be opt-in and never + touch the decode path." +2. Defaulting would slow `crates/manta-cli/tests/cli.rs`'s replay tests and + `manta soak --source` (whose entire purpose is compressed-time replay -- + see `crates/manta-cli/tests/soak_ci.rs`, which replays 120 s of scene as + a CI proxy for an hour-long run) by the same ~30x, defeating `soak`. +3. Determinism is a hard repo requirement. `--realtime` output is + byte-identical to unpaced output (`PacedSource` only ever decides *when* + a chunk is delivered, never *which* samples) -- + `crates/manta-cli/tests/cli.rs::realtime_replay_is_byte_identical_to_unpaced_replay` + enforces this -- so the flag is safe either way, but the default should + not move. + +`PacedSource` (`crates/manta-input/src/pace.rs`) tracks cumulative +delivered samples against a single `Instant` -- captured lazily on the +FIRST `read()`, not at construction, because `manta-cli` resolves the +replay epoch, hashes the whole recording for the session nonce, and binds +the telnet/JSON listeners in between, and charging that setup time to the +recording's own clock shortened the very window `--realtime` exists to +open -- and sleeps only for `due - elapsed` when positive. **Round-2 review +correction:** that sleep happens *after* the inner `read()`, against the +count that includes the buffer about to be returned. Sleeping first, against +the previous count, handed every chunk to the consumer a chunk before its own +recording interval had elapsed -- and `manta_engine::listen`'s first read is +the entire two-second calibration buffer, so a spot decoded from it could be +published before a client had any chance to connect, which is the exact +failure `--realtime` exists to prevent. A consumer that falls behind +the recording simply stops sleeping and degrades to unpaced -- it can never +stall the pipeline or accumulate drift from repeated small delays. + +`--realtime` and `--loop` both require `--source` (clap `requires = +"source"`): both are meaningless for a live device or a network SDR +(already realtime, never EOF), and a misuse should be a clap usage error, +not a silent no-op. + +## Decision 4 — `--loop` ships, with its dedupe interaction documented, not "fixed" + +`LoopingWavSource` (`crates/manta-input/src/replay.rs`) reopens the file at +EOF via `open_replay_wav` (rather than rewinding), which is what lets it +work for both replay flavours -- `WavIqSource` has no public rewind, and +`AudioIqSource` wraps an opaque `coppa_audio::AudioSource` with none either. +Loop wraps the *innermost* source; `--realtime` pacing wraps *around* the +looped stream (so pacing measures one continuous timeline rather than +restarting its clock every pass). + +`SUPPRESSION_SECONDS = 600.0` (`crates/manta-spot/src/dedupe.rs`) +suppresses a repeat spot for the same (callsign, 300 Hz freq bucket) for +10 minutes of *simulated* time. A short looped file therefore yields +roughly one spot per 10 minutes of looped playback, not one per pass -- +this is correct skimmer behavior (a real skimmer shouldn't re-spot the same +station every time it repeats CQ) and is not special-cased away. It's +stated in the flag's own `--help` text instead of "fixed," since fixing it +would mean weakening dedupe, an unrelated and unwanted change. + +**Amendment (round-14 review): `--loop` requires `--realtime` when +`--server-config` is given.** An unpaced loop is unbounded in both +directions -- it never reaches EOF, and its sample clock advances ~30-40x +faster than wall time. `Dedupe` therefore releases another spot every 600 s +of *simulated* time while `SpotBus::unix_ts_for` adds that runaway +`sample_ts` to the fixed replay epoch, so a networked unpaced loop +publishes an endless series of spots stamped progressively further into the +future -- to real telnet/JSON clients that have no way to distinguish them +from current observations. Rejected at startup, next to the `--dial-freq-hz` +gate and ahead of all file I/O, so it stays a flag error rather than a file +error. Looping *without* `--server-config` is unrestricted: it publishes to +nobody, and the local event stream is sample-relative. + +## Decision 5 — the M3 README demo needs `sh/dx`, not just perfect timing + +Measured empirically while writing the README's Quickstart update: with +`--realtime` alone, the vector's one spot is published to the live +`SpotBus` broadcast channel roughly 12 s into the run (2 s calibration + +SPEC §2.1's ~2.05 s warmup/confirm floor + the vector's own lead-in) and, +because `SUPPRESSION_SECONDS` exceeds V1's full 120 s duration, never +again. `SpotBus`'s live channel has explicit no-history-for-late-subscribers +semantics (`crates/manta-server/src/telnet.rs`) -- a client that connects +after that one moment sees nothing on the live stream, confirmed by hand: +connecting at t=15s (with or without `--loop`) produced only the login +banner, no spot. `sh/dx`, however, reads the bus's *retained history* +rather than the live subscription +(`crates/manta-server/tests/telnet_acceptance.rs::sh_dx_replays_recent_spot_history_in_rbn_format`), +and does return the spot regardless of when the client connected during the +run -- also confirmed by hand. The README's Quickstart therefore tells the +reader to type `sh/dx` at the prompt if the spot doesn't arrive on its own, +rather than relying on the reader switching terminals inside a ~12 s +window. This is a documentation fix, not a code change -- `sh/dx` already +existed; the gap was only that the README never told a hardware-free reader +about it. + +## Verification + +- `crates/manta-input/src/lib.rs` -- `open_replay_wav`/ + `replay_wav_center_freq_hz` unit tests (dispatch, sidecar parsing, the + never-fails contract). +- `crates/manta-input/src/pace.rs` -- `PacedSource` unit tests (sample + identity, minimum elapsed time, no drift compounding under a slow + consumer, EOF pass-through). +- `crates/manta-input/src/replay.rs` -- `LoopingWavSource` unit test + (reopens and reproduces the same samples past EOF). +- `crates/manta-cli/tests/cli.rs` -- CLI-level coverage for the format + dispatch, the relaxed/still-enforced `--dial-freq-hz` gate, realtime + byte-identity, `--loop` liveness, and `requires = "source"` usage errors. + The pre-existing + `server_config_without_dial_freq_for_audio_source_is_a_clean_error` and + `json_output_is_valid_and_deterministic_across_three_runs` regression + guards stay green. +- `crates/manta-cli/tests/telnet_e2e.rs` (new) -- the M3 acceptance gate, + automated: a real child process, real sockets, a real `manta gen`-shaped + fixture, a real RBN line on the telnet port and a real JSON Lines message + on the JSON port, no SDR and no hand-built recording. diff --git a/wiki/INDEX.md b/wiki/INDEX.md index 05494315..4a57fcf6 100644 --- a/wiki/INDEX.md +++ b/wiki/INDEX.md @@ -12,4 +12,5 @@ - [What spot-output contracts does manta expose (telnet RBN + JSON)?](pages/spot-output-contract.md) — manta produces spots on two surfaces: a **telnet DX cluster server** (default :7300) emitting s... - [Why is the weekly Cargo Dependabot run red with `unknown_error` / `null`?](pages/dependabot-cargo-unlock.md) — Diagnose Cargo updates that Dependabot cannot reach through its single-package unlock scope. - [What's the PR review convergence policy?](../docs/DECISIONS/2026-08-07-pr-review-convergence-policy.md) — P1 findings get fixed every review round; P2-and-lower findings raised after round one are tracked in a follow-up ticket instead of chased inline, so PRs converge instead of oscillating. +- [Why did `listen --source` reject `gen`'s own output, and how was it fixed?](pages/replay-input-dispatch.md) — `listen()` is rate-agnostic; the 48 kHz constraint belongs to `AudioIqSource` alone. Covers the IQ/audio WAV dispatch, `--realtime`/`--loop`, and `SpotBus`'s no-history-for-late-subscribers gotcha. - [What changed in the 2026-09-06 broad review?](../docs/DECISIONS/2026-09-06-broad-review-decisions.md) — RBN admission strategy, the wire SNR convention, multi-band identity, pausing the Pi4 CPU-budget story, and several other decisions that revise or supersede AGENTS.md, SPEC-decode-core.md, the legacy capability matrix, and the MAN-23 threat model on specific points. diff --git a/wiki/pages/replay-input-dispatch.md b/wiki/pages/replay-input-dispatch.md new file mode 100644 index 00000000..4424a189 --- /dev/null +++ b/wiki/pages/replay-input-dispatch.md @@ -0,0 +1,143 @@ +--- +id: replay-input-dispatch +title: Why did `listen --source` reject `gen`'s own output, and how was it fixed? +kind: gotcha +status: current +maintainer: agent +sources: + - crates/manta-input/src/lib.rs + - crates/manta-input/src/audio.rs + - crates/manta-input/src/pace.rs + - crates/manta-input/src/replay.rs + - docs/DECISIONS/2026-09-07-man121-hardware-free-replay.md +# `verified:` is deliberately UNSET. The only revision this page could +# have cited when it was written (49f05a4) predates the page itself and +# every module it describes -- `pace.rs`, `replay.rs`, `open_replay_wav` -- +# so recording it would have made the provenance marker actively +# misleading to a maintainer reading this field (MAN-121 review). Set it +# in a later pass, against a revision that actually contains the behaviour +# described below. +links: + - spot-output-contract +--- +`manta_engine::listen()` itself is **rate-agnostic** — `Channelizer::new` +accepts any `fs` where `fs / 93.75` is a power of two (96000 is; so is +48000). The 48 kHz constraint belongs to exactly one `IqSource` +implementation, `AudioIqSource` (`crates/manta-input/src/audio.rs`), not to +the streaming pipeline. Misreading that constraint as belonging to +`listen()` itself is what made MAN-121 look like a resampling problem for +longer than it was: it wasn't. `manta gen`/`decode` always use +`WavIqSource` (2-channel complex IQ, any channelizer rate, +sidecar-carried center freq); `listen --source`/`--device` used to +hard-code `AudioIqSource` (a real, single-channel rig-audio passband +converted to analytic form via Hilbert transform) regardless of what the file actually contained — so +`gen`'s own 96 kHz stereo IQ output couldn't feed the server the README's +quickstart told you to point it at. + +Note also that "rate-agnostic" means *any channelizer rate*, not any rate +at all: `fs / 93.75` must be a power of two (12/24/48/96/192/384 kHz and +so on). A 44.1 kHz or 100 kHz IQ WAV is still rejected — +now by `open_replay_wav` itself, up front. It used to be rejected only +later, by `Channelizer::new`, and `manta_engine::listen` sizes a +two-second complex calibration buffer from `sample_rate()` *before* that +check runs, so a malformed header claiming a rate near `u32::MAX` asked +the allocator for tens of GiB and aborted the process instead of printing +the unsupported-rate error (MAN-121 review). `Channelizer::supports_rate` +is the shared predicate. + +`supports_rate` is only half the guard, though, because it bounds the +rate's *shape* and not its *size*: 393216000 (= 93.75 × 2²²) is a +perfectly well-shaped channelizer rate, so it passes — and still asks for +a ~5.9 GiB calibration buffer and a 2²²-point FFT from a file that may be +a few hundred bytes long. `open_replay_wav` therefore also bounds the +magnitude, at `manta_input::MAX_REPLAY_RATE_HZ` (10 MS/s — the same +ceiling `--hpsdr-rate` already applies to a live source, so replay is +bounded no more tightly than real hardware). Both halves are needed; +neither implies the other (MAN-121 round-15 review). + +## The fix: dispatch on channel count, with a rate-and-sidecar tie-break at 48 kHz + +`manta_input::open_replay_wav(path)` picks the reader from the WAV header, +and **not** from channel count alone — the shipped rule is: + +| WAV header | Reader | +| --- | --- | +| 2 channels, rate ≠ 48 kHz | `WavIqSource` (complex IQ) | +| 2 channels, 48 kHz, parseable `.json` sidecar | `WavIqSource` (complex IQ) | +| 2 channels, 48 kHz, no/unparseable sidecar | `AudioIqSource` (downmix + Hilbert) | +| anything else (e.g. mono) | `AudioIqSource` (unchanged, still 48 kHz-only) | + +The 48 kHz row is the whole subtlety, and it is deliberate: 48 kHz is +`AudioIqSource`'s *only* rate, so a stereo soundcard recording of a +receiver's passband is indistinguishable from IQ by channel count alone, +and reading it as `Complex32::new(I, Q)` would silently corrupt it. Away +from 48 kHz there is no such collision — `AudioIqSource` never accepted +those rates — so channel count decides on its own. Do not "simplify" this +back to `channels == 2`. + +Two separate sidecar probes back that table, and conflating them was a +real round-2 review bug: + +- `manta_input::replay_wav_has_iq_sidecar(path)` — the **format** + question, and the one the 48 kHz tie-break asks. True for 2 channels + plus a parseable sidecar with a *finite* `center_freq_hz`, **including + `0.0`** (baseband, or a dial frequency the operator will pass with + `--dial-freq-hz`). +- `manta_input::replay_wav_center_freq_hz(path)` — the **RF** question, + and the one the CLI's `--dial-freq-hz` gate asks. `Some` only for a + finite *positive* `center_freq_hz`. + +Using the RF probe for the format tie-break (as the first cut did) sends a +48 kHz IQ file declaring `center_freq_hz: 0.0` down the Hilbert path, +discarding Q, even when `--dial-freq-hz` was supplied. + +Both probes are written to **never fail**, even for a nonexistent path — +the gate must report a missing flag before a missing file, and a probe +that could itself error would invert that ordering. See +`docs/DECISIONS/2026-09-07-man121-hardware-free-replay.md` for the full +rationale and rejected alternatives (resampling, a new 48 kHz vector +family). + +## The other half: file replay had no pacing at all + +`listen()`'s read loop had no pacing at all before MAN-121 — a file source +drained at whatever speed `read()` could return data (measured ~30-40x +realtime), so the process (and its telnet/JSON servers) could exit before +any client had a realistic chance to connect. `PacedSource` +(`crates/manta-input/src/pace.rs`) is a pure sleep wrapper around any +`IqSource`: it computes sleeps from *cumulative* delivered samples against +one `Instant` (started on the first `read()`, so CLI setup between +construction and the first sample — epoch resolution, whole-file hashing, +server bind — is not charged to the recording's clock), so a slow consumer +degrades to unpaced instead of ever +stalling or drifting — and because it never touches which samples are +delivered, `--realtime` output is byte-identical to unpaced output. The +sleep happens **after** the inner `read()`, against the count that +*includes* the buffer about to be returned: pacing against the previous +count (the first cut) returned every chunk a chunk early, and `listen()`'s +first read is the whole two-second calibration buffer, so a spot decoded +from it could reach the servers before any client could connect. Opt-in +only (`--realtime`), since the default path (tests, `soak --source`) still +wants full-speed drain. `--loop` (`LoopingWavSource`, +`crates/manta-input/src/replay.rs`) reopens the file at EOF for a +demo left running; combined with the 600 s spot-dedupe window, a short +looped file yields roughly one spot per 10 minutes, not one per pass — by +design, not a bug to chase. `--loop` **requires** `--realtime` when +`--server-config` is given: an unpaced loop never ends and runs its sample +clock ~30–40× faster than wall time, and `SpotBus::unix_ts_for` adds that +runaway `sample_ts` to the fixed replay epoch, so clients would receive +spots timestamped ever further into the future. Looping without +`--server-config` (nobody to publish to) stays unrestricted. + +## Gotcha: `SpotBus`'s live channel has no history for late subscribers + +A client that connects to the telnet/JSON servers *after* a file-replay +spot has already been published will **not** see it on the live stream — +confirmed by hand, connecting well after the one spot a short `--realtime` +run produces yields nothing further, `--loop` or not, until the next +10-minute dedupe cycle. `sh/dx`, however, reads the bus's retained +history rather than the live subscription, and *does* return a spot no +matter when the client connected during the run. This is why the README's +hardware-free demo tells the reader to try `sh/dx` if nothing arrives +within the first ~12 s, rather than relying on split-second terminal +switching.