fix(audio): hold a latest join's whole budget from the start

quest/m0/test-flakes-2/README.md

@@ -48,6 +48,5 @@ Public API: none. Wire: none.
 
 ## Required
 
-- [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

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<moq_net::Timestamp>,
-	/// 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<u128>, usize) {
+	async fn warmup_output(declared: bool) -> (Vec<u128>, 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()
 			},
@@ -960,7 +947,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,
 			[
@@ -972,16 +959,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::<f32>()].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)]

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;

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";

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

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

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.

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();

test/interop/interop.sh

@@ -1030,8 +1030,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)" \