fix(transcode): finish the open live group when demand ends

quest/m1/README.md

@@ -103,7 +103,6 @@ lands before the cut.
 - [GOAWAY coverage](/quest/m1/goaway-coverage.md) - every IETF request stream handles GOAWAY, a closed control stream is a violation, and the draft-18 cutoff is honored
 - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust
 - [Decode errors end the reader](/quest/m1/decode-errors.md) - a container decode error fails the reader in Rust and JS instead of reading as loss
-- [Transcode finishes its group](/quest/m1/transcode-open-group.md) - `moq-transcode` never aborts a group a fetch's live loop opened
 - [Rust request credit](/quest/m1/rs-request-credit.md) - a moq-net request fails at once without stream credit instead of waiting
 - [qmux no-wait opens](/quest/m1/qmux-no-wait.md) - @moq/qmux rejects an over-limit create when waitUntilAvailable is false, like Chrome
 - [Random-loss audit](/quest/m1/random-loss-audit.md) - every decision random-loss gates drove is kept, reworked, or reverted on shaped paths

rs/moq-transcode/src/lib.rs

@@ -10,7 +10,8 @@
 //! - Subscribing to a rung attaches it to a shared live decode of the source
 //!   (one subscription and one decoder per source, no matter how many rungs
 //!   are active); each rung resizes and encodes its own copy, group for group,
-//!   stopping when the last subscriber leaves.
+//!   stopping at the end of the group in progress once the last subscriber
+//!   leaves.
 //! - Fetching a specific group fetches that same group from the source and
 //!   transcodes just that group, unless every caller leaves before the
 //!   request is accepted, in which case nothing is encoded. Output groups
@@ -1536,6 +1537,186 @@ mod tests {
 		transcoder.abort();
 	}
 
+	/// A fetch is demand, so it starts the live path, which claims the group the
+	/// source has open. A reader that picked that group up from the cache is not
+	/// demand, so when the fetch ends the live path is left mid-group with
+	/// nobody asking. It finishes the group before leaving its source: a group
+	/// name always means the same content, so the reader gets the whole group
+	/// rather than one aborted under it.
+	///
+	/// Paused so the runtime settles before the source group ends: the live path
+	/// has seen the demand go away by then, which is the moment that used to
+	/// abort the group.
+	#[tokio::test(start_paused = true)]
+	async fn ended_demand_finishes_the_open_live_group() {
+		let source = source_catalog(320, 240);
+		let source_demand = source._track.demand();
+		// Parks the rung's own fetch of the group below, so that fetch is demand
+		// for as long as the caller holds it and nothing else serves it.
+		let _source_fetches = source._track.dynamic();
+
+		// Two frames, so the output's largest moves once both are encoded.
+		let mut group = source._track.create_group(0u64.into()).unwrap();
+		let mut encoder = moq_video::encode::Encoder::new(&{
+			let mut config = moq_video::encode::Config::new(320, 240, moq_video::Rate::new(30, 1).unwrap());
+			config.kind = moq_video::encode::Kind::Software;
+			config
+		})
+		.unwrap();
+		encoder.cut().unwrap();
+		let gray = vec![0x80u8; 320 * 240 * 4];
+		for timestamp in [0, 33_333] {
+			for encoded in encoder.encode(&gray_frame(&gray, timestamp)).unwrap() {
+				hang::container::Frame {
+					timestamp: encoded.timestamp,
+					payload: encoded.payload,
+				}
+				.write_to(&mut group)
+				.unwrap();
+			}
+		}
+
+		let config = Config {
+			ladder: Ladder::new([Rung::new(120, moq_net::bandwidth::Rate::from_bps(100_000))]).unwrap(),
+			encoder: moq_video::encode::Kind::Software,
+			decoder: moq_video::decode::Kind::Software,
+			source: None,
+			..Default::default()
+		};
+		let output = moq_net::broadcast::Info::default().produce();
+		let consumer = output.consume();
+		let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config));
+
+		let catalog = loop {
+			match consumer.track(hang::Catalog::DEFAULT_NAME) {
+				Ok(track) => break track,
+				Err(moq_net::Error::NotFound) => tokio::task::yield_now().await,
+				Err(err) => panic!("catalog track: {err}"),
+			}
+		};
+		let mut catalogs = moq_mux::catalog::hang::Consumer::<()>::new(catalog.subscribe(None).await.unwrap());
+		await_catalog(&mut catalogs, |snapshot| {
+			snapshot.video.renditions.contains_key("video/120p")
+		})
+		.await;
+
+		// The only demand on the rung. Sequence 7 is one the live path never
+		// reaches, so it parks in the source handler while the live path claims
+		// output group 0 and encodes both frames of the open source group.
+		let rung = consumer.track("video/120p").unwrap();
+		let fetching = rung.fetch_group(7, None);
+		let mut edge = rung.clone();
+		while edge.largest().await.map(|position| (position.group, position.frame)) != Some((0, 1)) {}
+
+		// Read straight from the cache while the group is still being written.
+		let mut fetched = rung.fetch_group(0, None).await.unwrap();
+
+		// The only demand ends with output group 0 still open.
+		drop(fetching);
+		tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+		assert!(source_demand.is_used(), "the live path left its source mid-group");
+
+		// Once the source group ends, the live path finishes its copy and leaves.
+		group.finish().unwrap();
+		source_demand.unused().await.unwrap();
+
+		let mut frames = 0;
+		while fetched.read_frame().await.expect("group 0 was aborted").is_some() {
+			frames += 1;
+		}
+		assert_eq!(frames, 2, "group 0 lost frames");
+
+		transcoder.abort();
+	}
+
+	/// Demand that ends between groups never opens another one. Here the next
+	/// source group is already queued for the live path when the last fetch
+	/// ends, so the live path sees both at once. It leaves its source rather
+	/// than opening that group and riding it out with nobody asking.
+	///
+	/// The live path picks between the two at random, so without the fix this
+	/// fails about half the rounds, so a run of several fails almost every time.
+	#[tokio::test(start_paused = true)]
+	async fn ended_demand_opens_no_new_live_group() {
+		for _ in 0..8 {
+			let source = source_catalog(320, 240);
+			let source_demand = source._track.demand();
+			// Parks the rung's own fetch of the group below, so that fetch is
+			// demand for as long as the caller holds it and nothing else serves it.
+			let _source_fetches = source._track.dynamic();
+
+			let mut encoder = moq_video::encode::Encoder::new(&{
+				let mut config = moq_video::encode::Config::new(320, 240, moq_video::Rate::new(30, 1).unwrap());
+				config.kind = moq_video::encode::Kind::Software;
+				config
+			})
+			.unwrap();
+			let gray = vec![0x80u8; 320 * 240 * 4];
+			let mut write = |group: &mut moq_net::group::Producer, timestamps: &[u64]| {
+				encoder.cut().unwrap();
+				for &timestamp in timestamps {
+					for encoded in encoder.encode(&gray_frame(&gray, timestamp)).unwrap() {
+						hang::container::Frame {
+							timestamp: encoded.timestamp,
+							payload: encoded.payload,
+						}
+						.write_to(group)
+						.unwrap();
+					}
+				}
+			};
+			let mut group = source._track.create_group(0u64.into()).unwrap();
+			write(&mut group, &[0, 33_333]);
+
+			let config = Config {
+				ladder: Ladder::new([Rung::new(120, moq_net::bandwidth::Rate::from_bps(100_000))]).unwrap(),
+				encoder: moq_video::encode::Kind::Software,
+				decoder: moq_video::decode::Kind::Software,
+				source: None,
+				..Default::default()
+			};
+			let output = moq_net::broadcast::Info::default().produce();
+			let consumer = output.consume();
+			let transcoder = tokio::spawn(run(source.broadcast.consume(), output, config));
+
+			let catalog = loop {
+				match consumer.track(hang::Catalog::DEFAULT_NAME) {
+					Ok(track) => break track,
+					Err(moq_net::Error::NotFound) => tokio::task::yield_now().await,
+					Err(err) => panic!("catalog track: {err}"),
+				}
+			};
+			let mut catalogs = moq_mux::catalog::hang::Consumer::<()>::new(catalog.subscribe(None).await.unwrap());
+			await_catalog(&mut catalogs, |snapshot| {
+				snapshot.video.renditions.contains_key("video/120p")
+			})
+			.await;
+
+			// The only demand on the rung, parked on a sequence the live path never
+			// reaches. Held until the live path has finished group 0.
+			let rung = consumer.track("video/120p").unwrap();
+			let fetching = rung.fetch_group(7, None);
+			let mut edge = rung.clone();
+			while edge.largest().await.map(|position| (position.group, position.frame)) != Some((0, 1)) {}
+			let mut fetched = rung.fetch_group(0, None).await.unwrap();
+			group.finish().unwrap();
+			while fetched.read_frame().await.expect("group 0 was aborted").is_some() {}
+
+			// Queue the next source group first, so it is waiting for the live
+			// path by the time it sees the demand go.
+			let mut next = source._track.create_group(1u64.into()).unwrap();
+			write(&mut next, &[66_666]);
+			drop(fetching);
+
+			tokio::time::timeout(std::time::Duration::from_secs(10), source_demand.unused())
+				.await
+				.expect("the live path opened a group nobody asked for")
+				.unwrap();
+
+			transcoder.abort();
+		}
+	}
+
 	/// Two transcoders fed the same source publish groups that mirror the
 	/// source's sequences and timestamps rather than anything numbered per
 	/// instance.
@@ -1578,9 +1759,7 @@ mod tests {
 			})
 			.await;
 
-			// A held query keeps the live path encoding across both fetches.
 			let rung = consumer.track("video/120p").unwrap();
-			let _query = rung.query();
 			let mut groups = Vec::new();
 			for sequence in 0..2 {
 				let mut fetched = rung.fetch_group(sequence, None).await.unwrap();

rs/moq-transcode/src/rung.rs

@@ -3,9 +3,11 @@
 //! Nothing is encoded until someone asks, via the two demand paths moq-net
 //! exposes on the output track:
 //!
-//! - A live subscription (`used`) starts a live loop that subscribes to the
-//!   source track (mirroring the aggregate subscription) and transcodes group
-//!   for group until the track goes `unused` again.
+//! - Any demand (`used`), including a fetch, starts a live loop that subscribes
+//!   to the source track (mirroring the aggregate subscription) and transcodes
+//!   group for group until the track goes `unused` again. The group open at
+//!   that point is finished first, so a later reader of that sequence gets the
+//!   whole group rather than an aborted one.
 //! - A fetch of a specific group (`requested_group`) fetches that same group
 //!   from the source and transcodes just that group with a fresh encoder. A
 //!   fetch that starts mid-group is refused: a fresh encode's frames are only
@@ -208,20 +210,40 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<En
 
 		// One listener + encoder per demand session: rate control persists
 		// across groups, while every group still opens with a forced IDR.
-		// Dropping them on unused releases the shared decode (if last) and the
-		// encoder session until someone subscribes again.
+		// Dropping them once demand is gone releases the shared decode (if last)
+		// and the encoder session until someone asks again.
 		let mut listener = rung.feed.listen();
 		let mut encoder = Encoder::default();
 
 		// The output group currently being written, if the feed is mid-group.
 		let mut current: Option<moq_net::group::Producer> = None;
+		// Set while demand is gone but a group is still open. A group name always
+		// means the same content, so the group is finished for whoever reads that
+		// sequence later instead of aborted, and the session ends at its boundary.
+		let mut leaving = false;
 
 		'session: loop {
 			let item = tokio::select! {
 				item = listener.recv() => item,
-				_ = demand.unused() => {
+				unused = demand.unused(), if !leaving => {
+					if unused.is_ok() && current.is_some() {
+						leaving = true;
+						continue;
+					}
+					// Between groups, or the track closed: nothing to finish.
+					if let Some(output) = current.take() {
+						output.abort(moq_net::Error::Cancel)?;
+					}
+					break 'session;
+				}
+				used = demand.used(), if leaving => {
+					if used.is_ok() {
+						// Back before the group ended: carry on as if never gone.
+						leaving = false;
+						continue;
+					}
+					// The track closed under the group.
 					if let Some(output) = current.take() {
-						// Signal downstream that the group is incomplete.
 						output.abort(moq_net::Error::Cancel)?;
 					}
 					break 'session;
@@ -237,9 +259,15 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<En
 				}
 			};
 
+			// Demand can change without the `unused` or `used` branch seeing it,
+			// when the select picked this item instead. A returning viewer keeps
+			// the session rather than missing the next group while a new one
+			// attaches, and a departed one never opens another group.
+			leaving = !demand.is_used();
+
 			match item {
 				Some(Item::Group(sequence)) => {
-					if retiring {
+					if retiring || leaving {
 						// The group we were riding out never ended, so it is
 						// incomplete however long we wait.
 						if let Some(output) = current.take() {
@@ -262,6 +290,11 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<En
 					// that cannot cut fails the rung here: its groups could never
 					// mirror the source's, which is what this rung promises.
 					encoder.cut().await?;
+					// Demand can also leave while the codec works above. Nothing is
+					// open yet, so there is nothing to finish.
+					if !demand.is_used() {
+						break 'session;
+					}
 					// Mirror the source sequence so fetches and rendition
 					// switches map 1:1.
 					let info = moq_net::group::Info { sequence };
@@ -282,8 +315,14 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<En
 				}
 				Some(Item::Frame(frame)) => {
 					// No open group: attached mid-group, skipped a duplicate, or
-					// recovering from a lag. Wait for the next boundary.
-					let Some(output) = &mut current else { continue };
+					// recovering from a lag. Wait for the next boundary, unless
+					// nobody is left to serve it.
+					let Some(output) = &mut current else {
+						if leaving {
+							break 'session;
+						}
+						continue;
+					};
 
 					// The feed decodes at the source's native size; size this rung's copy
 					// here. Supported GPU paths feed the encoder without touching the
@@ -302,7 +341,9 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<En
 						write(&mut output, &active, encoder.flush().await?)?;
 						output.finish()?;
 					}
-					if retiring {
+					// Or during the flush.
+					leaving = !demand.is_used();
+					if retiring || leaving {
 						break 'session;
 					}
 				}
@@ -312,7 +353,7 @@ async fn live(rung: &Rung, producer: &mut moq_net::track::Producer) -> Result<En
 					if let Some(output) = current.take() {
 						output.abort(moq_net::Error::Cancel)?;
 					}
-					if retiring {
+					if retiring || leaving {
 						break 'session;
 					}
 				}