diff --git a/src/audio/recorder.rs b/src/audio/recorder.rs index 2734567..473d11a 100644 --- a/src/audio/recorder.rs +++ b/src/audio/recorder.rs @@ -22,6 +22,8 @@ use std::path::{Path, PathBuf}; const SAMPLE_RATE: u32 = 48_000; 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; /// 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 @@ -34,7 +36,7 @@ const MAX_MIC_FIFO: usize = SAMPLE_RATE as usize / 5; pub struct WavWriter { file: File, /// Bytes of PCM data written so far (for the size fields). - data_bytes: u32, + data_bytes: u64, } impl WavWriter { @@ -42,7 +44,10 @@ impl WavWriter { pub fn new(path: &Path) -> io::Result { let mut file = File::create(path)?; file.write_all(&Self::header(0))?; - Ok(Self { file, data_bytes: 0 }) + Ok(Self { + file, + data_bytes: 0, + }) } /// The 44-byte canonical WAV/PCM header for the given data length in bytes. @@ -68,21 +73,40 @@ impl WavWriter { /// Append PCM samples to the data chunk. pub fn write_samples(&mut self, samples: &[i16]) -> io::Result<()> { + let added_bytes = u64::try_from(samples.len()) + .ok() + .and_then(|len| len.checked_mul(2)) + .ok_or_else(|| io::Error::other("WAV sample buffer too large"))?; + let new_data_bytes = self + .data_bytes + .checked_add(added_bytes) + .ok_or_else(|| io::Error::other("WAV data size overflow"))?; + if new_data_bytes > MAX_RIFF_DATA_BYTES { + return Err(io::Error::other("WAV too large for RIFF")); + } + let mut buf = Vec::with_capacity(samples.len() * 2); for &s in samples { buf.extend_from_slice(&s.to_le_bytes()); } self.file.write_all(&buf)?; - self.data_bytes += (samples.len() * 2) as u32; + self.data_bytes = new_data_bytes; Ok(()) } /// Patch the RIFF + data size fields and flush. Consumes the writer. pub fn finalize(mut self) -> io::Result<()> { + let data_bytes = u32::try_from(self.data_bytes) + .map_err(|_| io::Error::other("WAV too large for RIFF"))?; + let riff_size = self + .data_bytes + .checked_add(RIFF_DATA_OVERHEAD) + .and_then(|size| u32::try_from(size).ok()) + .ok_or_else(|| io::Error::other("WAV too large for RIFF"))?; self.file.seek(SeekFrom::Start(4))?; - self.file.write_all(&(36 + self.data_bytes).to_le_bytes())?; + self.file.write_all(&riff_size.to_le_bytes())?; self.file.seek(SeekFrom::Start(40))?; - self.file.write_all(&self.data_bytes.to_le_bytes())?; + self.file.write_all(&data_bytes.to_le_bytes())?; self.file.flush()?; Ok(()) } @@ -209,11 +233,29 @@ mod tests { let _ = std::fs::remove_file(&path); } + #[test] + fn wav_writer_rejects_data_that_would_overflow_riff_header() { + let dir = std::env::temp_dir(); + let path = dir.join(format!("peerspeak-overflow-{}.wav", std::process::id())); + let mut w = WavWriter::new(&path).unwrap(); + w.data_bytes = MAX_RIFF_DATA_BYTES - 1; + let before_len = std::fs::metadata(&path).unwrap().len(); + + let err = w.write_samples(&[0]).unwrap_err(); + + assert_eq!(err.kind(), io::ErrorKind::Other); + assert_eq!(w.data_bytes, MAX_RIFF_DATA_BYTES - 1); + assert_eq!(std::fs::metadata(&path).unwrap().len(), before_len); + drop(w); + let _ = std::fs::remove_file(&path); + } + #[test] fn mic_is_summed_with_mix_when_present() { let dir = std::env::temp_dir(); let mut r = Recorder { - writer: WavWriter::new(&dir.join(format!("ps-sum-{}.wav", std::process::id()))).unwrap(), + writer: WavWriter::new(&dir.join(format!("ps-sum-{}.wav", std::process::id()))) + .unwrap(), mic_fifo: VecDeque::new(), path: PathBuf::new(), }; @@ -223,7 +265,11 @@ mod tests { r.write_frame(&[10, 20]).unwrap(); assert_eq!(r.mic_fifo.len(), 1, "two samples consumed, one mic left"); r.write_frame(&[0, 0]).unwrap(); - assert_eq!(r.mic_fifo.len(), 0, "remaining mic sample consumed; rest is silence"); + assert_eq!( + r.mic_fifo.len(), + 0, + "remaining mic sample consumed; rest is silence" + ); let _ = r.finalize(); } @@ -231,7 +277,8 @@ mod tests { fn mic_fifo_is_capped() { let dir = std::env::temp_dir(); let mut r = Recorder { - writer: WavWriter::new(&dir.join(format!("ps-cap-{}.wav", std::process::id()))).unwrap(), + writer: WavWriter::new(&dir.join(format!("ps-cap-{}.wav", std::process::id()))) + .unwrap(), mic_fifo: VecDeque::new(), path: PathBuf::new(), }; diff --git a/src/core/jitter.rs b/src/core/jitter.rs index cc8724e..de526e0 100644 --- a/src/core/jitter.rs +++ b/src/core/jitter.rs @@ -59,6 +59,10 @@ const PRIME_TIMEOUT_TICKS: usize = 25; /// badly behind, so we drop the oldest and resync rather than grow unbounded. const MAX_BUFFERED_FRAMES: usize = 32; +/// Sequence discontinuities larger than this (~10s at 20ms/frame) are treated +/// as a restarted/new stream, not ordinary packet loss or reordering. +const MAX_REASONABLE_SEQ_GAP: u32 = 500; + pub struct JitterBuffer { decoder: OpusDecoder, /// Reorder window: sequence number -> encoded Opus payload. @@ -116,6 +120,14 @@ impl JitterBuffer { } } + fn reset_to_stream(&mut self, seq: u32, payload: Vec) { + self.packets.clear(); + self.packets.insert(seq, payload); + self.next_seq = None; + self.clean_run = 0; + self.buffering_ticks = 0; + } + /// Store a received packet, dropping ones we've already played past and /// bounding total depth. pub fn insert(&mut self, seq: u32, payload: Vec) { @@ -124,9 +136,19 @@ impl JitterBuffer { if let Some(next) = self.next_seq && seq_before(seq, next) { + if next.wrapping_sub(seq) > MAX_REASONABLE_SEQ_GAP { + self.reset_to_stream(seq, payload); + return; + } self.note_disruption(); return; } + if let Some(next) = self.next_seq + && seq.wrapping_sub(next) > MAX_REASONABLE_SEQ_GAP + { + self.reset_to_stream(seq, payload); + return; + } self.packets.insert(seq, payload); while self.packets.len() > MAX_BUFFERED_FRAMES { @@ -283,6 +305,44 @@ mod tests { assert_eq!(jb.packets.len(), 1); // only seq 7 remains buffered } + #[test] + fn far_behind_sequence_resets_as_restarted_stream() { + let mut jb = JitterBuffer::new().unwrap(); + jb.next_seq = Some(5_000); + jb.packets.insert(5_000, vec![9]); + jb.clean_run = 12; + jb.buffering_ticks = 4; + + jb.insert(0, vec![1]); + + assert_eq!(jb.next_seq, None); + assert_eq!(jb.packets.len(), 1); + assert_eq!(jb.packets.get(&0).map(Vec::as_slice), Some(&[1][..])); + assert_eq!(jb.clean_run, 0); + assert_eq!(jb.buffering_ticks, 0); + } + + #[test] + fn far_ahead_sequence_resets_to_bound_plc_run() { + let mut jb = JitterBuffer::new().unwrap(); + jb.next_seq = Some(10); + jb.packets.insert(10, vec![9]); + jb.clean_run = 12; + jb.buffering_ticks = 4; + + let jumped_seq = 10 + MAX_REASONABLE_SEQ_GAP + 1; + jb.insert(jumped_seq, vec![2]); + + assert_eq!(jb.next_seq, None); + assert_eq!(jb.packets.len(), 1); + assert_eq!( + jb.packets.get(&jumped_seq).map(Vec::as_slice), + Some(&[2][..]) + ); + assert_eq!(jb.clean_run, 0); + assert_eq!(jb.buffering_ticks, 0); + } + #[test] fn test_seq_before_ordering() { // Basic ordering @@ -358,7 +418,7 @@ mod tests { fn is_idle_reflects_buffer_state() { let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap(); let mut jb = JitterBuffer::new().unwrap(); - + // Fresh buffer assert!(jb.is_idle()); @@ -369,7 +429,7 @@ mod tests { // Prime (3 frames) jb.insert(1, frame(&mut enc, 1000)); jb.insert(2, frame(&mut enc, 1000)); - + // Drain past the end so it underruns assert!(jb.pop_frame().is_some()); assert!(jb.pop_frame().is_some()); @@ -635,4 +695,3 @@ mod tests { assert_eq!(jb.clean_run, 0); } } - diff --git a/task-report.md b/task-report.md index a12c48f..8abad41 100644 --- a/task-report.md +++ b/task-report.md @@ -51,11 +51,21 @@ Proposed future scope: Reason for not implementing: the current `run_playback` / `run_capture` code does not retain stream node or port ids, and changing `AUTOCONNECT` behavior plus adding manual `pw-link` calls could destabilize the working audio path. That matches the assignment's "bail if risky" instruction. +## Backlog A21/A22 - correctness fixes + +- Fixed A21 in `src/core/jitter.rs`: implausibly large sequence discontinuities now reset the per-peer jitter stream instead of being treated as ordinary late packets or packet loss. +- The reset threshold is `500` frames, about 10 seconds at 20 ms/frame. That covers both same-identity sender restart back to sequence 0 and a faulty/malicious jump far ahead that would otherwise force a long PLC run. +- Added jitter regression tests for both far-behind restart and far-ahead jump cases. +- Fixed A22 in `src/audio/recorder.rs`: `WavWriter` now tracks data bytes as `u64`, checks additions before writing, and rejects data that cannot fit both the RIFF size field and the `data` chunk size field. +- Added a WAV overflow regression test that exercises the limit without creating a huge file. + +Unverified: the same-identity peer restart has not been exercised in a live 2-machine call; the WAV fix is counter/size-field tested, not a real >12h recording. + ## Verification -- `cargo test --lib` passed: 285 passed, 0 failed, 2 ignored. +- `cargo test --lib` passed: 288 passed, 0 failed, 2 ignored. - `cargo clippy --all-targets` passed. - `cargo build --release` passed. -- `cargo fmt --check` reports broad repo-wide formatting diffs, including untouched files; I did not run `cargo fmt` to avoid unrelated churn. +- Formatted the touched Rust files with `rustfmt --edition 2024`; I did not run repo-wide `cargo fmt` to avoid unrelated formatting churn. -No new dependencies were added. I did not run git commands. +No new dependencies were added. Runtime/manual/field verification is still pending for audio-device and 2-machine behavior.