From cd02e0b8d7f242ca4f0fdb25c4b2440e95e881bc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 10 Nov 2026 01:16:32 -0800 Subject: [PATCH 1/6] chore(quest): claim quest/m1/audio-jitter-target/native-backpressure.md Quest-Claim: 33dd24c1d8137f11 From fcd21db23610a90b4861a51e5f9ea4fe5a5768c8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 10 Nov 2026 03:20:09 -0800 Subject: [PATCH 2/6] feat(audio): moq play decodes audio only when the ring has room Co-Authored-By: Claude Opus 5.5 --- quest/m1/audio-jitter-target/README.md | 1 - .../native-backpressure.md | 73 ------ quest/m1/audio-jitter-target/stretch.md | 1 - rs/moq-audio/src/decode/backpressure.rs | 71 ++++++ rs/moq-audio/src/decode/consumer.rs | 235 +++++++++++++++++- rs/moq-audio/src/decode/mod.rs | 2 + rs/moq-cli/src/play/media.rs | 108 +++++--- 7 files changed, 377 insertions(+), 114 deletions(-) delete mode 100644 quest/m1/audio-jitter-target/native-backpressure.md create mode 100644 rs/moq-audio/src/decode/backpressure.rs diff --git a/quest/m1/audio-jitter-target/README.md b/quest/m1/audio-jitter-target/README.md index 939e8b295e1..1dd9aa5bd66 100644 --- a/quest/m1/audio-jitter-target/README.md +++ b/quest/m1/audio-jitter-target/README.md @@ -96,7 +96,6 @@ native playback against its budgets. - [Capped budgets](/quest/m1/audio-jitter-target/capped-budgets.md) - the AAC browser budgets are re-measured from nightlies after the TS importer groups each PES - [Native A/V lag](/quest/m1/audio-jitter-target/native-av-lag.md) - `moq play --metrics` reports how far audio trails its presentation clock, so native rows grade `av_lag_ms` - [Chromium Opus stalled share](/quest/m1/audio-jitter-target/opus-stalled-share.md) - the Chromium Opus browser rows' stalled-share ceiling is measured with the stretch, and passes -- [Native backpressure](/quest/m1/audio-jitter-target/native-backpressure.md) - native playback decodes audio only when the ring has room, holding the rest of the target encoded as the browser does ## Closes diff --git a/quest/m1/audio-jitter-target/native-backpressure.md b/quest/m1/audio-jitter-target/native-backpressure.md deleted file mode 100644 index 5554e3c36c1..00000000000 --- a/quest/m1/audio-jitter-target/native-backpressure.md +++ /dev/null @@ -1,73 +0,0 @@ -# [M] Native playback decodes audio only when the ring has room - -## Goal - -Native playback (`moq_audio::decode::Consumer` feeding `moq play`'s sink) -decodes an audio frame when the browser's `Backpressure` would: while playing, -once the playhead is within the ring's floor of it, and, live, once the media -queued behind it covers the part of the target held encoded, so a backlog past -the target still reaches the ring to be caught up. The rest of the target -delay stays encoded upstream, so native memory follows the bytes actually -received and the ring never grows with the delay. Once this lands, Rust splits -the target between decoded PCM and encoded media the way JS does, and -`moq play` holds no decoded PCM outside the ring. - -## Plan - -Planned 2026-11-09 (maintainer, /quest-complete follow-up of -[moq-dev/moq#8362](https://github.com/moq-dev/moq/pull/8362): "quest-plan -backpressure like JS does. So we don't decode audio until there's space in -the ring buffer"). ✅ m1, in this epic (rejected: m2, m3). - -The browser already works this way: `Backpressure` in -`js/watch/src/audio/buffer.ts` decides when the decode loop may decode a frame, -holding the ring at its floor of PCM and the hold encoded in the container -consumer. Native decodes promptly instead: `Consumer::read` decodes each packet -as it is read, and the PCM waits in `play_audio`'s queue in -`rs/moq-cli/src/play/media.rs`. #8362 keeps a delay past the ring's budget in -sync by parking that PCM there until it is due. This quest replaces that queue -with the gate, so the hold stays encoded. - -Mirror the browser's rules, including the stalled refill that starts playback -with the whole target ahead of the playhead, rather than inventing new ones. -Share names and tests with the JS side where the two can match. - -The gate waits before decoding, never before reading. `Consumer::read` stamps -each packet's arrival for the jitter estimate (`target.observe` in -`rs/moq-audio/src/decode/consumer.rs`), and `Consumer::delay()` documents that -reads paced to the speaker make the estimate measure the pacing. Keep reading -and observing encoded packets as they arrive and hold them encoded, as -`js/watch` reads a frame and only then awaits `ring.wait()`. - -The gate has two inputs, and `play_audio` owns both, so the consumer needs -them passed in: - -- The playhead the JS worklet reports. Native can derive it from - `Sink::buffered()` (`rs/moq-audio/src/playback/sink.rs`) and the last - write's timestamp. -- The shared playout delay: the catalog `delay` plus the target, the largest - among the subscribed renditions, which the browser passes as - `sync.out.delay` and `play_audio` already holds the sink to through - `presentation.due`. `Consumer::delay()` is this track's jitter target alone; - a gate on it would decode at a depth the sink has already outrun and could - not replace #8362's hold. - -Decided 2026-11-09 (maintainer, /quest-unblock): the gate must not start -audio early when a shared catalog delay is under the ring's cap, with a test. -✅ Fix it here (rejected: pad the missing lead in -[moq-dev/moq#8362](https://github.com/moq-dev/moq/pull/8362)). With a shared delay -that fits the sink's cap, that PR writes the frame without its hold-path lead -padding, so audio can start up to ~150 ms early -(`rs/moq-cli/src/play/media.rs`). The browser-mirroring refill starts with the -whole target ahead of the playhead and replaces that padding path, so the fix -lands with the gate. - -Keep the gate inside moq-audio's consumer if that stays additive; if it needs -a public API change, it moves to m2. - -Public API: additive if the gate stays inside `moq_audio::decode::Consumer`. -Wire: none. - -## Related - -- [Stretch model](/quest/m1/audio-jitter-target/stretch.md) - requires this gate, then changes the stalled refill mirrored here to a small cushion in both players diff --git a/quest/m1/audio-jitter-target/stretch.md b/quest/m1/audio-jitter-target/stretch.md index 9e73984f2b1..e48ffc5f44b 100644 --- a/quest/m1/audio-jitter-target/stretch.md +++ b/quest/m1/audio-jitter-target/stretch.md @@ -126,7 +126,6 @@ Public API: none expected. Wire: none. ## Required - [The audio harness grades the stretch model](/quest/m1/audio-jitter-target/grading.md) - the gates that pushed toward zero underruns and zero lag are gone before this is measured against them -- [Native backpressure](/quest/m1/audio-jitter-target/native-backpressure.md) - native holds its ring at a floor and the rest of the target encoded, as the browser does, so one rule reads the same buffer in both ## Closes diff --git a/rs/moq-audio/src/decode/backpressure.rs b/rs/moq-audio/src/decode/backpressure.rs new file mode 100644 index 00000000000..13b81a68e45 --- /dev/null +++ b/rs/moq-audio/src/decode/backpressure.rs @@ -0,0 +1,71 @@ +use std::time::Duration; + +use kio::clock::Instant; + +/// Decides when [`Consumer::read`](super::Consumer::read) decodes a packet, which is how a +/// player splits its playout delay between decoded PCM in its playback ring and encoded +/// media held in the consumer. +/// +/// The ring holds the floor as PCM. The rest of the delay, the hold, stays encoded, so +/// memory follows the bytes actually received rather than the delay. The consumer keeps +/// reading packets as they arrive, so its delay estimate still times each one, and +/// decodes a packet at `timestamp`: +/// +/// - once the media queued behind it covers the hold, or nothing more continues it +/// (a gap, a declared end, or the track's end is queued after it). A backlog past +/// the delay still reaches the ring this way, to be caught up; +/// - while the playhead is known, also once it is within the floor of where the +/// packet's output ends, keeping the ring on its floor when the network runs late. +/// +/// With no hold every packet decodes at once and the ring bounds itself. The playhead is +/// unknown until the first [`observe`](Self::observe), and again after each gap or end +/// the consumer hands over, until an observation lands past it. +/// +/// Dropping the handle stops the gating, decoding every packet as it arrives. +pub struct Backpressure { + state: kio::Producer, +} + +#[derive(Debug, Default)] +pub(super) struct State { + pub floor: Duration, + pub hold: Duration, + /// Media at the first value sounds at the second, advancing in real time from there. + pub playhead: Option<(Duration, Instant)>, + /// Bumped on every update, so a waiting read notices one. + pub version: u64, +} + +impl Backpressure { + pub(super) fn new() -> (Self, kio::Consumer) { + let state = kio::Producer::new(State::default()); + let consumer = state.consume(); + (Self { state }, consumer) + } + + /// Split the playout delay: up to `floor` decoded in the ring, and `hold` past it encoded. + pub fn set_latency(&self, floor: Duration, hold: Duration) { + self.update(|state| { + state.floor = floor; + state.hold = hold; + }); + } + + /// Note that media at `timestamp`, on the consumer's timeline, sounds at `at`. + pub fn observe(&self, timestamp: moq_net::Timestamp, at: Instant) { + self.update(|state| state.playhead = Some((Duration::from(timestamp), at))); + } + + fn update(&self, f: impl FnOnce(&mut State)) { + if let Ok(mut state) = self.state.write() { + f(&mut state); + state.version += 1; + } + } +} + +impl std::fmt::Debug for Backpressure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Backpressure").finish_non_exhaustive() + } +} diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index 95ccb49848d..8fa3ae5f9cd 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -1,6 +1,7 @@ //! Subscribe to an encoded audio track and emit raw PCM. use std::collections::VecDeque; +use std::task::Poll; use kio::clock::Instant; use moq_mux::container::Read as Encoded; @@ -8,6 +9,7 @@ use std::time::Duration; use bytes::Bytes; +use super::backpressure::{self, Backpressure}; use super::decoder::{Config, Decoder}; use crate::jitter::{self, Target}; use crate::resample::{Remix, Resampler}; @@ -163,6 +165,24 @@ pub struct Consumer { target: Target, /// Where arrival times are measured from. origin: Instant, + /// Packets read and observed but not yet decoded, held by the [`Backpressure`] gate. + pending: VecDeque, + /// The track ended, or failed with this error, behind whatever is pending. + ended: Option>, + /// The gate, while its [`Backpressure`] handle lives, and the version last judged. + gate: Option>, + seen: u64, + /// The latest gap or end handed over: a playhead at or before it is from the timeline it ended. + boundary: Option, + /// The duration of the latest decoded packet, which the next one is taken to match. + frame: Duration, +} + +/// Whether a pending packet may decode. +enum Admit { + Now, + At(kio::clock::Instant), + Wait, } struct ActivitySpan { @@ -272,6 +292,12 @@ impl Consumer { end_group: 0, terminal_start: None, origin: now(), + pending: VecDeque::new(), + ended: None, + gate: None, + seen: 0, + boundary: None, + frame: Duration::ZERO, }) } @@ -312,6 +338,18 @@ impl Consumer { self.resolved_layout } + /// Gate decoding behind the returned [`Backpressure`], replacing any earlier gate. + /// + /// [`read`](Self::read) keeps reading packets as they arrive and holds them encoded + /// until the gate admits them. Dropping the handle decodes every packet as it arrives + /// again. + pub fn backpressure(&mut self) -> Backpressure { + let (backpressure, gate) = Backpressure::new(); + self.gate = Some(gate); + self.seen = 0; + backpressure + } + /// Read decoded PCM or a track boundary, or `None` when the track ends. /// /// [`Frame::activity`] reports whether the packet these samples came from @@ -330,7 +368,7 @@ impl Consumer { return Ok(Some(frame)); } - let (mux_frame, track_start) = match self.track.read().await? { + let (mux_frame, track_start) = match self.next().await? { Some(Encoded::Frame { group, frame }) => { let first = !std::mem::replace(&mut self.started, true); // A later group after a declared end comes from a fresh encoder; the end's @@ -348,6 +386,7 @@ impl Consumer { } // Behind whatever the resampler held from before the loss. Some(Encoded::Gap { timestamp, duration }) => { + self.boundary = Some(timestamp.into()); self.started = true; // The restart forgets the end, but the silence across it stays declared. if self.end.is_some() { @@ -363,6 +402,7 @@ impl Consumer { // after the end, starting before it, once a fresh encoder or the track's end // drains it. Some(Encoded::End { group, end }) => { + self.boundary = Some(end.into()); self.end = Some(end); self.end_group = group; self.frames_decoded = 0; @@ -373,12 +413,7 @@ impl Consumer { None => return Ok(self.flush()?.map(Read::Frame)), }; - // Every packet the container hands over is an arrival, including one the - // decoder is about to refuse, so it is observed before anything can fail. - let arrival = jitter::millis(now().duration_since(self.origin)); let timestamp = mux_frame.timestamp; - let media = timestamp.value() as f64 * 1000.0 / timestamp.scale().as_u64() as f64; - self.target.observe(arrival, media); // Undeclared holes are routine: a skipped stalled group, a packet the // decoder refused, an ingest that resynced. Drop every stage's state at @@ -438,6 +473,7 @@ impl Consumer { let covered = decoded.len() / self.decoder.layout().channels() as usize + trimmed; if covered > 0 { self.target.set_frame(covered as f64 * 1000.0 / rate as f64); + self.frame = Duration::from_secs_f64(covered as f64 / rate as f64); } if let Some(end) = self.end { let terminal_start = *self @@ -534,6 +570,118 @@ impl Consumer { } } + /// The next packet or boundary the gate admits, reading and observing every packet + /// that arrives meanwhile, or `None` once the track ends. + async fn next(&mut self) -> Result, Error> { + loop { + let deadline = match self.pending.front() { + Some(Encoded::Frame { frame, .. }) => match self.admit(frame.timestamp.into()) { + Admit::Now => return Ok(self.pending.pop_front()), + Admit::At(at) => Some(at), + Admit::Wait => None, + }, + Some(_) => return Ok(self.pending.pop_front()), + None => match self.ended.as_mut() { + Some(failed) => return failed.take().map_or(Ok(None), Err), + None => None, + }, + }; + + let gate = self.gate.as_ref().filter(|gate| !gate.is_closed()); + let seen = self.seen; + let mut sleep = std::pin::pin!(deadline.map(kio::clock::sleep_until)); + let track = &mut self.track; + let ended = self.ended.is_some(); + let read = kio::wait(|waiter| { + if !ended && let Poll::Ready(read) = track.poll_read(waiter) { + return Poll::Ready(Some(read)); + } + if let Some(gate) = gate + && gate + .poll(waiter, |state| match state.version != seen { + true => Poll::Ready(()), + false => Poll::Pending, + }) + .is_ready() + { + return Poll::Ready(None); + } + if let Some(sleep) = sleep.as_mut().as_pin_mut() + && waiter.poll_future(sleep).is_ready() + { + return Poll::Ready(None); + } + Poll::Pending + }) + .await; + + match read { + Some(Ok(Some(read))) => { + // Every packet the container hands over is an arrival, including one the + // decoder will refuse, so it is observed before anything can fail. + if let Encoded::Frame { frame, .. } = &read { + let arrival = jitter::millis(now().duration_since(self.origin)); + let media = frame.timestamp.value() as f64 * 1000.0 / frame.timestamp.scale().as_u64() as f64; + self.target.observe(arrival, media); + } + self.pending.push_back(read); + } + Some(Ok(None)) => self.ended = Some(None), + Some(Err(err)) => self.ended = Some(Some(err.into())), + None => {} + } + } + } + + /// Whether the pending packet at `timestamp` may decode now, and if not, when it may. + fn admit(&mut self, timestamp: Duration) -> Admit { + let Some(gate) = self.gate.as_ref().filter(|gate| !gate.is_closed()) else { + return Admit::Now; + }; + let state = gate.read(); + self.seen = state.version; + if state.hold.is_zero() || self.ended.is_some() { + return Admit::Now; + } + // Media queued behind the packet covers the hold, or nothing more continues it. + let mut queued = self.pending.iter().skip(1).rev(); + let covered = match queued.next() { + Some(Encoded::Frame { frame, .. }) => { + Duration::from(frame.timestamp) >= timestamp + state.hold + || queued.any(|read| !matches!(read, Encoded::Frame { .. })) + } + Some(_) => true, + None => false, + }; + if covered { + return Admit::Now; + } + + let boundary = self.boundary; + let Some((playhead, at)) = state + .playhead + .filter(|(playhead, _)| boundary.is_none_or(|boundary| *playhead > boundary)) + else { + return Admit::Wait; + }; + // Due once the playhead is within the floor of where this packet's output ends. + let end = timestamp + self.frame; + let reach = playhead + state.floor; + let due = match end.checked_sub(reach) { + Some(ahead) => at.checked_add(ahead), + // Before the clock's origin is long past. + None => match at.checked_sub(reach - end) { + Some(due) => Some(due), + None => return Admit::Now, + }, + }; + match due { + Some(due) if due > kio::clock::Instant::now() => Admit::At(due), + Some(_) => Admit::Now, + None => Admit::Wait, + } + } + /// Start a new timeline after a gap or past a declared end. The decoder is not reset: /// the next group already starts on a keyframe, and pre-skip is a play-path concern. /// @@ -1937,6 +2085,81 @@ mod tests { read } + /// Backpressure holds packets encoded while still reading and timing every one: + /// stalled, a packet decodes once the media behind it covers the hold, and with + /// a playhead, once that is within the floor of where the packet ends. + #[tokio::test(start_paused = true)] + async fn backpressure_holds_packets_encoded_until_the_ring_has_room() { + const PACKET: Duration = Duration::from_millis(20); + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, 48_000, 1); + let mut producer = moq_mux::container::Producer::new( + track, + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio), + ); + let mut consumer = Consumer::new( + &broadcast.consume(), + &catalog, + "audio", + Options { + delay: Delay { + max: Some(Duration::from_secs(1)), + ..Delay::default() + }, + ..Options::new() + }, + ) + .await + .unwrap(); + let backpressure = consumer.backpressure(); + backpressure.set_latency(Duration::from_millis(100), Duration::from_millis(200)); + + let payload: Bytes = vec![0u8; 960 * size_of::()].into(); + let mut write = |index: u32| { + producer + .write(moq_mux::container::Frame { + timestamp: Timestamp::try_from(PACKET * index).unwrap(), + duration: None, + payload: payload.clone(), + keyframe: true, + }) + .unwrap(); + }; + let read = async |consumer: &mut Consumer, wait: Duration| { + kio::clock::timeout(wait, next_frame(consumer)) + .await + .ok() + .map(|frame| Duration::from(frame.unwrap().unwrap().timestamp)) + }; + + // Ten packets cover 180 ms behind the first, short of the hold. + for index in 0..10 { + write(index); + } + assert_eq!(read(&mut consumer, Duration::from_secs(1)).await, None); + assert_eq!(consumer.pending.len(), 10, "every packet is read and held encoded"); + write(10); + assert_eq!(read(&mut consumer, Duration::ZERO).await, Some(Duration::ZERO)); + + // The first packet ends sounding in half a second, so the next one, ending + // 20 ms later, is within the 100 ms floor of the playhead 80 ms before that. + let sounds = kio::clock::Instant::now().checked_add(Duration::from_millis(500)).unwrap(); + backpressure.observe(Timestamp::try_from(PACKET).unwrap(), sounds); + assert_eq!(read(&mut consumer, Duration::from_millis(419)).await, None); + assert_eq!(read(&mut consumer, Duration::from_millis(1)).await, Some(PACKET)); + + // The track's end leaves nothing to wait for. + producer.finish().unwrap(); + let mut rest = 0; + while next_frame(&mut consumer).await.unwrap().is_some() { + rest += 1; + } + assert_eq!(rest, 9); + } + /// A packet that doesn't continue the last one is a hole, not a splice: the /// resampler hands back what it was holding from before the gap as its own /// frame, and the audio after it is stamped from the packet that carried it diff --git a/rs/moq-audio/src/decode/mod.rs b/rs/moq-audio/src/decode/mod.rs index a0e71455cd0..86220dd9fb9 100644 --- a/rs/moq-audio/src/decode/mod.rs +++ b/rs/moq-audio/src/decode/mod.rs @@ -15,10 +15,12 @@ //! PCM, and symphonia for AAC-LC). mod backend; +mod backpressure; mod consumer; mod decoded; mod decoder; +pub use backpressure::Backpressure; pub use consumer::{Consumer, Delay, Options, Output, Read, Start}; pub use decoded::Decoded; pub use decoder::{Config, Decoder, Kind}; diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index c18b80f353a..5f2e9546122 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -640,9 +640,27 @@ async fn play_audio( let lead = |delay: Duration| latency.max(delay.min(max_delay).max(AUDIO_BUFFER_MIN)); // The most the sink holds. A shared delay past it, such as a slow video - // encoder's, holds decoded media here until it is due instead. + // encoder's, stays encoded in the consumer until it is due instead, so the + // ring never grows with the delay. let cap = max_delay.max(AUDIO_BUFFER_MIN); let schedule = presentation.consume(); + // The consumer decodes a packet once the sink has room for it. Its playhead is + // where the shared clock says the last write sounds, rather than what the sink + // is sounding: audio follows that clock, so a raised delay holds media encoded + // until it is due instead of decoding toward a sink that would play it early. + let backpressure = consumer.backpressure(); + let steer = |wrote: Option<(moq_net::Timestamp, Duration)>| { + let shared = schedule.delay(); + backpressure.set_latency(shared.min(cap), shared.saturating_sub(cap)); + if let Some((end, mapped)) = wrote + && let Some(due) = moq_net::Timestamp::try_from(mapped) + .ok() + .and_then(|end| presentation.due(end)) + { + backpressure.observe(end, due); + } + }; + steer(None); // The depth that makes media ending at `end` sound when the shared clock says it is due. let hold = |end: Duration| { moq_net::Timestamp::try_from(end) @@ -788,8 +806,10 @@ async fn play_audio( let mut utterance = false; // The last frame written was trimmed, so a trim of this one continues that skip. let mut trimming = false; - // Where the media written so far ends. - let mut wrote = None; + // Where the media written so far ends, on the consumer's timeline and mapped. + let mut wrote: Option<(moq_net::Timestamp, Duration)> = None; + // The next write starts this sink, which plays it one opening cushion out. + let mut fresh = true; loop { let (event, delay, crossing, arrived_at) = loop { if let Some(event) = received.lock().unwrap().pop_front() { @@ -801,7 +821,8 @@ async fn play_audio( () = arrived.notified() => {} // No write may come to carry a delay change, so the speaker follows it now. _ = role.delay_changed(seen) => { - if let Some(held) = wrote.and_then(hold) { + steer(wrote); + if let Some(held) = wrote.and_then(|(_, end)| hold(end)) { sink.set_latency(held)?; } } @@ -908,6 +929,7 @@ async fn play_audio( role.reset(); playhead.restart(); sink = speaker.sink(input.clone())?; + fresh = true; } // Resumed media was received while the old tail played. Keep its @@ -933,19 +955,14 @@ async fn play_audio( // clock holds both while the sink keeps sizing for jitter alone. *jitter.lock().unwrap() = Some(input.latency); moved |= role.set_target(input.latency + stated.borrow().unwrap_or_default()); - // Media due further out than the sink holds waits here. The sink then - // holds only what was already written, its opening silence included, so - // the rest of the lead goes in as silence ahead of the frame, which already - // covers any hole before it. - let held_back = wait(&mut catalog, &|| { - presentation.due(mapped).and_then(|due| due.checked_sub(cap)) - }) - .await; - // A frame starting an utterance is instead placed one target out by the - // sink itself, and `hold` below sizes that target to the due time. + // A fresh sink plays its first write behind an opening cushion sized before + // the delay was known, so the rest of the lead goes in as silence ahead of + // the frame, which already covers any hole before it. A frame starting an + // utterance is instead placed one target out by the sink itself, and `hold` + // below sizes that target to the due time. let short = presentation .due(mapped) - .filter(|_| held_back && !utterance) + .filter(|_| fresh && !utterance) .map(|due| { due.saturating_duration_since(Instant::now()) .saturating_sub(sink.buffered()) @@ -955,7 +972,7 @@ async fn play_audio( // it is due, so the callback stretches the speaker onto the clock: behind // it after a stall, it plays faster until it catches up. Floored at what // the device needs, and capped at the age budget the sink was sized for; - // the wait above holds anything due later. + // the consumer holds anything due later encoded. let buffered = samples(sink.buffered()); let held = hold(timing.end).unwrap_or(input.latency.clamp(AUDIO_BUFFER_MIN, cap)); let target = samples(held); @@ -1001,7 +1018,12 @@ async fn play_audio( silent = true; } - wrote = Some(timing.end); + if written.accepted_sample_frames > 0 { + fresh = false; + } + let end = Duration::from(frame.timestamp) + Duration::from_secs_f64(length as f64 / sample_rate as f64); + wrote = moq_net::Timestamp::try_from(end).ok().map(|end| (end, timing.end)); + steer(wrote); playhead.wrote(timing.end, sink.buffered(), Instant::now()); // A move has to wake the window: it is asleep on a deadline computed from // the old schedule, and every queued frame is due earlier now. @@ -1132,12 +1154,36 @@ mod tests { } /// A video encoder delay past audio's own budget still plays audio in sync: - /// the media waits here until it is due rather than sounding early. + /// the media waits encoded until it is due rather than sounding early. #[tokio::test] async fn audio_holds_a_shared_delay_past_its_budget() { - tokio::time::pause(); let budget = Duration::from_millis(200); let encoder = Duration::from_secs(1); + assert_on_time(first_sound(budget, budget, None, Some(encoder)).await, encoder); + } + + /// A catalog delay the sink can hold still plays audio when it is due: the + /// sink opens on a cushion sized before the delay was known, so the first + /// write is placed behind the rest of the lead rather than sounding early. + #[tokio::test] + async fn audio_holds_a_catalog_delay_within_its_budget() { + let jitter = Duration::from_millis(200); + let catalog = Duration::from_millis(150); + let sounded = first_sound(jitter, Duration::from_secs(1), Some(catalog), None).await; + assert_on_time(sounded, jitter + catalog); + } + + fn assert_on_time(sounded: Duration, due: Duration) { + assert!( + sounded.max(due) - sounded.min(due) <= Duration::from_millis(1), + "audio sounded at {sounded:?}, due {due:?}" + ); + } + + /// How long after it arrives the first packet sounds, holding `min` as the + /// jitter target within a `max` budget, plus a catalog and a video delay. + async fn first_sound(min: Duration, max: Duration, catalog: Option, video: Option) -> Duration { + tokio::time::pause(); let broadcast = moq_net::broadcast::Info::new().produce(); let track = broadcast .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) @@ -1145,8 +1191,8 @@ mod tests { let mut producer = moq_mux::container::Producer::new(track, Container::Legacy(moq_mux::container::Kind::Audio)); let mut options = moq_audio::decode::Options::new(); options.delay = moq_audio::decode::Delay { - min: Some(budget), - max: Some(budget), + min: Some(min), + max: Some(max), }; options.estimate = false; options.output.format = moq_audio::Format::F32; @@ -1158,15 +1204,17 @@ mod tests { ) .await .unwrap(); - let presentation = Presentation::new(budget); - let video = presentation.track(moq_mux::container::Kind::Video); - video.set_target(encoder); + let presentation = Presentation::new(min); + if let Some(video) = video { + presentation.track(moq_mux::container::Kind::Video).set_target(video); + } + let (_catalog, stated) = tokio::sync::watch::channel(catalog); let recorder = Recorder::default(); let playback = AudioPlayback { - delay: tokio::sync::watch::channel(None).1, + delay: stated, changed: Default::default(), speaker: recorder.clone(), - presentation: presentation.clone(), + presentation, playhead: Default::default(), // What `--delay auto` opens on, well short of the lead the held media needs. latency: AUDIO_BUFFER_MIN, @@ -1190,13 +1238,7 @@ mod tests { .into_iter() .find(|part| part.sample == 0.25) .expect("played audio"); - let due = start.checked_add(encoder).expect("test deadline overflows"); - assert!( - first.from.max(due) - first.from.min(due) <= Duration::from_millis(1), - "audio sounded at {:?}, due {:?}", - first.from - start, - encoder - ); + first.from - start } /// A push-to-talk mute declares its silence and sends nothing until the next From 5339b0f6f63ba4c1e0f13295822b1c088726063b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 10 Nov 2026 07:27:45 -0800 Subject: [PATCH 3/6] fix(audio): declare a silence to the estimator as its packets arrive Co-Authored-By: Claude Opus 5.5 --- rs/moq-audio/src/decode/consumer.rs | 29 +++++++++++++++++++++-------- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index 8fa3ae5f9cd..ea173f6d524 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -176,6 +176,8 @@ pub struct Consumer { boundary: Option, /// The duration of the latest decoded packet, which the next one is taken to match. frame: Duration, + /// The group of the latest end read, until a later group or a loss arrives. + declared: Option, } /// Whether a pending packet may decode. @@ -298,6 +300,7 @@ impl Consumer { seen: 0, boundary: None, frame: Duration::ZERO, + declared: None, }) } @@ -373,10 +376,7 @@ impl Consumer { let first = !std::mem::replace(&mut self.started, true); // A later group after a declared end comes from a fresh encoder; the end's // own group carries the stopped encoder's final packets, trimmed below. - // Those packets end the speech before the silence, so only this frame - // tells the estimator the silence it measures across was declared. if self.end.is_some() && group != self.end_group { - self.target.end(); self.restart()?; } // Nothing precedes group 0, and a group delivers from its frame 0 or not at @@ -388,10 +388,6 @@ impl Consumer { Some(Encoded::Gap { timestamp, duration }) => { self.boundary = Some(timestamp.into()); self.started = true; - // The restart forgets the end, but the silence across it stays declared. - if self.end.is_some() { - self.target.end(); - } self.restart()?; self.warmup_pending = !self.warmup.is_zero(); self.ready.push_back(Read::Gap { timestamp, duration }); @@ -617,6 +613,21 @@ impl Consumer { match read { Some(Ok(Some(read))) => { + // The estimator times arrivals, so it learns of a declared silence as the + // packets around it arrive rather than as they decode. The end's own group + // carries the stopped encoder's final packets, which end the speech before + // the silence, so only a later group, or a loss, says the silence it + // measures across was declared. A loss forgets the end, but the silence + // across it stays declared. + match &read { + Encoded::End { group, .. } => self.declared = Some(*group), + Encoded::Frame { group, .. } if self.declared.is_some_and(|end| end != *group) => { + self.declared = None; + self.target.end(); + } + Encoded::Gap { .. } if self.declared.take().is_some() => self.target.end(), + _ => {} + } // Every packet the container hands over is an arrival, including one the // decoder will refuse, so it is observed before anything can fail. if let Encoded::Frame { frame, .. } = &read { @@ -2146,7 +2157,9 @@ mod tests { // The first packet ends sounding in half a second, so the next one, ending // 20 ms later, is within the 100 ms floor of the playhead 80 ms before that. - let sounds = kio::clock::Instant::now().checked_add(Duration::from_millis(500)).unwrap(); + let sounds = kio::clock::Instant::now() + .checked_add(Duration::from_millis(500)) + .unwrap(); backpressure.observe(Timestamp::try_from(PACKET).unwrap(), sounds); assert_eq!(read(&mut consumer, Duration::from_millis(419)).await, None); assert_eq!(read(&mut consumer, Duration::from_millis(1)).await, Some(PACKET)); From 8d4de17c8efd93eaf366a9b17e08e121a1ec68f4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 10 Nov 2026 08:23:43 -0800 Subject: [PATCH 4/6] fix(audio): judge the gate once per wait, and hold a timeline's early tail Co-Authored-By: Claude Opus 5.5 --- rs/moq-audio/src/decode/consumer.rs | 11 ++++------- rs/moq-cli/src/play/media.rs | 27 +++++++++++++++++++-------- 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index ea173f6d524..bc3a21b9deb 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -169,9 +169,8 @@ pub struct Consumer { pending: VecDeque, /// The track ended, or failed with this error, behind whatever is pending. ended: Option>, - /// The gate, while its [`Backpressure`] handle lives, and the version last judged. + /// The gate, while its [`Backpressure`] handle lives. gate: Option>, - seen: u64, /// The latest gap or end handed over: a playhead at or before it is from the timeline it ended. boundary: Option, /// The duration of the latest decoded packet, which the next one is taken to match. @@ -297,7 +296,6 @@ impl Consumer { pending: VecDeque::new(), ended: None, gate: None, - seen: 0, boundary: None, frame: Duration::ZERO, declared: None, @@ -349,7 +347,6 @@ impl Consumer { pub fn backpressure(&mut self) -> Backpressure { let (backpressure, gate) = Backpressure::new(); self.gate = Some(gate); - self.seen = 0; backpressure } @@ -570,6 +567,8 @@ impl Consumer { /// that arrives meanwhile, or `None` once the track ends. async fn next(&mut self) -> Result, Error> { loop { + // Read before judging the head, so any update after that wakes the wait. + let seen = self.gate.as_ref().map_or(0, |gate| gate.read().version); let deadline = match self.pending.front() { Some(Encoded::Frame { frame, .. }) => match self.admit(frame.timestamp.into()) { Admit::Now => return Ok(self.pending.pop_front()), @@ -584,7 +583,6 @@ impl Consumer { }; let gate = self.gate.as_ref().filter(|gate| !gate.is_closed()); - let seen = self.seen; let mut sleep = std::pin::pin!(deadline.map(kio::clock::sleep_until)); let track = &mut self.track; let ended = self.ended.is_some(); @@ -645,12 +643,11 @@ impl Consumer { } /// Whether the pending packet at `timestamp` may decode now, and if not, when it may. - fn admit(&mut self, timestamp: Duration) -> Admit { + fn admit(&self, timestamp: Duration) -> Admit { let Some(gate) = self.gate.as_ref().filter(|gate| !gate.is_closed()) else { return Admit::Now; }; let state = gate.read(); - self.seen = state.version; if state.hold.is_zero() || self.ended.is_some() { return Admit::Now; } diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 5f2e9546122..ad983517f45 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -955,14 +955,22 @@ async fn play_audio( // clock holds both while the sink keeps sizing for jitter alone. *jitter.lock().unwrap() = Some(input.latency); moved |= role.set_target(input.latency + stated.borrow().unwrap_or_default()); + // The consumer releases media due further out than the sink holds only + // when its timeline ends before the media behind it covers the hold, so + // that little waits here instead of overflowing the sink. + let held_back = wait(&mut catalog, &|| { + presentation.due(mapped).and_then(|due| due.checked_sub(cap)) + }) + .await; // A fresh sink plays its first write behind an opening cushion sized before - // the delay was known, so the rest of the lead goes in as silence ahead of - // the frame, which already covers any hole before it. A frame starting an - // utterance is instead placed one target out by the sink itself, and `hold` - // below sizes that target to the due time. + // the delay was known, and a held frame behind only what was already + // written, so the rest of the lead goes in as silence ahead of the frame, + // which already covers any hole before it. A frame starting an utterance is + // instead placed one target out by the sink itself, and `hold` below sizes + // that target to the due time. let short = presentation .due(mapped) - .filter(|_| fresh && !utterance) + .filter(|_| (fresh || held_back) && !utterance) .map(|due| { due.saturating_duration_since(Instant::now()) .saturating_sub(sink.buffered()) @@ -1205,9 +1213,12 @@ mod tests { .await .unwrap(); let presentation = Presentation::new(min); - if let Some(video) = video { - presentation.track(moq_mux::container::Kind::Video).set_target(video); - } + // Held for the whole test: a dropped track stops asking for its delay. + let _video = video.map(|video| { + let track = presentation.track(moq_mux::container::Kind::Video); + track.set_target(video); + track + }); let (_catalog, stated) = tokio::sync::watch::channel(catalog); let recorder = Recorder::default(); let playback = AudioPlayback { From b30af87eb10110f1b9ae81fa063f9cd9feb2bddf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 10 Nov 2026 09:46:58 -0800 Subject: [PATCH 5/6] fix(audio): anchor on encoded arrivals and bound the gate by arrival time Co-Authored-By: Claude Opus 5.5 --- rs/moq-audio/src/decode/consumer.rs | 133 ++++++++++++++++------------ rs/moq-cli/src/play/media.rs | 21 +++-- 2 files changed, 88 insertions(+), 66 deletions(-) diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index bc3a21b9deb..14b0d28bd44 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -165,8 +165,12 @@ pub struct Consumer { target: Target, /// Where arrival times are measured from. origin: Instant, - /// Packets read and observed but not yet decoded, held by the [`Backpressure`] gate. - pending: VecDeque, + /// Packets read and observed but not yet decoded, held by the [`Backpressure`] gate, + /// each with when it arrived, and how many of them are gaps or ends. + pending: VecDeque<(Encoded, kio::clock::Instant)>, + breaks: usize, + /// When the latest packet decoded arrived. + arrived: Option, /// The track ended, or failed with this error, behind whatever is pending. ended: Option>, /// The gate, while its [`Backpressure`] handle lives. @@ -179,13 +183,6 @@ pub struct Consumer { declared: Option, } -/// Whether a pending packet may decode. -enum Admit { - Now, - At(kio::clock::Instant), - Wait, -} - struct ActivitySpan { end: moq_net::Timestamp, activity: Activity, @@ -294,6 +291,8 @@ impl Consumer { terminal_start: None, origin: now(), pending: VecDeque::new(), + breaks: 0, + arrived: None, ended: None, gate: None, boundary: None, @@ -350,6 +349,12 @@ impl Consumer { backpressure } + /// When the packet behind the latest decoded frame arrived, which a gate held + /// encoded may have kept well before [`read`](Self::read) handed the frame over. + pub fn arrival(&self) -> Option { + self.arrived + } + /// Read decoded PCM or a track boundary, or `None` when the track ends. /// /// [`Frame::activity`] reports whether the packet these samples came from @@ -570,12 +575,11 @@ impl Consumer { // Read before judging the head, so any update after that wakes the wait. let seen = self.gate.as_ref().map_or(0, |gate| gate.read().version); let deadline = match self.pending.front() { - Some(Encoded::Frame { frame, .. }) => match self.admit(frame.timestamp.into()) { - Admit::Now => return Ok(self.pending.pop_front()), - Admit::At(at) => Some(at), - Admit::Wait => None, + Some((Encoded::Frame { frame, .. }, arrived)) => match self.admit(frame.timestamp.into(), *arrived) { + None => return Ok(self.pop()), + due => due, }, - Some(_) => return Ok(self.pending.pop_front()), + Some(_) => return Ok(self.pop()), None => match self.ended.as_mut() { Some(failed) => return failed.take().map_or(Ok(None), Err), None => None, @@ -628,12 +632,16 @@ impl Consumer { } // Every packet the container hands over is an arrival, including one the // decoder will refuse, so it is observed before anything can fail. - if let Encoded::Frame { frame, .. } = &read { - let arrival = jitter::millis(now().duration_since(self.origin)); - let media = frame.timestamp.value() as f64 * 1000.0 / frame.timestamp.scale().as_u64() as f64; - self.target.observe(arrival, media); + match &read { + Encoded::Frame { frame, .. } => { + let arrival = jitter::millis(now().duration_since(self.origin)); + let media = + frame.timestamp.value() as f64 * 1000.0 / frame.timestamp.scale().as_u64() as f64; + self.target.observe(arrival, media); + } + _ => self.breaks += 1, } - self.pending.push_back(read); + self.pending.push_back((read, kio::clock::Instant::now())); } Some(Ok(None)) => self.ended = Some(None), Some(Err(err)) => self.ended = Some(Some(err.into())), @@ -642,52 +650,52 @@ impl Consumer { } } - /// Whether the pending packet at `timestamp` may decode now, and if not, when it may. - fn admit(&self, timestamp: Duration) -> Admit { - let Some(gate) = self.gate.as_ref().filter(|gate| !gate.is_closed()) else { - return Admit::Now; - }; + /// Take the oldest pending packet or boundary. + fn pop(&mut self) -> Option { + let (read, arrived) = self.pending.pop_front()?; + match read { + Encoded::Frame { .. } => self.arrived = Some(arrived), + _ => self.breaks -= 1, + } + Some(read) + } + + /// When the pending packet at `timestamp`, which arrived at `arrived`, may decode, or + /// `None` if it may now. + fn admit(&self, timestamp: Duration, arrived: kio::clock::Instant) -> Option { + let gate = self.gate.as_ref().filter(|gate| !gate.is_closed())?; let state = gate.read(); if state.hold.is_zero() || self.ended.is_some() { - return Admit::Now; + return None; } // Media queued behind the packet covers the hold, or nothing more continues it. - let mut queued = self.pending.iter().skip(1).rev(); - let covered = match queued.next() { - Some(Encoded::Frame { frame, .. }) => { - Duration::from(frame.timestamp) >= timestamp + state.hold - || queued.any(|read| !matches!(read, Encoded::Frame { .. })) - } - Some(_) => true, - None => false, - }; + let covered = self.breaks > 0 + || matches!(self.pending.back(), Some((Encoded::Frame { frame, .. }, _)) + if Duration::from(frame.timestamp) >= timestamp + state.hold); if covered { - return Admit::Now; + return None; } + // A packet waits at most the hold, which bounds what the gate keeps by arrival time, + // whatever the timestamps say. + let mut due = arrived.checked_add(state.hold); let boundary = self.boundary; - let Some((playhead, at)) = state + if let Some((playhead, at)) = state .playhead .filter(|(playhead, _)| boundary.is_none_or(|boundary| *playhead > boundary)) - else { - return Admit::Wait; - }; - // Due once the playhead is within the floor of where this packet's output ends. - let end = timestamp + self.frame; - let reach = playhead + state.floor; - let due = match end.checked_sub(reach) { - Some(ahead) => at.checked_add(ahead), - // Before the clock's origin is long past. - None => match at.checked_sub(reach - end) { - Some(due) => Some(due), - None => return Admit::Now, - }, - }; - match due { - Some(due) if due > kio::clock::Instant::now() => Admit::At(due), - Some(_) => Admit::Now, - None => Admit::Wait, + { + // Due once the playhead is within the floor of where this packet's output ends. + let end = timestamp + self.frame; + let reach = playhead + state.floor; + let played = match end.checked_sub(reach) { + Some(ahead) => at.checked_add(ahead), + // Before the clock's origin is long past, so it decodes now. + None => Some(at.checked_sub(reach - end)?), + }; + due = due.into_iter().chain(played).min(); } + // Past the clock's range never comes, so decode now rather than hold it forever. + due.filter(|due| *due > kio::clock::Instant::now()) } /// Start a new timeline after a gap or past a declared end. The decoder is not reset: @@ -2094,8 +2102,9 @@ mod tests { } /// Backpressure holds packets encoded while still reading and timing every one: - /// stalled, a packet decodes once the media behind it covers the hold, and with - /// a playhead, once that is within the floor of where the packet ends. + /// stalled, a packet decodes once the media behind it covers the hold, with a + /// playhead, once that is within the floor of where the packet ends, and either + /// way once it has waited the hold, whatever its timestamp says. #[tokio::test(start_paused = true)] async fn backpressure_holds_packets_encoded_until_the_ring_has_room() { const PACKET: Duration = Duration::from_millis(20); @@ -2123,7 +2132,7 @@ mod tests { .await .unwrap(); let backpressure = consumer.backpressure(); - backpressure.set_latency(Duration::from_millis(100), Duration::from_millis(200)); + backpressure.set_latency(Duration::from_millis(100), Duration::from_secs(2)); let payload: Bytes = vec![0u8; 960 * size_of::()].into(); let mut write = |index: u32| { @@ -2149,7 +2158,7 @@ mod tests { } assert_eq!(read(&mut consumer, Duration::from_secs(1)).await, None); assert_eq!(consumer.pending.len(), 10, "every packet is read and held encoded"); - write(10); + write(100); assert_eq!(read(&mut consumer, Duration::ZERO).await, Some(Duration::ZERO)); // The first packet ends sounding in half a second, so the next one, ending @@ -2161,13 +2170,19 @@ mod tests { assert_eq!(read(&mut consumer, Duration::from_millis(419)).await, None); assert_eq!(read(&mut consumer, Duration::from_millis(1)).await, Some(PACKET)); + // With the playhead far off, the next packet waits out the hold from its arrival. + let far = kio::clock::Instant::now().checked_add(Duration::from_secs(10)).unwrap(); + backpressure.observe(Timestamp::try_from(PACKET * 2).unwrap(), far); + assert_eq!(read(&mut consumer, Duration::from_millis(579)).await, None); + assert_eq!(read(&mut consumer, Duration::from_millis(1)).await, Some(PACKET * 2)); + // The track's end leaves nothing to wait for. producer.finish().unwrap(); let mut rest = 0; while next_frame(&mut consumer).await.unwrap().is_some() { rest += 1; } - assert_eq!(rest, 9); + assert_eq!(rest, 8); } /// A packet that doesn't continue the last one is a hole, not a splice: the diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index ad983517f45..9cf6c46fbd3 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -760,10 +760,12 @@ async fn play_audio( } _ => None, }; - received - .lock() - .unwrap() - .push_back((event, delay, crossing, Instant::now())); + received.lock().unwrap().push_back(( + event, + delay, + crossing, + consumer.arrival().unwrap_or_else(Instant::now), + )); arrived.notify_one(); if ended { return Ok::<_, anyhow::Error>(()); @@ -1188,8 +1190,8 @@ mod tests { ); } - /// How long after it arrives the first packet sounds, holding `min` as the - /// jitter target within a `max` budget, plus a catalog and a video delay. + /// How long after it arrives the first packet of a live track sounds, holding + /// `min` as the jitter target within a `max` budget, plus a catalog and a video delay. async fn first_sound(min: Duration, max: Duration, catalog: Option, video: Option) -> Duration { tokio::time::pause(); let broadcast = moq_net::broadcast::Info::new().produce(); @@ -1239,7 +1241,12 @@ mod tests { .await; }); let start = Instant::now(); - for index in 0..5 { + // Live, then a burst the track's end releases at once. + for index in 0..60 { + producer.write(packet(index, 0.25)).unwrap(); + kio::clock::sleep(PACKET_DURATION).await; + } + for index in 60..65 { producer.write(packet(index, 0.25)).unwrap(); } producer.finish().unwrap(); From 697fdb6752ecf8b417a7447eaf82e239bd74e841 Mon Sep 17 00:00:00 2001 From: quest fixer Date: Tue, 10 Nov 2026 20:25:49 -0800 Subject: [PATCH 6/6] test(audio): retain finished track handle after base API change --- rs/moq-audio/src/decode/consumer.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index 4e65f0f7c0f..352c3aba245 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -2177,7 +2177,7 @@ mod tests { assert_eq!(read(&mut consumer, Duration::from_millis(1)).await, Some(PACKET * 2)); // The track's end leaves nothing to wait for. - producer.finish().unwrap(); + let _finished = producer.finish().unwrap(); let mut rest = 0; while next_frame(&mut consumer).await.unwrap().is_some() { rest += 1;