From 3c01c00494f222cef49362e40782b18890d70850 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 18:49:11 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m0/test-flakes-2/ffi-audio-gap.md Quest-Claim: ff98f3c2265d96c7 From 4a412038a032066b2e739d2c815f8e7f47ab48c6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 20:26:29 -0700 Subject: [PATCH 2/2] fix(audio): hold a latest join's whole budget from the start A `Start::Latest` join with a catalog warmup subscribed with the warmup alone as its max delay and widened to `max_delay + warmup` only once it read its first packet. The transport enforces that budget too, so a reader starved between subscribing and its first read had the groups queued behind it reset as `Old`, and its first frames were followed by a gap. moq-c decodes audio this way, which is why only the `-> c` interop cells failed the tone check. Subscribe with the whole budget, move the local cursor to the live edge as a join without warmup already did, and prime the decoder by trimming the warmup from that group, as every other join does. The gap-checked interop readers also give themselves the 5 s budget the native JS audio reader has, since a reader starved past 1 s skips on its own. Co-Authored-By: Claude Opus 5.5 --- quest/m0/test-flakes-2/README.md | 1 - quest/m0/test-flakes-2/ffi-audio-gap.md | 15 ---- rs/moq-audio/src/decode/consumer.rs | 97 ++++++++++++++------- test/interop/clients/c/subscribe.c | 11 ++- test/interop/clients/cpp/main.cpp | 7 +- test/interop/clients/go/main.go | 7 +- test/interop/clients/js-native/subscribe.ts | 8 +- test/interop/clients/python/interop.py | 6 +- test/interop/clients/rust/audio.rs | 7 +- test/interop/interop.sh | 4 +- 10 files changed, 102 insertions(+), 61 deletions(-) delete mode 100644 quest/m0/test-flakes-2/ffi-audio-gap.md diff --git a/quest/m0/test-flakes-2/README.md b/quest/m0/test-flakes-2/README.md index 9c567438d01..7fb9b90e2ad 100644 --- a/quest/m0/test-flakes-2/README.md +++ b/quest/m0/test-flakes-2/README.md @@ -49,7 +49,6 @@ Public API: none. Wire: none. ## Required - [moq-cli relay tests under load](/quest/m0/test-flakes-2/cli-relay-load.md) - the fetch and announced tests wait on events, not timing -- [FFI publisher audio gaps](/quest/m0/test-flakes-2/ffi-audio-gap.md) - interop tone checks stop finding gaps from FFI publishers - [IETF22 fallback](/quest/m0/test-flakes-2/ietf22-fallback.md) - an IETF22 browser viewer follows a fallback within the media lane's bound, keeping its session and its audio - [Swapped Rust to JS Opus](/quest/m0/test-flakes-2/swapped-js-opus.md) - the `rust-swapped -> js` interop cell stops timing out under load - [JS host-task yields follow the clock](/quest/m0/test-flakes-2/js-host-yields.md) - `Budget` and `nextTask` yield through the injected clock, so `Manual` tests stop racing them diff --git a/quest/m0/test-flakes-2/ffi-audio-gap.md b/quest/m0/test-flakes-2/ffi-audio-gap.md deleted file mode 100644 index dc19831cc6f..00000000000 --- a/quest/m0/test-flakes-2/ffi-audio-gap.md +++ /dev/null @@ -1,15 +0,0 @@ -# [S] FFI publisher audio gaps in interop - -## Goal - -The interop tone check (`test/interop/tone.py`, "timestamp gap") stops -failing for FFI publishers: `python -> c`, `go -> c`, and the `cpp ->` -cells seen in [moq-dev/moq#6464](https://github.com/moq-dev/moq/pull/6464) and -[moq-dev/moq#6648](https://github.com/moq-dev/moq/pull/6648). - -## Plan - -- [moq-dev/moq#6658](https://github.com/moq-dev/moq/pull/6658)'s max-age cause - does not apply: FFI publishers keep hang's 30 s max-age. -- Reproduce under load and find which group or frame jumps, then fix the - cause at its source. diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index 706fc7c0c83..3d6df82b970 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -51,8 +51,8 @@ pub enum Start { /// Decode every cached group. #[default] Oldest, - /// Start at the newest group, including earlier groups for decoder warmup. Opening - /// waits for the track's live edge, or its end. + /// Start at the newest group, skipping whatever is already cached. Opening waits for + /// the track's live edge, or its end. Latest, } @@ -114,8 +114,6 @@ pub struct Consumer { warmup: Duration, warmup_pending: bool, warmup_until: Option, - /// Initial subscription reaches only the warmup window before restoring the skip budget. - joining_latest: bool, resolved_sample_rate: u32, resolved_layout: Layout, /// Where the next packet's timestamp should land: the last packet's timestamp @@ -176,27 +174,23 @@ impl Consumer { let name = name.into(); let warmup = catalog.warmup.unwrap_or_default(); - let joining_latest = options.start == Start::Latest && !warmup.is_zero(); - let budget = if joining_latest { - warmup - } else { - options.max_delay.saturating_add(warmup) - }; let track = broadcast.track(hang::catalog::Reference::track(catalog, &name))?; + // The whole budget from the start: the transport enforces it too, so a narrower one + // while joining would drop the groups queued behind a reader slow to take its first. let mut subscriber = track .subscribe( moq_net::track::Subscription::default() .with_priority(hang::catalog::PRIORITY.audio) - .with_max_delay(budget), + .with_max_delay(options.max_delay.saturating_add(warmup)), ) .await?; - // Latest skips cached history unless decoder warmup needs it. Move the - // local cursor, not Subscription::ranges: the latter is aggregated - // across subscribers and can ask the publisher to rewind for everyone. The live edge - // is known only once the feed is live, so wait for it rather than read the backlog - // of a warm copy still waiting on its source; the wait ends with the track. + // Latest skips cached history; a warmup is primed by trimming the newest group's + // start, like any other join. Move the local cursor, not Subscription::ranges: the + // latter is aggregated across subscribers and can ask the publisher to rewind for + // everyone. The live edge is known only once the feed is live, so wait for it rather + // than read the backlog of a warm copy still waiting on its source; the wait ends + // with the track. if options.start == Start::Latest - && warmup.is_zero() && let Some(live_edge) = subscriber.largest().await { subscriber = subscriber.with_range(live_edge.group..); @@ -230,7 +224,6 @@ impl Consumer { warmup, warmup_pending: !warmup.is_zero(), warmup_until: None, - joining_latest, resolved_sample_rate: sample_rate, resolved_layout: layout, next_start: None, @@ -337,11 +330,6 @@ impl Consumer { // 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; - if self.joining_latest { - self.track - .set_max_delay(self.options.max_delay.saturating_add(self.warmup)); - self.joining_latest = false; - } let media = timestamp.value() as f64 * 1000.0 / timestamp.scale().as_u64() as f64; self.target.observe(arrival, media); @@ -797,7 +785,7 @@ mod tests { } } - async fn warmup_output(start: Start, declared: bool) -> (Vec, usize) { + async fn warmup_output(declared: bool) -> (Vec, usize) { let broadcast = moq_net::broadcast::Info::new().produce(); let track = broadcast .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) @@ -830,7 +818,6 @@ mod tests { &catalog, "audio", Options { - start, max_delay: Duration::from_secs(10), ..Options::new() }, @@ -961,7 +948,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn warmup_trims_the_join_without_trimming_groups_or_declared_returns() { - let (stamps, frames) = warmup_output(Start::Oldest, true).await; + let (stamps, frames) = warmup_output(true).await; assert_eq!( stamps, [ @@ -973,16 +960,64 @@ mod tests { #[tokio::test(start_paused = true)] async fn warmup_restarts_after_an_undeclared_gap() { - let (stamps, frames) = warmup_output(Start::Oldest, false).await; + let (stamps, frames) = warmup_output(false).await; assert_eq!(stamps, [80_000, 100_000, 1_080_000, 1_100_000]); assert_eq!(frames, 4 * 960); } + /// A latest join skips the cached backlog and primes the decoder on the live edge's own + /// group, as any other join does, while asking the publisher for its whole budget: the + /// transport enforces it too, so a narrower one would drop the groups queued behind a + /// reader slow to take its first. #[tokio::test(start_paused = true)] - async fn warmup_latest_joins_early_without_replaying_the_skip_budget() { - let (stamps, frames) = warmup_output(Start::Latest, true).await; - assert_eq!(stamps, [1_100_000]); - assert_eq!(frames, 960); + async fn warmup_latest_joins_at_the_live_edge_with_its_whole_budget() { + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let mut producer = moq_mux::container::Producer::new( + track.clone(), + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio), + ); + for (start, packets) in [(0, 5), (1_000_000, 6)] { + for packet in 0..packets { + producer + .write(moq_mux::container::Frame { + timestamp: Timestamp::from_micros(start + packet * 20_000).unwrap(), + duration: None, + payload: vec![0; 960 * size_of::()].into(), + keyframe: packet == 0, + }) + .unwrap(); + } + producer.cut(None).unwrap(); + } + producer.finish().unwrap(); + + let mut catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, 48_000, 1); + catalog.warmup = Some(Duration::from_millis(80)); + let mut consumer = Consumer::new( + &broadcast.consume(), + &catalog, + "audio", + Options { + start: Start::Latest, + max_delay: Duration::from_secs(10), + ..Options::new() + }, + ) + .await + .unwrap(); + assert_eq!( + track.subscription().unwrap().max_delay, + Some(Duration::from_millis(10_080)) + ); + + let mut stamps = Vec::new(); + while let Some(frame) = next_frame(&mut consumer).await.unwrap() { + stamps.push(frame.timestamp.as_micros()); + } + assert_eq!(stamps, [1_080_000, 1_100_000]); } #[tokio::test(start_paused = true)] diff --git a/test/interop/clients/c/subscribe.c b/test/interop/clients/c/subscribe.c index 0353ce0ddaf..8e552fe9a62 100644 --- a/test/interop/clients/c/subscribe.c +++ b/test/interop/clients/c/subscribe.c @@ -55,6 +55,13 @@ typedef struct { // The most audio renditions this client decodes; a catalog offering more fails it. #define MAX_AUDIO 4 +// How far a read may fall behind live before it skips, in microseconds. A gap fails the run, which +// a reader skipping behind live would cause itself: the matrix starts every subscriber at once, and +// a runtime starved by the others can fall a second behind. That is this reader's speed, not the +// delivery under test. A track's max-age caps it, so interop.sh's Rust publisher keeps its media at +// least this long. +#define MAX_DELAY_US 5000000 + struct ctx { pthread_mutex_t mu; pthread_cond_t cv; @@ -238,7 +245,7 @@ static void start_audio(ctx_t *c, int32_t catalog) { moq_audio_decoder_output out; memset(&out, 0, sizeof(out)); out.format = MOQ_AUDIO_SAMPLE_FORMAT_F32; - out.max_delay_us = 1000000; + out.max_delay_us = MAX_DELAY_US; int32_t track = moq_decode_audio((uint32_t)catalog, i, &out, on_audio, &c->audio[i]); pthread_mutex_lock(&c->mu); if (track > 0) c->audio[i].track = track; @@ -317,7 +324,7 @@ static void on_catalog(void *ud, int32_t catalog) { moq_video_config vcfg; memset(&vcfg, 0, sizeof(vcfg)); if (moq_consume_video_config((uint32_t)catalog, 0, &vcfg) == 0) { - int32_t track = moq_consume_video((uint32_t)catalog, 0, 1000000, on_frame, ud); + int32_t track = moq_consume_video((uint32_t)catalog, 0, MAX_DELAY_US, on_frame, ud); if (track > 0) { pthread_mutex_lock(&c->mu); c->video_track = track; diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index 1a7f5b50f78..62d91522bbe 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -49,8 +49,11 @@ using Clock = std::chrono::steady_clock; constexpr size_t READ_CHUNK = 64 * 1024; -// SubscribeMedia max delay: how much reordering the jitter buffer tolerates. -constexpr std::chrono::nanoseconds MAX_DELAY = std::chrono::seconds(1); +// How far a read may fall behind live before it skips. A gap fails the run, which a reader skipping +// behind live would cause itself: the matrix starts every subscriber at once, and a runtime starved +// by the others can fall a second behind. That is this reader's speed, not the delivery under test. +// A track's max-age caps it, so interop.sh's Rust publisher keeps its media at least this long. +constexpr std::chrono::nanoseconds MAX_DELAY = std::chrono::seconds(5); // Synthetic audio: a 48 kHz tone per channel at 0.5 amplitude, encoded as Opus. constexpr const char *AUDIO_TRACK = "tone"; diff --git a/test/interop/clients/go/main.go b/test/interop/clients/go/main.go index 196b3f2bdd1..73b707f7622 100644 --- a/test/interop/clients/go/main.go +++ b/test/interop/clients/go/main.go @@ -41,8 +41,11 @@ import ( const readChunk = 64 * 1024 -// media.TrackConsumer max delay: how much reordering the jitter buffer tolerates. -const maxDelay = time.Second +// How far a read may fall behind live before it skips. A gap fails the run, which a reader skipping +// behind live would cause itself: the matrix starts every subscriber at once, and a runtime starved +// by the others can fall a second behind. That is this reader's speed, not the delivery under test. +// A track's max-age caps it, so interop.sh's Rust publisher keeps its media at least this long. +const maxDelay = 5 * time.Second // finishLinger is how long a finished publisher stays up for the late readers. const finishLinger = 10 * time.Second diff --git a/test/interop/clients/js-native/subscribe.ts b/test/interop/clients/js-native/subscribe.ts index e578b11789b..69521bae9ca 100644 --- a/test/interop/clients/js-native/subscribe.ts +++ b/test/interop/clients/js-native/subscribe.ts @@ -39,10 +39,10 @@ import { readFinish } from "./finish"; // has to run before run() below. install(); -// How stale a group may get before it is skipped, matching the Go and Python -// subscribers. A relay drops a superseded group (RESET_STREAM Old) rather than -// finish sending it, e.g. a cached group a fresh one lands right behind, so a -// subscriber has to move on to the next group instead of failing. +// How stale a video group may get before it is skipped. A relay drops a +// superseded group (RESET_STREAM Old) rather than finish sending it, e.g. a +// cached group a fresh one lands right behind, so a subscriber has to move on +// to the next group instead of failing. const MAX_DELAY = Milli(1000); // Audio is checked for gaps, which a reader skipping behind live would cause itself: the matrix diff --git a/test/interop/clients/python/interop.py b/test/interop/clients/python/interop.py index 23a071c9815..f6c6aa1d943 100644 --- a/test/interop/clients/python/interop.py +++ b/test/interop/clients/python/interop.py @@ -32,7 +32,11 @@ import moq READ_CHUNK = 64 * 1024 -MAX_DELAY = timedelta(seconds=1) # container consumer congestion-control / lookahead window +# How far a read may fall behind live before it skips. A gap fails the run, which a reader skipping +# behind live would cause itself: the matrix starts every subscriber at once, and a runtime starved +# by the others can fall a second behind. That is this reader's speed, not the delivery under test. +# A track's max-age caps it, so interop.sh's Rust publisher keeps its media at least this long. +MAX_DELAY = timedelta(seconds=5) LINGER = timedelta(seconds=10) # how long a finished publisher stays up for the late readers # Synthetic audio: a 48 kHz tone per channel at 0.5 amplitude, encoded as Opus. diff --git a/test/interop/clients/rust/audio.rs b/test/interop/clients/rust/audio.rs index 73e86881dec..f7fdacc9053 100644 --- a/test/interop/clients/rust/audio.rs +++ b/test/interop/clients/rust/audio.rs @@ -89,7 +89,12 @@ async fn decode( let rendition = source.resolve(audio.broadcast.as_ref(), audio.epoch.as_ref()).await?; let mut options = moq_audio::decode::Options::new(); options.output.format = moq_audio::Format::F32; - options.max_delay = Duration::from_secs(1); + // How far the read may fall behind live before it skips. A gap fails the run, which a reader + // skipping behind live would cause itself: the matrix starts every subscriber at once, and a + // runtime starved by the others can fall a second behind. That is this reader's speed, not the + // delivery under test. A track's max-age caps it, so interop.sh's Rust publisher keeps its + // media at least this long. + options.max_delay = Duration::from_secs(5); let mut decoder = moq_audio::decode::Consumer::new(&rendition, audio, name, options).await?; let rate = decoder.sample_rate(); let channels = decoder.layout().channels(); diff --git a/test/interop/interop.sh b/test/interop/interop.sh index 7d0d8a41de2..c0a868eaa33 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -1016,8 +1016,8 @@ run_publisher() { local connect rust_connect "$url" if [[ "$AUDIO" -eq 1 ]]; then - # The max-age caps every reader's max delay, so it covers the longest, the native - # JS audio readers' 5 s. Each 1 s fragment lands as one burst, so with a 1 s + # The max-age caps every reader's max delay, so it covers the gap-checked + # readers' 5 s. Each 1 s fragment lands as one burst, so with a 1 s # max-age the oldest group a joining reader is served goes stale as soon as the # next group lands, cutting it mid-read and failing the gap check. ffmpeg_fmp4 "$tone" "$(publisher_codecs rust)" | RUST_LOG="$CLI_LOG" "$MOQ" "${connect[@]}" --connect-version "$(cell_version rust)" \