diff --git a/ac-rs/ZMQ.md b/ac-rs/ZMQ.md index c1ae7521..7bccafab 100644 --- a/ac-rs/ZMQ.md +++ b/ac-rs/ZMQ.md @@ -1230,12 +1230,30 @@ same applied value. // topic: data (measurement/frequency_response/point frame, see Shared types) ``` -**DATA** — terminal: +**DATA** — terminal, success: ```json // topic: done { "cmd": "plot", "n_points": , "xruns": } ``` +`xruns` is the session delta — `AudioEngine::xruns()` sampled once at +engine creation and once at sweep completion, then subtracted +(wrapping-safe) — not a sum of that per-point cumulative reading (#428). + +**DATA** — terminal, analyzer failure (#428): +```json +// topic: error +{ "cmd": "plot", "message": "", "requested_points": , "completed_points": } +``` + +An analyzer failure at any point aborts the sweep atomically: the engine +stops, this `error` is published, and none of `measurement/frequency_ +response/complete`, `measurement/report`, the report file, or `done` +follow — a failed sweep never archives its completed prefix as a +successful measurement. `requested_points` is the full sweep's point +count; `completed_points` is how many points had already published a +`measurement/frequency_response/point` frame before the failure. + --- ### `plot_level` @@ -1270,12 +1288,22 @@ exceeds the ceiling flattens there rather than running unclamped. includes `"freq_hz"` and `"drive_db"` fields — `drive_db` is the applied, post-clamp level for that step). -**DATA** — terminal: +**DATA** — terminal, success: ```json // topic: done { "cmd": "plot_level", "n_points": , "xruns": } ``` +`xruns` is the session delta, same accounting as `plot`'s (#428). + +**DATA** — terminal, analyzer failure (#428): same shape and same +atomic-failure guarantee as `plot`'s, above, with `"cmd": "plot_level"` +and `requested_points` the level-step count (`steps`). +```json +// topic: error +{ "cmd": "plot_level", "message": "", "requested_points": , "completed_points": } +``` + --- ### `monitor_spectrum` @@ -2963,6 +2991,8 @@ When the guard fires: // topic: error { "cmd": "", "message": "" } ``` +`plot`/`plot_level` add `requested_points`/`completed_points` to this +shape on an analyzer failure (#428) — see their sections above. ### Unparseable config.json (#370) ```json diff --git a/ac-rs/crates/ac-cli/src/commands/plot.rs b/ac-rs/crates/ac-cli/src/commands/plot.rs index b18fbe1f..074e589d 100644 --- a/ac-rs/crates/ac-cli/src/commands/plot.rs +++ b/ac-rs/crates/ac-cli/src/commands/plot.rs @@ -68,8 +68,8 @@ pub fn run( launch_ui(LaunchKind::SweepFreq, cfg, None); } - let results = collect_sweep(client, "plot"); - if results.is_empty() { + let (results, outcome) = collect_sweep(client, "plot"); + if outcome != SweepOutcome::Done || results.is_empty() { return; } io::print_summary(&results, "DUT", have_cal); @@ -140,8 +140,8 @@ pub fn run_level( launch_ui(LaunchKind::SweepLevel, cfg, None); } - let results = collect_sweep(client, "plot_level"); - if results.is_empty() { + let (results, outcome) = collect_sweep(client, "plot_level"); + if outcome != SweepOutcome::Done || results.is_empty() { return; } io::print_summary(&results, "DUT", have_cal); @@ -415,11 +415,34 @@ fn print_ir_notes(report_frame: Option<&serde_json::Value>) { } } -fn collect_sweep(client: &mut AcClient, cmd_name: &str) -> Vec { +/// Whether a sweep reached its terminal `done` frame. Anything else — a +/// terminal `error` (analyzer failure, #428) or a timeout — leaves +/// `results` holding only a prefix that must never be treated as a +/// complete artifact: no summary printed, no CSV written. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SweepOutcome { + Done, + Failed, +} + +fn collect_sweep(client: &mut AcClient, cmd_name: &str) -> (Vec, SweepOutcome) { + collect_sweep_frames(|| client.recv_data(300_000), cmd_name) +} + +/// Core of `collect_sweep`, generic over the frame source so the +/// atomic-failure gating (a terminal `error` must never leave `outcome == +/// Done`, however many `measurement/frequency_response/point` frames +/// preceded it) can be unit-tested without a real `AcClient`/socket — +/// see `tests::error_after_points_is_failed_not_done` below (#428 QA). +fn collect_sweep_frames( + mut next_frame: impl FnMut() -> Option<(String, serde_json::Value)>, + cmd_name: &str, +) -> (Vec, SweepOutcome) { let mut results = Vec::new(); + let mut outcome = SweepOutcome::Failed; loop { - let frame = match client.recv_data(300_000) { + let frame = match next_frame() { Some(f) => f, None => { eprintln!("\n error: timeout waiting for {cmd_name} data"); @@ -441,17 +464,27 @@ fn collect_sweep(client: &mut AcClient, cmd_name: &str) -> Vec { + format!(" ({completed} of {requested} points completed; no report written)") + } + _ => String::new(), + }; + eprintln!("\n !! {msg}{partial}"); break; } } - results + (results, outcome) } fn save_results(results: &[serde_json::Value], label: &str, cfg: &ac_core::config::Config) { @@ -462,6 +495,69 @@ fn save_results(results: &[serde_json::Value], label: &str, cfg: &ac_core::confi io::save_csv(results, &path); } +#[cfg(test)] +mod tests { + use super::{collect_sweep_frames, SweepOutcome}; + use std::collections::VecDeque; + + fn point(freq_hz: f64) -> serde_json::Value { + serde_json::json!({ + "type": "measurement/frequency_response/point", + "freq_hz": freq_hz, + }) + } + + /// PR #451 QA finding (#428): a terminal `error` after some points had + /// already streamed must report `SweepOutcome::Failed` and only the + /// completed prefix — `run`/`run_level` gate `print_summary`/ + /// `save_results` on `outcome == Done`, so this is what makes the + /// atomic-failure guarantee reach the CLI's own summary/CSV output, + /// not just the daemon's wire frames. + #[test] + fn error_after_points_is_failed_not_done() { + let mut frames: VecDeque<(String, serde_json::Value)> = VecDeque::from([ + ("data".to_string(), point(100.0)), + ("data".to_string(), point(200.0)), + ( + "error".to_string(), + serde_json::json!({ + "cmd": "plot", + "message": "capture at 1000 Hz has 48 samples; minimum is 256", + "requested_points": 5, + "completed_points": 2, + }), + ), + // Must never be reached: a real daemon does not publish a + // point or `done` after a terminal `error`, and the loop must + // not either. + ("done".to_string(), serde_json::json!({"xruns": 0})), + ]); + + let (results, outcome) = collect_sweep_frames(|| frames.pop_front(), "plot"); + + assert_eq!(outcome, SweepOutcome::Failed); + assert_eq!( + results.len(), + 2, + "only the pre-failure points should be retained: {results:?}" + ); + } + + #[test] + fn done_after_points_is_done() { + let mut frames: VecDeque<(String, serde_json::Value)> = VecDeque::from([ + ("data".to_string(), point(100.0)), + ("data".to_string(), point(200.0)), + ("done".to_string(), serde_json::json!({"xruns": 0})), + ]); + + let (results, outcome) = collect_sweep_frames(|| frames.pop_front(), "plot"); + + assert_eq!(outcome, SweepOutcome::Done); + assert_eq!(results.len(), 2); + } +} + /// What `launch_ui` should do post-command. The GPU viewer this used to /// spawn is gone; `Monitor` now always renders via the /// terminal (`monitor_tui`), and the sweep variants just note that no diff --git a/ac-rs/crates/ac-daemon/src/audio/fake/hooks.rs b/ac-rs/crates/ac-daemon/src/audio/fake/hooks.rs index cfde9f7b..14f4c4da 100644 --- a/ac-rs/crates/ac-daemon/src/audio/fake/hooks.rs +++ b/ac-rs/crates/ac-daemon/src/audio/fake/hooks.rs @@ -107,11 +107,13 @@ pub(super) fn tau_noise_amplitude_override() -> f32 { /// go red under `--fake-audio`. /// /// `AC_FAKE_XRUNS_OVERRIDE`: comma-separated delta list, one value -/// consumed per `play_and_capture` call in this process (0-based — same -/// call indexing as [`TAU_DELAY_CALL_COUNT`] above, so slot *N* of this -/// list and slot *N* of the delay override line up with the same -/// `measure_tau_twice` lifecycle). A call past the end of the list adds 0. -/// Unset ⇒ every call adds 0, byte-identical to today's hardcoded-0 count. +/// consumed per `play_and_capture` call in this process (0-based: the +/// first call gets the first value); a call past the end of the list adds +/// 0. Unset ⇒ every call adds 0, byte-identical to today's hardcoded-0 +/// count. Deliberately scoped to `play_and_capture` alone — sharing this +/// counter with `capture_block` (below) would shift call indices for +/// every unrelated calibrate/monitor path that also captures via +/// `capture_block`, breaking the fixed indexing this doc promises. static XRUNS_CALL_COUNT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); fn xruns_override_list() -> &'static [u32] { @@ -130,3 +132,41 @@ pub(super) fn next_xruns_delta() -> u32 { let call_idx = XRUNS_CALL_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); xruns_override_list().get(call_idx).copied().unwrap_or(0) } + +/// Opt-in, fake-only test hook (#428): lets a test drive a `plot`/ +/// `plot_level` sweep's `capture_block` calls across a nonzero xrun count, +/// independent of [`next_xruns_delta`] above. Without this, a sweep that +/// only calls `capture_block` (`plot`, `plot_level` — never +/// `play_and_capture`) has no way to exercise a nonzero session xrun +/// delta under `--fake-audio`, and the #428 fix (report the delta since +/// baseline, not a per-point cumulative sum) has no reproduction outside +/// unit tests. +/// +/// `AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE`: comma-separated delta list, one +/// value consumed per `capture_block` call in this process (0-based). A +/// `plot`/`plot_level` point issues two calls — a discarded 0.1 s warm-up, +/// then the real capture — so both consume a slot. A call past the end of +/// the list adds 0. Unset ⇒ every call adds 0, unchanged from before #428. +static CAPTURE_BLOCK_XRUNS_CALL_COUNT: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); + +fn capture_block_xruns_override_list() -> &'static [u32] { + static LIST: std::sync::OnceLock> = std::sync::OnceLock::new(); + LIST.get_or_init(|| { + std::env::var("AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE") + .ok() + .map(|s| s.split(',').filter_map(|v| v.trim().parse().ok()).collect()) + .unwrap_or_default() + }) +} + +/// Next `capture_block` xrun delta, consuming one slot of the override +/// list (see [`CAPTURE_BLOCK_XRUNS_CALL_COUNT`] doc above). +pub(super) fn next_capture_block_xruns_delta() -> u32 { + let call_idx = + CAPTURE_BLOCK_XRUNS_CALL_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + capture_block_xruns_override_list() + .get(call_idx) + .copied() + .unwrap_or(0) +} diff --git a/ac-rs/crates/ac-daemon/src/audio/fake/mod.rs b/ac-rs/crates/ac-daemon/src/audio/fake/mod.rs index 75910512..1ea6342b 100644 --- a/ac-rs/crates/ac-daemon/src/audio/fake/mod.rs +++ b/ac-rs/crates/ac-daemon/src/audio/fake/mod.rs @@ -39,8 +39,8 @@ use anyhow::Result; use std::time::Duration; use self::hooks::{ - next_loopback_delay_samples, next_xruns_delta, period_size_override, tau_gain_override, - tau_noise_amplitude_override, + next_capture_block_xruns_delta, next_loopback_delay_samples, next_xruns_delta, + period_size_override, tau_gain_override, tau_noise_amplitude_override, }; use self::ring_mode::{FakeRings, RingDrain}; use self::stimulus::{Stimulus, StimulusGen, Synth}; @@ -234,6 +234,10 @@ impl AudioEngine for FakeEngine { if let Some(out) = self.ring_capture(n, duration, RingDrain::Block) { return Ok(out?.into_iter().next().unwrap_or_default()); } + // Opt-in xrun injection (#428) — see + // `hooks::next_capture_block_xruns_delta`'s doc. Inert (adds 0) + // unless `AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE` is set. + self.xruns += next_capture_block_xruns_delta(); std::thread::sleep(Duration::from_secs_f64(duration)); let port = self.input_port.clone(); let gain = tau_gain_override(); diff --git a/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs b/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs index 2124fbbf..9b1d7326 100644 --- a/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs +++ b/ac-rs/crates/ac-daemon/src/handlers/audio/plot.rs @@ -68,6 +68,12 @@ pub fn plot(state: &ServerState, cmd: &Value) -> Value { Err(e) => return json!({"ok": false, "error": e}), }; let backend = eng.backend_name(); + // Baseline taken right at engine creation (#428): the terminal xruns + // count is the wrapping-safe delta against this snapshot, not a sum of + // repeated cumulative reads — `AudioEngine::xruns()` already counts + // since start, so summing it once per point double- (then triple-, + // quadruple-...) counts every xrun that happened before the last point. + let xruns_start = eng.xruns(); let out_ch = cfg.output_channel; let in_ch = cfg.input_channel; let cal = cal_guard!(out_ch, in_ch); @@ -101,7 +107,6 @@ pub fn plot(state: &ServerState, cmd: &Value) -> Value { let sr = eng.sample_rate(); let mut n = 0usize; - let mut xruns = 0u32; let mut points: Vec = Vec::with_capacity(freqs.len()); let mut concat_capture: Vec = Vec::new(); for freq in &freqs { @@ -122,7 +127,6 @@ pub fn plot(state: &ServerState, cmd: &Value) -> Value { return; } }; - xruns += eng.xruns(); match ac_core::measurement::thd::analyze(&samples, sr, *freq, 10) { Ok(mut r) => { @@ -170,7 +174,27 @@ pub fn plot(state: &ServerState, cmd: &Value) -> Value { send_pub(&pub_tx, "data", &frame); n += 1; } - Err(e) => eprintln!("plot: analyze error at {freq}Hz: {e}"), + Err(e) => { + // Atomic failure exit (#428): an analyzer failure must + // not archive the successful prefix as a complete sweep. + // Stop the engine and publish the terminal error before + // any of `frequency_response/complete`, `measurement/ + // report`, the report file, or `done` — none of those + // run past this `return`. + eng.set_silence(); + eng.stop(); + send_pub( + &pub_tx, + "error", + &json!({ + "cmd": "plot", + "message": format!("{e}"), + "requested_points": freqs.len(), + "completed_points": n, + }), + ); + return; + } } if bpo.is_some() { concat_capture.extend_from_slice(&samples); @@ -179,6 +203,11 @@ pub fn plot(state: &ServerState, cmd: &Value) -> Value { eng.set_silence(); eng.stop(); + // Session xrun delta (#428): a single wrapping-safe subtraction + // against the baseline taken at engine creation, not a sum of + // repeated cumulative reads. + let xruns = eng.xruns().wrapping_sub(xruns_start); + let timestamp = ac_core::shared::time::now_utc_iso8601(); // Snapshot the processing-chain state at report-build time so a // re-loaded report tells the reader what was active during this @@ -346,6 +375,9 @@ pub fn plot_level(state: &ServerState, cmd: &Value) -> Value { Err(e) => return json!({"ok": false, "error": e}), }; let backend = eng.backend_name(); + // Baseline taken right at engine creation (#428) — see `plot`'s + // identical comment. + let xruns_start = eng.xruns(); let worker = spawn_worker(state, "plot_level", move |stop| { let mic_curve_opt = cal.as_ref().and_then(|c| c.mic_response.clone()); let spl_offset = cal.as_ref().and_then(Calibration::spl_offset_db); @@ -366,7 +398,6 @@ pub fn plot_level(state: &ServerState, cmd: &Value) -> Value { let sr = eng.sample_rate(); let mut n = 0usize; - let mut xruns = 0u32; for &level_req in &levels { if stop.load(Ordering::Relaxed) { break; @@ -389,7 +420,6 @@ pub fn plot_level(state: &ServerState, cmd: &Value) -> Value { return; } }; - xruns += eng.xruns(); match ac_core::measurement::thd::analyze(&samples, sr, freq_hz, 10) { Ok(mut r) => { @@ -423,11 +453,29 @@ pub fn plot_level(state: &ServerState, cmd: &Value) -> Value { send_pub(&pub_tx, "data", &frame); n += 1; } - Err(e) => eprintln!("plot_level: analyze error at {level_dbfs}dBFS: {e}"), + Err(e) => { + // Atomic failure exit (#428) — see `plot`'s identical + // comment: no `done` past this point for this sweep. + eng.set_silence(); + eng.stop(); + send_pub( + &pub_tx, + "error", + &json!({ + "cmd": "plot_level", + "message": format!("{e}"), + "requested_points": levels.len(), + "completed_points": n, + }), + ); + return; + } } } eng.set_silence(); eng.stop(); + // Session xrun delta (#428) — see `plot`'s identical comment. + let xruns = eng.xruns().wrapping_sub(xruns_start); send_pub( &pub_tx, "done", diff --git a/ac-rs/crates/ac-daemon/tests/it_protocol/basics.rs b/ac-rs/crates/ac-daemon/tests/it_protocol/basics.rs index ff5cf73f..9c3b9b0e 100644 --- a/ac-rs/crates/ac-daemon/tests/it_protocol/basics.rs +++ b/ac-rs/crates/ac-daemon/tests/it_protocol/basics.rs @@ -366,6 +366,47 @@ fn plot_frames_carry_processing_context_envelope() { assert!(mr["imported_at"].is_string()); } +/// #428: `AudioEngine::xruns()` is cumulative "since start" (see the trait +/// doc), so summing it once per sweep point double-, triple-, ...-counts +/// every xrun that happened before the sweep's last point. One real xrun +/// injected mid-sweep must show up as exactly 1 in the terminal `done` +/// frame, not accumulate across the remaining points. +/// +/// The fake backend's `capture_block` consumes one +/// `AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE` slot per call +/// (`audio/fake/hooks.rs`), and each sweep point issues two calls — a +/// discarded 0.1 s warm-up, then the real capture — so the 4th call +/// (index 3, 0-based) is point 1's real capture. +#[test] +fn plot_reports_session_xrun_delta_not_cumulative_sum() { + let d = Daemon::spawn_with_env(&[("AC_FAKE_CAPTURE_BLOCK_XRUNS_OVERRIDE", "0,0,0,1")]); + let c = Client::new(&d); + let r = c.call(json!({ + "cmd": "plot", + "start_hz": 100.0, + "stop_hz": 1000.0, + "level_dbfs": -20.0, + "ppd": 3, + "duration": 0.05, + })); + assert_eq!(r["ok"], json!(true), "plot ack: {r}"); + + let done = c + .wait_for_topic("done", Duration::from_secs(10)) + .expect("plot never finished"); + assert_eq!(done["cmd"], json!("plot")); + assert_eq!( + done["n_points"], + json!(3), + "all 3 points should have completed cleanly: {done}" + ); + assert_eq!( + done["xruns"], + json!(1), + "session xrun delta must be 1, not a per-point cumulative sum: {done}" + ); +} + // --------------------------------------------------------------------------- // server_idle_timeout — daemon folds the public bind back to localhost after // the configured idle CTRL-activity window expires. See issue #58. diff --git a/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs b/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs index aea41d00..d8fe712d 100644 --- a/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs +++ b/ac-rs/crates/ac-daemon/tests/it_protocol/out_of_range.rs @@ -1,5 +1,6 @@ use serde_json::json; use serde_json::Value; +use std::time::{Duration, Instant}; use crate::common::{Client, Daemon}; @@ -144,6 +145,70 @@ fn in_range_channels_are_unaffected() { c.call(json!({"cmd":"stop"})); } +/// #428: `duration: 0` makes `plot`'s per-point capture length +/// `max(duration, 3.0 / freq)`, which falls under `analyze`'s 256-sample +/// minimum for any point above 562.5 Hz at the fake backend's 48 kHz rate +/// (`3.0 / 562.5 * 48_000 == 256`). A sweep that fails partway through +/// must not archive the successful prefix as a complete measurement: it +/// must terminate on `error`, carrying how much of the request actually +/// completed, and never reach `done` or `measurement/report` for this run. +#[test] +fn plot_duration_zero_fails_atomically_instead_of_archiving_partial_sweep() { + let d = Daemon::spawn(); + let c = Client::new(&d); + let r = c.call(json!({ + "cmd": "plot", + "start_hz": 100.0, + "stop_hz": 2000.0, + "level_dbfs": -20.0, + "ppd": 5, + "duration": 0.0, + })); + assert_eq!(r["ok"], json!(true), "plot ack: {r}"); + + let deadline = Instant::now() + Duration::from_secs(10); + let mut error_frame: Option = None; + while Instant::now() < deadline && error_frame.is_none() { + let remaining = deadline + .saturating_duration_since(Instant::now()) + .as_millis() as i32; + match c.recv_pub(remaining.max(1)) { + Some((t, v)) if t == "error" => error_frame = Some(v), + Some((t, _)) if t == "done" => { + panic!("a sweep that failed partway through must not reach done") + } + Some((_, v)) if v["type"] == json!("measurement/report") => panic!( + "a sweep that failed partway through must not archive a measurement/report: {v}" + ), + Some(_) => continue, + None => break, + } + } + let err = error_frame.expect("plot never published a terminal error"); + assert_eq!(err["cmd"], json!("plot"), "frame: {err}"); + let message = err["message"].as_str().unwrap_or_default(); + assert!( + message.contains("256"), + "error message should name the analyzer's 256-sample minimum, got: {message:?}" + ); + let requested = err["requested_points"] + .as_u64() + .expect("requested_points missing"); + let completed = err["completed_points"] + .as_u64() + .expect("completed_points missing"); + assert!( + completed < requested, + "completed_points ({completed}) should be less than requested_points \ + ({requested}) — some points at/below 562.5 Hz must have succeeded \ + before the failure: {err}" + ); + assert!( + completed > 0, + "the low-frequency prefix should have completed: {err}" + ); +} + // --------------------------------------------------------------------------- // Multi-time-window ladder (handoff-mtw-live-spectrum) // ---------------------------------------------------------------------------