fix(net): keep a lite subscription's demand until its groups drain

quest/m1/README.md

@@ -46,7 +46,6 @@ ladders.
 - [Publishing never invents a timestamp](/quest/m1/publish-timestamp.md) - no Rust or binding publish API fills in a timestamp; an untimed payload goes out untimed
 - [Hard fork](/quest/m1/quic/fork/README.md) - quinn hard-forked in-tree as `moq-quic`, ranked ahead of perf; the rest of the QUIC line follows it
 - [Cluster routing](/quest/m1/cluster-routing/README.md) - any node routes toward a broadcast's origin over CDN and P2P links alike, with per-origin routes and path-less announces
-- [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop`
 - [Flat questlines](/quest/m1/quest-flat-lines.md) - moq pins the current quest CLI, lands its questline branches on main, and retires them
 - [main merges through a squash queue](/quest/m1/merge-queue-settings.md) - condition: the maintainer enables the squash merge queue on `main`
 - [Binding audio delay](/quest/m1/binding-surface.md) - moq-ffi and every wrapper configure and observe audio playout delay

quest/m1/js-pending-tail.md

@@ -3,22 +3,17 @@
 ## Goal
 
 A `@moq/net` reader of a received track waits for missing groups below a
-declared end until the tail settles, as Rust readers do once #4225 lands,
+declared end until the tail settles, as Rust readers do since #4225,
 instead of ending at the end with groups missing (`doc/lib/js/net.md`).
 
 ## Plan
 
-Mirror the Rust hold from [Track tail interop](/quest/m1/track-tail-interop.md)
-(#4225) once it merges, and test it against the Rust behavior with mocked
-time. Split from [JS session parity](/quest/m1/js-session-parity.md) on
+Mirror the Rust pending-tail hold from #4225 (`rs/moq-net/src/model/track.rs`)
+and test it against the Rust behavior with mocked time. Split from [JS session parity](/quest/m1/js-session-parity.md) on
 2026-10-08 so the caps work stays ready.
 
 Public API: none. Wire: none.
 
-## Required
-
-- [Track tail interop](/quest/m1/track-tail-interop.md) - the Rust hold this mirrors
-
 ## Related
 
 - [JS session parity](/quest/m1/js-session-parity.md) - per-session caps, the other half

quest/m1/subscribe-drop.md

@@ -81,14 +81,9 @@ its header; on each version the subscriber settles without waiting out the
 grace. A skipped sequence at the end is covered on lite-05 through lite-07
 (`track_tail::skipped_groups_end_without_the_grace`).
 
-Add the lite-07 drop case to the tail interop harness from
-[track tail interop](/quest/m1/track-tail-interop.md): Rust and JS
+Add the lite-07 drop case to the tail interop harness
+(`just test interop --tail`, `test/interop/clients/*/tail.*`): Rust and JS
 subscribers both settle on SUBSCRIBE_DROP through the relay, so a group the
 publisher skipped or never opened ends the track without waiting out the
 grace. Decided in the 2026-09-30 audit: the case moved here so the basic
 tail interop could land first.
-
-
-## Related
-
-- [Track tail interop](/quest/m1/track-tail-interop.md) - the Rust-JS tail harness the lite-07 drop case extends

rs/moq-cli/Cargo.toml

@@ -154,3 +154,7 @@ vt100 = "0.16.2"
 [[bench]]
 name = "play_buffer"
 harness = false
+
+[[example]]
+name = "interop-tail"
+path = "../../test/interop/clients/rust/tail.rs"

rs/moq-net/src/ietf/subscriber.rs

@@ -2371,6 +2371,8 @@ where
 										fill.end();
 									}
 								}
+								// The tail settled, so readers may end at an END_OF_TRACK's boundary.
+								track.set_tail_pending(false);
 								// A no-op once an END_OF_TRACK declared the end.
 								let _ = track.finish();
 							}
@@ -3082,7 +3084,8 @@ fn end_track(track: &mut track::Producer, end: u64) -> Result<(), Error> {
 	if track.final_sequence().is_some() {
 		return Ok(());
 	}
-	if let Err(err) = track.finish_at(end) {
+	// Lower groups may still be on the wire, behind the one that carried the end.
+	if let Err(err) = track.finish_at_pending(end) {
 		tracing::warn!(%err, end, "invalid END_OF_TRACK");
 		let _ = track.clone().abort(Error::ProtocolViolation);
 		return Err(Error::ProtocolViolation);

rs/moq-net/src/lite/publisher.rs

@@ -1827,7 +1827,13 @@ enum SubscribeServe<S: crate::transport::poll::Session> {
 	/// is moved on every transition.
 	Run(Box<TrackRun<S>>),
 	/// The track finished: draining the in-flight group streams before the FIN.
-	Drain { children: kio::Tasks<GroupServe<S>> },
+	Drain {
+		children: kio::Tasks<GroupServe<S>>,
+		/// The subscription's demand lasts until its groups drain. A relay cancels its
+		/// upstream subscription once nobody subscribes, and the publisher then resets
+		/// every group still on the wire. `None` only in passing between states.
+		_track: Option<Box<track::Subscriber>>,
+	},
 }
 
 impl<S: crate::transport::poll::Session> Request<S> for SubscribeServe<S> {
@@ -1893,6 +1899,7 @@ impl<S: crate::transport::poll::Session> Request<S> for SubscribeServe<S> {
 						self,
 						Self::Drain {
 							children: Default::default(),
+							_track: None,
 						},
 					)
 					else {
@@ -1958,10 +1965,22 @@ impl<S: crate::transport::poll::Session> Request<S> for SubscribeServe<S> {
 					if ready!(run.poll_step(writer, waiter))?.is_continue() {
 						return Poll::Ready(Ok(ControlFlow::Continue(())));
 					}
-					let children = std::mem::take(&mut run.children);
-					*self = Self::Drain { children };
+					let Self::Run(run) = std::mem::replace(
+						self,
+						Self::Drain {
+							children: Default::default(),
+							_track: None,
+						},
+					) else {
+						unreachable!()
+					};
+					let TrackRun { children, track, .. } = *run;
+					*self = Self::Drain {
+						children,
+						_track: Some(Box::new(track)),
+					};
 				}
-				Self::Drain { children } => {
+				Self::Drain { children, .. } => {
 					ready!(children.poll(waiter));
 					return Poll::Ready(Ok(ControlFlow::Break(())));
 				}

rs/moq-net/src/lite/subscriber.rs

@@ -4515,6 +4515,7 @@ impl<S: crate::transport::poll::Session> kio::Task for TrackServeRun<S> {
 							}
 						},
 						ServeEnd::Finished => {
+							serve_loop.serving.set_tail_pending(false);
 							let _ = serve_loop.serving.finish();
 						}
 						ServeEnd::GiveBack(err) => {
@@ -4870,13 +4871,15 @@ impl<S: crate::transport::poll::Session> ServeLoop<S> {
 									// consumers learn the boundary early; the later stream FIN
 									// then finds the track already finished.
 									lite::SubscribeResponse::End(end) => {
-										// finish_at rejects a boundary at or below a group
+										// Lower groups may still be on the wire, behind a higher
+										// one, so the end holds readers at a hole until the tail
+										// settles. finish_at rejects a boundary at or below a group
 										// already received. lite-05 specified an inclusive end,
 										// and `@moq/net` 0.1.3 to 0.1.9 sent one, so there it
 										// only costs the early boundary: warn, and let the FIN
 										// finish the track. Later drafts made it exclusive, so
 										// the publisher contradicted its own end.
-										if let Err(err) = self.serving.finish_at(end.group) {
+										if let Err(err) = self.serving.finish_at_pending(end.group) {
 											match serve.subscriber.version {
 												Version::Lite05 => {
 													tracing::warn!(track = %serve.name, group = end.group, %err, "invalid subscribe end")