fix(ietf): cancel an unclaimed fill head when a subscription ends

quest/m1/README.md

@@ -161,7 +161,6 @@ per-frame CPU growth since #5926.
 - [A moq-transport request token is one auth per distinct token](/quest/m1/ietf-token-auth.md) - requests naming the same token share one verification and grant, on any request type, ending with the last such request
 - [moq-quic's API improves on quinn's](/quest/m1/quic-api/README.md) - moq-quic and both QUIC drivers ship one mirrored API: `send::`/`recv::` namespaces, one error per direction, consuming terminal operations, plain configs, no quinn baggage
 - [On-demand JavaScript FETCH](/quest/m1/js-fetch-on-demand.md) - browser publishers answer IETF FETCH for evicted or uncached groups through the JS ranges request surface
-- [Rust cancels an unclaimed fill head when a subscription ends](/quest/m1/ietf-fill-end-cancel.md) - SUBSCRIPTION_ENDED cancels, not finishes, a fill no group was sent for, as in JS
 - [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume
 - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the playout target is default in both languages; a manual browser proof and a native trace replay remain
 - [A/V clock](/quest/m1/av-clock.md) - the audio playhead drives Sync.reference while audio plays, through per-track sync handles

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

@@ -276,7 +276,8 @@ enum Fill {
 	/// The tail is what ends the group, and a publisher serving the subscription's range
 	/// opens a stream for it even when the group ended at the join point, since that empty
 	/// stream is how the group ends. One that opens none instead leaves this head unfinished
-	/// until the publisher ends the subscription, which is what publishes it.
+	/// until the publisher ends the subscription: TRACK_ENDED publishes it, and any other
+	/// end cancels it.
 	///
 	/// Nothing shorter is safe to infer. A later group arriving looks like proof that no
 	/// tail is coming, but streams are independent: the tail's own can still be behind it.
@@ -298,7 +299,7 @@ enum Fill {
 	/// the join to the next group boundary.
 	Failed { sequence: u64 },
 
-	/// The publisher ended the subscription cleanly, so no tail is coming. Unlike
+	/// The publisher ended the track, so no tail is coming. Unlike
 	/// [`Fill::Done`], a head still being written is the whole group once it lands: a
 	/// draft-18 joining FETCH is not in PUBLISH_DONE's Stream Count, so the settle can
 	/// end before it does.
@@ -373,7 +374,7 @@ impl Fill {
 	/// Release a head nothing claimed, publishing the objects it did carry.
 	///
 	/// The tail is what normally ends the group, so this is the fallback for when none is
-	/// coming: the publisher ended the subscription, or the tail that arrived could not be
+	/// coming: the publisher ended the track, or the tail that arrived could not be
 	/// stitched. Finishing rather than aborting, because the head is a valid prefix of the
 	/// group: it starts at the group's first object and has no holes.
 	fn release(&mut self) {
@@ -382,8 +383,11 @@ impl Fill {
 		}
 	}
 
-	/// The publisher ended the subscription cleanly: release the head, and any still being
-	/// written when it lands.
+	/// The publisher ended the track: release the head, and any still being written when it
+	/// lands.
+	///
+	/// Only TRACK_ENDED, since any other end, such as SUBSCRIPTION_ENDED, leaves the group
+	/// open upstream and so [`Self::abort`]s the head instead.
 	///
 	/// [`Fill::Failed`] stays: a joining FETCH is not in PUBLISH_DONE's Stream Count, so its
 	/// head can land after the end, and the tail it owed may already have been dropped.
@@ -3865,10 +3869,15 @@ impl Subscriber {
 										poll_settled(&mut settle, waiter, &fill, count)
 									})
 									.await;
-									// The publisher ended without a tail for the head, so the head
-									// is the whole group, even one that lands after this.
+									// No tail is coming for the head. Only TRACK_ENDED makes it the
+									// whole group, even one that lands after this. Any other end
+									// leaves the group open upstream, so the head is cut short
+									// rather than passed off as the whole group.
 									if let Ok(mut fill) = fill.write() {
-										fill.end();
+										match &end {
+											Ok(()) => fill.end(),
+											Err(err) => fill.abort(err.clone()),
+										}
 									}
 								}
 								// Only TRACK_ENDED ends the track. Any other status ends just this
@@ -15694,6 +15703,88 @@ mod joining_fetch_tests {
 		}
 	}
 
+	/// A head nothing claimed when the publisher ends the subscription is the whole group only
+	/// on TRACK_ENDED. SUBSCRIPTION_ENDED leaves the group open upstream, so finishing the head
+	/// would pass a prefix off as the whole group: it is cancelled instead.
+	#[kio::test]
+	async fn an_unclaimed_head_finishes_only_when_the_track_ends() {
+		let largest = Some(ietf::Location { group: 4, object: 0 });
+		for version in JOINING_DRAFTS {
+			for (status_code, whole) in [(ietf::PublishDoneStatus::TrackEnded.code(), true), (0x3, false)] {
+				let mut ok = subscribe_ok(version, largest);
+				ok.extend(message_bytes(
+					ietf::PublishDone::ID,
+					&ietf::PublishDone {
+						status_code,
+						stream_count: 0,
+						reason_phrase: "done".into(),
+					},
+					version,
+				));
+				let mut run = JoinRun::start(version, None, ok, fetch_ok(version)).await;
+				let fetch_id = *run
+					.subscriber
+					.state
+					.lock()
+					.fetches
+					.keys()
+					.next()
+					.expect("joining FETCH");
+
+				let mut objects = Vec::new();
+				let w = &mut Encoder::new(&mut objects, version.into());
+				w.varint(ietf::FetchHeader::TYPE).unwrap();
+				ietf::FetchHeader { request_id: fetch_id }.encode(w, version).unwrap();
+				ietf::FetchObject::Object {
+					subgroup: ietf::FetchSubgroup::Zero,
+					group: Some(4),
+					object: Some(0),
+					priority: Some(0),
+					properties: None,
+				}
+				.encode(w, version)
+				.unwrap();
+				w.bytes(b"x").unwrap();
+				let data = ScriptedSession::new(objects);
+				data.close(crate::lite::test_transport::Close::Fin);
+				let (_, recv) = data.clone().open_bi(Default::default()).await.unwrap();
+				let mut reader = Reader::new(crate::transport::boxed::RecvStream::new(recv), version);
+
+				// The whole head lands before the grace runs out, and no tail ever claims it.
+				run.subscriber.clone().recv_fill(&mut reader).await.expect("fill");
+				assert!(
+					matches!(
+						*run.subscriber
+							.state
+							.lock()
+							.subscribes
+							.values()
+							.next()
+							.unwrap()
+							.fill
+							.read(),
+						Fill::Ready { .. }
+					),
+					"{version}/{status_code}: the head waits for its tail"
+				);
+
+				let mut subscriber = (&mut run._hold.2).await.expect("subscribed");
+				let mut group = subscriber.recv_group().await.unwrap().unwrap();
+				assert_eq!(group.sequence, 4, "{version}/{status_code}");
+				assert_eq!(group.read_frame().await.unwrap().unwrap().payload.as_ref(), b"x");
+
+				while !run.subscriber.state.lock().subscribes.is_empty() {
+					settle().await;
+				}
+				let end = group.read_frame().await;
+				match whole {
+					true => assert!(matches!(end, Ok(None)), "{version}: the head was cut short: {end:?}"),
+					false => assert!(end.is_err(), "{version}: the head read as the whole group: {end:?}"),
+				}
+			}
+		}
+	}
+
 	/// SUBSCRIPTION_ENDED ends only our subscription, not the track, so the copy is released
 	/// for a later reader to subscribe afresh rather than lingering as if the track ended. Its
 	/// reader sees an error, not a clean end: the fresh copy's groups continue past this one's,