fix(net): a relay's rejoin never shows an older cached group as the live edge

doc/concept/moq-lite.md

@@ -345,8 +345,11 @@ reader or the next fetch finds its cache. That copy is not live meanwhile:
 readers get nothing from its cache until the source answers again, since how
 stale it is cannot be told. lite-07 and moq-transport answer with their largest
 position, and older lite versions are asked from the head of the newest cached
-group, so their first frame says where their feed is. A feed that has moved on
-past everything cached leaves the cache below it to fetches.
+group, so their first frame says where their feed is. Readers resume once the
+cache shows the answered position: a received group stays hidden until its
+first frame lands or its stream ends, since a bare group header would leave the
+older cached group looking like the live edge. A feed that has moved on past
+everything cached leaves the cache below it to fetches.
 
 The publisher may declare a retention window per track. Omission sets no limit;
 zero keeps only the live edge. The origin cache ceiling and cache pool may still

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

@@ -32,7 +32,7 @@ Public API: none. Wire: none.
 
 ## Required
 
-- [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed
+- [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - a late joiner shows video promptly and catches up to live, and the check asserts that under load
 - [Import catalog finish](/quest/m1/test-flakes-2/import-catalog-finish.md) - `moq-cli`'s subprocess EOF catalog-finish test holds up under load with event-based fixture coordination
 - [Relay restart rebind](/quest/m1/test-flakes-2/relay-restart-rebind.md) - the crash drill restarts on its original UDP address under concurrent load
 - [Impaired handshake](/quest/m1/test-flakes-2/impaired-handshake.md) - the impaired cluster drills' clients never time out while connecting

quest/m1/test-flakes-2/media-late-join.md

@@ -1,30 +1,54 @@
-# [S] Media late join within one GOP
+# [S] Media late join catches up to live
 
 ## Goal
 
-Late join holds its one-GOP budget under load in both `just test media` and
-the interop browser lane, or the regression behind it is fixed.
+A viewer joining a broadcast through a relay after a demand gap shows video
+promptly and reaches the live edge within a bound, and the late-join check
+asserts exactly that and passes under load.
 
 ## Plan
 
-It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181):
-"joined at frame 111, 16 frames behind 127", against a budget of one GOP
-(15). First decide whether 16 frames is a real regression (the player
-joining at the previous GOP's keyframe) or an off-by-one in how the fixture
-samples the live edge. Fix whichever it is; don't widen the budget without a
-reason.
-
-The interop browser check "late join starts live"
-(`test/interop/clients/js/media.ts`, `MAX_LATE_JOIN_LAG_FRAMES`, one GOP
-compared with `<=`) fails the same way on `main`: 16 frames behind with a
-15-frame GOP on 09-24, 09-25, 09-28, and in
-[#4577](https://github.com/moq-dev/moq/pull/4577). It samples `live` just
-before the page opens. Treat it as the same flake (decided 2026-10-01).
-
-Recorded in the 2026-10-05 audit: on the test-flakes-2 line,
-[#4719](https://github.com/moq-dev/moq/pull/4719) replaced
-`MAX_LATE_JOIN_LAG_FRAMES` in the interop browser check with a
-published-keyframe check: the latecomer's presented timestamp must be at or
-after the published track's newest keyframe. That fixes the browser lane's
-sampling. The original 16-frame failure was never reproduced, so what remains
-is the Rust `just test media` late join under load.
+It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181) ("16
+frames behind" a 15-frame GOP), and the interop browser check failed the
+same way on `main`. `just test media` is that browser lane through a Rust
+`moq-relay`; there is no separate Rust media test.
+[#4719](https://github.com/moq-dev/moq/pull/4719) now compares the
+latecomer against the published keyframe instead of the painted counter.
+
+It is real, not fixture sampling: the latecomer is handed a GOP from before
+the demand gap. Reproduced by looping `bun media.ts --cases late-join`
+against a local relay with debug logs. In the fixture, the old viewer
+leaves, the publisher pauses (closing its open GOP and writing an empty
+marker group), and the latecomer joins about a second later.
+
+[#4914](https://github.com/moq-dev/moq/pull/4914) fixed two relay windows,
+each with a deterministic test:
+
+- A received group stays hidden from readers until its first frame lands or
+  its stream ends, and an idle copy goes live only once the cache shows the
+  route's answer (`rejoin_waits_for_the_answers_first_frame`).
+- A rejoin's answer is judged against the groups that survived the leave,
+  not the idle snapshot, since the cancel resets the group in flight
+  (`an_answer_past_a_reset_group_skips_the_cache`).
+
+What remains (1 of 30 runs after both fixes): the latecomer's max delay rises
+to its playout delay (about 145ms) right after it subscribes. When the
+pre-gap GOP was short, the GOP before it still reaches within that budget of
+the marker, so the relay hands it over by design. The player then presents
+from that GOP's keyframe, not from its playout position. Relay log: groups
+7, 8, 9 served in 1ms with a 145ms budget, and the player shows frames 360ms
+behind the published keyframe.
+
+Decided 2026-10-06 by the maintainer: a joiner showing the reachable pre-gap
+GOP at once and then fast-forwarding to live is the wanted behavior, so no
+player or consumer change. The check is what's too strict: it asserts the
+first presented frame is at or after the newest published keyframe.
+
+Remaining work: the late-join check (`just test media` and the interop
+browser lane) asserts what a viewer cares about instead: video shows within a
+start bound, and the player reaches the live edge within a catch-up bound.
+Set both bounds from measurements, looped under synthetic CPU load (the
+maintainer approved a bounded load generator such as `stress-ng` for these
+runs), and keep the relay fixes' deterministic tests as the regression guard.
+
+Public API: none. Wire: none.

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

@@ -474,11 +474,13 @@ impl<S: crate::transport::poll::Session> Subscriber<S> {
 		if let Ok(mut tail) = entry.tail.write() {
 			tail.account(dg.sequence..dg.sequence.saturating_add(1), self.runtime.now());
 		}
-		// A datagram says where the live feed is as well as a group does.
-		if !entry.producer.is_live() {
+		// A datagram says where the live feed is as well as a group does. Inserted first, so
+		// the copy goes live with it already showing.
+		let live = entry.producer.is_live();
+		entry.producer.insert_datagram(dg.sequence, timestamp, dg.payload)?;
+		if !live {
 			entry.producer.set_live(Some(Position::group(dg.sequence)));
 		}
-		entry.producer.insert_datagram(dg.sequence, timestamp, dg.payload)?;
 		Ok(())
 	}
 
@@ -792,6 +794,8 @@ enum GroupRecvState {
 		group: crate::recv::Group,
 		track: track::Producer,
 		ingest: FrameIngest,
+		/// Whether readers see the group yet; see [`track::Producer::receive_group`].
+		shown: bool,
 		_reading: Reading,
 	},
 	Done,
@@ -819,19 +823,21 @@ impl<S: crate::transport::poll::Session> GroupRecv<S> {
 						// The subscription's end waits until this stream is read.
 						let reading = Reading::open(&entry.tail, Some(hdr.sequence), self.subscriber.runtime.now());
 
+						let group_info = group::Info { sequence: hdr.sequence };
+						// Stats (groups/frames/bytes) are counted in the model as the group
+						// is written, through the tagged `track::Producer`. Withheld from
+						// readers until its first frame lands.
+						let received = entry.producer.receive_group(group_info);
 						// A route's first group says where its live feed is, when its answer did
 						// not: even one the cache already holds, as an idle copy asked from its
-						// head gets back.
+						// head gets back. The copy goes live once the cache shows that group.
 						if !entry.producer.is_live() {
 							entry.producer.set_live(Some(Position {
 								group: hdr.sequence,
 								frame: hdr.frame_start,
 							}));
 						}
-						let group_info = group::Info { sequence: hdr.sequence };
-						// Stats (groups/frames/bytes) are counted in the model as the group
-						// is written, through the tagged `track::Producer`.
-						let mut group = match entry.producer.create_group(group_info) {
+						let mut group = match received {
 							Ok(group) => group,
 							// The group is at or past the end the publisher declared, which no
 							// later stream can repair. Before lite-05 only a local finish sets
@@ -865,11 +871,16 @@ impl<S: crate::transport::poll::Session> GroupRecv<S> {
 						group: crate::recv::Group::new(group),
 						track,
 						ingest: FrameIngest::new(&self.subscriber, timescale),
+						shown: false,
 						_reading: reading,
 					};
 				}
 				GroupRecvState::Serve {
-					group, track, ingest, ..
+					group,
+					track,
+					ingest,
+					shown,
+					..
 				} => {
 					// The track or group dying cancels the stream; the peer's own close
 					// arrives through the ingest's reads.
@@ -880,7 +891,13 @@ impl<S: crate::transport::poll::Session> GroupRecv<S> {
 						if let Poll::Ready(err) = group.poll_closed(waiter) {
 							break 'serve Err(err);
 						}
-						match ingest.poll(&mut self.reader, group, waiter) {
+						let res = ingest.poll(&mut self.reader, group, waiter);
+						// Shown once its first frame opens, even mid-payload.
+						if !*shown && group.is_started() {
+							track.reveal_group(group);
+							*shown = true;
+						}
+						match res {
 							Poll::Ready(res) => break 'serve res,
 							Poll::Pending => return Poll::Pending,
 						}
@@ -890,27 +907,35 @@ impl<S: crate::transport::poll::Session> GroupRecv<S> {
 					// short drops into an already-aborted group instead of reporting a loss.
 					let GroupRecvState::Serve {
 						group,
+						track,
 						ingest: _ingest,
+						shown,
 						_reading,
-						..
 					} = std::mem::replace(&mut self.state, GroupRecvState::Done)
 					else {
 						unreachable!()
 					};
-					match res {
+					// A group that ended without a frame shows once it settles.
+					let handle = (!shown).then(|| (*group).clone());
+					let res = match res {
 						Ok(()) => {
 							let _ = group.finish();
+							Ok(())
 						}
 						Err(err @ (Error::Cancel | Error::Stream(crate::StreamError::Cancel))) => {
 							let _ = group.abort(err);
+							Ok(())
 						}
 						Err(err) => {
 							tracing::debug!(%err, group = %group.sequence, "group error");
 							let _ = group.abort(err.clone());
-							return Poll::Ready(Err(err));
+							Err(err)
 						}
+					};
+					if let Some(handle) = handle {
+						track.reveal_group(&handle);
 					}
-					return Poll::Ready(Ok(()));
+					return Poll::Ready(res);
 				}
 				GroupRecvState::Done => return Poll::Ready(Ok(())),
 			}

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

@@ -192,6 +192,11 @@ enum Timeline {
 }
 
 impl GroupState {
+	/// Whether the first frame opened, or the group ended without one.
+	fn started(&self) -> bool {
+		!matches!(self.timeline, Timeline::Empty) || self.fin.is_some() || self.abort.is_some()
+	}
+
 	/// Content still available to a reader of this group.
 	///
 	/// Counts the in-flight frame at its declared size, like [`Self::content_range`]: a
@@ -895,6 +900,16 @@ impl Producer {
 		(state.committed > state.offset).then_some(state.committed)
 	}
 
+	/// Whether `other` is a handle to this same group.
+	pub(crate) fn is_clone(&self, other: &Self) -> bool {
+		self.state.same_channel(&other.state)
+	}
+
+	/// Whether the group opened its first frame, or ended without one.
+	pub(crate) fn is_started(&self) -> bool {
+		self.state.read().started()
+	}
+
 	/// Where the group starts in presentation time: its first frame's timestamp, or
 	/// `None` while no frame has been opened or when that frame is untimed.
 	///
@@ -998,7 +1013,7 @@ impl Producer {
 	/// none has been opened.
 	pub(crate) fn poll_started(&self, waiter: &kio::Waiter) -> Poll<()> {
 		match self.state.poll(waiter, |state| {
-			if !matches!(state.timeline, Timeline::Empty) || state.fin.is_some() || state.abort.is_some() {
+			if state.started() {
 				Poll::Ready(())
 			} else {
 				Poll::Pending