From bf694fb707d15e8704a3f9f1aa06f0cd62cf4eb4 Mon Sep 17 00:00:00 2001 From: Thomas Drapier Date: Tue, 29 Sep 2026 17:04:14 +0100 Subject: [PATCH] fix(moq-mux): one TS re-anchor shift per program An unflagged loop wrap grew each elementary stream's shift to its own live edge, so audio and video moved apart at every wrap. The shift is now shared: the first stream below its edge holds each stream's new pass until every live stream has shown its new PTS, or the PCR has run a second past, and the shift grows once by the largest need. Co-authored-by: Claude Opus 5.5 Co-authored-by: Cursor --- quest/m1/3489-ts-import-stream-liveness.md | 2 +- quest/m1/README.md | 1 - quest/m1/hop-aligned-import.md | 1 - quest/m1/remove-live.md | 4 +- quest/m1/ts-import-shared-shift.md | 51 -- quest/m1/ts-programs.md | 4 - rs/moq-mux/src/container/ts/import.rs | 854 ++++++++++++++++----- 7 files changed, 679 insertions(+), 238 deletions(-) delete mode 100644 quest/m1/ts-import-shared-shift.md diff --git a/quest/m1/3489-ts-import-stream-liveness.md b/quest/m1/3489-ts-import-stream-liveness.md index 8107a665b5..2dd6f1633b 100644 --- a/quest/m1/3489-ts-import-stream-liveness.md +++ b/quest/m1/3489-ts-import-stream-liveness.md @@ -35,7 +35,7 @@ audio half is visible, late, through #3372's resync line. `PID_error` check, and leave the catalog `stalled` bit alone: the importer already sets it for a quiet video PID (`Stream::tick` after each decode batch, #3630). These counters add no timeout; anything that must bound a - wait on a silent PID (the shared-shift quest) brings its own. + wait on a silent PID brings its own, as the TS re-anchor hold does on the PCR. - Tests with the issue's stimulus shape: suppress one PID's PES while keeping its PCR and continuity legal, assert the row's count stops and the gap grows; audio and SCTE-35 arms. diff --git a/quest/m1/README.md b/quest/m1/README.md index bfd771be85..23148512e0 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -59,7 +59,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Full codec string](/quest/m1/publish-codec-string.md) - browser-published video carries the encoder's full RFC 6381 codec string, so native players decode it - [TS program selection](/quest/m1/ts-programs.md) - `import ts` refuses a multi-program stream unless `--program ` picks one or publishes each - [TS export jitter](/quest/m1/ts-export-jitter.md) - the video reorder bound follows later catalogs and the declared reorder depth, so a late B-frame never reorders TS output; an undeclared stream can reorder once per new maximum depth -- [TS import shared shift](/quest/m1/ts-import-shared-shift.md) - unflagged loop wraps move audio and video by one shift, so A/V sync holds across wraps - [ffmpeg muxdelay](/quest/m1/ts-publish-muxdelay.md) - the documented MPEG-TS publish line adds `-muxdelay 0`, so quiet audio is not clumped - [Same-hop importers](/quest/m1/hop-aligned-import.md) - importers sharing a `--hop` and fed one stream publish identical groups and timestamps, so failover survives - [PipeWire duplicate cameras](/quest/m1/pipewire-dup-cameras.md) - a webcam lists once with PipeWire enabled diff --git a/quest/m1/hop-aligned-import.md b/quest/m1/hop-aligned-import.md index 6f6f68b6c6..64c2ef0bc8 100644 --- a/quest/m1/hop-aligned-import.md +++ b/quest/m1/hop-aligned-import.md @@ -41,5 +41,4 @@ survives the standby joining and the incumbent stopping. ## Related - [Redundant ingest](/quest/m2/redundant-ingest.md) - splicing across first hops and two encoders, which this does not attempt -- [TS import shared shift](/quest/m1/ts-import-shared-shift.md) - the TS re-anchor shift that must stay input-derived - [Broadcast epochs](/quest/m1/broadcast-epoch/README.md) - a redundant pair shares one epoch diff --git a/quest/m1/remove-live.md b/quest/m1/remove-live.md index 99c040ec2c..8355d9f8ae 100644 --- a/quest/m1/remove-live.md +++ b/quest/m1/remove-live.md @@ -36,6 +36,9 @@ the opposite: every gateway opting into `live()`): crate-private `clock::Anchor`, and `SourceMap` if nothing else uses it. fMP4 passthrough stops rewriting `tfdt`. A published `moq-mux` API break, so this targets `dev`. +- The TS importer's re-anchor shift for unflagged loop wraps is not that + anchor: it is shared by the program, bounded on the PCR, and must stay + input-derived. Tests: per importer, a source starting at a large PTS publishes that PTS and a catalog clock that maps it to near the arrival time; a rewind ends the @@ -47,4 +50,3 @@ replace `ts_import_publishes_on_the_broadcast_clock` in moq-cli. - [Broadcast epochs](/quest/m1/broadcast-epoch/README.md) - the new epoch an encoder restart becomes - [Catalog wall clock](/quest/m1/catalog-wall-clock.md) - the PTS-to-wall conversion this relies on, at full precision -- [TS import shared shift](/quest/m1/ts-import-shared-shift.md) - the TS re-anchor shift that must stay input-derived diff --git a/quest/m1/ts-import-shared-shift.md b/quest/m1/ts-import-shared-shift.md deleted file mode 100644 index e5c9190bd3..0000000000 --- a/quest/m1/ts-import-shared-shift.md +++ /dev/null @@ -1,51 +0,0 @@ -# [S] moq import ts: one re-anchor shift per program - -## Goal - -An unflagged loop wrap in `moq import ts` moves audio and video forward by the -same amount, so A/V sync holds across any number of wraps. Today each -elementary stream owns a `Reanchor` and grows its shift to reach its own live -edge ([#3997](https://github.com/moq-dev/moq/pull/3997)). Audio and video -edges sit on their own last frame starts, so each wrap drifts A/V by the -difference in frame durations: about 12 ms at 30 fps with 48 kHz AAC, or -1.7 s a day on a 10-minute loop. - -## Plan - -Decided: a program-level shared shift in `rs/moq-mux/src/container/ts/import.rs`. -Every stream of a program applies the same shift, grown once per wrap by the -largest amount any stream needs to clear its edge, which preserves the source's -inter-stream offsets. A timebase break (PCR discontinuity) clears it for the -whole program, as it already does per stream. - -Guidance: - -- The shift must be known before any stream emits a frame of the new - generation (Codex on #4307): a first stream that grows only enough for - itself leaves a later-arriving stream below its edge, and growing again - then drifts the two. Decided: hold each stream's post-wrap frames until - every live stream of the program has shown its new PTS, then take the - maximum growth once. -- The hold needs its own bound: #3489 adds per-PID counters, not a timeout, - and the catalog `stalled` bit (`Stream::tick`, #3630) covers video only and - runs on the catalog's timer. Decided: bound the hold on the program clock - (PCR advance since the first stream's new generation, not wall time), and - commit the shift over the streams seen so far when it expires. A stream - that returns later applies the committed shift, clamped to its edge as - below. The `Anchor`/`Lane` split behind `live()` in `moq_mux::clock` solves - a similar problem for restarts, but the remove-live quest deletes it, so - copy what helps rather than depending on it. -- A stream whose own edge is still above the shifted timestamp after the - shared growth (its tail ran longer) is the case that forces growing by the - maximum. Landing on its edge is accepted today; keep that trade-off. -- Sections already take the video shift, so cues keep following pictures. - MPEG-2 video (`Stream::Clock`) has no track and so no edge, but it should - read the shared shift too. -- Tests beside the existing loop-wrap tests: a muxed H.264 + AAC loop whose - period is not a multiple of either frame duration keeps the first audio and - video timestamps of each pass at the source offset across three wraps. - -## Related - -- [#3489](/quest/m1/3489-ts-import-stream-liveness.md) - per-PID liveness in the same importer; touches `Stream` but not the shift -- [Remove live()](/quest/m1/remove-live.md) - deletes the restart anchor; a wrap shift stays input-derived diff --git a/quest/m1/ts-programs.md b/quest/m1/ts-programs.md index 14e17a964c..2baad24005 100644 --- a/quest/m1/ts-programs.md +++ b/quest/m1/ts-programs.md @@ -31,7 +31,3 @@ clock; `--program all` publishes both broadcasts, each timed correctly. Update ## Closes - [#4353](https://github.com/moq-dev/moq/issues/4353) - close this issue when the quest finishes - -## Related - -- [TS import shared shift](/quest/m1/ts-import-shared-shift.md) - its per-program shift assumes the single program this settles diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 0cfa35a533..2e597115b5 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -87,6 +87,8 @@ pub struct Import { /// consecutive markers (a repeated flag, a retransmitted clock packet) declare one /// break rather than one each. published: bool, + /// The program's re-anchor shift, and the wrap it is holding media back to measure. + shift: Shift, /// True once a PMT with at least one supported stream has been parsed. initialized: bool, @@ -128,11 +130,13 @@ pub struct Import { /// the PAT into the catalog service record yet (set once; the PAT is stable). identity_recorded: bool, /// Latest video PTS: the media clock used to timestamp private sections, which - /// carry no PES PTS of their own. Unwrapped independently of the video stream, and - /// shifted with it, so a wrap the video re-anchors moves the cues along with it. - /// SPTS scope: one clock for the whole input. Under MPTS every program's video - /// advances it, so a cue could be stamped with another program's PTS. + /// carry no PES PTS of their own. Unwrapped independently of the video stream but not + /// shifted: a section takes the program's shift when it is stamped, so a wrap moves the + /// cues along with the pictures. SPTS scope: one clock for the whole input. Under MPTS + /// every program's video advances it, so a cue could be stamped with another program's PTS. last_pts: Option, + /// The video PID that last set `last_pts`, whose shift the sections take. + clock_pid: Option, media_unwrap: PtsUnwrap, /// The source's mapping onto the broadcast clock, set by [`live`](Self::live). `None` /// publishes the source's unwrapped PTS verbatim. @@ -166,6 +170,7 @@ impl Import { pcr_pid: None, mux_rate: Default::default(), published: false, + shift: Shift::default(), initialized: false, scratch: Vec::new(), synced: false, @@ -178,6 +183,7 @@ impl Import { si, identity_recorded: false, last_pts: None, + clock_pid: None, media_unwrap: PtsUnwrap::default(), anchor: None, } @@ -289,15 +295,17 @@ impl Import { if self.pcr_pid.is_some_and(|p| p.as_u16() == pid) && pkt[1] & 0x80 == 0 { if discontinuity_indicator(&pkt) { self.timebase_break()?; - } else if self.supports_mpegts - && let Some(pcr) = pcr(&pkt) - && self.mux_rate.pcr(pcr) - { - self.record_mux_rate()?; + } else if let Some(pcr) = pcr(&pkt) { + if self.shift.expired(pcr) { + self.release()?; + } + if self.supports_mpegts && self.mux_rate.pcr(pcr) { + self.record_mux_rate()?; + } } } - if let Some(section) = self.sections.get_mut(&pid) { - self.published |= section.packet(&pkt, self.last_pts)?; + if self.sections.contains_key(&pid) { + self.section_packet(pid, &pkt)?; continue; } // Intercept the standalone SI PIDs before the routing gate below drops them: @@ -367,8 +375,11 @@ impl Import { self.scratch.drain(..off); // Cut the snapshot groups for whatever SI committed in this batch. Batching per // decode call (plus the store's own host-clock debounce) coalesces a junction's - // burst of sub-table commits into few groups instead of one per commit. - self.si.flush(self.last_pts.unwrap_or(Timestamp::ZERO), false)?; + // burst of sub-table commits into few groups instead of one per commit. A hold defers + // it too, until the clock it would stamp them with is on the new pass. + if self.shift.hold.is_none() { + self.si.flush(self.clock()?.unwrap_or(Timestamp::ZERO), false)?; + } // Video PIDs that went quiet while the mux kept flowing mark themselves stalled. for stream in self.streams.values_mut() { stream.tick()?; @@ -473,7 +484,6 @@ impl Import { split: h264::Split::new(), import: Box::new(h264::Import::new(track, self.reserve(), self.video_hint())?), unwrap: PtsUnwrap::default(), - reanchor: Reanchor::default(), } } StreamType::H265 => { @@ -484,7 +494,6 @@ impl Import { split: h265::Split::new(), import: Box::new(h265::Import::new(track, self.reserve(), self.video_hint())?), unwrap: PtsUnwrap::default(), - reanchor: Reanchor::default(), } } // Only ADTS-framed AAC (0x0F). 0x11 is LATM/LOAS, which uses a different @@ -495,7 +504,6 @@ impl Import { reserved: Some(self.reserve()), container: self.container.clone(), unwrap: PtsUnwrap::default(), - reanchor: Reanchor::default(), tail: Vec::new(), tail_pts: None, resync: Resync::new(pid.as_u16(), ".aac"), @@ -523,7 +531,6 @@ impl Import { Stream::Opus(Box::new(OpusStream { import: opus::Import::new(track, self.reserve(), config)?, unwrap: PtsUnwrap::default(), - reanchor: Reanchor::default(), })) } StreamType::Mpeg1Video | StreamType::Mpeg2Video => Stream::Clock, @@ -574,7 +581,6 @@ impl Import { tail: Vec::new(), tail_pts: None, resync: Resync::new(pid.as_u16(), descriptor.track_suffix), - reanchor: Reanchor::default(), })) } @@ -652,43 +658,31 @@ impl Import { let is_video = matches!(stream, Stream::H264 { .. } | Stream::H265 { .. } | Stream::Clock); let is_clock = matches!(stream, Stream::Clock); if is_video { - for stream in self.streams.values_mut() { - if let Stream::Aac(audio) = stream { - audio.burst = std::time::Duration::ZERO; - } + // Both for the audio going out now and, replayed in order, for the audio held. + self.end_bursts(); + if let Some(hold) = &mut self.shift.hold { + hold.queue.push(Held::Picture); } // Advance the media clock here, not at flush: unbounded video only // flushes on the next PES, so a SCTE-35 section arriving during this // frame must be timestamped with this frame's PTS ("now"), not the // previous one's. if pes.header.pts.is_some() { - let pts = unwrap_pts( + self.last_pts = unwrap_pts( &mut self.media_unwrap, pes.header.pts.map(|t| t.as_u64()), self.anchor.as_mut(), )?; - let video = match self.streams.get(&pid) { - Some(Stream::H264 { reanchor, import, .. }) => { - Some((reanchor, import.floor(false), import.floor(true))) - } - Some(Stream::H265 { reanchor, import, .. }) => { - Some((reanchor, import.floor(false), import.floor(true))) - } - _ => None, - }; - self.last_pts = match (pts, video) { - (Some(pts), Some((reanchor, edge, floor))) => { - let pts = reanchor.shifted(pts)?; - // Below the live edge is a wrap or restart, which only a keyframe re-anchored to - // the floor continues. That write waits for this PES to end, so stamp sections - // arriving meanwhile there too rather than on the old timeline. - match floor { - Some(floor) if edge.is_some_and(|edge| pts.as_micros() < edge.as_micros()) => Some(floor), - _ => Some(pts), - } - } - (pts, _) => pts, - }; + self.clock_pid = Some(pid); + // Below the closed groups is a wrap or restart, whose write waits for this PES to + // end. Hold from here, so the cues arriving meanwhile wait for the shift too. + if let Some(pts) = self.last_pts + && let Some(edge) = self.streams.get(&pid).and_then(Stream::picture_edge) + && !self.shift.late.contains(&pid) + && shifted(pts, self.shift.of(pid)?)? < edge + { + self.shift.open(); + } } } @@ -729,7 +723,49 @@ impl Import { let Some(pending) = self.pending.remove(&pid) else { return Ok(()); }; + let Some(stream) = self.streams.get_mut(&pid) else { + return Ok(()); + }; + let unit = stream.stamp(pending, self.anchor.as_mut())?; + let need = stream.need(&unit, self.shift.of(pid)?)?; + + // A stream a hold went without applies the shift it committed, clamped to its edge. It + // keeps the clamp, or its next frame would land below the one it just published. + if need.is_some() && self.shift.late.remove(&pid) { + if let Some(need) = need.filter(|need| !need.is_zero()) { + self.shift.lag.insert(pid, need); + } + return self.write(pid, unit); + } + let wrapped = need.is_some_and(|need| need.as_micros() > SLACK); + if wrapped { + self.shift.open(); + } + let Some(hold) = &mut self.shift.hold else { + return self.write(pid, unit); + }; + if !hold.seen.contains(&pid) { + // Still on the pass before the wrap, and on the shift it had: it goes out now, so + // its edge is current when its own wrap is measured. + if !wrapped { + return self.write(pid, unit); + } + hold.seen.insert(pid); + hold.grow = hold.grow.max(need); + } + hold.queue.push(Held::Unit(pid, unit)); + if self + .streams + .iter() + .all(|(pid, stream)| hold.seen.contains(pid) || !stream.live()) + { + self.release()?; + } + Ok(()) + } + /// Publish one stamped PES on the shift committed so far. + fn write(&mut self, pid: Pid, unit: Unit) -> anyhow::Result<()> { let batched = self .streams .values() @@ -737,13 +773,81 @@ impl Import { let Some(stream) = self.streams.get_mut(&pid) else { return Ok(()); }; - self.published |= stream.write(pending, batched, self.anchor.as_mut())?; + self.published |= stream.write(unit, batched, self.shift.of(pid)?)?; // Record the decoded media track's PID + PMT descriptors (language, ...) once // its lazily created track exists, so export can preserve them. self.record_media_track(pid) } + /// Commit an open hold's shift and publish everything it kept back, in arrival order. + fn release(&mut self) -> anyhow::Result<()> { + let live: Vec = self + .streams + .iter() + .filter(|(_, stream)| stream.live()) + .map(|(pid, _)| *pid) + .collect(); + let Some(queue) = self.shift.commit(live)? else { + return Ok(()); + }; + for held in queue { + match held { + Held::Unit(pid, unit) => self.write(pid, unit)?, + Held::Section(pid, section, clock) => self.emit_section(pid, section, clock)?, + Held::Picture => self.end_bursts(), + } + } + Ok(()) + } + + /// A video PES started: the audio completed since the last one is one burst. + fn end_bursts(&mut self) { + for stream in self.streams.values_mut() { + if let Stream::Aac(audio) = stream { + audio.burst = std::time::Duration::ZERO; + } + } + } + + /// The media clock on the published timeline, moved as the video that set it. + fn clock(&self) -> anyhow::Result> { + self.stamp_clock(self.last_pts) + } + + fn stamp_clock(&self, clock: Option) -> anyhow::Result> { + let shift = match self.clock_pid { + Some(pid) => self.shift.of(pid)?, + None => self.shift.value, + }; + clock.map(|pts| shifted(pts, shift)).transpose() + } + + /// Feed one TS packet on a section-framed PID to its reassembler, publishing or holding + /// each section it completes. + fn section_packet(&mut self, pid: u16, pkt: &[u8; 188]) -> anyhow::Result<()> { + let mut sections = Vec::new(); + if let Some(stream) = self.sections.get_mut(&pid) { + stream.reassembler.push(pkt, &mut sections); + } + for section in sections { + match &mut self.shift.hold { + Some(hold) => hold.queue.push(Held::Section(pid, section, self.last_pts)), + None => self.emit_section(pid, section, self.last_pts)?, + } + } + Ok(()) + } + + fn emit_section(&mut self, pid: u16, section: Vec, clock: Option) -> anyhow::Result<()> { + let pts = self.stamp_clock(clock)?; + if let Some(stream) = self.sections.get_mut(&pid) { + stream.emit(section, pts)?; + self.published = true; + } + Ok(()) + } + /// The packet chain on `pid` was cut: salvage the truncated PES only where its bytes /// stand on their own, drop whatever is left mid-unit, and require the next frame to /// prove its boundary. @@ -779,14 +883,18 @@ impl Import { for pid in self.streams.keys().copied().collect::>() { self.broken(pid)?; } + // What a hold kept back ran on the old clock too, so it closes the old timeline. + self.release()?; for stream in self.streams.values_mut() { stream.discontinuity(self.published)?; } for section in self.sections.values_mut() { section.discontinuity(self.published)?; } + self.shift = Shift::default(); self.media_unwrap.discontinuity(); self.last_pts = None; + self.clock_pid = None; self.published = false; if self.mux_rate.discontinuity() { self.record_mux_rate()?; @@ -874,6 +982,7 @@ impl Import { /// Close the current group on every track and reopen at `sequence`. pub fn seek(&mut self, sequence: u64) -> anyhow::Result<()> { + self.release()?; for stream in self.streams.values_mut() { stream.seek(sequence)?; } @@ -889,13 +998,14 @@ impl Import { for pid in pids { self.flush(pid)?; } + self.release()?; for stream in self.streams.values_mut() { stream.finish()?; } for section in self.sections.values_mut() { section.finish()?; } - self.si.finish(self.last_pts.unwrap_or(Timestamp::ZERO))?; + self.si.finish(self.clock()?.unwrap_or(Timestamp::ZERO))?; Ok(()) } @@ -999,6 +1109,14 @@ struct Pending { data_len: Option, } +/// One PES on the source's timeline: unwrapped, and translated when live, but not yet shifted. +enum Unit { + /// The access units a video PES split into, and its reorder delay. + Video(Vec, Option), + /// Any other PES, and its PTS. + Pes(Pending, Option), +} + impl Pending { /// A PES carrying nothing, used to drain a stream's carried tail at end of stream. fn empty() -> Self { @@ -1137,20 +1255,9 @@ impl SectionStream { }) } - /// Consume one 188-byte TS packet, publishing each completed section. `pts` is - /// the current media clock used to timestamp a section (its arrival on the - /// timeline; the splice time itself is inside the section bytes), if one is running. - fn packet(&mut self, pkt: &[u8], pts: Option) -> anyhow::Result { - let mut sections = Vec::new(); - self.reassembler.push(pkt, &mut sections); - let published = !sections.is_empty(); - for section in sections { - self.emit(section, pts)?; - } - Ok(published) - } - - /// Publish one complete section as a frame in its own group. + /// Publish one complete section as a frame in its own group. `pts` is the media clock + /// used to timestamp it (its arrival on the timeline; the splice time itself is inside + /// the section bytes), if one is running. /// /// The clock is already on the video's re-anchored timeline, so it only steps back by a /// reorder: a B-frame presents before the P-frame decoded ahead of it. Such a cue lands on @@ -1203,7 +1310,6 @@ struct VerbatimStream { track: crate::container::Producer, entry: VerbatimEntry, unwrap: PtsUnwrap, - reanchor: Reanchor, /// Whether the PES stream_id has been recorded into the catalog yet (once). stream_id_recorded: bool, } @@ -1232,14 +1338,13 @@ impl VerbatimStream { track, entry, unwrap: PtsUnwrap::default(), - reanchor: Reanchor::default(), stream_id_recorded: false, }) } /// Publish one reassembled PES payload verbatim, in its own group, stamped with /// its PTS (or the live edge when the PES carried none). - fn write(&mut self, pending: Pending, anchor: Option<&mut crate::clock::Anchor>) -> anyhow::Result { + fn write(&mut self, pending: Pending, pts: Option, shift: Option) -> anyhow::Result { // Record the original PES stream_id once, from the first PES, so export // re-emits the stream under its real id (e.g. 0xBD for teletext/DVB AC-3). if !self.stream_id_recorded { @@ -1253,8 +1358,8 @@ impl VerbatimStream { } let edge = self.track.live_edge(); - let pts = match unwrap_pts(&mut self.unwrap, pending.pts, anchor)? { - Some(pts) => self.reanchor.apply(pts, edge)?, + let pts = match pts { + Some(pts) => lift(pts, shift, edge)?, // No clock to shift, so land on the edge and leave the shift alone. None => edge.unwrap_or(Timestamp::ZERO), }; @@ -1271,7 +1376,6 @@ impl VerbatimStream { fn discontinuity(&mut self, published: bool) -> anyhow::Result<()> { self.unwrap.discontinuity(); - self.reanchor.discontinuity(); if published { self.track.discontinuity()?; } @@ -1547,13 +1651,11 @@ enum Stream { split: h264::Split, import: Box, unwrap: PtsUnwrap, - reanchor: Reanchor, }, H265 { split: h265::Split, import: Box, unwrap: PtsUnwrap, - reanchor: Reanchor, }, Aac(Box>), Opus(Box), @@ -1567,27 +1669,108 @@ enum Stream { } impl Stream { - fn write( - &mut self, - pending: Pending, - batched: bool, - anchor: Option<&mut crate::clock::Anchor>, - ) -> anyhow::Result { - match self { - Stream::H264 { - split, - import, - unwrap, - reanchor, - } => { + /// Put one PES on the source's timeline, as it arrives. + fn stamp(&mut self, pending: Pending, anchor: Option<&mut crate::clock::Anchor>) -> anyhow::Result { + Ok(match self { + Stream::H264 { split, unwrap, .. } => { let reorder = reorder_delay(pending.pts, pending.dts); let pts = unwrap_pts(unwrap, pending.pts, anchor)?; // Each PES is one access unit, so flush to emit it immediately. let mut frames = split.decode(&pending.data, pts)?; frames.extend(split.flush(pts)?); + Unit::Video(frames, reorder) + } + Stream::H265 { split, unwrap, .. } => { + let reorder = reorder_delay(pending.pts, pending.dts); + let pts = unwrap_pts(unwrap, pending.pts, anchor)?; + let mut frames = split.decode(&pending.data, pts)?; + frames.extend(split.flush(pts)?); + Unit::Video(frames, reorder) + } + Stream::Aac(stream) => { + let pts = unwrap_pts(&mut stream.unwrap, pending.pts, anchor)?; + Unit::Pes(pending, pts) + } + Stream::Opus(stream) => { + let pts = unwrap_pts(&mut stream.unwrap, pending.pts, anchor)?; + Unit::Pes(pending, pts) + } + Stream::Legacy(stream) => { + let pts = unwrap_pts(&mut stream.unwrap, pending.pts, anchor)?; + Unit::Pes(pending, pts) + } + Stream::Verbatim(stream) => { + let pts = unwrap_pts(&mut stream.unwrap, pending.pts, anchor)?; + Unit::Pes(pending, pts) + } + Stream::Clock | Stream::Ignored => Unit::Pes(pending, None), + }) + } + + /// How much more `shift` must grow for `unit` to clear this stream's edge, or `None` when + /// the unit carries no timestamp to tell. + fn need(&self, unit: &Unit, shift: Option) -> anyhow::Result> { + let (pts, edge) = match (self, unit) { + (Stream::H264 { import, .. }, Unit::Video(frames, _)) => match frames.first() { + Some(frame) => (frame.timestamp, import.floor(frame.keyframe)), + None => return Ok(None), + }, + (Stream::H265 { import, .. }, Unit::Video(frames, _)) => match frames.first() { + Some(frame) => (frame.timestamp, import.floor(frame.keyframe)), + None => return Ok(None), + }, + (stream, Unit::Pes(_, Some(pts))) => (*pts, stream.edge()), + _ => return Ok(None), + }; + Ok(Some(clearance(shifted(pts, shift)?, edge)?)) + } + + /// The lowest timestamp the next PES may start at, for a stream that is not video. + fn edge(&self) -> Option { + match self { + // A carried tail publishes ahead of the PES that completes it. + Stream::Aac(stream) => stream + .import + .as_ref() + .and_then(aac::Import::live_edge) + .max(stream.tail_pts), + Stream::Legacy(stream) => stream + .import + .as_ref() + .and_then(legacy::Import::live_edge) + .max(stream.tail_pts), + Stream::Opus(stream) => stream.import.live_edge(), + Stream::Verbatim(stream) => stream.track.live_edge(), + Stream::H264 { .. } | Stream::H265 { .. } | Stream::Clock | Stream::Ignored => None, + } + } + + /// Whether this stream has published, and so has an edge a wrap must clear. + fn live(&self) -> bool { + match self { + Stream::H264 { import, .. } => import.floor(true).is_some(), + Stream::H265 { import, .. } => import.floor(true).is_some(), + stream => stream.edge().is_some(), + } + } + + /// The edge a video PES starting below can only continue from as a wrap or restart: the + /// groups already closed, which a B-frame never dips under. + fn picture_edge(&self) -> Option { + match self { + Stream::H264 { import, .. } => import.floor(false), + Stream::H265 { import, .. } => import.floor(false), + _ => None, + } + } + + /// Publish `unit` moved by `shift`, landing on the edge where that still leaves it below. + fn write(&mut self, unit: Unit, batched: bool, shift: Option) -> anyhow::Result { + match (self, unit) { + (Stream::H264 { import, .. }, Unit::Video(frames, reorder)) => { let mut published = false; for mut frame in frames { - frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; + frame.timestamp = lift(frame.timestamp, shift, import.floor(frame.keyframe))?; published |= skip_missing_keyframe(import.decode([frame]))?; } // After decode, so the track (and its catalog rendition) exists. @@ -1596,20 +1779,10 @@ impl Stream { } Ok(published) } - Stream::H265 { - split, - import, - unwrap, - reanchor, - } => { - let reorder = reorder_delay(pending.pts, pending.dts); - let pts = unwrap_pts(unwrap, pending.pts, anchor)?; - // Each PES is one access unit, so flush to emit it immediately. - let mut frames = split.decode(&pending.data, pts)?; - frames.extend(split.flush(pts)?); + (Stream::H265 { import, .. }, Unit::Video(frames, reorder)) => { let mut published = false; for mut frame in frames { - frame.timestamp = reanchor.apply(frame.timestamp, import.floor(frame.keyframe))?; + frame.timestamp = lift(frame.timestamp, shift, import.floor(frame.keyframe))?; published |= skip_missing_keyframe(import.decode([frame]))?; } if let Some(reorder) = reorder { @@ -1617,11 +1790,12 @@ impl Stream { } Ok(published) } - Stream::Aac(stream) => stream.write(pending, batched, anchor), - Stream::Opus(stream) => stream.write(pending, anchor), - Stream::Legacy(stream) => stream.write(pending, anchor), - Stream::Verbatim(stream) => stream.write(pending, anchor), - Stream::Clock | Stream::Ignored => Ok(false), + (Stream::Aac(stream), Unit::Pes(pending, pts)) => stream.write(pending, pts, batched, shift), + (Stream::Opus(stream), Unit::Pes(pending, pts)) => stream.write(pending, pts, shift), + (Stream::Legacy(stream), Unit::Pes(pending, pts)) => stream.write(pending, pts, shift), + (Stream::Verbatim(stream), Unit::Pes(pending, pts)) => stream.write(pending, pts, shift), + // Held across a PMT that remapped the PID: the codec it was stamped for is gone. + _ => Ok(false), } } @@ -1675,27 +1849,15 @@ impl Stream { /// next PTS against a sample from the timebase that just ended. fn discontinuity(&mut self, published: bool) -> anyhow::Result<()> { match self { - Stream::H264 { - import, - unwrap, - reanchor, - .. - } => { + Stream::H264 { import, unwrap, .. } => { unwrap.discontinuity(); - reanchor.discontinuity(); if published { import.discontinuity()?; } Ok(()) } - Stream::H265 { - import, - unwrap, - reanchor, - .. - } => { + Stream::H265 { import, unwrap, .. } => { unwrap.discontinuity(); - reanchor.discontinuity(); if published { import.discontinuity()?; } @@ -2017,7 +2179,6 @@ struct AacStream { /// The container this importer publishes decoded renditions with. container: hang::catalog::Container, unwrap: PtsUnwrap, - reanchor: Reanchor, /// Partial frame left at the end of the previous PES. ISO 13818-1 doesn't require /// audio frames to align with PES boundaries, so a legitimate mux can split one. tail: Vec, @@ -2033,11 +2194,10 @@ impl AacStream { fn write( &mut self, pending: Pending, + pes_base: Option, batched: bool, - anchor: Option<&mut crate::clock::Anchor>, + shift: Option, ) -> anyhow::Result { - let pes_base = unwrap_pts(&mut self.unwrap, pending.pts, anchor)?; - // Prepend the partial frame left by the previous PES, if any. let carried = self.tail.len(); let joined; @@ -2055,7 +2215,7 @@ impl AacStream { let mut pts = if carried > 0 { self.tail_pts.take() } else { - self.reanchor(pes_base)? + self.reanchor(pes_base, shift)? }; let mut in_tail = carried > 0; @@ -2067,7 +2227,7 @@ impl AacStream { let mut fallback = None; while offset + adts::MIN_HEADER_LEN <= data.len() { if in_tail && offset >= carried { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; in_tail = false; } @@ -2132,7 +2292,7 @@ impl AacStream { match self.resync.recover(data, offset, &SyncWord::ADTS) { Recover::At(next) => { if in_tail { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; in_tail = false; } offset = next; @@ -2150,7 +2310,7 @@ impl AacStream { } None => { if in_tail { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; in_tail = false; } offset = next; @@ -2209,7 +2369,7 @@ impl AacStream { // with the PTS it should carry. if offset < data.len() { if in_tail && offset >= carried { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; } self.tail = data[offset..].to_vec(); self.tail_pts = pts; @@ -2220,15 +2380,14 @@ impl AacStream { /// The PES PTS on the published timeline. Resolved where each frame starts rather than /// once per PES, so the frames a carried tail publishes first count toward the edge. - fn reanchor(&mut self, pts: Option) -> anyhow::Result> { + fn reanchor(&self, pts: Option, shift: Option) -> anyhow::Result> { let edge = self.import.as_ref().and_then(aac::Import::live_edge); - pts.map(|pts| self.reanchor.apply(pts, edge)).transpose() + pts.map(|pts| lift(pts, shift, edge)).transpose() } fn discontinuity(&mut self, published: bool) -> anyhow::Result<()> { self.desync(); self.unwrap.discontinuity(); - self.reanchor.discontinuity(); if published && let Some(import) = &mut self.import { import.discontinuity()?; } @@ -2262,8 +2421,8 @@ impl AacStream { // of it. if !self.tail.is_empty() && self.import.is_some() { self.resync.drain(); - // No PTS to translate, so no mapping needed. - self.write(Pending::empty(), true, None)?; + // No PTS to shift: the tail keeps the one it was cut with. + self.write(Pending::empty(), None, true, None)?; } // A partial frame at end of stream isn't emissible; drop it, but leave a trace for // diagnosing truncated captures. @@ -2295,14 +2454,12 @@ impl AacStream { struct OpusStream { import: opus::Import, unwrap: PtsUnwrap, - reanchor: Reanchor, } impl OpusStream { - fn write(&mut self, pending: Pending, anchor: Option<&mut crate::clock::Anchor>) -> anyhow::Result { - let base = unwrap_pts(&mut self.unwrap, pending.pts, anchor)?; + fn write(&mut self, pending: Pending, base: Option, shift: Option) -> anyhow::Result { let edge = self.import.live_edge(); - let base = base.map(|base| self.reanchor.apply(base, edge)).transpose()?; + let base = base.map(|base| lift(base, shift, edge)).transpose()?; let data = &pending.data; let mut offset = 0; @@ -2339,7 +2496,6 @@ impl OpusStream { fn discontinuity(&mut self, published: bool) -> anyhow::Result<()> { self.unwrap.discontinuity(); - self.reanchor.discontinuity(); if published { self.import.discontinuity()?; } @@ -2451,13 +2607,16 @@ struct LegacyStream { /// only covers frames that begin in that PES. tail_pts: Option, resync: Resync, - reanchor: Reanchor, } impl LegacyStream { - fn write(&mut self, pending: Pending, anchor: Option<&mut crate::clock::Anchor>) -> anyhow::Result { + fn write( + &mut self, + pending: Pending, + pes_base: Option, + shift: Option, + ) -> anyhow::Result { let mut published = false; - let pes_base = unwrap_pts(&mut self.unwrap, pending.pts, anchor)?; // Prepend the partial frame left by the previous PES, if any. let carried = self.tail.len(); @@ -2479,7 +2638,7 @@ impl LegacyStream { let mut pts = if carried > 0 { self.tail_pts.take() } else { - self.reanchor(pes_base)? + self.reanchor(pes_base, shift)? }; let mut in_tail = carried > 0; @@ -2489,7 +2648,7 @@ impl LegacyStream { let mut fallback = None; while offset + self.descriptor.min_header_len <= data.len() { if in_tail && offset >= carried { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; in_tail = false; } @@ -2566,7 +2725,7 @@ impl LegacyStream { match self.resync.recover(data, offset, &self.descriptor.into()) { Recover::At(next) => { if in_tail { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; in_tail = false; } offset = next; @@ -2584,7 +2743,7 @@ impl LegacyStream { } None => { if in_tail { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; in_tail = false; } offset = next; @@ -2633,7 +2792,7 @@ impl LegacyStream { // PES, with the PTS it should carry. if offset < data.len() { if in_tail && offset >= carried { - pts = self.reanchor(pes_base)?; + pts = self.reanchor(pes_base, shift)?; } self.tail = data[offset..].to_vec(); self.tail_pts = pts; @@ -2644,15 +2803,14 @@ impl LegacyStream { /// The PES PTS on the published timeline. Resolved where each frame starts rather than /// once per PES, so the frames a carried tail publishes first count toward the edge. - fn reanchor(&mut self, pts: Option) -> anyhow::Result> { + fn reanchor(&self, pts: Option, shift: Option) -> anyhow::Result> { let edge = self.import.as_ref().and_then(legacy::Import::live_edge); - pts.map(|pts| self.reanchor.apply(pts, edge)).transpose() + pts.map(|pts| lift(pts, shift, edge)).transpose() } fn discontinuity(&mut self, published: bool) -> anyhow::Result<()> { self.desync(); self.unwrap.discontinuity(); - self.reanchor.discontinuity(); if published && let Some(import) = &mut self.import { import.discontinuity()?; } @@ -2685,8 +2843,8 @@ impl LegacyStream { // of it. if !self.tail.is_empty() && self.import.is_some() { self.resync.drain(); - // No PTS to translate, so no mapping needed. - self.write(Pending::empty(), None)?; + // No PTS to shift: the tail keeps the one it was cut with. + self.write(Pending::empty(), None, None)?; } // A partial frame at end of stream isn't emissible verbatim; drop it, but // leave a trace for diagnosing truncated captures. @@ -2820,51 +2978,132 @@ impl PtsUnwrap { } } -/// Keeps one track's timeline at or past its live edge when the source clock steps back. +/// How far the program clock runs past the first stream of a wrap before its hold gives up on +/// the streams still missing. The T-STD delivers a unit at most a second before it decodes +/// (ISO 13818-1 2.4.2.6), so streams starting together over a wrap arrive within about that. +const HOLD: u64 = 27_000_000; + +/// The most, in microseconds, a PES may overlap its stream's edge and still be clamped where it +/// lands rather than measured as a wrap. Rounding a PTS to 90 kHz, or a muxer's PTS jitter, +/// overlaps by a tick or so; a wrap or restart steps back by a period. +const SLACK: u128 = 10_000; + +/// The re-anchor shift of the program, shared by every stream so a wrap moves them together. /// -/// A content join, a playout loop wrap, or a restarted encoder lands a timestamp below what -/// the track already published, and the producer refuses a rewind. TS ingest trusts its -/// source, so it shifts forward instead, by just enough to reach the edge. The shift only -/// grows: a looping source wraps again every period, and each wrap lands below the edge the -/// last one reached. A timebase break clears it, since the new clock is measured against the -/// edge afresh. +/// A content join, a playout loop wrap, or a restarted encoder lands a timestamp below what a +/// track already published, and the producer refuses a rewind. TS ingest trusts its source, so +/// it shifts forward instead. Each stream's tail ends on its own last frame, so each needs a +/// different amount to clear its edge, and growing each by its own need would move them apart +/// at every wrap. So the shift is shared and grows by the largest need, measured before any +/// stream publishes the new pass: from the first stream below its edge, each stream's new pass, +/// and every section, is held until each live stream has shown its new PTS, or the program clock +/// has run [`HOLD`] past it. The shift only grows, since a looping source wraps again every +/// period. A timebase break clears it, since the new clock is measured against the edges afresh. #[derive(Default)] -struct Reanchor { - shift: Option, +struct Shift { + value: Option, + hold: Option, + /// Live streams a hold committed without. Their next PES applies the shift, clamped to their + /// edge, rather than holding the program again. + late: HashSet, + /// What the clamp added for a late stream that came back below its edge: it stays that far + /// behind the rest of the program until the next timebase break. + lag: HashMap, } -impl Reanchor { - /// `pts` on the published timeline, as shifted so far. - fn shifted(&self, pts: Timestamp) -> anyhow::Result { - Ok(match self.shift { - Some(shift) => pts.checked_add(shift.convert(pts.scale())?)?, - None => pts, +/// A wrap being measured. +#[derive(Default)] +struct Hold { + /// The most any stream needs to clear its edge. + grow: Option, + /// The streams whose new PTS has arrived. + seen: HashSet, + /// The first PCR since the hold opened. + pcr: Option, + queue: Vec, +} + +/// What a [`Hold`] keeps back, in arrival order. +enum Held { + Unit(Pid, Unit), + /// A reassembled section and the video clock it arrived on. + Section(u16, Vec, Option), + /// A video PES started, which ends each audio stream's burst. + Picture, +} + +impl Shift { + /// The shift `pid` publishes on. + fn of(&self, pid: Pid) -> anyhow::Result> { + Ok(match (self.value, self.lag.get(&pid)) { + (Some(value), Some(lag)) => Some(value.convert(lag.scale())?.checked_add(*lag)?), + (value, lag) => value.or(lag.copied()), }) } - /// `pts` on the published timeline, at or past `edge`. - fn apply(&mut self, pts: Timestamp, edge: Option) -> anyhow::Result { - let scale = pts.scale(); - let pts = self.shifted(pts)?; - let Some(edge) = edge.filter(|edge| pts.as_micros() < edge.as_micros()) else { - return Ok(pts); + fn open(&mut self) { + self.hold.get_or_insert_with(Hold::default); + } + + /// Whether an open hold has run [`HOLD`] on the program clock. + fn expired(&mut self, pcr: u64) -> bool { + const FIELD: u64 = 300 << 33; + let Some(hold) = &mut self.hold else { + return false; }; - // Round up to the source's scale: landing a tick short of the edge is still a rewind. - let mut target = Timestamp::new(u64::try_from(edge.as_scale(scale))?, scale)?; - if target.as_micros() < edge.as_micros() { - target = Timestamp::new(target.value() + 1, scale)?; + let start = *hold.pcr.get_or_insert(pcr); + let advance = (pcr + FIELD - start) % FIELD; + // The clock stepped back under the hold, as a second wrap does: count from there. + if advance > FIELD / 2 { + hold.pcr = Some(pcr); + return false; } - let grow = target.checked_sub(pts)?; - self.shift = Some(match self.shift { - Some(shift) => shift.convert(scale)?.checked_add(grow)?, - None => grow, - }); - Ok(target) + advance >= HOLD } - fn discontinuity(&mut self) { - self.shift = None; + /// Close the hold, growing the shift by the largest need it saw, and return what it held. + fn commit(&mut self, live: impl IntoIterator) -> anyhow::Result>> { + let Some(hold) = self.hold.take() else { + return Ok(None); + }; + if let Some(grow) = hold.grow.filter(|grow| !grow.is_zero()) { + self.value = Some(match self.value { + Some(value) => value.convert(grow.scale())?.checked_add(grow)?, + None => grow, + }); + } + self.late + .extend(live.into_iter().filter(|pid| !hold.seen.contains(pid))); + Ok(Some(hold.queue)) + } +} + +/// `pts` moved by `shift`. +fn shifted(pts: Timestamp, shift: Option) -> anyhow::Result { + Ok(match shift { + Some(shift) => pts.checked_add(shift.convert(pts.scale())?)?, + None => pts, + }) +} + +/// How far `pts` must move to reach `edge`: zero when it already has. +fn clearance(pts: Timestamp, edge: Option) -> anyhow::Result { + let scale = pts.scale(); + let Some(edge) = edge.filter(|edge| pts.as_micros() < edge.as_micros()) else { + return Ok(Timestamp::new(0, scale)?); + }; + // Round up to the source's scale: landing a tick short of the edge is still a rewind. + let mut target = Timestamp::new(u64::try_from(edge.as_scale(scale))?, scale)?; + if target.as_micros() < edge.as_micros() { + target = Timestamp::new(target.value() + 1, scale)?; } + Ok(target.checked_sub(pts)?) +} + +/// `pts` moved by `shift`, landing on `edge` when that still leaves it below. +fn lift(pts: Timestamp, shift: Option, edge: Option) -> anyhow::Result { + let pts = shifted(pts, shift)?; + Ok(pts.checked_add(clearance(pts, edge)?)?) } /// A cloneable, refillable [`Read`] source backed by a shared buffer. @@ -5764,6 +6003,263 @@ mod test { ); } + const AAC: u16 = 0x0060; + /// 1024 samples at 48 kHz, in 90 kHz ticks. + const AAC_FRAME: u64 = 1024 * 90_000 / 48_000; + /// How far the source puts the first audio frame after the first picture. + const AV_OFFSET: u64 = 600; + + /// One pass of a looping H.264 + AAC file from `base`, muxed in decode order: `gops` GOPs + /// of video, and `audio` AAC frames from `AV_OFFSET` past the first picture. Two GOPs and + /// fourteen frames make a pass whose audio outruns its video by a length on neither grid. + fn av_pass(mux: &mut Mux, base: u64, gops: u64, audio: u64) { + let mut units = Vec::new(); + for g in 0..gops { + let b = base + g * 4 * FRAME; + for (keyframe, pts, dts) in [ + (true, b + FRAME, Some(b)), + (false, b + 4 * FRAME, Some(b + FRAME)), + (false, b + 2 * FRAME, None), + (false, b + 3 * FRAME, None), + ] { + units.push((dts.unwrap_or(pts), Some((keyframe, dts)), pts)); + } + } + for i in 0..audio { + let pts = base + FRAME + AV_OFFSET + i * AAC_FRAME; + units.push((pts, None, pts)); + } + units.sort_by_key(|(at, _, _)| *at); + for (i, (_, video, pts)) in units.into_iter().enumerate() { + let packet = match video { + Some((keyframe, dts)) => video_pes(VIDEO, mux.cc(VIDEO), pts, dts, &annexb_au(keyframe)), + None => audio_pes_packet(AAC, mux.cc(AAC), pts, &adts_frame(17, 0xA0 | (i & 0x0f) as u8)), + }; + mux.out.extend_from_slice(&packet); + } + } + + #[allow(clippy::type_complexity)] + fn av_import() -> ( + moq_net::broadcast::Consumer, + crate::catalog::Producer, + super::Import, + ) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let import = super::Import::new(broadcast, catalog.reserve()); + (consumer, catalog, import) + } + + async fn av_frames( + consumer: &moq_net::broadcast::Consumer, + catalog: &crate::catalog::Producer, + ) -> (Vec, Vec) { + let snapshot = catalog.snapshot(); + let video = snapshot.video.renditions.keys().next().unwrap().clone(); + let audio = snapshot.audio.renditions.keys().next().unwrap().clone(); + ( + read_track(consumer, &video, crate::container::Kind::Video).await, + read_track(consumer, &audio, crate::container::Kind::Audio).await, + ) + } + + /// The first audio frame's time less the first picture's, in microseconds. + fn av_offset(video: &crate::container::Frame, audio: &crate::container::Frame) -> i128 { + audio.timestamp.as_micros() as i128 - video.timestamp.as_micros() as i128 + } + + /// Each track of a muxed loop ends on its own last frame, so each needs a different shift to + /// clear its edge. The program moves by the largest, together, so every pass starts with its + /// audio where the source put it against the video. + #[tokio::test(start_paused = true)] + async fn muxed_loop_keeps_av_offset_across_wraps() { + let (consumer, catalog, mut import) = av_import(); + let mut mux = Mux { + out: synth_pmt(&[(StreamType::H264, VIDEO), (StreamType::AdtsAac, AAC)], false), + ..Default::default() + }; + for _ in 0..4 { + av_pass(&mut mux, 90_000, 2, 14); + } + import.decode(&mux.out).expect("the wraps must not end the import"); + import.finish().unwrap(); + let (video, audio) = av_frames(&consumer, &catalog).await; + assert_eq!((video.len(), audio.len()), (32, 56), "every pass publishes"); + assert_forward(&video); + assert_forward(&audio); + + let source = (AV_OFFSET * 1_000_000 / 90_000) as i128; + let offsets: Vec = (0..4) + .map(|pass| av_offset(&video[8 * pass], &audio[14 * pass])) + .collect(); + // Within a microsecond: the wire is in microseconds and the source in 90 kHz ticks. + assert!( + offsets.iter().all(|offset| offset.abs_diff(source) <= 1), + "a wrap moved the audio against the video: {offsets:?}, source {source}" + ); + } + + /// A PES of the pass before the wrap that completes after another stream's wrap has opened + /// the hold still belongs to that pass: it goes out on the old shift, and its stream's wrap is + /// measured from the edge it leaves. + #[tokio::test(start_paused = true)] + async fn muxed_loop_publishes_the_old_pass_through_a_hold() { + let (consumer, catalog, mut import) = av_import(); + let mut mux = Mux { + out: synth_pmt(&[(StreamType::H264, VIDEO), (StreamType::AdtsAac, AAC)], false), + ..Default::default() + }; + av_pass(&mut mux, 90_000, 2, 13); + let last = adts_frame(17, 0xCD); + let pts = 90_000 + FRAME + AV_OFFSET + 13 * AAC_FRAME; + let cc = mux.cc(AAC); + mux.out + .extend_from_slice(&audio_pes_open(AAC, cc, pts, last.len(), &last[..10])); + let cc = mux.cc(AAC); + // The rest of it arrives after the next pass's keyframe, the first packet of that pass. + let wrap = mux.out.len() + 188; + for _ in 0..3 { + av_pass(&mut mux, 90_000, 2, 14); + } + let rest = mux.out.split_off(wrap); + mux.out.extend_from_slice(&ts_continuation(AAC, cc, &last[10..])); + mux.out.extend_from_slice(&rest); + + import.decode(&mux.out).expect("the wraps must not end the import"); + import.finish().unwrap(); + let (video, audio) = av_frames(&consumer, &catalog).await; + assert_eq!((video.len(), audio.len()), (32, 56), "every pass publishes"); + assert_forward(&video); + assert_forward(&audio); + let frame = (AAC_FRAME * 1_000_000 / 90_000) as u128; + assert!( + (audio[13].timestamp.as_micros() - audio[12].timestamp.as_micros()).abs_diff(frame) <= 1, + "the old pass's last frame took the new shift" + ); + let source = (AV_OFFSET * 1_000_000 / 90_000) as i128; + let offsets: Vec = (0..4) + .map(|pass| av_offset(&video[8 * pass], &audio[14 * pass])) + .collect(); + assert!( + offsets.iter().all(|offset| offset.abs_diff(source) <= 1), + "a wrap moved the audio against the video: {offsets:?}, source {source}" + ); + } + + /// A stream whose new pass has not shown a second of program clock after the first one does + /// no longer holds the program: the shift commits without it, and it applies that shift when + /// it returns, clamped to its edge and keeping the clamp. + #[tokio::test(start_paused = true)] + async fn wrap_hold_expires_on_the_program_clock() { + // Eleven frames end the audio short of the video, so the shift clears its edge. + let (video, audio) = late_audio_after_a_wrap(11).await; + assert_forward(&audio); + let source = (AV_OFFSET * 1_000_000 / 90_000) as i128; + assert!( + av_offset(&video[8], &audio[11]).abs_diff(source) <= 1, + "the late audio did not take the committed shift" + ); + + // Fourteen run past it, so the shift the video committed leaves the audio below its edge. + let (_, audio) = late_audio_after_a_wrap(14).await; + assert_forward(&audio); + assert_eq!( + audio[14].timestamp, audio[13].timestamp, + "the late audio did not land on its edge" + ); + let frame = (AAC_FRAME * 1_000_000 / 90_000) as u128; + let steps: Vec = audio[14..] + .windows(2) + .map(|pair| pair[1].timestamp.as_micros() - pair[0].timestamp.as_micros()) + .collect(); + assert!( + steps.iter().all(|step| step.abs_diff(frame) <= 1), + "the clamp was not kept, so the pass piled onto its edge: {steps:?}" + ); + } + + /// A pass of `audio` frames, then a wrap whose audio arrives after the hold has expired. + async fn late_audio_after_a_wrap(audio: u64) -> (Vec, Vec) { + let (consumer, catalog, mut import) = av_import(); + let mut mux = Mux { + out: synth_pmt(&[(StreamType::H264, VIDEO), (StreamType::AdtsAac, AAC)], false), + ..Default::default() + }; + av_pass(&mut mux, 90_000, 2, audio); + // The wrap's video, and a second and a bit of clock, with its audio still to come. + mux.gops(VIDEO, 90_000, 2); + mux.out.extend_from_slice(&pcr_packet(VIDEO, 90_000 * 300)); + mux.out + .extend_from_slice(&pcr_packet(VIDEO, 90_000 * 300 + super::HOLD / 2)); + import.decode(&mux.out).unwrap(); + let (video, _) = av_frames(&consumer, &catalog).await; + assert_eq!(video.len(), 8, "half a second of clock released the hold early"); + + mux.out.clear(); + mux.out + .extend_from_slice(&pcr_packet(VIDEO, 90_000 * 300 + super::HOLD)); + import.decode(&mux.out).unwrap(); + let (video, _) = av_frames(&consumer, &catalog).await; + assert_eq!( + video.len(), + 16, + "the video waited past the hold for audio that never came" + ); + + mux.out.clear(); + for i in 0..audio { + let cc = mux.cc(AAC); + let pts = 90_000 + FRAME + AV_OFFSET + i * AAC_FRAME; + mux.out + .extend_from_slice(&audio_pes_packet(AAC, cc, pts, &adts_frame(17, 0xB0 | i as u8))); + } + import.decode(&mux.out).unwrap(); + import.finish().unwrap(); + let (video, frames) = av_frames(&consumer, &catalog).await; + assert_eq!(frames.len() as u64, 2 * audio, "every pass publishes"); + (video, frames) + } + + /// A timebase break while a wrap is held publishes what was held on the old clock and clears + /// the shift, which the new clock then measures afresh for the whole program. + #[tokio::test(start_paused = true)] + async fn wrap_hold_released_by_a_timebase_break() { + let (consumer, catalog, mut import) = av_import(); + let mut mux = Mux { + out: synth_pmt(&[(StreamType::H264, VIDEO), (StreamType::AdtsAac, AAC)], false), + ..Default::default() + }; + av_pass(&mut mux, 90_000, 2, 14); + // The wrap's first GOP, held for its audio when the break arrives. + mux.gops(VIDEO, 90_000, 1); + import.decode(&mux.out).unwrap(); + let (video, _) = av_frames(&consumer, &catalog).await; + assert_eq!(video.len(), 8, "the wrap was not held"); + + mux.out.clear(); + mux.out.extend_from_slice(&clock_break_packet(VIDEO)); + av_pass(&mut mux, 90_000, 2, 14); + import + .decode(&mux.out) + .expect("a break under a hold must not end the import"); + import.finish().unwrap(); + let (video, audio) = av_frames(&consumer, &catalog).await; + assert_eq!((video.len(), audio.len()), (20, 28), "the held GOP was lost"); + assert_forward(&video); + assert_forward(&audio); + let source = (AV_OFFSET * 1_000_000 / 90_000) as i128; + assert!( + av_offset(&video[12], &audio[14]).abs_diff(source) <= 1, + "the new clock was not re-measured for the program together" + ); + } + /// An adaptation-only clock packet on `pid` carrying `ticks` of the 27 MHz PCR. fn pcr_packet(pid: u16, ticks: u64) -> Vec { let (base, ext) = (ticks / 300, ticks % 300);