From 6b0b23ef697965fe98234108a10139ee50a35fa8 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Mon, 22 Jun 2026 03:34:29 -0400 Subject: [PATCH] fix(audio,game): Tier A bug-sweep fixes (S-01, F-04, F-08, F-09, S-02) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five confirmed findings from the 2026-06-22 adversarial bug sweep: - S-01: clamp PipeWire capture chunk size to the mapped slice before indexing, so a bad reported size can't panic (= process abort) from the RT capture callback. Extracted testable for_each_capture_sample. - F-04: reserve ring occupancy before publishing a frame on the PipeWire playback path (mirrors the cpal fix), preventing the RT consumer from popping an uncounted sample and wrapping fill_gauge to usize::MAX, which permanently wedged mixer pacing. Extracted publish_frame. - F-09: GameDetector::spawn now returns io::Result and retains its JoinHandle (joined on Drop); core fuses a closed watch receiver to None via next_game_change so a dead detector can't busy-loop select!. - F-08: collision-free recording paths — Recorder::create and the multitrack session dir use create_new/create_dir with bounded suffix retry, so two recordings in the same second no longer truncate the first. - S-02: bound the Windows SteamPath registry read (<=4 KiB, even length, re-checked type/returned length) before allocating/decoding. 403 lib tests pass (+6), clippy --all-targets clean. Implemented by Codex, reviewed + gates re-run by senior. Co-Authored-By: Codex Co-Authored-By: Claude Opus 4.8 --- src/audio/multitrack.rs | 39 ++++++++++++++++++ src/audio/pipewire_impl.rs | 83 +++++++++++++++++++++++++++++++++----- src/audio/recorder.rs | 66 +++++++++++++++++++++++++----- src/core/mod.rs | 70 +++++++++++++++++++++++++------- src/game/detector.rs | 25 +++++++++--- src/game/steam.rs | 52 ++++++++++++++++++++---- 6 files changed, 288 insertions(+), 47 deletions(-) diff --git a/src/audio/multitrack.rs b/src/audio/multitrack.rs index 5c5228f..497b7ab 100644 --- a/src/audio/multitrack.rs +++ b/src/audio/multitrack.rs @@ -29,6 +29,32 @@ const SILENCE_CHUNK: usize = FRAME_SAMPLES * 256; /// can drift if the capture clock runs ahead of the mixer cycle; past it the /// oldest mic audio is dropped. Mirrors `recorder::MAX_MIC_FIFO`. const MAX_MIC_FIFO: usize = 48_000 / 5; +const MAX_SESSION_DIR_ATTEMPTS: usize = 1_000; + +/// Create a collision-free session directory for a timestamp. The base +/// timestamp is tried first, followed by `-2`, `-3`, and so on; an existing +/// recording is never reopened or overwritten. +pub fn create_session_dir(base: &Path, now_unix_secs: u64) -> io::Result { + let filename = crate::audio::recorder::timestamp_filename(now_unix_secs); + let stem = filename.trim_end_matches(".wav"); + for attempt in 1..=MAX_SESSION_DIR_ATTEMPTS { + let name = if attempt == 1 { + stem.to_string() + } else { + format!("{stem}-{attempt}") + }; + let path = base.join(name); + match std::fs::create_dir(&path) { + Ok(()) => return Ok(path), + Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e), + } + } + Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "multitrack directory suffixes exhausted", + )) +} /// One output track: its WAV writer plus whether it has been written *this* /// cycle (so `end_cycle` knows which tracks to pad with silence). @@ -263,6 +289,19 @@ mod tests { assert_eq!(track_filename("!!!", &id), format!("peer-{short}.wav")); } + #[test] + fn same_second_sessions_get_unique_directories_without_reuse() { + let base = tmpdir("collision"); + let first = create_session_dir(&base, 1_700_000_000).unwrap(); + std::fs::write(first.join("sentinel"), b"keep me").unwrap(); + + let second = create_session_dir(&base, 1_700_000_000).unwrap(); + + assert_ne!(second, first); + assert_eq!(std::fs::read(first.join("sentinel")).unwrap(), b"keep me"); + let _ = std::fs::remove_dir_all(&base); + } + #[test] fn all_tracks_equal_length_after_n_cycles() { let dir = tmpdir("equal"); diff --git a/src/audio/pipewire_impl.rs b/src/audio/pipewire_impl.rs index b782ee8..d7b2506 100644 --- a/src/audio/pipewire_impl.rs +++ b/src/audio/pipewire_impl.rs @@ -151,11 +151,9 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender>, target_n let data = &mut datas[0]; let size = data.chunk().size() as usize; if let Some(slice) = data.data() { - // Each sample is 2 bytes (S16LE) - for chunk in slice[..size].chunks_exact(2) { - let sample = i16::from_le_bytes([chunk[0], chunk[1]]); + for_each_capture_sample(slice, size, |sample| { let _ = user_data.producer.try_push(sample); - } + }); } } } @@ -224,6 +222,16 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender>, target_n Ok(()) } +/// Visit the complete S16LE samples in the portion PipeWire reports as filled. +/// Clamp the reported byte count to the mapped slice before indexing: a bad +/// chunk size must not panic from the realtime capture callback. +fn for_each_capture_sample(slice: &[u8], size: usize, mut visit: impl FnMut(i16)) { + let size = size.min(slice.len()); + for chunk in slice[..size].chunks_exact(2) { + visit(i16::from_le_bytes([chunk[0], chunk[1]])); + } +} + /// Frames the playback RT callback should produce this cycle. /// /// `requested` is the graph's per-cycle quantum from `Buffer::requested()` (0 if @@ -263,6 +271,25 @@ fn drain_loop( } } +/// Reserve exact occupancy before making a frame visible to the consumer. +/// `after_reserve` is empty in production and lets the regression test force a +/// consumer interleaving at the critical ordering boundary. +fn publish_frame>( + fill: &AtomicUsize, + dropped: &AtomicU64, + producer: &mut P, + frame: &[i16], + after_reserve: impl FnOnce(), +) { + fill.fetch_add(frame.len(), Ordering::Relaxed); + after_reserve(); + let pushed = producer.push_slice(frame); + if pushed != frame.len() { + fill.fetch_sub(frame.len() - pushed, Ordering::Relaxed); + dropped.fetch_add(1, Ordering::Relaxed); + } +} + fn frames_to_produce(requested: usize, mapped_frames: usize) -> usize { /// Safe per-cycle fallback when the graph doesn't report a quantum. const FALLBACK_FRAMES: usize = 1024; @@ -522,10 +549,12 @@ fn run_playback( worker_dropped.fetch_add(1, Ordering::Relaxed); return; } - for &sample in &frame { - let _ = producer.try_push(sample); - } - worker_fill.fetch_add(frame.len(), Ordering::Relaxed); + // Reserve occupancy BEFORE publishing samples. Otherwise the RT + // consumer can pop a newly-visible sample before it is counted and + // wrap the exact fill gauge to usize::MAX, wedging mixer pacing. + // `push_slice` also publishes the frame as one operation rather than + // exposing a half-written stereo pair. + publish_frame(&worker_fill, &worker_dropped, &mut producer, &frame, || {}); }); }); @@ -577,8 +606,9 @@ fn run_playback( #[cfg(test)] mod tests { - use super::{drain_loop, frames_to_produce}; - use std::sync::atomic::{AtomicBool, Ordering}; + use super::{drain_loop, for_each_capture_sample, frames_to_produce, publish_frame}; + use ringbuf::{HeapRb, traits::{Consumer, Producer, Split}}; + use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use std::{sync::mpsc, thread}; @@ -614,6 +644,39 @@ mod tests { assert_eq!(frames_to_produce(1024, 0), 0); } + #[test] + fn capture_size_larger_than_mapping_is_clamped() { + let mut samples = Vec::new(); + for_each_capture_sample(&[1, 0, 2, 0, 3], usize::MAX, |sample| { + samples.push(sample) + }); + assert_eq!(samples, vec![1, 2]); + } + + #[test] + fn occupancy_is_reserved_before_frame_is_published() { + let rb = HeapRb::::new(8); + let (mut producer, mut consumer) = rb.split(); + assert!(producer.try_push(7).is_ok()); + + let fill = AtomicUsize::new(1); + let dropped = AtomicU64::new(0); + publish_frame(&fill, &dropped, &mut producer, &[10, 11], || { + // Force the consumer to drain the old sample after the new frame's + // occupancy is reserved but before that frame is published. + assert_eq!(consumer.try_pop(), Some(7)); + assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 3); + }); + + assert_eq!(fill.load(Ordering::Relaxed), 2); + assert_eq!(consumer.try_pop(), Some(10)); + assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 2); + assert_eq!(consumer.try_pop(), Some(11)); + assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 1); + assert_eq!(fill.load(Ordering::Relaxed), 0); + assert_eq!(dropped.load(Ordering::Relaxed), 0); + } + // --- drain_loop (A7: worker must not hang shutdown) --- #[test] diff --git a/src/audio/recorder.rs b/src/audio/recorder.rs index 473d11a..7c3615f 100644 --- a/src/audio/recorder.rs +++ b/src/audio/recorder.rs @@ -14,7 +14,7 @@ //! and patches the two size fields on [`Recorder::finalize`]. use std::collections::VecDeque; -use std::fs::File; +use std::fs::{File, OpenOptions}; use std::io::{self, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; @@ -24,6 +24,7 @@ const BITS_PER_SAMPLE: u16 = 16; const CHANNELS: u16 = 1; const RIFF_DATA_OVERHEAD: u64 = 36; const MAX_RIFF_DATA_BYTES: u64 = u32::MAX as u64 - RIFF_DATA_OVERHEAD; +const MAX_NAME_ATTEMPTS: usize = 1_000; /// Cap on buffered mic samples (~200ms). Bounds how far recording lag can drift /// if the capture clock runs persistently faster than playout — past this we drop @@ -42,7 +43,12 @@ pub struct WavWriter { impl WavWriter { /// Create the file and write the 44-byte header with zeroed size fields. pub fn new(path: &Path) -> io::Result { - let mut file = File::create(path)?; + Self::from_file(File::create(path)?) + } + + /// Start a WAV in an already-opened file. This lets callers choose atomic + /// create-new semantics instead of the truncating behavior of `File::create`. + fn from_file(mut file: File) -> io::Result { file.write_all(&Self::header(0))?; Ok(Self { file, @@ -125,13 +131,31 @@ impl Recorder { /// Create a recording at `dir/.wav`. The directory is assumed to /// exist (the caller creates it). pub fn create(dir: &Path, now_unix_secs: u64) -> io::Result { - let path = dir.join(timestamp_filename(now_unix_secs)); - let writer = WavWriter::new(&path)?; - Ok(Self { - writer, - mic_fifo: VecDeque::new(), - path, - }) + let filename = timestamp_filename(now_unix_secs); + let stem = filename.trim_end_matches(".wav"); + for attempt in 1..=MAX_NAME_ATTEMPTS { + let name = if attempt == 1 { + filename.clone() + } else { + format!("{stem}-{attempt}.wav") + }; + let path = dir.join(name); + match OpenOptions::new().write(true).create_new(true).open(&path) { + Ok(file) => { + return Ok(Self { + writer: WavWriter::from_file(file)?, + mic_fifo: VecDeque::new(), + path, + }); + } + Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e), + } + } + Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "recording filename suffixes exhausted", + )) } /// The path being written. @@ -210,6 +234,30 @@ mod tests { assert_eq!(timestamp_filename(0), "peerspeak-1970-01-01_000000.wav"); } + #[test] + fn same_second_recordings_get_unique_files_without_truncation() { + let dir = std::env::temp_dir().join(format!( + "peerspeak-collision-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + + let mut first = Recorder::create(&dir, 1_700_000_000).unwrap(); + first.write_frame(&[123, 456]).unwrap(); + let first_path = first.path().to_path_buf(); + first.finalize().unwrap(); + let original = std::fs::read(&first_path).unwrap(); + + let second = Recorder::create(&dir, 1_700_000_000).unwrap(); + let second_path = second.path().to_path_buf(); + assert_ne!(second_path, first_path); + assert_eq!(std::fs::read(&first_path).unwrap(), original); + second.finalize().unwrap(); + + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn wav_header_round_trips_sizes() { let dir = std::env::temp_dir(); diff --git a/src/core/mod.rs b/src/core/mod.rs index 58a1feb..96787cd 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -91,6 +91,22 @@ fn game_presence_label(game: Option<&crate::game::DetectedGame>) -> Option>>, +) -> Option> { + let Some(rx) = game_rx.as_mut() else { + return std::future::pending().await; + }; + if rx.changed().await.is_err() { + *game_rx = None; + return None; + } + Some(rx.borrow_and_update().clone()) +} + fn arm_discovery_retry( discovery_deadline: &mut Option, now: tokio::time::Instant, @@ -943,11 +959,19 @@ async fn run_core_loop( // the game is *broadcast* as presence is gated by `game_presence_enabled` (opt-in, // seeded false; the GUI sends `SetGamePresenceEnabled` from config at startup). // The override + process map start at their defaults and are set via commands. - let game_detector = crate::game::detector::GameDetector::spawn( + let (game_detector, mut game_rx) = match crate::game::detector::GameDetector::spawn( crate::game::ManualOverride::Auto, std::collections::BTreeMap::new(), - ); - let mut game_rx = game_detector.subscribe(); + ) { + Ok(detector) => { + let rx = detector.subscribe(); + (Some(detector), Some(rx)) + } + Err(e) => { + crate::log_msg(&format!("game detector unavailable: {e}")); + (None, None) + } + }; let mut game_presence_enabled = false; // The latest debounced detection, kept regardless of the broadcast toggle so a // later opt-in can immediately publish whatever is currently running. @@ -1058,13 +1082,14 @@ async fn run_core_loop( Some(cmd) => cmd, None => break, }, - changed = game_rx.changed() => { + game_change = next_game_change(&mut game_rx) => { // The detector worker published a new debounced game (or `None`). - if changed.is_err() { - // Worker gone (shouldn't happen before shutdown); stop watching. + let Some(detected) = game_change else { + // Worker gone unexpectedly. The helper fused this source, so + // this logs once and the closed channel cannot spin select!. + crate::log_msg("game detector stopped; disabling game detection"); continue; - } - let detected = game_rx.borrow_and_update().clone(); + }; current_game = detected.clone(); // Always tell the GUI for the local per-game background + indicator. let _ = ui_tx.send(UiEvent::GameChanged(detected.clone())).await; @@ -2255,11 +2280,15 @@ async fn run_core_loop( CoreCommand::SetGameOverride(override_) => { // Applied on the detector's next poll, immediately (bypasses debounce). - game_detector.set_override(override_); + if let Some(detector) = &game_detector { + detector.set_override(override_); + } } CoreCommand::SetGameProcessMap(map) => { - game_detector.set_process_map(map); + if let Some(detector) = &game_detector { + detector.set_process_map(map); + } } CoreCommand::SetRecordingMode(mode) => { @@ -2283,11 +2312,13 @@ async fn run_core_loop( .unwrap_or(0); let result: Result = if recording_mode.is_multitrack() { // Multitrack/Both: a per-session directory of stems. - let stamp = crate::audio::recorder::timestamp_filename(now); - let session_dir = base.join(stamp.trim_end_matches(".wav")); - std::fs::create_dir_all(&session_dir) + std::fs::create_dir_all(&base) .map_err(|e| e.to_string()) .and_then(|_| { + crate::audio::multitrack::create_session_dir(&base, now) + .map_err(|e| e.to_string()) + }) + .and_then(|session_dir| { MultitrackRecorder::create( &session_dir, FRAME_SAMPLES, @@ -2484,7 +2515,8 @@ async fn run_core_loop( mod tests { use super::{ apply_volume, audio_datagram_len_ok, frame_level, mix_frames, mix_stereo_frames, - stereo_to_mono, MicLevelMeter, MAX_OPUS_PAYLOAD, MIC_LEVEL_REPORT_SAMPLES, + next_game_change, stereo_to_mono, MicLevelMeter, MAX_OPUS_PAYLOAD, + MIC_LEVEL_REPORT_SAMPLES, }; /// A frame of constant amplitude with the given sample count. @@ -2492,6 +2524,16 @@ mod tests { vec![amp; len] } + #[tokio::test] + async fn closed_game_watch_is_fused_after_one_ready_event() { + let (tx, rx) = tokio::sync::watch::channel(None); + let mut rx = Some(rx); + drop(tx); + + assert_eq!(next_game_change(&mut rx).await, None); + assert!(rx.is_none(), "closed receiver must disable its select source"); + } + #[test] fn mic_meter_reports_only_after_enough_samples() { let mut m = MicLevelMeter::new(); diff --git a/src/game/detector.rs b/src/game/detector.rs index 98e49ed..dcf3ede 100644 --- a/src/game/detector.rs +++ b/src/game/detector.rs @@ -15,8 +15,10 @@ use super::{ use super::scan; use super::steam::SteamProbe; use std::collections::BTreeMap; +use std::io; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; use std::time::Duration; use tokio::sync::watch; @@ -62,13 +64,17 @@ pub struct GameDetector { inputs: Arc, rx: watch::Receiver>, stop: Arc, + worker: Option>, } impl GameDetector { /// Spawn the detector worker. `process_map` seeds the non-Steam mappings; /// `override_` seeds the manual override (usually `Auto`). The worker runs /// until [`stop`](Self::stop) or the returned `GameDetector` is dropped. - pub fn spawn(override_: ManualOverride, process_map: BTreeMap) -> Self { + pub fn spawn( + override_: ManualOverride, + process_map: BTreeMap, + ) -> io::Result { let inputs = Arc::new(DetectorInputs { override_: Mutex::new(override_), process_map: Mutex::new(process_map), @@ -78,12 +84,16 @@ impl GameDetector { let worker_inputs = inputs.clone(); let worker_stop = stop.clone(); - std::thread::Builder::new() + let worker = std::thread::Builder::new() .name("game-detector".to_string()) - .spawn(move || worker_loop(worker_inputs, tx, worker_stop)) - .ok(); + .spawn(move || worker_loop(worker_inputs, tx, worker_stop))?; - Self { inputs, rx, stop } + Ok(Self { + inputs, + rx, + stop, + worker: Some(worker), + }) } /// A clone of the watch receiver for detected-game changes. The current value @@ -112,6 +122,9 @@ impl GameDetector { impl Drop for GameDetector { fn drop(&mut self) { self.stop(); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } } } @@ -223,7 +236,7 @@ mod tests { fn spawn_and_stop_is_clean() { // Smoke test the lifecycle: spawning and stopping must not panic, and the // initial published value is None. - let det = GameDetector::spawn(ManualOverride::Auto, BTreeMap::new()); + let det = GameDetector::spawn(ManualOverride::Auto, BTreeMap::new()).unwrap(); assert_eq!(*det.subscribe().borrow(), None); det.set_override(ManualOverride::ForceNone); det.set_process_map(map(&[("x", "X")])); diff --git a/src/game/steam.rs b/src/game/steam.rs index dae87bf..581ddee 100644 --- a/src/game/steam.rs +++ b/src/game/steam.rs @@ -18,6 +18,29 @@ use std::time::SystemTime; /// (a manifest is a few KB); the cap stops a corrupt/hostile giant file from being /// slurped into memory before the parser's own depth guard kicks in. const MAX_STEAM_FILE_BYTES: u64 = 4 * 1024 * 1024; +/// SteamPath is a local filesystem path. Four KiB is deliberately generous and +/// prevents a corrupt registry length from driving an enormous allocation. +#[cfg(any(windows, test))] +const MAX_STEAM_PATH_BYTES: u32 = 4 * 1024; + +#[cfg(any(windows, test))] +fn validate_reg_len(len: u32) -> Option { + (len != 0 && len.is_multiple_of(2) && len <= MAX_STEAM_PATH_BYTES) + .then_some(len as usize / 2) +} + +#[cfg(any(windows, test))] +fn decode_reg_sz(mut buf: Vec, returned_bytes: u32) -> Option { + let units = validate_reg_len(returned_bytes)?; + if units > buf.len() { + return None; + } + buf.truncate(units); + while buf.last() == Some(&0) { + buf.pop(); + } + Some(String::from_utf16_lossy(&buf)) +} /// Parse the live `RunningAppID` out of a Steam `registry.vdf` (the Linux/macOS /// client's emulated-registry text file). Returns the appid only when present and @@ -333,6 +356,7 @@ mod win { //! Windows registry reads via direct Win32 FFI (windows-sys), no `winreg` //! crate. Steam stores both the live `RunningAppID` and its install path under //! `HKCU\Software\Valve\Steam`. + use super::{decode_reg_sz, validate_reg_len}; use std::path::PathBuf; use windows_sys::Win32::Foundation::ERROR_SUCCESS; use windows_sys::Win32::System::Registry::{ @@ -401,12 +425,17 @@ mod win { &mut len, ) }; - if rc != ERROR_SUCCESS || kind != REG_SZ || len == 0 { + if rc != ERROR_SUCCESS || kind != REG_SZ { // SAFETY: valid handle. unsafe { RegCloseKey(hkey) }; return None; } - let mut buf = vec![0u16; (len as usize).div_ceil(2)]; + let Some(units) = validate_reg_len(len) else { + // SAFETY: valid handle. + unsafe { RegCloseKey(hkey) }; + return None; + }; + let mut buf = vec![0u16; units]; let mut len2 = len; // SAFETY: buffer sized to the queried byte length. let rc = unsafe { @@ -421,14 +450,10 @@ mod win { }; // SAFETY: valid handle. unsafe { RegCloseKey(hkey) }; - if rc != ERROR_SUCCESS { + if rc != ERROR_SUCCESS || kind != REG_SZ || len2 > len { return None; } - // Trim the trailing NUL(s). - while buf.last() == Some(&0) { - buf.pop(); - } - Some(PathBuf::from(String::from_utf16_lossy(&buf))) + Some(PathBuf::from(decode_reg_sz(buf, len2)?)) } } @@ -436,6 +461,17 @@ mod win { mod tests { use super::*; + #[test] + fn registry_string_lengths_are_bounded_and_trimmed() { + assert_eq!(validate_reg_len(5), None, "odd byte lengths are invalid UTF-16"); + assert_eq!(validate_reg_len(MAX_STEAM_PATH_BYTES + 2), None); + assert_eq!(validate_reg_len(8), Some(4)); + + let raw = "C:\\Steam\0ignored".encode_utf16().collect::>(); + let returned_bytes = ("C:\\Steam\0".encode_utf16().count() * 2) as u32; + assert_eq!(decode_reg_sz(raw, returned_bytes).as_deref(), Some("C:\\Steam")); + } + #[test] fn running_app_id_reads_nonzero_and_rejects_zero() { let running = r#""Registry" { "HKCU" { "Software" { "Valve" { "Steam" {