fix(net): a lite-06 viewer joining a relay reads its first group again

js/net/src/cache.test.ts

@@ -581,6 +581,20 @@ test("a kept local skip expires once the edge passes it or the track closes", as
 	});
 });
 
+// A final drop ends a group still arriving, but leaves one that finished here to every reader, as a
+// group arriving after its drop replaces it. Mirrors Rust `a_final_drop_leaves_a_finished_group`.
+test("a final drop leaves a finished group", async () => {
+	const producer = pooled(new cache.Pool({ capacity: 1 << 30 }));
+	hooks.receiveTrack(producer);
+	producer.createGroup(0n).finish();
+	const open = producer.createGroup(1n);
+	hooks.dropTrackGroups(producer, 0n, 2n, { error: new StreamError(StreamCode.Cancel) });
+
+	expect(dropsHeld(producer)).toEqual([1n]);
+	expect(open.closed.peek()).toBeInstanceOf(Error);
+	expect(await cached(producer)).toContain(0n);
+});
+
 // A received stale run the pool forgets is accounted for, as an evicted group is: no reader waits
 // on it from the live feed, so a reader that wants it asks a fill. Mirrors Rust
 // `a_reader_asks_a_fill_for_a_forgotten_run_it_judged`.

js/net/src/lite/publisher.test.ts

@@ -1511,7 +1511,10 @@ test("lite draft-07: a final drop replaces an expired stream", async () => {
 		expect("start" in (await decodeSubscribeResponse(sub.client.reader, Version.DRAFT_07))).toBe(true);
 
 		const reset = sub.stall();
-		sub.serve(1n);
+		// Still arriving, as a relay's copy from upstream: the upstream's final drop ends it. One
+		// held whole would stay, since a group stream wins over its drop.
+		const arriving = sub.track.createGroup(1n);
+		arriving.writeFrame(textFrame("hello", Timestamp.fromMillis(1)));
 		expect(await sub.servedHeader()).toBe(1n);
 		await flush();
 		await clock.advance(Duration.millis(budget * 10));

js/net/src/track.ts

@@ -1482,33 +1482,29 @@ export class Producer {
 	// Record `[start, end)` given up as `hole` says, returning the runs recorded. A skip is final
 	// and the most authoritative verdict, so a drop never replaces one. An upstream drop (`open`)
 	// also ends the groups still arriving in it with its error, so their readers settle now
-	// rather than once the stream's reset lands; one that finished here stays whole. A skip names
-	// nothing produced.
+	// rather than once the stream's reset lands. A skip names nothing produced.
 	#drop(start: bigint, end: bigint, hole: Hole, open = false): Run[] {
 		if (start >= end) return [];
 		const skipped = hole.skip ? [] : this.#state.drops.overlapping(start, end).filter((run) => run.skip);
 		let runs = runGaps(skipped, start, end);
-		// A received stale verdict leaves a group that arrived and was not aborted: a lite-07
-		// publisher may serve a group after its stale drop, and the group's stream may land first.
-		// The group wins, still arriving or not.
-		if (open && hole.stale !== undefined) {
+		const arriving: GroupProducer[] = [];
+		if (open && !hole.skip) {
 			for (const [sequence, entry] of this.#cached) {
-				if (sequence >= start && sequence < end && !(entry.group.closed.peek() instanceof Error)) {
-					runs = removeRun(runs, sequence, sequence + 1n);
-				}
+				const closed = entry.group.closed.peek();
+				if (sequence < start || sequence >= end || closed instanceof Error) continue;
+				// A lite-07 publisher may serve a group after its stale drop, and the group's stream
+				// may land first. A final drop ends a group still arriving, but one that finished here
+				// stays whole, as a re-sent copy the subscriber refused leaves it: naming it given up
+				// would hide it from every reader.
+				if (hole.stale === undefined && closed === undefined) arriving.push(entry.group);
+				else runs = removeRun(runs, sequence, sequence + 1n);
 			}
 		}
 		// A fill no longer owes what the track gave up for good. A stale verdict settles only
 		// the readers it covers, so a fill asked by any other is still owed.
 		if (hole.stale === undefined) this.#fills.dropped(start, end);
 		for (const run of runs) this.#setDrops(run.start, run.end, hole);
-		if (open && !hole.skip && hole.stale === undefined) {
-			for (const [sequence, entry] of this.#cached) {
-				if (sequence >= start && sequence < end && entry.group.closed.peek() === undefined) {
-					entry.group.abort(hole.error);
-				}
-			}
-		}
+		for (const group of arriving) group.abort(hole.error);
 		// A pending tail owes nothing the publisher gave up. A stale verdict is not receipt: new
 		// terms may supersede it, leaving those groups unknown again, as in Rust.
 		if (hole.stale === undefined) this.#arrive(start, end);

quest/m0/nightly/lite06-join-hang.md

@@ -1,36 +1,19 @@
-# [M] A lite-06 viewer joining a relay reads its first group again
+# [XS] A scheduled Nightly's relay session job finishes inside its hour
 
 ## Goal
 
-`relay_session_join_broadcasts` and `relay_session_join_stale` finish their
-moq-lite-06 cells, as they did before
-[moq-dev/moq#8476](https://github.com/moq-dev/moq/pull/8476), so
 `Nightly (Relay session benchmarks)` (`just gh nightly rust-relay`) finishes
-inside its hour with every cell.
+inside its hour with every cell, including the moq-lite-06 cells of
+`relay_session_join_broadcasts` and `relay_session_join_stale`.
 
 ## Plan
 
-Found 2026-11-12 while timing the relay session group for
-[moq-dev/moq#8555](https://github.com/moq-dev/moq/pull/8555). The first
-moq-lite-06 cell of `relay_session_join_broadcasts`
-(`relays=1/publishers=1/broadcasts=1/viewers=16`) and of
-`relay_session_join_stale` never leaves its warm-up: the bench thread parks in
-the cell's `join` loop (`rs/moq-net/benches/session/mod.rs`) at 0% CPU,
-waiting on a join, watch, or the first group read that never completes. The
-moq-transport and moq-lite-07 cells of the same groups finish in seconds, and
-`relay_session_join_relays` finishes on every version.
+The lite-06 join hang that kept those cells in their warm-up is fixed: a final
+drop no longer hides a group the relay holds whole. Locally, every
+`relay_session_join_*` cell finishes, the first lite-06 one in about 0.2 ms per
+join. Confirm the first scheduled Nightly after the fix lands; never dispatch
+Nightly.
 
-`git bisect` over `c916855d16..75f00f92c8` (the 11-11 Nightly's commit, whose
-lite-06 join cells finished in 0.2 s, to main on 2026-11-12) names
-f3edb0b8d9 ([moq-dev/moq#8476](https://github.com/moq-dev/moq/pull/8476), "a
-failed fetch below the end no longer holds it open") as the first hanging
-commit. Reproduce with
-`cargo bench -p moq-relay --features test-support --bench session -- --quick 'relay_session_join_broadcasts/moq-lite-06/'`.
+## Waiting
 
-Find which await hangs, reproduce it as a moq-net integration test of a
-lite-06 viewer joining a relay that holds history, and fix the cause. A
-viewer that never receives a group it subscribed to is a product bug, not
-only a bench one, so do not skip the cells.
-
-Then confirm a scheduled Nightly's relay session job finishes inside its
-hour; never dispatch Nightly.
+- moq-dev/moq: the first scheduled Nightly on `main` after the fix lands finishes its `Relay session benchmarks` job inside the hour; check with `gh run list --workflow nightly.yml --event schedule`, then delete this quest

rs/moq-net/src/model/track.rs

@@ -4447,8 +4447,10 @@ impl Producer {
 	}
 
 	/// Record `groups` given up as `hole` describes, returning the runs recorded: a drop
-	/// leaves a skipped sequence skipped, and a stale verdict leaves a group that arrived and
-	/// was not aborted, since a group stream wins over its drop.
+	/// leaves a skipped sequence skipped, and a group a copy still serves whole, since a group
+	/// stream wins over its drop: a stale drop leaves it still arriving or not, and a final one
+	/// ends the live copy still arriving but leaves a group finished here or served by a
+	/// fetched head.
 	#[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))]
 	pub(crate) fn give_up(&self, groups: ops::Range<u64>, hole: Hole) -> RangeSet<u64> {
 		let error = match &hole {
@@ -4467,38 +4469,38 @@ impl Producer {
 		}
 		let (runs, open): (_, Vec<group::Handle>) = match self.modify() {
 			Ok(mut state) => {
-				// Only the live feed's copy: a drop judges the subscription, so a fetched head
-				// or fetch-only backfill may still complete.
-				let open: Vec<_> = state
-					.lookup
-					.range(groups.clone())
-					.filter(|(_, slot)| slot.visible && !slot.group.is_aborted())
-					.map(|(&sequence, slot)| (sequence, slot.group.clone()))
-					.collect();
-				let recorded = match hole.stale_budget() {
+				let stale = hole.stale_budget().is_some();
+				let mut left = RangeSet::new();
+				left.insert(groups.clone());
+				let mut open = Vec::new();
+				for (&sequence, slot) in state.lookup.range(groups) {
+					if !slot.visible {
+						continue;
+					}
+					// A final drop ends only the live feed's copy still arriving: a drop judges
+					// the subscription, so a fetched head or fetch-only backfill may still complete.
+					let arriving = !slot.group.is_aborted() && !slot.group.is_settled();
+					if !stale && arriving {
+						open.push(slot.group.clone());
+					}
 					// A lite-07 publisher may serve a group after its stale drop, and the group's
-					// stream may land first: the group wins, still arriving or not.
-					Some(_) => {
-						let mut left = RangeSet::new();
-						left.insert(groups);
-						for (sequence, _) in &open {
-							if let Some(end) = sequence.checked_add(1) {
-								left.remove(*sequence..end);
-							}
-						}
-						let mut runs = RangeSet::new();
-						for run in left.iter() {
-							runs.extend(state.drop_groups(run.clone(), hole.clone()).iter().cloned());
-						}
-						(runs, Vec::new())
+					// stream may land first. A group a copy still serves whole stays, whichever
+					// copy serves it, as a re-sent copy the subscriber refused leaves it: naming
+					// it given up would hide it from every reader.
+					let ended = !stale && arriving && std::ptr::eq(slot.serving(), &slot.group);
+					if !ended
+						&& slot.serves_end()
+						&& let Some(end) = sequence.checked_add(1)
+					{
+						left.remove(sequence..end);
 					}
-					None => (
-						state.drop_groups(groups, hole),
-						open.into_iter().map(|(_, group)| group).collect(),
-					),
-				};
+				}
+				let mut runs = RangeSet::new();
+				for run in left.iter() {
+					runs.extend(state.drop_groups(run.clone(), hole.clone()).iter().cloned());
+				}
 				state.charge_debt();
-				recorded
+				(runs, open)
 			}
 			Err(_) => return RangeSet::new(),
 		};
@@ -10927,11 +10929,8 @@ mod test {
 		indexed();
 		producer.drop_groups(15..25, StreamError::Cancel);
 		indexed();
-		producer
-			.create_group(group::Info { sequence: 17 })
-			.unwrap()
-			.finish()
-			.unwrap();
+		// Still arriving, so the next drop ends it and covers it again.
+		let _recreated = producer.create_group(group::Info { sequence: 17 }).unwrap();
 		indexed();
 		producer.drop_groups(14..26, StreamError::Old);
 		indexed();
@@ -10941,6 +10940,65 @@ mod test {
 		);
 	}
 
+	/// A final drop ends a group still arriving, but leaves one that finished here to every
+	/// reader, as a group arriving after its drop replaces it.
+	#[test]
+	fn a_final_drop_leaves_a_finished_group() {
+		let producer = track_producer("test", Info::default());
+		producer
+			.create_group(group::Info { sequence: 0 })
+			.unwrap()
+			.finish()
+			.unwrap();
+		let open = producer.create_group(group::Info { sequence: 1 }).unwrap();
+		producer.drop_groups(0..2, StreamError::Cancel);
+
+		let drops: Vec<_> = producer.state.read().drops.iter().map(|(run, _)| run.clone()).collect();
+		assert_eq!(drops, vec![1..2]);
+		assert!(open.is_aborted(), "the group still arriving ends");
+		let mut consumer = producer.subscribe(None);
+		assert_eq!(consumer.assert_group().sequence, 0);
+	}
+
+	/// A drop leaves a group a fetched head serves whole, whatever its live copy is doing: a
+	/// final drop still ends a live copy still arriving, and the head serves in its place.
+	#[kio::test]
+	#[cfg(feature = "async")]
+	async fn a_drop_leaves_a_group_a_fetched_head_serves() {
+		let producer = track_producer("test", Info::timed());
+		let dynamic = producer.requests();
+		let consumer = producer.consume();
+		let mut lives = Vec::new();
+		for sequence in 0..2 {
+			let mut live = producer.create_group(group::Info { sequence }).unwrap();
+			live.start_at(1).unwrap();
+			live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta"))
+				.unwrap();
+			let fetch = consumer.fetch_group(sequence, None);
+			let request = dynamic.requested_fill_group().await.unwrap();
+			let mut head = request.accept(Info::timed()).unwrap();
+			for payload in [&b"snapshot"[..], b"delta"] {
+				head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(payload))
+					.unwrap();
+			}
+			head.finish().unwrap();
+			fetch.await.unwrap();
+			lives.push(live);
+		}
+		lives.pop().unwrap().abort(Error::Cancel).unwrap();
+
+		producer.drop_groups(0..1, StreamError::App(5));
+		producer.give_up(1..2, Hole::stale(Duration::from_millis(500)));
+
+		assert!(producer.state.read().drops.iter().next().is_none());
+		assert!(lives[0].is_aborted(), "the live copy still arriving ends");
+		let mut sub = producer.subscribe(None);
+		assert!(sub.poll_drop(&kio::Waiter::noop()).is_pending());
+		for sequence in 0..2 {
+			assert_eq!(sub.assert_group().sequence, sequence);
+		}
+	}
+
 	/// Let `duration` pass on the cache pool `producer`'s groups charge into.
 	fn elapse(producer: &Producer, duration: Duration) {
 		producer.broadcast.pool.step(duration);

rs/moq-net/tests/integration/history_join.rs

@@ -0,0 +1,67 @@
+#![cfg(feature = "async")]
+
+//! Viewers joining a relay that already holds a track's history, each on its own session.
+
+use std::time::Duration;
+
+use moq_net::{Timestamp, Version, cache, origin, track};
+
+use crate::support::harness::{MockConnectOptions, connect_mock, spawn};
+
+const TIMEOUT: Duration = Duration::from_secs(10);
+
+fn origin(hop: u64) -> origin::Producer {
+	let mut config = origin::Config::new(std::num::NonZeroU64::new(hop).unwrap());
+	config.pool = cache::Pool::unbounded();
+	let (origin, driver) = origin::Producer::new(config);
+	spawn(driver);
+	origin.with_proxy(true)
+}
+
+/// Every viewer that joins reads the group the relay holds. Each join subscribes upstream
+/// again once the last viewer left, so the publisher serves the held group again; the relay
+/// refuses that copy, and a lite publisher before lite-07 then drops the group for good.
+/// That drop must not hide the group the relay holds whole.
+#[kio::test]
+async fn each_joining_viewer_reads_the_held_group() {
+	for version in ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip", "moq-transport-22"] {
+		let version: Version = version.parse().unwrap();
+		let publisher = origin(1);
+		let relay = origin(2);
+		let mut options = MockConnectOptions::new(version);
+		options.client_publish = Some(publisher.consume());
+		options.server_subscribe = Some(relay.clone());
+		let _upstream = connect_mock(options).await;
+
+		let broadcast = publisher.route("live", origin::RouteOptions::default()).unwrap();
+		let info = track::Info::default()
+			.with_timescale(moq_net::Timescale::MILLI)
+			.with_max_age(None);
+		let source = broadcast.broadcast().create_track("video", info).unwrap();
+		let mut group = source.append_group().unwrap();
+		group
+			.write_frame(Timestamp::from_millis(0).unwrap(), &b"x"[..])
+			.unwrap();
+		group.finish().unwrap();
+
+		for join in 0..3 {
+			let viewer = origin(3 + join);
+			let mut options = MockConnectOptions::new(version);
+			options.server_publish = Some(relay.consume());
+			options.client_subscribe = Some(viewer.clone());
+			let downstream = connect_mock(options).await;
+			kio::clock::timeout(TIMEOUT, async {
+				let remote = viewer.consume().follow("live", None).unwrap().available().await;
+				let remote = remote.unwrap().expect("the broadcast resolves");
+				let mut subscriber = remote.track("video").unwrap().subscribe(None).await.unwrap();
+				let mut group = subscriber.recv_group().await.unwrap().expect("a group");
+				assert_eq!(group.sequence, 0, "{version}: viewer {join}");
+				let frame = group.read_frame().await.unwrap().expect("a frame");
+				assert_eq!(&frame.payload[..], b"x", "{version}: viewer {join}");
+			})
+			.await
+			.unwrap_or_else(|_| panic!("{version}: viewer {join} never read the held group"));
+			drop(downstream);
+		}
+	}
+}

rs/moq-net/tests/integration/main.rs

@@ -38,6 +38,7 @@ mod flights;
 mod goaway;
 mod group_order;
 mod history_groups;
+mod history_join;
 mod idle_cancel_refetch;
 mod idle_session;
 mod implementation;

rs/moq-net/verify/subscribe/SubscribeAccounting.tla

@@ -342,7 +342,9 @@ TailFold(g) ==
 \* `apply_response` for SUBSCRIBE_DROP. A stale drop under a replaced update is no
 \* verdict. Otherwise the tail accounts for the group; a stale drop records it in the
 \* copy unless its stream finished under this update or the copy holds it
-\* (`give_up` keeps an open group), and a final drop records it and ends its readers.
+\* (`give_up` keeps an open group), and a final drop records it and ends its readers
+\* unless the copy holds it whole (`give_up` keeps a group a finished copy serves,
+\* the live one or a fetched head).
 \* A stale drop of a group inside the span a fold forgot that the ledger already
 \* accounted (`Tail::overtaken`) may be one a finished stream overtook, so the next
 \* update owes it nothing again.
@@ -365,7 +367,7 @@ OnDrop(m) ==
                /\ UNCHANGED <<cached, violation>>
        ELSE /\ acct' = acct \cup {g}
             /\ sstale' = sstale \ {g}
-            /\ rec' = [rec EXCEPT ![g] = IF @ = "skip" THEN "skip" ELSE "final"]
+            /\ rec' = IF rec[g] = "skip" \/ cached[g] = "whole" THEN rec ELSE [rec EXCEPT ![g] = "final"]
             /\ cached' = [cached EXCEPT ![g] = IF @ = "open" THEN "aborted" ELSE @]
             /\ UNCHANGED violation