fix(audio,game): Tier A bug-sweep fixes (S-01, F-04, F-08, F-09, S-02)

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 <codex@openai.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-22 03:34:29 -04:00
co-authored by Codex Claude Opus 4.8
parent f422150c84
commit 6b0b23ef69
6 changed files with 288 additions and 47 deletions
+39
View File
@@ -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<PathBuf> {
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");
+73 -10
View File
@@ -151,11 +151,9 @@ fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender<Vec<i16>>, 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<Vec<i16>>, 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<P: Producer<Item = i16>>(
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::<i16>::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]
+57 -9
View File
@@ -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<Self> {
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<Self> {
file.write_all(&Self::header(0))?;
Ok(Self {
file,
@@ -125,13 +131,31 @@ impl Recorder {
/// Create a recording at `dir/<timestamped>.wav`. The directory is assumed to
/// exist (the caller creates it).
pub fn create(dir: &Path, now_unix_secs: u64) -> io::Result<Self> {
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();