fix(video): trim orphan leading pictures after delivery breaks

.github/workflows/ts.yml

@@ -141,11 +141,11 @@ jobs:
           MOQ_TEST_RUNS: ${{ runner.temp }}/moq-test-runs
         shell: bash -leo pipefail {0}
 
-      # The same round-trip over an open-GOP clip: every leading picture after a
-      # recovery point must come back, in decode order.
+      # The same round-trip over an open-GOP clip: recovery-group orphans are trimmed,
+      # continuous leading pictures stay in decode order, and all shape checks gate.
       - name: TS open GOP
         if: ${{ !cancelled() }}
-        run: nix develop --command just test ts --open-gop
+        run: nix develop --command just test ts --open-gop --strict
         env:
           MOQ_TEST_RUNS: ${{ runner.temp }}/moq-test-runs
         shell: bash -leo pipefail {0}

js/watch/src/video/decoder.test.ts

@@ -700,3 +700,37 @@ it("a reset drops a parked video frame even when the next anchor restores the re
 		clock.restore();
 	}
 });
+
+for (const kind of ["legacy", "cmaf"] as const) {
+	it(`${kind} trims orphan leading pictures on join and later gap, preserving continuous groups`, async () => {
+		const reads: Read[] = [
+			sample(10, 1000),
+			sample(10, 800, false),
+			sample(10, 900, false),
+			sample(10, 1100, false),
+			sample(11, 2000),
+			sample(11, 1800, false),
+			sample(11, 1900, false),
+			sample(11, 2100, false),
+			gap,
+			sample(13, 4000),
+			sample(13, 3800, false),
+			sample(13, 3900, false),
+			sample(13, 4100, false),
+			{ kind: "end", group: 14, end: Micro(4200) },
+			sample(15, 5000),
+			sample(15, 4800, false),
+			sample(15, 4900, false),
+			sample(15, 5100, false),
+		];
+		const playback = await guardedPlayback(kind, reads);
+		try {
+			await microtasks();
+			expect(playback.submitted).toEqual([
+				1000, 1100, 2000, 1800, 1900, 2100, 4000, 4100, 5000, 4800, 4900, 5100,
+			]);
+		} finally {
+			playback.close();
+		}
+	});
+}

js/watch/src/video/decoder.ts

@@ -26,6 +26,7 @@ import {
 	supportCacheKey,
 } from "./config";
 import { caughtUp, renditionJitter, switchJitter } from "./playhead";
+import { Recovery } from "./recovery";
 import type { Source } from "./source";
 
 // The amount of time to wait before considering the video to be buffering.
@@ -552,6 +553,7 @@ class DecoderTrack {
 
 		let previous: Micro | undefined;
 		let latest: number | undefined;
+		const recovery = new Recovery();
 
 		effect.spawn(async () => {
 			for (;;) {
@@ -561,7 +563,10 @@ class DecoderTrack {
 				// A gap or a declared end breaks the timeline: re-anchor before decoding further, and
 				// never bridge the buffer across it.
 				if (next.kind !== "frame") {
-					if (next.kind === "gap") this.arrivals?.write(gap(next));
+					if (next.kind === "gap") {
+						recovery.gap();
+						this.arrivals?.write(gap(next));
+					}
 					if (latest !== undefined) this.#onBreak();
 					previous = undefined;
 					continue;
@@ -572,11 +577,12 @@ class DecoderTrack {
 				// sequences only grow within a subscription, so a break (which a stale marker can
 				// raise) never resets this guard. Rejected payloads still count as received bytes.
 				const stale = latest !== undefined && next.group < latest;
+				const accepted = !stale && recovery.accept(frame);
 				this.stats.update((current) => ({
-					frameCount: (current?.frameCount ?? 0) + (stale ? 0 : 1),
+					frameCount: (current?.frameCount ?? 0) + (accepted ? 1 : 0),
 					bytesReceived: (current?.bytesReceived ?? 0) + frame.payload.byteLength,
 				}));
-				if (stale) continue;
+				if (!accepted) continue;
 
 				// Classify ingress before decoder backpressure changes the delivery clock.
 				const timestamp = Milli.fromMicro(frame.timestamp as Micro);
@@ -637,6 +643,7 @@ class DecoderTrack {
 
 		let previous: Micro | undefined;
 		let latest: number | undefined;
+		const recovery = new Recovery();
 
 		effect.spawn(async () => {
 			for (;;) {
@@ -645,7 +652,10 @@ class DecoderTrack {
 
 				// A gap or a declared end breaks the timeline (see #runLegacy).
 				if (next.kind !== "frame") {
-					if (next.kind === "gap") this.arrivals?.write(gap(next));
+					if (next.kind === "gap") {
+						recovery.gap();
+						this.arrivals?.write(gap(next));
+					}
 					if (latest !== undefined) this.#onBreak();
 					previous = undefined;
 					continue;
@@ -656,11 +666,12 @@ class DecoderTrack {
 				// sequences only grow within a subscription, so a break (which a stale marker can
 				// raise) never resets this guard. Rejected payloads still count as received bytes.
 				const stale = latest !== undefined && next.group < latest;
+				const accepted = !stale && recovery.accept(frame);
 				this.stats.update((current) => ({
-					frameCount: (current?.frameCount ?? 0) + (stale ? 0 : 1),
+					frameCount: (current?.frameCount ?? 0) + (accepted ? 1 : 0),
 					bytesReceived: (current?.bytesReceived ?? 0) + frame.payload.byteLength,
 				}));
-				if (stale) continue;
+				if (!accepted) continue;
 
 				// Classify ingress before decoder backpressure changes the delivery clock.
 				const timestamp = Milli.fromMicro(frame.timestamp);

js/watch/src/video/recovery.ts

@@ -0,0 +1,22 @@
+import type { Micro } from "@moq/clock";
+import type { Container } from "@moq/hang";
+
+/** Refuse leading pictures whose previous GOP was not delivered to this decoder. @internal */
+export class Recovery {
+	#needed = true;
+	#floor?: Micro;
+
+	/** Arm recovery after media was lost, preserving continuous groups and declared ends. */
+	gap(): void {
+		this.#needed = true;
+	}
+
+	/** Whether this access unit has the references available at the current delivery boundary. */
+	accept(frame: Container.ReceivedFrame): boolean {
+		if (frame.keyframe) {
+			this.#floor = this.#needed ? frame.timestamp : undefined;
+			this.#needed = false;
+		}
+		return this.#floor === undefined || frame.timestamp >= this.#floor;
+	}
+}

quest/m1/README.md

@@ -140,7 +140,6 @@ above them.
 - [Encoder colour](/quest/m1/color-model.md) - every moq-video encode path signals the colour its output actually has, or refuses instead of mislabelling
 - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide
 - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the supported T-STD buffer models, starting with a fixed `--delay`
-- [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - a viewer joining at a recovery point drops the leading pictures it cannot decode; continuous viewers keep them
 - [Draft 14-16 updates](/quest/m1/ietf-legacy-updates.md) - Rust and JS apply and answer moq-transport 14-16 request updates without ending or leaking the request
 - [Tail arrivals](/quest/m1/tail-arrivals.md) - a track judges its pending tail from recorded arrivals, not a cache scan, and errors a reader whose tail ends short
 - [Resume reorder](/quest/m1/resume-reorder.md) - a group whose stream header arrives after its successor's still resumes on failover; only a dropped or long-gone group fails

quest/m1/ts-reliability/recovery.md

@@ -43,5 +43,4 @@ and measured release timing as separate checks.
 ## Related
 
 - [Automated validation](/test/ts/README.md) - shared graders, profiles and known-bad controls
-- [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - trims after a gap
 - [Gradual recovery](/quest/m2/intra-refresh/README.md) - recovery counts and per-group warmup

quest/m2/intra-refresh/README.md

@@ -56,7 +56,3 @@ make these presentation guarantees true even when sweep lengths change.
 - [H.264 import](/quest/m2/intra-refresh/h264-import.md) - the splitter keeps `recovery_frame_cnt` and import publishes `warmup` from it
 - [H.265 import](/quest/m2/intra-refresh/h265-import.md) - the splitter reads the recovery-point SEI so an HEVC intra-refresh stream forms groups and publishes `warmup`
 - [Export sync flags](/quest/m2/intra-refresh/export-sync-flags.md) - fmp4, MKV, and HLS stop advertising a refresh group start as a sync sample
-
-## Related
-
-- [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - frames stamped before the group's keyframe are the other tune-in trim

quest/m2/intra-refresh/consumer-warmup.md

@@ -59,5 +59,4 @@ The fix is history rather than a withhold rule:
 
 ## Related
 
-- [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - trims frames stamped before the keyframe, the other tune-in case
 - [Native dual decoder](/quest/m2/intra-refresh/native-dual-decoder.md) - the player path, which shows stripes instead

rs/moq-cli/src/play/buffer.rs

@@ -6,19 +6,36 @@ use std::collections::VecDeque;
 use std::time::Duration;
 
 /// Encoded access units stay in decode order, even when their PTS reorders.
-#[derive(Default)]
 pub(super) struct Buffer {
 	pub frames: VecDeque<Frame>,
 	pub bytes: usize,
 	edge: Option<moq_net::Timestamp>,
 	// Monotone minimum: PTS can reorder behind the front of decode order.
 	oldest: VecDeque<moq_net::Timestamp>,
 	pub generation: u64,
+	/// The keyframe the buffer last resumed at. A picture presenting before it references
+	/// media the buffer never had, so it is neither decoded nor shown.
 	pub floor: Option<moq_net::Timestamp>,
 	waiting_keyframe: bool,
 	pub ended: bool,
 }
 
+impl Default for Buffer {
+	fn default() -> Self {
+		Self {
+			frames: VecDeque::new(),
+			bytes: 0,
+			edge: None,
+			oldest: VecDeque::new(),
+			generation: 0,
+			floor: None,
+			// A join resumes at its first keyframe, like any other break.
+			waiting_keyframe: true,
+			ended: false,
+		}
+	}
+}
+
 impl Buffer {
 	pub fn clear(&mut self) {
 		self.frames.clear();
@@ -53,15 +70,10 @@ impl Buffer {
 			}
 			self.waiting_keyframe = false;
 		}
-		if self.waiting_keyframe {
+		if self.waiting_keyframe || self.below_floor(&frame) {
 			return;
 		}
-		while self.oldest.back().is_some_and(|at| *at > frame.timestamp) {
-			self.oldest.pop_back();
-		}
-		self.oldest.push_back(frame.timestamp);
-		self.bytes += frame.payload.len();
-		self.frames.push_back(frame);
+		self.index(frame);
 		let edge = self.edge.unwrap();
 		let stale =
 			|oldest: &moq_net::Timestamp| edge.as_micros().saturating_sub(oldest.as_micros()) > max_delay.as_micros();
@@ -75,8 +87,30 @@ impl Buffer {
 			self.generation += 1;
 			self.waiting_keyframe = self.frames.is_empty();
 			self.floor = self.frames.front().map(|frame| frame.timestamp);
+			// The retained keyframe's leading pictures referenced what was just dropped.
+			let mut frames = std::mem::take(&mut self.frames);
+			frames.retain(|frame| !self.below_floor(frame));
+			self.bytes = 0;
+			self.oldest.clear();
+			for frame in frames {
+				self.index(frame);
+			}
 		}
 	}
+
+	/// Append a frame, tracking its bytes and the earliest presentation time.
+	fn index(&mut self, frame: Frame) {
+		while self.oldest.back().is_some_and(|at| *at > frame.timestamp) {
+			self.oldest.pop_back();
+		}
+		self.oldest.push_back(frame.timestamp);
+		self.bytes += frame.payload.len();
+		self.frames.push_back(frame);
+	}
+
+	fn below_floor(&self, frame: &Frame) -> bool {
+		self.floor.is_some_and(|floor| frame.timestamp < floor)
+	}
 }
 
 #[cfg(test)]
@@ -113,6 +147,30 @@ mod tests {
 		assert_eq!(buffer.bytes, 7);
 	}
 
+	#[test]
+	fn eviction_drops_the_retained_keyframes_leading_pictures() {
+		let mut buffer = Buffer::default();
+		let age = Duration::from_millis(100);
+		for (ms, keyframe) in [
+			(0, true),
+			(50, false),
+			(100, true),
+			(80, false),
+			(90, false),
+			(110, false),
+		] {
+			buffer.push(frame(ms, keyframe), age);
+		}
+		buffer.push(frame(95, false), age);
+		let kept: Vec<_> = buffer.frames.iter().map(|f| f.timestamp.as_millis()).collect();
+		assert_eq!(kept, [100, 110], "80, 90 and 95 reference the evicted group");
+		assert_eq!(buffer.bytes, 14);
+		assert_eq!(buffer.oldest().unwrap().as_millis(), 100);
+		buffer.push(frame(200, true), age);
+		buffer.push(frame(180, false), age);
+		assert_eq!(buffer.frames.len(), 4, "a continuous group keeps its leading pictures");
+	}
+
 	#[test]
 	fn reordered_timestamps_keep_decode_order_and_the_newest_edge() {
 		let mut buffer = Buffer::default();