From be61be2f237e88d8e1dd41af9835212fec95b90d Mon Sep 17 00:00:00 2001 From: Franco Carballar Date: Wed, 30 Sep 2026 17:26:43 -0300 Subject: [PATCH 1/4] fix: prioritize streaming playback and preserve buffered speech --- native/src/audio/engine.rs | 101 +++++++++++++++++++++++++++++++---- native/src/audio/io.rs | 58 +++++++++++++++++++- native/src/audio/resample.rs | 27 +++++++++- 3 files changed, 171 insertions(+), 15 deletions(-) diff --git a/native/src/audio/engine.rs b/native/src/audio/engine.rs index 0a93f28..004d822 100644 --- a/native/src/audio/engine.rs +++ b/native/src/audio/engine.rs @@ -37,6 +37,7 @@ const FRAME: usize = 960; // 20 ms at 48 kHz const MAX_DECODED: usize = 5760; // 120 ms at 48 kHz const MAX_CONCEALED_PACKETS: u16 = 5; const LEVEL_INTERVAL: Duration = Duration::from_millis(50); +const MAX_BLOCKS_PER_SERVICE: usize = 2; const SERVICE_INTERVAL: Duration = Duration::from_millis(5); #[derive(Default)] @@ -210,6 +211,16 @@ impl State { let mut last_levels = Instant::now(); let mut was_muted = control.muted.load(Ordering::Acquire); while !control.stop.load(Ordering::Acquire) { + // Service incoming audio before DSP work: a capture backlog must not delay + // speech until the transcript (which arrives on another thread) is complete. + for _ in 0..32 { + let Ok(packet) = incoming.try_recv() else { + break; + }; + self.receive(packet)?; + } + self.flush_playback(&mut rings.playback); + // Speaker reference first so capture blocks see an up-to-date echo model. loop { let read = rings.reference.pop_slice(&mut self.scratch); @@ -220,7 +231,10 @@ impl State { } let mut block = [0.0f32; BLOCK]; let mut processed = [0.0f32; BLOCK]; - while self.reference.take_exact(&mut block) { + for _ in 0..MAX_BLOCKS_PER_SERVICE { + if !self.reference.take_exact(&mut block) { + break; + } self.apm .process_render_f32(&[&block], &mut [&mut processed]) .map_err(|_| anyhow::anyhow!("echo reference processing failed"))?; @@ -242,7 +256,10 @@ impl State { let playback_delay_ms = rings.playback.occupied_len() as f64 * 1000.0 / f64::from(self.output_rate); let delay = (playback_delay_ms + 20.0).clamp(0.0, 500.0) as i32; - while self.capture.take_exact(&mut block) { + for _ in 0..MAX_BLOCKS_PER_SERVICE { + if !self.capture.take_exact(&mut block) { + break; + } let _ = self.apm.set_stream_delay_ms(delay); self.apm .process_capture_f32(&[&block], &mut [&mut processed]) @@ -264,15 +281,6 @@ impl State { } } - // Playback. - while let Ok(packet) = incoming.try_recv() { - self.receive(packet)?; - } - if self.playback.available() > 0 { - let samples: Vec = self.playback.drain_all().collect(); - rings.playback.push_slice(&samples); - } - if last_levels.elapsed() >= LEVEL_INTERVAL { last_levels = Instant::now(); emitter.emit(serde_json::json!({ @@ -288,6 +296,15 @@ impl State { Ok(()) } + fn flush_playback(&mut self, sink: &mut ringbuf::HeapProd) { + let capacity = sink.vacant_len().min(self.playback.available()); + if capacity > 0 { + let samples: Vec = self.playback.drain(capacity).collect(); + let written = sink.push_slice(&samples); + debug_assert_eq!(written, samples.len()); + } + } + fn receive(&mut self, packet: IncomingPacket) -> anyhow::Result<()> { let mut decoded = [0.0f32; MAX_DECODED]; if let Some(expected) = self.expected_sequence { @@ -326,3 +343,65 @@ impl State { self.playback.push(samples) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn speaker_backpressure_preserves_every_sample_in_order() { + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let (mut prod, mut cons) = ringbuf::HeapRb::::new(2).split(); + state.playback.push(&[1.0, 2.0, 3.0, 4.0, 5.0]).unwrap(); + state.flush_playback(&mut prod); + assert_eq!(state.playback.available(), 3); + // A second service while full must not consume queued samples. + state.flush_playback(&mut prod); + assert_eq!(state.playback.available(), 3); + let mut actual = Vec::new(); + while state.playback.available() > 0 || !cons.is_empty() { + while let Some(sample) = cons.try_pop() { + actual.push(sample); + } + state.flush_playback(&mut prod); + } + assert_eq!(actual, vec![1.0, 2.0, 3.0, 4.0, 5.0]); + } + + // Offline: no microphone, devices, credentials or network. Run explicitly to + // check the DSP budget when evaluating a helper build on a target machine. + #[test] + #[ignore = "manual DSP timing diagnostic"] + fn one_second_dsp_budget() { + let mut state = State::new(RATE, RATE).unwrap(); + let input: Vec = (0..BLOCK) + .map(|i| ((i as f32 / RATE as f32) * 440.0 * std::f32::consts::TAU).sin() * 0.1) + .collect(); + let mut processed = [0.0; BLOCK]; + let mut encoded = [0u8; 1275]; + let started = Instant::now(); + for _ in 0..100 { + state + .apm + .process_render_f32(&[&input], &mut [&mut processed]) + .unwrap(); + state + .apm + .process_capture_f32(&[&input], &mut [&mut processed]) + .unwrap(); + state.pending.extend_from_slice(&processed); + if state.pending.len() >= FRAME { + state + .encoder + .encode_float(&state.pending[..FRAME], &mut encoded) + .unwrap(); + state.pending.drain(..FRAME); + } + } + eprintln!( + "DSP processing for 1000 ms of audio: {} ms", + started.elapsed().as_millis() + ); + } +} diff --git a/native/src/audio/io.rs b/native/src/audio/io.rs index ea5ee5b..3c95bc2 100644 --- a/native/src/audio/io.rs +++ b/native/src/audio/io.rs @@ -320,6 +320,8 @@ struct Renderer { control: Arc, starved: bool, prebuffer: usize, + waited: usize, + max_wait: usize, } impl Renderer { @@ -336,6 +338,9 @@ impl Renderer { starved: true, // 60 ms of jitter absorption before speech starts playing. prebuffer: rate as usize * 60 / 1000, + waited: 0, + // A short response or tail must not wait for another utterance to reach 60 ms. + max_wait: rate as usize / 10, } } @@ -343,9 +348,19 @@ impl Renderer { if self.control.clear.swap(false, Ordering::AcqRel) { self.playback.clear(); self.starved = true; + self.waited = 0; } - if self.starved && self.playback.occupied_len() >= self.prebuffer { - self.starved = false; + if self.starved { + let available = self.playback.occupied_len(); + if available > 0 { + self.waited = self.waited.saturating_add(out.len()); + if available >= self.prebuffer || self.waited >= self.max_wait { + self.starved = false; + self.waited = 0; + } + } else { + self.waited = 0; + } } let read = if self.starved { 0 @@ -543,3 +558,42 @@ fn run_virtual_output( let _ = writer.finalize(); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn short_response_plays_without_waiting_for_the_next_turn() { + let (mut prod, cons) = HeapRb::::new(200).split(); + let (reference, _) = HeapRb::::new(200).split(); + let mut renderer = + Renderer::new(cons, reference, Arc::new(PlaybackControl::default()), 1000); + prod.push_slice(&[0.5; 20]); + let mut out = [0.0; 10]; + for _ in 0..9 { + renderer.render(&mut out); + assert_eq!(out, [0.0; 10]); + } + renderer.render(&mut out); + assert_eq!(out, [0.5; 10]); + renderer.render(&mut out); + assert_eq!(out, [0.5; 10]); + } + + #[test] + fn normal_prebuffer_starts_immediately_and_clear_discards_it() { + let (mut prod, cons) = HeapRb::::new(200).split(); + let (reference, _) = HeapRb::::new(200).split(); + let control = Arc::new(PlaybackControl::default()); + let mut renderer = Renderer::new(cons, reference, control.clone(), 1000); + prod.push_slice(&[0.5; 60]); + let mut out = [0.0; 10]; + renderer.render(&mut out); + assert_eq!(out, [0.5; 10]); + control.clear.store(true, Ordering::Release); + renderer.render(&mut out); + assert_eq!(out, [0.0; 10]); + assert!(renderer.playback.is_empty()); + } +} diff --git a/native/src/audio/resample.rs b/native/src/audio/resample.rs index 8528d6a..18ed87d 100644 --- a/native/src/audio/resample.rs +++ b/native/src/audio/resample.rs @@ -83,7 +83,30 @@ impl Converter { true } - pub fn drain_all(&mut self) -> std::collections::vec_deque::Drain<'_, f32> { - self.output.drain(..) + /// Drain only samples the speaker ring can accept, preserving any overflow. + pub fn drain(&mut self, limit: usize) -> std::collections::vec_deque::Drain<'_, f32> { + self.output.drain(..limit.min(self.output.len())) + } +} + +#[cfg(test)] +mod tests { + use super::Converter; + + #[test] + fn partial_playback_drain_preserves_order_and_overflow() { + let mut converter = Converter::new(48_000, 48_000).unwrap(); + converter.push(&[1.0, 2.0, 3.0, 4.0]).unwrap(); + assert_eq!(converter.drain(2).collect::>(), vec![1.0, 2.0]); + assert_eq!(converter.available(), 2); + assert_eq!(converter.drain(8).collect::>(), vec![3.0, 4.0]); + } + + #[test] + fn full_speaker_ring_does_not_discard_pending_audio() { + let mut converter = Converter::new(48_000, 48_000).unwrap(); + converter.push(&[1.0, 2.0]).unwrap(); + assert_eq!(converter.drain(0).count(), 0); + assert_eq!(converter.available(), 2); } } From 6edda52a0be4bc2d470a4758015fe3481ef7f96f Mon Sep 17 00:00:00 2001 From: Franco Carballar Date: Wed, 30 Sep 2026 19:01:39 -0300 Subject: [PATCH 2/4] fix: clear retained playback before refilling the speaker ring Reset pending resampler audio and wait for the renderer to acknowledge clearing its ring. Cover overflow, repeated clears and real sample-rate conversion without devices. Co-authored-by: OpenAI --- native/src/audio/engine.rs | 77 ++++++++++++++++++++++++++++++------ native/src/audio/io.rs | 10 +++-- native/src/audio/resample.rs | 27 +++++++++++++ native/src/main.rs | 2 +- 4 files changed, 100 insertions(+), 16 deletions(-) diff --git a/native/src/audio/engine.rs b/native/src/audio/engine.rs index 004d822..1ccc698 100644 --- a/native/src/audio/engine.rs +++ b/native/src/audio/engine.rs @@ -44,11 +44,11 @@ const SERVICE_INTERVAL: Duration = Duration::from_millis(5); pub struct EngineControl { pub muted: AtomicBool, pub stop: AtomicBool, + pub clear: AtomicBool, } pub struct Engine { pub control: Arc, - playback: Arc, thread: Option>, } @@ -67,7 +67,6 @@ impl Engine { .take() .ok_or_else(|| anyhow::anyhow!("audio already started"))?; let state = State::new(io.input.rate, io.output.rate)?; - let playback = io.control.clone(); let thread_control = control.clone(); let thread = std::thread::Builder::new() .name("audio-engine".into()) @@ -89,7 +88,6 @@ impl Engine { })?; Ok(Self { control, - playback, thread: Some(thread), }) } @@ -99,7 +97,7 @@ impl Engine { } pub fn clear_output(&self) { - self.playback.clear.store(true, Ordering::Release); + self.control.clear.store(true, Ordering::Release); } } @@ -213,13 +211,7 @@ impl State { while !control.stop.load(Ordering::Acquire) { // Service incoming audio before DSP work: a capture backlog must not delay // speech until the transcript (which arrives on another thread) is complete. - for _ in 0..32 { - let Ok(packet) = incoming.try_recv() else { - break; - }; - self.receive(packet)?; - } - self.flush_playback(&mut rings.playback); + self.service_playback(&mut rings.playback, incoming, control, &io.control)?; // Speaker reference first so capture blocks see an up-to-date echo model. loop { @@ -296,6 +288,31 @@ impl State { Ok(()) } + fn service_playback( + &mut self, + sink: &mut ringbuf::HeapProd, + incoming: &std_mpsc::Receiver, + control: &EngineControl, + playback: &super::io::PlaybackControl, + ) -> anyhow::Result<()> { + if control.clear.swap(false, Ordering::AcqRel) { + self.playback.clear(); + playback.clear.store(true, Ordering::Release); + } + // Leave the producer idle until the renderer acknowledges clearing its ring. + if playback.clear.load(Ordering::Acquire) { + return Ok(()); + } + for _ in 0..32 { + let Ok(packet) = incoming.try_recv() else { + break; + }; + self.receive(packet)?; + } + self.flush_playback(sink); + Ok(()) + } + fn flush_playback(&mut self, sink: &mut ringbuf::HeapProd) { let capacity = sink.vacant_len().min(self.playback.available()); if capacity > 0 { @@ -369,6 +386,44 @@ mod tests { assert_eq!(actual, vec![1.0, 2.0, 3.0, 4.0, 5.0]); } + #[test] + fn clear_discards_overflow_and_waits_for_the_renderer_before_refilling() { + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let control = EngineControl::default(); + let playback = Arc::new(super::super::io::PlaybackControl::default()); + let (mut prod, cons) = ringbuf::HeapRb::::new(2).split(); + let (reference, _) = ringbuf::HeapRb::::new(20).split(); + let mut renderer = super::super::io::Renderer::new(cons, reference, playback.clone(), 10); + let (_tx, incoming) = std_mpsc::channel(); + state.playback.push(&[1.0, 2.0, 3.0, 4.0, 5.0]).unwrap(); + state.flush_playback(&mut prod); + assert_eq!(state.playback.available(), 3); + for _ in 0..2 { + control.clear.store(true, Ordering::Release); + state + .service_playback(&mut prod, &incoming, &control, &playback) + .unwrap(); + assert_eq!(state.playback.available(), 0); + assert!(playback.clear.load(Ordering::Acquire)); + } + state.playback.push(&[6.0, 7.0]).unwrap(); + state + .service_playback(&mut prod, &incoming, &control, &playback) + .unwrap(); + assert_eq!(state.playback.available(), 2); + let mut out = [0.0; 2]; + renderer.render(&mut out); + assert_eq!(out, [0.0; 2]); + assert!(!playback.clear.load(Ordering::Acquire)); + state + .service_playback(&mut prod, &incoming, &control, &playback) + .unwrap(); + renderer.render(&mut out); + assert_eq!(out, [6.0, 7.0]); + assert_eq!(state.playback.available(), 0); + } + // Offline: no microphone, devices, credentials or network. Run explicitly to // check the DSP budget when evaluating a helper build on a target machine. #[test] diff --git a/native/src/audio/io.rs b/native/src/audio/io.rs index 3c95bc2..2704619 100644 --- a/native/src/audio/io.rs +++ b/native/src/audio/io.rs @@ -314,7 +314,7 @@ where } /// Renders mono speaker samples, handling prebuffering after underruns and clear requests. -struct Renderer { +pub(super) struct Renderer { playback: HeapCons, reference: HeapProd, control: Arc, @@ -325,7 +325,7 @@ struct Renderer { } impl Renderer { - fn new( + pub(super) fn new( playback: HeapCons, reference: HeapProd, control: Arc, @@ -344,11 +344,13 @@ impl Renderer { } } - fn render(&mut self, out: &mut [f32]) { - if self.control.clear.swap(false, Ordering::AcqRel) { + pub(super) fn render(&mut self, out: &mut [f32]) { + if self.control.clear.load(Ordering::Acquire) { self.playback.clear(); self.starved = true; self.waited = 0; + // The engine must not refill the ring until it has been emptied. + self.control.clear.store(false, Ordering::Release); } if self.starved { let available = self.playback.occupied_len(); diff --git a/native/src/audio/resample.rs b/native/src/audio/resample.rs index 18ed87d..c88229d 100644 --- a/native/src/audio/resample.rs +++ b/native/src/audio/resample.rs @@ -71,6 +71,14 @@ impl Converter { self.output.len() } + pub fn clear(&mut self) { + self.output.clear(); + if let Some(inner) = &mut self.inner { + inner.input.clear(); + inner.resampler.reset(); + } + } + /// Takes exactly `out.len()` samples when enough are buffered. pub fn take_exact(&mut self, out: &mut [f32]) -> bool { if self.output.len() < out.len() { @@ -109,4 +117,23 @@ mod tests { assert_eq!(converter.drain(0).count(), 0); assert_eq!(converter.available(), 2); } + + #[test] + fn clear_discards_output_partial_input_and_resampler_history() { + for rate in [48_000, 44_100] { + let mut converter = Converter::new(48_000, rate).unwrap(); + converter.push(&[1.0; 4_801]).unwrap(); + assert!(converter.available() > 0); + converter.clear(); + assert_eq!(converter.available(), 0); + let mut fresh = Converter::new(48_000, rate).unwrap(); + let samples = [0.25; 4_800]; + converter.push(&samples).unwrap(); + fresh.push(&samples).unwrap(); + assert_eq!( + converter.drain(usize::MAX).collect::>(), + fresh.drain(usize::MAX).collect::>() + ); + } + } } diff --git a/native/src/main.rs b/native/src/main.rs index 5749a68..510e939 100644 --- a/native/src/main.rs +++ b/native/src/main.rs @@ -204,7 +204,7 @@ impl Session { emitter.emit(serde_json::json!({ "type": "muted", "muted": muted })); } Command::Clear {} => { - // Speaker queue is owned by the output endpoint; the engine forwards the flag. + // The engine clears its pending audio before asking the endpoint to clear. if let Some(engine) = &self.engine { engine.clear_output(); } From ff02622be83d3d44dfaacd1f6bccd559834ca6e6 Mon Sep 17 00:00:00 2001 From: Franco Carballar Date: Tue, 6 Oct 2026 13:23:29 -0300 Subject: [PATCH 3/4] =?UTF-8?q?fix(audio):=20=F0=9F=90=9B=20limitar=20back?= =?UTF-8?q?log=20y=20vencer=20espera=20de=20limpieza?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Acota a un segundo el audio pendiente del resampler y descarta las muestras más antiguas. Si el renderer no confirma una limpieza en un segundo, el engine falla para cerrar la llamada en vez de quedar silenciado. Co-Authored-By: OpenAI --- native/src/audio/engine.rs | 54 ++++++++++++++++++++++++++++++++++-- native/src/audio/resample.rs | 52 ++++++++++++++++++++++++++++++++-- 2 files changed, 102 insertions(+), 4 deletions(-) diff --git a/native/src/audio/engine.rs b/native/src/audio/engine.rs index 1ccc698..12cddcc 100644 --- a/native/src/audio/engine.rs +++ b/native/src/audio/engine.rs @@ -36,6 +36,8 @@ const BLOCK: usize = 480; // 10 ms at 48 kHz const FRAME: usize = 960; // 20 ms at 48 kHz const MAX_DECODED: usize = 5760; // 120 ms at 48 kHz const MAX_CONCEALED_PACKETS: u16 = 5; +const MAX_PENDING_PLAYBACK_MS: usize = 1_000; +const CLEAR_ACK_TIMEOUT: Duration = Duration::from_secs(1); const LEVEL_INTERVAL: Duration = Duration::from_millis(50); const MAX_BLOCKS_PER_SERVICE: usize = 2; const SERVICE_INTERVAL: Duration = Duration::from_millis(5); @@ -152,6 +154,7 @@ struct State { pending: Vec, expected_sequence: Option, output_rate: u32, + clear_started: Option, mic: Meter, speaker: Meter, scratch: Vec, @@ -184,13 +187,18 @@ impl State { Ok(Self { capture: Converter::new(input_rate, RATE)?, reference: Converter::new(output_rate, RATE)?, - playback: Converter::new(RATE, output_rate)?, + playback: Converter::new_bounded( + RATE, + output_rate, + (output_rate as usize * MAX_PENDING_PLAYBACK_MS) / 1000, + )?, apm, encoder, decoder, pending: Vec::with_capacity(FRAME), expected_sequence: None, output_rate, + clear_started: None, mic: Meter::new(), speaker: Meter::new(), scratch: vec![0.0; 8192], @@ -297,12 +305,22 @@ impl State { ) -> anyhow::Result<()> { if control.clear.swap(false, Ordering::AcqRel) { self.playback.clear(); - playback.clear.store(true, Ordering::Release); + if !playback.clear.swap(true, Ordering::AcqRel) { + self.clear_started = Some(Instant::now()); + } } // Leave the producer idle until the renderer acknowledges clearing its ring. if playback.clear.load(Ordering::Acquire) { + if self + .clear_started + .is_some_and(|started| started.elapsed() >= CLEAR_ACK_TIMEOUT) + && playback.clear.load(Ordering::Acquire) + { + anyhow::bail!("speaker did not acknowledge playback clear within 1 second"); + } return Ok(()); } + self.clear_started = None; for _ in 0..32 { let Ok(packet) = incoming.try_recv() else { break; @@ -424,6 +442,38 @@ mod tests { assert_eq!(state.playback.available(), 0); } + #[test] + fn playback_pending_audio_is_bounded_and_keeps_the_newest_samples() { + let mut state = State::new(RATE, RATE).unwrap(); + let samples: Vec = (0..RATE as usize + 10) + .map(|sample| sample as f32) + .collect(); + state.playback.push(&samples).unwrap(); + assert_eq!(state.playback.available(), RATE as usize); + assert_eq!(state.playback.drain(1).next(), Some(10.0)); + } + + #[test] + fn clear_ack_timeout_fails_instead_of_leaving_audio_silently_stalled() { + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let control = EngineControl::default(); + let playback = Arc::new(super::super::io::PlaybackControl::default()); + playback.clear.store(true, Ordering::Release); + state.clear_started = Some(Instant::now() - CLEAR_ACK_TIMEOUT - Duration::from_millis(1)); + let (mut sink, _) = ringbuf::HeapRb::::new(2).split(); + let (_tx, incoming) = std_mpsc::channel(); + + let error = state + .service_playback(&mut sink, &incoming, &control, &playback) + .unwrap_err(); + assert!( + error + .to_string() + .contains("did not acknowledge playback clear") + ); + } + // Offline: no microphone, devices, credentials or network. Run explicitly to // check the DSP budget when evaluating a helper build on a target machine. #[test] diff --git a/native/src/audio/resample.rs b/native/src/audio/resample.rs index c88229d..d9a2d8d 100644 --- a/native/src/audio/resample.rs +++ b/native/src/audio/resample.rs @@ -8,6 +8,7 @@ use rubato::audioadapter_buffers::direct::InterleavedSlice; pub struct Converter { inner: Option, output: VecDeque, + output_limit: Option, } struct Inner { @@ -40,12 +41,36 @@ impl Converter { Ok(Self { inner, output: VecDeque::new(), + output_limit: None, }) } + pub fn new_bounded(from: u32, to: u32, max_output: usize) -> anyhow::Result { + let mut converter = Self::new(from, to)?; + converter.output_limit = Some(max_output); + Ok(converter) + } + + fn append_output(output: &mut VecDeque, limit: Option, samples: &[f32]) { + let Some(limit) = limit else { + output.extend(samples); + return; + }; + let skip = samples.len().saturating_sub(limit); + let samples = &samples[skip..]; + let excess = output + .len() + .saturating_add(samples.len()) + .saturating_sub(limit); + if excess > 0 { + output.drain(..excess); + } + output.extend(samples); + } + pub fn push(&mut self, samples: &[f32]) -> anyhow::Result<()> { let Some(inner) = &mut self.inner else { - self.output.extend(samples); + Self::append_output(&mut self.output, self.output_limit, samples); return Ok(()); }; inner.input.extend(samples); @@ -62,7 +87,11 @@ impl Converter { .process_into_buffer(&input, &mut output, None) .map_err(|error| anyhow::anyhow!("resampling failed: {error}"))?; inner.input.drain(..consumed); - self.output.extend(&inner.scratch[..produced]); + Self::append_output( + &mut self.output, + self.output_limit, + &inner.scratch[..produced], + ); } Ok(()) } @@ -118,6 +147,25 @@ mod tests { assert_eq!(converter.available(), 2); } + #[test] + fn bounded_output_drops_oldest_samples() { + let mut converter = Converter::new_bounded(48_000, 48_000, 4).unwrap(); + converter.push(&[1.0, 2.0, 3.0, 4.0]).unwrap(); + converter.push(&[5.0, 6.0]).unwrap(); + assert_eq!(converter.available(), 4); + assert_eq!( + converter.drain(usize::MAX).collect::>(), + vec![3.0, 4.0, 5.0, 6.0] + ); + } + + #[test] + fn bounded_output_stays_within_limit_after_resampling() { + let mut converter = Converter::new_bounded(48_000, 44_100, 32).unwrap(); + converter.push(&[0.25; 4_800]).unwrap(); + assert_eq!(converter.available(), 32); + } + #[test] fn clear_discards_output_partial_input_and_resampler_history() { for rate in [48_000, 44_100] { From 41fba4220cfb58ef1d3013e4fec5e83df036602e Mon Sep 17 00:00:00 2001 From: "M. Adel Alhashemi" Date: Thu, 8 Oct 2026 00:31:16 +0300 Subject: [PATCH 4/4] fix(audio): flush playback after each packet so a large batch isn't dropped early The engine decoded up to 32 packets before moving any audio to the speaker ring, so a batch over one second overflowed the pending cap and dropped its oldest audio even when the ring had room. Flush after every packet instead. --- native/src/audio/engine.rs | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/native/src/audio/engine.rs b/native/src/audio/engine.rs index 12cddcc..d9c0498 100644 --- a/native/src/audio/engine.rs +++ b/native/src/audio/engine.rs @@ -321,11 +321,14 @@ impl State { return Ok(()); } self.clear_started = None; + // Flush after every packet, so a large batch moves into the speaker ring as it is + // decoded instead of overflowing the pending cap while the ring still has room. for _ in 0..32 { let Ok(packet) = incoming.try_recv() else { break; }; self.receive(packet)?; + self.flush_playback(sink); } self.flush_playback(sink); Ok(()) @@ -442,6 +445,32 @@ mod tests { assert_eq!(state.playback.available(), 0); } + #[test] + fn a_full_batch_of_packets_reaches_the_speaker_ring_without_dropping() { + use ringbuf::traits::Observer; + use ringbuf::traits::Split; + let mut state = State::new(RATE, RATE).unwrap(); + let control = EngineControl::default(); + let playback = super::super::io::PlaybackControl::default(); + let (mut sink, _cons) = ringbuf::HeapRb::::new(RATE as usize * 4).split(); + let (tx, incoming) = std_mpsc::channel(); + let mut encoder = + opus::Encoder::new(RATE, opus::Channels::Mono, opus::Application::Voip).unwrap(); + let packet = vec![0.1f32; FRAME * 2]; // 40 ms + for sequence in 0..32u16 { + let payload = encoder.encode_vec_float(&packet, 4000).unwrap(); + tx.send(IncomingPacket { sequence, payload }).unwrap(); + } + state + .service_playback(&mut sink, &incoming, &control, &playback) + .unwrap(); + // 32 x 40 ms is more than the 1 s pending cap, but fits the 4 s speaker ring. + assert_eq!( + sink.occupied_len() + state.playback.available(), + 32 * packet.len() + ); + } + #[test] fn playback_pending_audio_is_bounded_and_keeps_the_newest_samples() { let mut state = State::new(RATE, RATE).unwrap();