From 1dcefe49061ac8514a090b2bc8d63fc1161dbba6 Mon Sep 17 00:00:00 2001 From: t0ms Date: Wed, 7 Oct 2026 13:12:01 +0100 Subject: [PATCH 1/3] feat(moq-mux): import ts --passthrough carries the multiplex whole `moq import ts --passthrough [--pcr-pid ]` publishes every 188-byte packet of a transport stream, in order, on one track that a new `m2ts` root section of the hang catalog names, for what demultiplexing cannot carry: TS-level scrambling, private PIDs, and PSI and SI as authored. It writes through `container::Producer`, so the track joins the broadcast's timelines. The section stays in the catalog for as long as the importer lives, however the track ends, as the demultiplexed lane's entries do. Objects run from one PCR on the clock PID to the next and are stamped with the PCR time of their first byte, unwrapped into a 64-bit tick count across the 33-bit wrap, so the timestamps never step back. A flagged PCR that jumps forward carries the jump and starts a group behind a break marker. A PCR that steps back ends the import, flagged or not: a flagged one is a restart, which is new content, so a supervisor that carries on publishes it as a new broadcast under a new epoch. Groups start at video random access points, at a flagged forward jump, after a second without either, or where the next object would take a group past half of moq-net's group cache. Bytes before the first group are dropped, so byte identity holds from there. The draft specifies the section and maps it onto MSFTS; the comparison now uses MSFTS's `mpeg2ts` names. This lands the import half of ts-passthrough, so that quest is replaced by ts-passthrough-export, which holds the export half. Co-authored-by: Claude Opus 5.5 --- doc/bin/cli.md | 5 +- doc/concept/hang.md | 21 + drafts/draft-lcurley-moq-mpegts.md | 102 +- quest/m1/README.md | 2 +- quest/m1/ts-passthrough-export.md | 54 ++ quest/m1/ts-passthrough.md | 122 --- quest/m1/tstd/README.md | 2 +- quest/m2/ts-hitless.md | 2 +- quest/m3/msfts-convergence.md | 5 +- rs/hang/src/catalog/m2ts.rs | 84 ++ rs/hang/src/catalog/mod.rs | 2 + rs/hang/src/catalog/root.rs | 27 +- rs/moq-cli/src/args.rs | 70 ++ rs/moq-cli/src/main.rs | 1 + rs/moq-cli/src/publish.rs | 258 ++++- rs/moq-mux/src/container/ts/import.rs | 10 +- rs/moq-mux/src/container/ts/mod.rs | 8 + rs/moq-mux/src/container/ts/passthrough.rs | 788 +++++++++++++++ .../src/container/ts/passthrough_test.rs | 905 ++++++++++++++++++ 19 files changed, 2322 insertions(+), 146 deletions(-) create mode 100644 quest/m1/ts-passthrough-export.md delete mode 100644 quest/m1/ts-passthrough.md create mode 100644 rs/hang/src/catalog/m2ts.rs create mode 100644 rs/moq-mux/src/container/ts/passthrough.rs create mode 100644 rs/moq-mux/src/container/ts/passthrough_test.rs diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 94041600c8..f43b273858 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -76,7 +76,10 @@ each as its own broadcast (`event.hang` becomes `event/1.hang`, `event/2.hang`). A damaged packet is dropped on its own PID, so video freezes until its next keyframe while the other tracks carry on. The importer also logs feed health (stalled streams, TR 101 290 errors) without changing what it -publishes. FLV covers H.264 + AAC. +publishes. `--passthrough` publishes the multiplex whole instead, every packet +in order on one [`m2ts`](/concept/hang#transport-streams) track with every +program aboard, for what demultiplexing cannot carry, such as a scrambled +service; `export ts` does not read it. FLV covers H.264 + AAC. ## Export diff --git a/doc/concept/hang.md b/doc/concept/hang.md index 82e6b75434..e430d27b7c 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -126,6 +126,27 @@ Name the section with a namespaced key such as `com.example.mavlink`. A generic consumer only finds tracks in `json` and `binary`. +## Transport streams + +A broadcast can carry an MPEG transport stream whole instead of demultiplexed +into renditions, for what demultiplexing cannot carry, such as a scrambled +service. The root `m2ts` section names the one track holding it: + +```json +{ + "m2ts": { "track": "0.m2ts", "randomAccess": true, "muxRate": 2499999 } +} +``` + +Each frame is a run of whole 188-byte packets in input order, timestamped on +the multiplex's own PCR clock, which never steps back: a source whose clock +steps back is a new broadcast. `randomAccess` says whether every group begins +where a receiver can start decoding, and `muxRate` is the constant multiplex +rate in bits per second, absent for a variable-rate source. A player has +nothing to decode here. `moq import ts --passthrough` writes this section (see +[moq-cli](/bin/cli)); the encoding is in +[draft-lcurley-moq-mpegts](/draft/moq-mpegts). + ## Container The `container.kind` on each rendition says how frames are framed: diff --git a/drafts/draft-lcurley-moq-mpegts.md b/drafts/draft-lcurley-moq-mpegts.md index 02c2018ab6..fd70e07fa0 100644 --- a/drafts/draft-lcurley-moq-mpegts.md +++ b/drafts/draft-lcurley-moq-mpegts.md @@ -50,6 +50,7 @@ informative: This document defines the `mpegts` catalog section, which records what demultiplexing an MPEG-2 Transport Stream {{mpeg2}} into a MoQ broadcast would otherwise lose: each track's PID and PMT descriptors, the program identity, the service information tables, and a carriage record for every elementary stream the publisher did not decode. It is a root member of either the hang catalog {{hang}} or the MSF catalog {{msf}}, so a subscriber that ignores it still plays the broadcast and one that reads it can rebuild the source multiplex. +It also defines the `m2ts` section of the hang catalog, which names a track carrying the transport stream whole, and maps that track onto MSFTS {{msfts}}. --- note_Note_to_Readers @@ -68,16 +69,17 @@ A **verbatim** track carries an elementary stream the publisher did not decode, # Introduction {#introduction} A transport stream reaches MoQ in one of two shapes. -A publisher can leave the multiplex intact and carry the packet stream as opaque payload, which is what {{msfts}} specifies. -Or it can demultiplex, which is what this document addresses: each elementary stream becomes its own MoQ track with a codec description, so a relay can drop and prioritize per track and a subscriber can decode without a transport-stream parser ({{comparison}}). +A publisher can leave the multiplex intact and carry the packet stream as opaque payload, which is what {{msfts}} specifies for MSF and the `m2ts` section ({{m2ts}}) describes in a hang catalog. +Or it can demultiplex, which is what most of this document addresses: each elementary stream becomes its own MoQ track with a codec description, so a relay can drop and prioritize per track and a subscriber can decode without a transport-stream parser ({{comparison}}). Demultiplexing loses everything that is not media: the PID layout, the PMT descriptors, the program identity, the service information, and any stream the publisher has no decoder for. The `mpegts` section records that. It defines no packaging and no container: decoded media keeps the container its catalog entry declares, and verbatim tracks use the framing in {{verbatim}}. -# Comparison with m2ts Packaging {#comparison} +# Comparison with mpeg2ts Packaging {#comparison} {{msfts}} carries the transport stream as it arrives: each object is a run of whole 188- or 192-byte packets, and the receiver demultiplexes as it always has. +The `m2ts` section ({{m2ts}}) carries the same shape in a hang catalog: one track per multiplex, with a timestamp before each object's packets. This document takes the stream apart at the publisher and discards the packetization layer, because MoQ already provides what it was there for: - **Packet framing** interleaves elementary streams onto one serial pipe. MoQ delivers each track separately. @@ -88,17 +90,17 @@ This document takes the stream apart at the publisher and discards the packetiza What remains is media in a codec-neutral catalog, plus the signaling a demultiplexer cannot reconstruct: that is the `mpegts` section. -| | m2ts packaging | mpegts section | +| | mpeg2ts packaging | mpegts section | |:--------|:--------|:--------| -| MoQ track | one per program | one per elementary stream | +| MoQ track | one per program or multiplex | one per elementary stream | | Object payload | whole source packets | one media frame, or one verbatim PES payload or section | -| Packaging | `m2ts` | none defined; media keeps its own | -| Catalog | `m2ts*` members on each track | one `mpegts` root member | +| Packaging | `mpeg2ts` | none defined; media keeps its own | +| Catalog | `mpeg2ts*` members on each track | one `mpegts` root member | | Relay | caches and drops whole objects | drops and prioritizes per elementary stream | -The two overlap only on program identity: `m2tsProgramNumber` and `m2tsPmtPid` against {{field-program}}, and `m2tsPsiInterval` against a per-PID `interval` ({{field-si}}). +The two overlap on program identity, `mpeg2tsProgramNumber` against {{field-program}}, and on the multiplex rate, `mpeg2tsMuxRate` against {{field-mux-rate}}. They do not collide, since those members sit on a track and this section sits at the root. -A track is one shape or the other. +A track is one shape or the other, and a broadcast may carry either. # The mpegts Section {#section} @@ -245,11 +247,89 @@ A verbatim track has no codec, dimensions, or decoder configuration, so a publis Nothing there declares a container, so its frames use hang's `legacy` container ({{hang}}): a varint microsecond timestamp followed by the payload. +# The m2ts Section {#m2ts} +A publisher that carries a transport stream whole, without demultiplexing it, names its track in an `m2ts` root member of the hang catalog ({{carriage-hang}}): + +~~~ +type M2ts = { + "track": TrackName, + "randomAccess": boolean, + "muxRate": number | undefined, +} +~~~ + +`track` names the one track carrying the multiplex, in the same broadcast as the catalog. +A second multiplex is a second broadcast. + +`randomAccess` says that every group starts at a random access point ({{m2ts-groups}}). +A publisher MUST write it, and a consumer MUST read an absent `randomAccess` as false. + +`muxRate` follows {{field-mux-rate}}, measured over every packet of the track. + +The section records no program, PID, descriptor, or table: they all reach the consumer inside the packets, as the publisher received them. +A track named here MUST NOT also be described as a rendition or in an `mpegts` section. + +## Objects {#m2ts-objects} +Each frame is a run of whole 188-byte packets in the order the publisher received them, in hang's `legacy` container ({{hang}}) with the timestamp of {{m2ts-timestamps}}. +A publisher MUST refuse a source of 192- or 204-byte packets rather than reframe it, since removing the extra bytes would change the stream. + +The multiplex is paced on one clock PID: the `PCR_PID` in the PMT of the first program the PAT lists, unless the publisher is configured with another. +A frame runs from one packet carrying a PCR on the clock PID up to the next, and is also cut wherever a group starts. + +A publisher publishes nothing until it has read the PAT, the PMT of the clock's program, and a PCR, and its first group then waits for the next group start ({{m2ts-groups}}). +It drops the bytes before that group, and any bytes off the 188-byte packet grid, and they are outside the track's identity with its input. +From the first group on, the frames of consecutive groups concatenate to the input: a publisher MUST NOT drop, reorder, or modify a packet after that point. + +## Groups {#m2ts-groups} +A group starts at: + +- a packet with `random_access_indicator` set on the video PID of the clock's program (the first PID its PMT declares with a video `stream_type`), or on the clock PID when that program has no video; +- a PCR on the clock PID with `discontinuity_indicator` set that steps forward ({{m2ts-timestamps}}), since a group MUST NOT contain a PCR discontinuity; +- otherwise, the first PCR at least one second of PCR time after the group's start; +- the frame that would take a group past a size bound the publisher chooses below what a relay caches of one group; frames are never split to meet it. + +`randomAccess` is true only while the PAT lists one program, its PMT declares at most one video stream, and the clock PID is that PMT's `PCR_PID`, and only if every group so far has started at a `random_access_indicator`. +The first group that starts any other way republishes the section with `randomAccess` false for the rest of the track. +With more programs or video streams the random access points can stagger, so a group start is not one for all of them. + +Both indicators sit in the adaptation field, which TS-level scrambling leaves in the clear, so a scrambled multiplex groups like its clear twin. +A publisher parses the PAT and the PMT and nothing else past the adaptation field. + +## Timestamps {#m2ts-timestamps} +A frame's timestamp is the PCR time of its first byte, in microseconds rounded down. +A PCR gives the time of the byte holding the last bit of its `program_clock_reference_base` ({{mpeg2}}), ten bytes into its packet. +A byte between two PCRs is placed at the rate between them; a byte past the last PCR of a timebase takes the rate of the interval before it, and one before the first PCR of a timebase takes the rate of the interval after it. + +PCRs are unwrapped into a 64-bit count of 27 MHz ticks, and timestamps never decrease. +A step forward of at most half the 2^33 × 300 tick period adds to the count, across the wrap included, so a wrap is not a discontinuity. +A PCR with `discontinuity_indicator` set whose step is forward by that measure adds its step too, after the bytes before it are timed at the rate before it, and the group it starts MUST follow a skipped group sequence, hang's break marker. +A PCR that steps backwards MUST end the track with an error, flagged or not: a flagged one is a restart, which is new content, so a publisher that carries on publishes it as a new broadcast. + +The timeline depends only on the bytes since the publisher's first PCR, so two publishers that start at the same packet of one input publish identical frames. + +## Mapping onto MSFTS {#m2ts-msfts} +Stripping each frame's timestamp leaves an {{msfts}} object of whole source packets, and the groups carry over unchanged. +An MSF catalog does not list the `m2ts` track itself, whose frames hold more than packets; a converter that republishes the stripped objects describes its track as follows: + +| MSFTS track field | Value | +|:--------|:--------| +| `packaging` | `mpeg2ts` | +| `mpeg2tsMode` | `unmodified-program` when the PAT lists one program, otherwise `unmodified-multiplex` | +| `mpeg2tsPacketSize` | 188 | +| `mpeg2tsRandomAccess` | `randomAccess` | +| `mpeg2tsMuxRate` | `muxRate`, in `unmodified-program` mode only | +| `mpeg2tsProgramNumber`, `mpeg2tsPcrPid`, `mpeg2tsScte35Pid` | read from the PAT and the PMT in the packets, where {{msfts}} permits them in the mode | +| `mpeg2tsTimestampMode`, `mpeg2tsEsPid`, `mpeg2tsSiPids` | absent | + +{{msfts}} asks for the PAT and the PMT in `initDataList` when random access is true; the `m2ts` section does not carry them, so a converter takes them from the packets. + + # Carriage {#carriage} The section is the same JSON in either catalog format. ## hang {#carriage-hang} A root member named `mpegts`, alongside `video` and `audio` ({{hang}}). +The `m2ts` section ({{m2ts}}) is a root member of the hang catalog only. hang carries a decoder config's raw bytes as hex while every binary field here is base64; the alphabets overlap, so the encoding cannot be detected and is stated per field above. ## MSF {#carriage-msf} @@ -258,7 +338,7 @@ A publisher MUST NOT name a section after a member MSF itself defines. The keys of the section's own `tracks` member are MSF track names; a decoded track also has an MSF track object, a verbatim track does not. -This document defines no packaging value; {{msfts}} registers `m2ts` for the passthrough shape ({{comparison}}). +This document defines no packaging value; {{msfts}} defines `mpeg2ts` for the passthrough shape ({{comparison}}), and {{m2ts-msfts}} maps an `m2ts` track onto it. # Rebuilding a Transport Stream {#rebuild} @@ -279,6 +359,7 @@ A consumer MUST NOT derive one that contradicts a recorded descriptor. # Security Considerations Every binary field here is re-emitted without inspection, by a publisher that never parsed it and a relay that never looked, so a consumer MUST treat all of it as untrusted input. +The same holds for every packet of an `m2ts` track, which the publisher forwards as it received them. Nothing bounds the number of entries in the section, nor the size of an SI track: a publisher controls both, and an EPG is large by nature. A consumer MUST bound what it accepts of each, and MUST reject a catalog it cannot bound rather than truncate it into a stream that silently differs from what was described. @@ -346,6 +427,7 @@ A broadcast demultiplexed from a DVB transport stream: video and audio described - A consumer refuses a catalog with an unrecognized `framing` or an invalid `si` map key. - Added `muxRate`, the source's constant multiplex rate. - A consumer may read the pre-`table_id` inline `sections` form; writing stays track-only. +- Added the `m2ts` section, a transport stream carried whole, with its mapping onto MSFTS; the comparison uses MSFTS's `mpeg2ts` names. # Acknowledgments diff --git a/quest/m1/README.md b/quest/m1/README.md index d5a22c1091..5bc3f3e8ec 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -156,7 +156,7 @@ were done or not worth their cost. - [Catalog warmup](/quest/m1/catalog-warmup.md) - `warmup` on video and audio renditions, in the catalog and the draft - [Audio warmup](/quest/m1/audio-warmup.md) - a viewer joining an Opus rendition mid-stream never hears the unconverged first 80 ms - [#3021](/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md) - moq-gst picks the broadcast wall epoch; a restarted source is a new epoch, not a forward re-anchor -- [TS passthrough](/quest/m1/ts-passthrough.md) - `--passthrough` carries the multiplex as whole packets, listed in an `m2ts` catalog section, and writes it back byte-identical (less late drops) on a fixed delay +- [TS passthrough export](/quest/m1/ts-passthrough-export.md) - `export ts --passthrough` writes the `m2ts` track back byte-identical (less late drops) on a fixed delay - [FLV and MKV export delay](/quest/m1/export-delay.md) - FLV and MKV interleave through the shared jitter buffer on a fixed delay, breaking the CLI once - [MKV lacing](/quest/m1/mkv-lacing.md) - laced MKV blocks import as one timed frame each, refused without DefaultDuration - [Release profile](/quest/m1/release-profile.md) - every release build gets fat LTO, one codegen unit, and stripping from the workspace profile instead of three script exports diff --git a/quest/m1/ts-passthrough-export.md b/quest/m1/ts-passthrough-export.md new file mode 100644 index 0000000000..098caf4f6c --- /dev/null +++ b/quest/m1/ts-passthrough-export.md @@ -0,0 +1,54 @@ +# [M] TS passthrough export writes the multiplex back byte-identical on a fixed delay + +## Goal + +`moq export ts --passthrough --delay ` reads the track a broadcast's +`m2ts` catalog section names and writes the same bytes back out, paced on the +source's own PCR and released at a fixed delay like an SRT receiver. From the +first group it releases, the output is byte-identical to the input, less any +object dropped late, so it is exactly as T-STD-conformant as the source. +There is no JS player for it. + +## Plan + +Decided (2026-10-01), from a discussion with t0ms, and split from the TS +passthrough quest on #5003, whose import half it completes: + +- The export reuses the TS export's jitter buffer (`rs/moq-mux/src/jitter.rs`), + keyed on each object's PCR time instead of a DTS, which also spreads each + object's bytes at the PCR-implied rate. It must pace on the source's PCR, + not on arrival: a pacer that re-clocks on arrival moves the PCR-to-PTS + offset over a long capture and fails the decoder buffers, even though every + byte is intact. +- An object that misses its deadline is dropped and counted, as the release + stage does. The output shows a continuity error there; passthrough never + rewrites continuity counters. +- Source clock drift is handled by the fixed-delay release's clock recovery, + shared with the demultiplexed export, not here. +- A flagged forward jump arrives as a break marker before the group it + starts; the release keeps the jump in the output, since the bytes carry it. +- Passthrough gives 1+1 identity for free: two exporters fed the same objects + emit identical TS packets for every object both release, continuity + counters included; an object one leg drops late is a gap in that leg only. + Aligning the legs in time is left to the `--sync` anchor planned for the + demultiplexed export's 2022-7 legs, which passthrough can adopt. That is TS + identity, not ST 2022-7 recovery, which also needs matching RTP headers from + a coordinated RTP egress. +- Rust only. + +Test: an export of a broadcast capture is byte-identical to the input from the +first released group, and the strict T-STD check gives the same verdict on +output and input. Two exporters fed the same objects with different arrival +skew emit identical bytes; when only one misses a deadline, its output is the +other's less that object's packets. A dropped object is counted, and the rest +still go out on time. Rerun the #4613 netem rig (10% loss, 120 s) against it. + +Update `doc/bin/cli.md` for the flag, which today says `export ts` does not +read the passthrough track. + +Public API: a new export mode; nothing existing breaks. + +## Related + +- [TS hitless](/quest/m2/ts-hitless.md) - the demultiplexed lane's 2022-7 legs and the `--sync` anchor +- [T-STD TS export](/quest/m1/tstd/README.md) - the jitter buffer and clock recovery this release shares diff --git a/quest/m1/ts-passthrough.md b/quest/m1/ts-passthrough.md deleted file mode 100644 index 38e21af912..0000000000 --- a/quest/m1/ts-passthrough.md +++ /dev/null @@ -1,122 +0,0 @@ -# [M] TS passthrough carries the multiplex verbatim on a fixed delay - -## Goal - -`moq import ts --passthrough` publishes a transport stream without -demultiplexing it, and `moq export ts --passthrough --delay ` writes the -same bytes back out, paced on the source's own PCR and released at a fixed -delay like an SRT receiver. From the first group it releases, the output is -byte-identical to the input, so it is exactly as T-STD-conformant as the -source. It carries what the demultiplexed lane cannot: TS-level scrambling, -private PIDs, and PSI/SI as authored. There is no JS player for it. - -## Plan - -Decided (2026-10-01), from a discussion with t0ms: - -- Why: the [T-STD line](/quest/m1/tstd/README.md) names a passthrough lane as - the only one that can carry primary distribution until the remux passes, and - none exists. Some feeds will want it permanently: the demultiplexed lane - cannot parse a scrambled elementary stream into frames, and an operator - who must hand on SI and private PIDs as authored has nothing to gain from - a remux. -- The track is listed in a new `m2ts` root section of the hang catalog, - not in `video`/`audio` (it has no codec to describe) and not in the - `mpegts` section, whose members (`tracks`, `program`, `si`, ...) record - what demultiplexing loses and are all in-band here. `m2ts` is one object, - `{ track, randomAccess, muxRate }`, naming the broadcast's one passthrough - track. Passthrough always carries the whole multiplex: splitting out a - program would rewrite the PAT, so there is no scope field, and a second - multiplex is a second broadcast. `muxRate` is present only while the - source holds a constant rate, as in `mpegts`. 188-byte packets only; a - 192-byte source is refused, so there is no packet-size field. This quest - specifies the section in `drafts/draft-lcurley-moq-mpegts.md` next to - `mpegts`, with its mapping onto MSFTS's track fields. A track is one shape - or the other. -- Objects are runs of whole 188-byte packets on one track. A group starts at - a packet with `random_access_indicator` set on the program's video PID - (its `stream_type` read from the PMT), or on the PCR PID when the program - has no video, so a joiner lands on a decodable picture whichever PID - carries the PCR. A group also starts at any PCR with - `discontinuity_indicator` set, since MSFTS forbids a PCR discontinuity - inside a group. A source that never sets the random-access indicator - falls back to a group every N PCRs (N to be fixed in the PR, with a group - spanning at most 1 s of PCR time). Pacing stays on the PCR PID throughout. -- Any PCR rewind, flagged or not, is fatal, as on the demultiplexed lane: an - importer never restarts a broadcast in-process (decided 2026-10-08; a - supervisor that wants to continue publishes a new broadcast under a new - epoch). A flagged forward jump starts a group. -- `randomAccess` is true only while the PAT lists a single program with at - most one video PID, and every group has started at a - `random_access_indicator`. The first group that starts without one (a - fallback, or a `discontinuity_indicator` alone) republishes the catalog - with it false. Any other layout always publishes false, since programs or - video streams sharing a clock can stagger their GOPs. - Both indicators sit in the adaptation field, which TS scrambling leaves in - clear, so a scrambled feed still groups and paces. -- The publisher reads the PAT and PMT (with the existing `ts/psi.rs` - parsers) to find the PCR and video PIDs and count programs, and parses - nothing else past the adaptation field. PSI is never scrambled, and every - byte is still published as received. -- Each object's timestamp is the PCR time of its first byte, interpolated - between PCRs at the stream's own rate, carried in hang's `legacy` - container (a varint timestamp before the packets) as verbatim tracks are. - Object boundaries and timestamps then depend only on the bytes, so two - publishers of one feed publish identical objects, which 1+1 needs. -- A multiplex is paced on one PCR PID: the `PCR_PID` in the PMT of the PAT's - first program, or `--pcr-pid`. Every byte stays in order behind it. Programs - on independent clocks are out of scope. -- The export reuses the TS export's jitter buffer (`rs/moq-mux/src/jitter.rs`, - #4645), keyed on each object's PCR time instead of a DTS, which also - spreads each object's bytes at the PCR-implied rate. It must pace on the - source's PCR, not on arrival: a pacer that re-clocks on arrival moves the - PCR-to-PTS offset over a long capture and fails the decoder buffers, even - though every byte is intact. -- An object that misses its deadline is dropped and counted, as the release - stage does. The output shows a continuity error there; passthrough never - rewrites continuity counters. -- Source clock drift is handled by the fixed-delay release's clock recovery - (#4645), shared with the demultiplexed export, not here. -- Passthrough gives 1+1 identity for free: two exporters fed the same - objects emit identical TS packets for every object both release, - continuity counters included, since passthrough never rewrites them; an - object one leg drops late is a gap in that leg only. Aligning the legs in - time is left to the `--sync` anchor planned for the demultiplexed export's - 2022-7 legs, which passthrough can adopt. - That is TS identity, not ST 2022-7 recovery, which also needs matching RTP - headers from a coordinated RTP egress. -- Rust only. `js/hang` does not parse `m2ts`, and a player sees no - rendition for the track. -- `--passthrough` publishes only the passthrough track. Publishing it - alongside the demultiplexed tracks is out of scope. - -Test: an export of a broadcast capture is byte-identical to the input from the -first released group, and the strict T-STD check gives the same verdict on -output and input. A scrambled fixture (`transport_scrambling_control` set on -its elementary PIDs) groups and paces the same as its clear twin. A -two-program fixture on one clock with staggered GOPs publishes `randomAccess` -false, as does one program with two staggered video PIDs. A single-program -fixture with the PCR on its audio PID starts groups at video random access -points only. PCR discovery finds a PCR PID that differs from the PMT PID, with -the PMT section split across packets. Two exporters fed the same objects with -different arrival skew emit identical bytes; when only one misses a deadline, -its output matches the other's except for the missed object's packets. A -backward PCR errors, flagged or not. A dropped object is counted, and the -rest still go out on time. Rerun the -#4613 netem rig (10% loss, 120 s) against it. - -Lands in two PRs (decided 2026-10-08): #5003 is the import half and the -`m2ts` catalog section, refusing any PCR rewind; the export half follows on -the fixed-delay release stage. This quest finishes with the export. - -Update `doc/bin/cli.md` for both flags, `doc/concept` for the section, and -the draft's comparison section to say this repository now publishes both -shapes. - -Public API: new import and export modes; nothing existing breaks. Wire: the -hang catalog gains an `m2ts` root section; additive. - -## Related - -- [TS hitless](/quest/m2/ts-hitless.md) - the demultiplexed lane's 2022-7 legs and the `--sync` anchor -- [MSFTS convergence](/quest/m3/msfts-convergence.md) - the ES-level side of the same mapping diff --git a/quest/m1/tstd/README.md b/quest/m1/tstd/README.md index 5d8eb04b4d..a748afd30c 100644 --- a/quest/m1/tstd/README.md +++ b/quest/m1/tstd/README.md @@ -40,4 +40,4 @@ nightly. ## Related -- [TS passthrough](/quest/m1/ts-passthrough.md) - the passthrough lane named in the Goal +- [TS passthrough export](/quest/m1/ts-passthrough-export.md) - the passthrough lane named in the Goal diff --git a/quest/m2/ts-hitless.md b/quest/m2/ts-hitless.md index 1b578656bd..d8f9200f61 100644 --- a/quest/m2/ts-hitless.md +++ b/quest/m2/ts-hitless.md @@ -40,4 +40,4 @@ requirement, so this waits until the T-STD line settles. ## Related -- [TS passthrough](/quest/m1/ts-passthrough.md) - byte-identical legs for free in its own lane; adopts `--sync` for alignment +- [TS passthrough export](/quest/m1/ts-passthrough-export.md) - byte-identical legs for free in its own lane; adopts `--sync` for alignment diff --git a/quest/m3/msfts-convergence.md b/quest/m3/msfts-convergence.md index 38597b2599..1e9cfdcf39 100644 --- a/quest/m3/msfts-convergence.md +++ b/quest/m3/msfts-convergence.md @@ -5,8 +5,9 @@ A subscriber author can map this repository's demultiplexed TS lane (access units, Hang catalog `mpegts` section) onto MSFTS ES-level carriage without guessing. Transporting TS verbatim is out of scope here: -[TS passthrough](/quest/m1/ts-passthrough.md) carries it in an `m2ts` -catalog section that maps onto MSFTS's track fields. +`moq import ts --passthrough` carries it in an `m2ts` catalog section that +[the TS draft](/drafts/draft-lcurley-moq-mpegts.md) maps onto MSFTS's track +fields. ## Plan diff --git a/rs/hang/src/catalog/m2ts.rs b/rs/hang/src/catalog/m2ts.rs new file mode 100644 index 0000000000..6857f9ed31 --- /dev/null +++ b/rs/hang/src/catalog/m2ts.rs @@ -0,0 +1,84 @@ +use serde::{Deserialize, Serialize}; + +/// A transport stream carried whole, as one track of verbatim packets. +/// +/// The track's frames are runs of whole 188-byte packets in source order, each in the +/// [`Legacy`](crate::catalog::Container::Legacy) container with the PCR time of its first byte. +/// Nothing about the multiplex is described here because all of it stays in-band: the PAT, PMT, +/// and SI ride the track as authored, scrambled payloads included. A demultiplexed broadcast +/// describes each elementary stream as a rendition instead; a track is one shape or the other. +/// +/// Marked `#[non_exhaustive]` so additional optional fields can be added without bumping the major +/// version. External callers build one with [`M2ts::new`] and assign the fields they need. +#[serde_with::skip_serializing_none] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[non_exhaustive] +pub struct M2ts { + /// The track carrying the packets, published in the same broadcast as the catalog. + pub track: String, + + /// Whether every group starts at a random access point of the multiplex's one video stream, + /// so a subscriber can decode from the first object of any group. + /// + /// Only ever true for a single program with at most one video stream: programs or video + /// streams sharing a clock can stagger their GOPs, leaving no packet that starts all of them. + /// Always written; a reader treats it as false when absent. + #[serde(default)] + pub random_access: bool, + + /// The rate the source's PCR paced the whole multiplex at, in bits per second, null packets + /// included. Present only while the source holds a constant rate. + #[serde(default)] + pub mux_rate: Option, +} + +impl M2ts { + /// A section naming `track`, with no random access promised and no rate recorded. + pub fn new(track: impl Into) -> Self { + Self { + track: track.into(), + random_access: false, + mux_rate: None, + } + } +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn roundtrip() { + let mut section = M2ts::new("0.m2ts"); + section.random_access = true; + section.mux_rate = Some(2_499_999); + + let json = serde_json::to_string(§ion).unwrap(); + assert_eq!(json, r#"{"track":"0.m2ts","randomAccess":true,"muxRate":2499999}"#); + assert_eq!(serde_json::from_str::(&json).unwrap(), section); + } + + #[test] + fn random_access_is_always_written_and_mux_rate_only_when_known() { + let json = serde_json::to_string(&M2ts::new("0.m2ts")).unwrap(); + assert_eq!(json, r#"{"track":"0.m2ts","randomAccess":false}"#); + } + + #[test] + fn absent_random_access_reads_as_false() { + let section: M2ts = serde_json::from_str(r#"{"track":"0.m2ts"}"#).unwrap(); + assert_eq!(section, M2ts::new("0.m2ts")); + } + + #[test] + fn unknown_fields_are_ignored() { + let section: M2ts = serde_json::from_str(r#"{"track":"0.m2ts","packetSize":188}"#).unwrap(); + assert_eq!(section.track, "0.m2ts"); + } + + #[test] + fn a_section_without_a_track_is_refused() { + serde_json::from_str::(r#"{"randomAccess":true}"#).expect_err("a section must name its track"); + } +} diff --git a/rs/hang/src/catalog/mod.rs b/rs/hang/src/catalog/mod.rs index 74a7a0f575..bd0ac127c3 100644 --- a/rs/hang/src/catalog/mod.rs +++ b/rs/hang/src/catalog/mod.rs @@ -14,6 +14,7 @@ mod consumer; mod container; mod hex; mod json; +mod m2ts; mod millis; mod mode; mod priority; @@ -29,6 +30,7 @@ pub use compression::*; pub use consumer::*; pub use container::*; pub use json::*; +pub use m2ts::*; pub use mode::*; pub use priority::*; pub use root::*; diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index b54196d93c..f8fd16d0be 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -11,8 +11,9 @@ use serde::{Deserialize, Serialize}; /// /// The base catalog carries the media sections (`video`, `audio`, `text`), the optional /// `archive` (each track's timeline and any durable recording), the optional `clock` (the -/// broadcast's one wall-clock mapping), and the data sections -/// (`json`, `binary`) for application tracks that aren't media. +/// broadcast's one wall-clock mapping), the data sections +/// (`json`, `binary`) for application tracks that aren't media, and the optional `m2ts` (a +/// transport stream carried whole). /// Applications extend it with their own root sections (e.g. `scte35`) through `E`. The catalog /// does not deny unknown fields, so a base consumer ignores the extra sections and an extended /// catalog stays wire-compatible. See the `extension_roundtrip` test. @@ -99,6 +100,11 @@ pub struct Catalog { )] pub binary: Binary, + /// A transport stream carried whole rather than demultiplexed: the one track holding its + /// packets, verbatim. See [`M2ts`](crate::catalog::M2ts). Omitted from the wire when absent. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub m2ts: Option, + /// Application-defined root sections, flattened alongside the base catalog sections. #[serde(flatten)] pub ext: E, @@ -700,6 +706,23 @@ mod test { ); } + #[test] + fn m2ts_roundtrips_at_the_root() { + let mut m2ts = crate::catalog::M2ts::new("0.m2ts"); + m2ts.random_access = true; + let catalog = Catalog::<()> { + m2ts: Some(m2ts.clone()), + ..Default::default() + }; + + let json = catalog.to_json().expect("failed to encode"); + assert_eq!(json, r#"{"m2ts":{"track":"0.m2ts","randomAccess":true}}"#); + assert_eq!( + Catalog::<()>::from_str(&json).expect("failed to decode").m2ts, + Some(m2ts) + ); + } + #[test] fn clock_stays_off_the_wire_when_absent() { // Absent clock stays off the wire. Empty video and audio are omitted as well. diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 69eb2f08cc..2831a60755 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -769,6 +769,9 @@ impl ImportSource { Some(match self { Self::Avc3 => PublishFormat::Avc3, Self::Fmp4 => PublishFormat::Fmp4, + Self::Ts(args) if args.passthrough => PublishFormat::TsPassthrough { + pcr_pid: args.pcr_pid.map(|pid| pid.0), + }, Self::Ts(args) => PublishFormat::Ts { program: args.program.and_then(TsProgram::number), }, @@ -798,6 +801,37 @@ pub struct TsImport { /// `event/2.hang`, ...). Without it, a stream carrying more than one program is refused. #[usage(long)] pub program: Option, + + /// Publish the multiplex whole instead of demultiplexing it: every 188-byte packet verbatim, + /// on one track the catalog's `m2ts` section names. Scrambled streams, private PIDs, and the + /// PSI and SI ride through as authored. Bytes before the first group are dropped. + #[usage(long, conflicts = "--program")] + pub passthrough: bool, + + /// Pace `--passthrough` on this PID's PCR rather than the `PCR_PID` of the PAT's first + /// program. Decimal, or hex with `0x`. + #[usage(long, requires = "--passthrough")] + pub pcr_pid: Option, +} + +/// An `import ts --pcr-pid` value: a PID that can carry a PCR, in decimal or `0x` hex. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct TsPid(pub u16); + +impl std::str::FromStr for TsPid { + type Err = String; + + fn from_str(arg: &str) -> Result { + let parsed = match arg.strip_prefix("0x").or_else(|| arg.strip_prefix("0X")) { + Some(hex) => u16::from_str_radix(hex, 16), + None => arg.parse(), + }; + match parsed { + Ok(pid @ 0x0001..=0x1ffe) => Ok(Self(pid)), + Ok(pid) => Err(format!("PID {pid:#06x} cannot carry a PCR; expected 0x0001..=0x1ffe")), + Err(_) => Err(format!("expected a PID, got `{arg}`")), + } + } } /// An `import ts --program` or `import srt --program` value. @@ -1239,6 +1273,42 @@ mod tests { assert_eq!(program("two"), None); } + #[test] + fn import_ts_passthrough_takes_an_optional_pcr_pid() { + // `None` when the command line is refused. + let format = |args: &[&str]| { + let cli = Invocation::try_parse_from([&["moq", "import", "ts"], args].concat()).ok()?; + let Command::Import(import) = &cli.stages[0] else { + panic!("an import stage"); + }; + import.source.stdin_format() + }; + assert!(matches!( + format(&["--passthrough"]), + Some(PublishFormat::TsPassthrough { pcr_pid: None }) + )); + assert!(matches!( + format(&["--passthrough", "--pcr-pid", "0x21"]), + Some(PublishFormat::TsPassthrough { pcr_pid: Some(0x21) }) + )); + assert!(matches!( + format(&["--passthrough", "--pcr-pid", "33"]), + Some(PublishFormat::TsPassthrough { pcr_pid: Some(33) }) + )); + assert!(format(&["--pcr-pid", "33"]).is_none(), "--pcr-pid needs --passthrough"); + assert!( + format(&["--passthrough", "--program", "1"]).is_none(), + "nothing to select" + ); + assert!( + format(&["--passthrough", "--pcr-pid", "0x1fff"]).is_none(), + "null packets" + ); + assert!(format(&["--passthrough", "--pcr-pid", "0"]).is_none(), "the PAT"); + assert!(format(&["--passthrough", "--pcr-pid", "clock"]).is_none()); + assert!(matches!(format(&[]), Some(PublishFormat::Ts { program: None }))); + } + /// `import srt` takes the same `--program` as `import ts`; `export srt` has no program to pick. #[test] fn import_srt_takes_a_program() { diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 7d3a9d65ff..68178ea6e3 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -652,6 +652,7 @@ fn spawn_import( .with_bandwidth(bandwidth.clone()); let publish = if let ImportSource::Ts(TsImport { program: Some(TsProgram::All), + .. }) = &import.source { let name = require_broadcast(name, "import ts --program all")?; diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 07c46cdf6f..ce0f6f2ab8 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -11,6 +11,8 @@ pub enum PublishFormat { Fmp4, /// MPEG-TS (transport stream), limited to one `program` of a multiplex when given. Ts { program: Option }, + /// MPEG-TS published whole, paced on `pcr_pid` when given rather than the PMT's clock. + TsPassthrough { pcr_pid: Option }, /// FLV (Flash Video / RTMP). Flv, } @@ -143,6 +145,8 @@ enum PublishDecoder { Ts(Box>), /// `import ts --program all`: one importer, broadcast, and catalog per program. TsPrograms(Box), + /// `import ts --passthrough`: the multiplex whole, on one track. + TsPassthrough(Box), Flv(Box), } @@ -159,6 +163,7 @@ impl PublishDecoder { Self::Fmp4(d) => d.decode(chunk)?, Self::Ts(d) => d.decode(chunk).map_err(suggest_program)?, Self::TsPrograms(d) => d.decode(chunk)?, + Self::TsPassthrough(d) => d.decode(chunk)?, Self::Flv(d) => d.decode(chunk)?, } Ok(()) @@ -170,7 +175,7 @@ impl PublishDecoder { match self { Self::Ts(d) => Some(d.stats()), Self::TsPrograms(d) => Some(d.stats()), - Self::Avc3 { .. } | Self::Fmp4(_) | Self::Flv(_) => None, + Self::Avc3 { .. } | Self::Fmp4(_) | Self::TsPassthrough(_) | Self::Flv(_) => None, } } @@ -187,6 +192,7 @@ impl PublishDecoder { Self::Fmp4(d) => d.finish()?, Self::Ts(d) => d.finish()?, Self::TsPrograms(d) => d.finish()?, + Self::TsPassthrough(d) => d.finish()?, Self::Flv(d) => d.finish()?, } Ok(()) @@ -200,6 +206,7 @@ impl PublishDecoder { Self::Fmp4(d) => d.abort(err), Self::Ts(d) => d.abort(err), Self::TsPrograms(d) => d.abort(err), + Self::TsPassthrough(d) => d.abort(err), Self::Flv(d) => d.abort(err), } } @@ -326,6 +333,15 @@ impl Publish { PublishDecoder::Fmp4(Box::new(fmp4)) } PublishFormat::Ts { .. } => unreachable!("TS is handled above with the mpegts catalog extension"), + // The `m2ts` section is in the base catalog: nothing about the multiplex is described + // beyond the track that carries it. + PublishFormat::TsPassthrough { pcr_pid } => { + let mut passthrough = ts::Passthrough::new(broadcast.clone(), catalog.reserve())?; + if let Some(pid) = pcr_pid { + passthrough = passthrough.with_pcr_pid(*pid); + } + PublishDecoder::TsPassthrough(Box::new(passthrough)) + } PublishFormat::Flv => { let flv = flv::Import::new(broadcast.clone(), catalog.reserve()); PublishDecoder::Flv(Box::new(flv)) @@ -842,6 +858,246 @@ pub(crate) mod tests { ); } + /// `moq import ts --passthrough` publishes no renditions, only the `m2ts` section naming one + /// track, and that track reads back as the input from its first group on. + #[tokio::test(start_paused = true)] + async fn ts_passthrough_publishes_the_multiplex_on_one_track() { + const KYRION: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/kyrion_mpeg2av_ac3.ts"); + let broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut publish = Publish::new( + broadcast, + &PublishFormat::TsPassthrough { pcr_pid: None }, + Default::default(), + ) + .unwrap(); + #[allow(irrefutable_let_patterns)] + let Source::Stream { decoder, .. } = &mut publish.source else { + panic!("expected a stream source"); + }; + for chunk in KYRION.chunks(64 * 1024) { + decoder.decode_chunk(chunk).unwrap(); + } + decoder.finish().unwrap(); + + let catalog = hang::catalog::Catalog::<()>::subscribe(&consumer) + .await + .unwrap() + .next() + .await + .unwrap() + .expect("a catalog"); + assert!(catalog.video.renditions.is_empty() && catalog.audio.renditions.is_empty()); + let m2ts = catalog.m2ts.expect("the m2ts section"); + assert!(m2ts.random_access); + + let mut track = consumer + .track(&m2ts.track) + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(RECORDING_MAX_AGE)) + .await + .unwrap(); + let mut output = Vec::new(); + while let Some(mut group) = track.recv_group().await.unwrap() { + while let Some(frame) = group.read_frame().await.unwrap() { + output.extend_from_slice(&hang::container::Frame::decode(frame.payload).unwrap().payload); + } + } + assert!(!output.is_empty()); + assert!(KYRION.ends_with(&output), "byte-identical from the first group"); + } + + /// The passthrough track crosses a relay byte for byte on the lite and the IETF wire, through + /// a flagged PCR jump twenty seconds forward. The jump crosses as one skipped sequence, which + /// neither the relay nor a late subscriber waits on, and the timestamps carry it, so a budget + /// shorter than the jump drops the groups before it. + #[tokio::test] + async fn ts_passthrough_crosses_a_relay_through_a_flagged_jump() { + const KYRION: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/kyrion_mpeg2av_ac3.ts"); + // The capture twice, the second copy's clock twenty seconds ahead of the first, its first + // PCR flagged as the discontinuity. Inside the track's max age, so the recording keeps + // every group. + const STEP: u64 = 20 * 90_000; + let mut second = KYRION.to_vec(); + let mut pcrs = second + .as_chunks_mut::<188>() + .0 + .iter_mut() + .filter(|pkt| pkt[3] & 0x20 != 0 && pkt[4] > 0 && pkt[5] & 0x10 != 0) + .peekable(); + pcrs.peek_mut().expect("a PCR")[5] |= 0x80; + for pkt in pcrs { + let base = pkt[6..11].iter().fold(0u64, |base, byte| base << 8 | u64::from(*byte)) >> 7; + let base = ((base + STEP) & ((1 << 33) - 1)) << 7 | u64::from(pkt[10] & 0x7f); + pkt[6..11].copy_from_slice(&base.to_be_bytes()[3..]); + } + let input = [KYRION, &second].concat(); + + let _env = crate::test_env::EnvGuard::clear(&["MOQ_CONNECT", "MOQ_HOP"]); + let _ = moq_tokio::crypto::install_default(); + for version in ["moq-lite-06", "moq-transport-14"] { + let fixture = moq_relay::test_relay().await.unwrap(); + let ready = fixture.relay.ready(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.unwrap(); + let net = crate::Net { + quic: Default::default(), + #[cfg(feature = "iroh")] + iroh: None, + }; + let url = fixture.url.to_string(); + let client = || { + let cli = crate::args::Invocation::try_parse_from([ + "moq", + "--connect", + &url, + "--connect-tls-fingerprint", + &fixture.fingerprint, + "--connect-version", + version, + "fetch", + "x", + ]) + .unwrap(); + net.client(cli.moq.client.clone()).unwrap().with_reconnect(false) + }; + + let origin = moq_tokio::origin::spawn(); + let broadcast = origin.create_broadcast("pt.hang").unwrap(); + broadcast.announce(Default::default()).unwrap(); + let mut publish = Publish::new( + broadcast, + &PublishFormat::TsPassthrough { pcr_pid: None }, + Default::default(), + ) + .unwrap(); + #[allow(irrefutable_let_patterns)] + let Source::Stream { decoder, .. } = &mut publish.source else { + panic!("expected a stream source"); + }; + let _publisher = client() + .with_publisher(origin.consume()) + .connect(fixture.url.clone()) + .established() + .await + .unwrap(); + let remote = moq_tokio::origin::spawn(); + let _subscriber = client() + .with_subscriber(remote.clone()) + .connect(fixture.url.clone()) + .established() + .await + .unwrap(); + + // The catalog names the track once the first group is out, which the first copy holds; + // the jump is in the second. + const WAIT: Duration = Duration::from_secs(10); + decoder.decode_chunk(KYRION).unwrap(); + let consumer = tokio::time::timeout(WAIT, remote.consume().routed_broadcast("pt.hang")) + .await + .unwrap() + .unwrap(); + let mut catalogs = hang::catalog::Catalog::<()>::subscribe(&consumer).await.unwrap(); + let catalog = tokio::time::timeout(WAIT, catalogs.next()) + .await + .unwrap() + .unwrap() + .expect("a catalog"); + let name = catalog.m2ts.expect("the m2ts section").track; + + // Each group's sequence and frames, read until the track ends. A read cut off as + // stale leaves `None` for that group's frames. + let read = |subscription: moq_net::track::Subscription| { + let track = consumer.track(&name).unwrap(); + tokio::spawn(tokio::time::timeout(WAIT, async move { + let mut track = track.subscribe(subscription).await.unwrap(); + let mut groups = Vec::new(); + while let Some(mut group) = track.recv_group().await.unwrap() { + let mut frames = Some(Vec::new()); + loop { + match group.read_frame().await { + Ok(Some(frame)) => frames + .as_mut() + .unwrap() + .push(hang::container::Frame::decode(frame.payload).unwrap()), + Ok(None) => break, + // Cut here, or by the relay. + Err(moq_net::Error::Old | moq_net::Error::Stream(moq_net::StreamError::Old)) => { + frames = None; + break; + } + Err(err) => panic!("{err}"), + } + } + groups.push((group.sequence, frames)); + } + groups.sort_by_key(|(sequence, _)| *sequence); + groups + })) + }; + let recording = read(moq_net::track::Subscription::default().with_max_delay(RECORDING_MAX_AGE)); + // About twice real time, so the recording keeps up and only staleness can cut it. + for chunk in second.chunks(8 * 1024) { + decoder.decode_chunk(chunk).unwrap(); + tokio::time::sleep(Duration::from_millis(15)).await; + } + decoder.finish().unwrap(); + + let recording: Vec<_> = recording + .await + .unwrap() + .expect("the recording ends with the track") + .into_iter() + .map(|(sequence, frames)| (sequence, frames.expect("the recording reads every group whole"))) + .collect(); + let output: Vec = recording + .iter() + .flat_map(|(_, frames)| frames.iter().flat_map(|frame| frame.payload.iter().copied())) + .collect(); + assert!(output.len() > KYRION.len(), "{version}: both copies crossed"); + assert!( + input.ends_with(&output), + "{version}: byte-identical from the first group" + ); + let first = |frames: &[hang::container::Frame]| frames[0].timestamp; + let jump = recording + .windows(2) + .find(|pair| first(&pair[1].1).as_micros() > first(&pair[0].1).as_micros() + 10_000_000) + .map(|pair| pair[1].0) + .unwrap_or_else(|| panic!("{version}: no jump crossed")); + let last = recording.last().unwrap().0; + assert_eq!( + recording.iter().map(|(sequence, _)| *sequence).collect::>(), + (recording[0].0..jump - 1).chain(jump..=last).collect::>(), + "{version}: no group missing but the break marker" + ); + + // Joining once every group is out, with a budget half the jump, reads every group + // after it past the skipped sequence, and none of the groups the jump left more than + // the budget behind. An IETF subscription starts at the newest group, so only lite + // has the groups behind it to read. + if version.starts_with("moq-lite") { + let late = read(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(10))) + .await + .unwrap() + .expect("the late read ends with the track"); + let whole: Vec = late + .iter() + .filter(|(_, frames)| frames.is_some()) + .map(|(sequence, _)| *sequence) + .collect(); + assert!( + whole.ends_with(&(jump..=last).collect::>()), + "{version}: a late subscriber reads every group from the jump on: {whole:?}" + ); + assert!( + whole.iter().all(|sequence| *sequence + 2 >= jump), + "{version}: only the group the jump follows is in budget before it: {whole:?}" + ); + } + } + } + /// `moq import ts` publishes the feed's own PTS (1.4s into bbb) and anchors the catalog /// clock on it, so the first frame is live on arrival. #[tokio::test] diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 0a16305d92..1727eca5e8 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -1537,7 +1537,7 @@ impl PatReader { } /// Whether every adaptation field fits inside the length its packet declares. -fn adaptation_valid(pkt: &[u8; TsPacket::SIZE]) -> bool { +pub(super) fn adaptation_valid(pkt: &[u8; TsPacket::SIZE]) -> bool { if pkt[3] & 0x20 == 0 { return true; } @@ -1587,10 +1587,10 @@ pub(super) fn payload(pkt: &[u8; TsPacket::SIZE]) -> Payload<'_> { /// Private sections (SCTE-35 table_id 0xFC and others) are not PES. This handles /// pointer_field alignment, sections split across packets (including a 3-byte /// header split, where section_length is not yet known), continuity-counter -/// gaps, and adaptation-field discontinuities. Deliberately private and minimal: -/// just enough to recover whole sections verbatim. +/// gaps, and adaptation-field discontinuities. Deliberately minimal: just enough to +/// recover whole sections verbatim. #[derive(Default)] -struct SectionReassembler { +pub(super) struct SectionReassembler { /// Bytes of the section currently being reassembled. Its 3-byte header (and /// thus section_length) may not all be present yet, so completeness is /// re-checked as bytes arrive; empty means no section in progress. @@ -1601,7 +1601,7 @@ struct SectionReassembler { impl SectionReassembler { /// Consume one 188-byte TS packet, appending every completed section to `out`. - fn push(&mut self, pkt: &[u8], out: &mut Vec>) { + pub(super) fn push(&mut self, pkt: &[u8], out: &mut Vec>) { let pkt: &[u8; 188] = pkt.try_into().expect("section packet must be 188 bytes"); match self.continuity.observe(pkt) { Continuation::Duplicate => return, diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 4217979b74..02b8459b4a 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -17,6 +17,10 @@ //! a complete snapshot (one frame per sub-table, sections concatenated verbatim); //! frames apply in order with later-wins by sub-table identity, and a joiner reads //! only the newest group. +//! +//! [`Passthrough`] is the other shape: it publishes the multiplex whole, every packet +//! verbatim on one track named by the hang catalog's `m2ts` section, for what +//! demultiplexing cannot carry (TS-level scrambling, PSI and SI as authored). mod adts; mod export; @@ -36,8 +40,10 @@ mod catalog; pub use catalog::{Catalog, Descriptor, Ext, Framing, Mpegts, Program, SiEntry, Track, Verbatim}; pub use export::Export; pub use import::*; +pub use passthrough::Passthrough; pub use programs::Programs; +pub mod passthrough; pub mod stats; #[cfg(test)] @@ -46,3 +52,5 @@ mod export_test; mod export_timing_test; #[cfg(test)] mod import_test; +#[cfg(test)] +mod passthrough_test; diff --git a/rs/moq-mux/src/container/ts/passthrough.rs b/rs/moq-mux/src/container/ts/passthrough.rs new file mode 100644 index 0000000000..f5f6ed8008 --- /dev/null +++ b/rs/moq-mux/src/container/ts/passthrough.rs @@ -0,0 +1,788 @@ +//! Carrying a transport stream whole, without demultiplexing it. +//! +//! [`Passthrough`] publishes every 188-byte packet of its input, in order, on one track named by +//! the catalog's [`m2ts`](hang::catalog::M2ts) section. It reads the PAT and PMT to find the +//! program clock and the video stream, and reads nothing else past an adaptation field, so a +//! scrambled multiplex groups and paces exactly like its clear twin. +//! +//! # Objects and groups +//! +//! An object runs from one PCR packet on the clock PID to the next, and is also cut wherever a +//! group starts. A group starts at: +//! +//! - a packet with `random_access_indicator` on the video PID, or on the clock PID when the +//! program has no video; +//! - a PCR with `discontinuity_indicator` that jumps forward, since no group may hold a PCR +//! discontinuity; +//! - the first PCR at least [`FALLBACK`] after the group's start, for a source that never marks +//! random access; +//! - the first object that would take the group past [`GROUP_BYTES_MAX`] or +//! [`GROUP_OBJECTS_MAX`], so a long GOP cannot outgrow what a relay caches. Objects are never +//! split to meet the bound. +//! +//! Only the first of these makes a group a random access point, so any other start publishes +//! [`random_access`](hang::catalog::M2ts::random_access) false from then on. +//! +//! Bytes before the first group are dropped: until the PMT, a PCR, and a group start have all +//! been seen there is no group to put them in. The track is byte-identical to the input from its +//! first group on, and only from there. Bytes off the 188-byte packet grid are dropped too. +//! +//! # Timestamps +//! +//! An object's timestamp is the PCR time of its first byte, interpolated at the rate between the +//! PCRs either side of it, floored to microseconds. A PCR stamps the byte holding the last bit of +//! its base, ten bytes into its packet, so even an object that starts at a PCR packet sits ten +//! bytes of that rate before its own PCR. +//! +//! PCRs are unwrapped into a 64-bit count of 27 MHz ticks that never goes back. A step forward of +//! at most half the 2^33 * 300 tick period adds to the count, across the wrap included. A PCR with +//! `discontinuity_indicator` whose step is forward by that measure adds the step too, after the +//! bytes before it are timed at the rate before it, and starts a group behind a break marker. Any +//! step back ends the import: [`PcrRewindError`] without the flag, [`PcrRestartError`] with it, +//! since a signalled restart is new content and needs a new broadcast. The timeline therefore +//! depends only on the bytes since the importer's first PCR, and two importers of one feed started +//! at the same packet publish identical objects. + +use std::collections::{BTreeMap, HashMap, VecDeque}; + +use bytes::BytesMut; +use mpeg2ts::ts::{Pid, TsPacket}; + +use super::import::{Framer, PatReader, SectionReassembler, adaptation_valid, discontinuity_indicator, pcr}; +use super::mux_rate::{Meter, PCR_WRAP}; +use super::psi::{self, Pmt}; + +/// The PCR clock, in Hz. +const PCR_HZ: u64 = 27_000_000; + +/// The longest a group runs without a random access point before a PCR starts the next. +pub const FALLBACK: std::time::Duration = std::time::Duration::from_secs(1); +const FALLBACK_TICKS: u64 = PCR_HZ; + +/// The most bytes a group holds before its next object starts another: half of what `moq-net` +/// caches of one group, so the frame headers and an object of slack never reach the cap. +pub const GROUP_BYTES_MAX: u64 = moq_net::group::MAX_CACHE_BYTES / 2; + +/// The most objects a group holds before its next object starts another: half of `moq-net`'s +/// frame cap, which a source stamping PCRs far more often than the 40 ms norm could reach. +pub const GROUP_OBJECTS_MAX: usize = moq_net::group::MAX_GROUP_FRAMES / 2; + +/// The bytes a PCR sits into its packet: four of header, the adaptation field's length and flags, +/// and the five bytes the 33-bit base ends in. +const PCR_OFFSET: u64 = 10; + +/// The packet sizes a sync byte can recur at: 188, BDAV's 192, and 204 with Reed-Solomon parity. +const PACKET_SIZES: [usize; 3] = [TsPacket::SIZE, 192, 204]; + +/// The sync bytes in a row, at one stride, that settle the packet size: payload rarely fakes that. +const SYNC_RUN: usize = 4; + +/// The bytes from a candidate first sync byte to the last one a run of the widest stride needs. +const SYNC_WINDOW: usize = 204 * (SYNC_RUN - 1) + 1; + +/// The stream types a PMT declares video with. A video stream carried under a private stream +/// type (0x06 with a registration descriptor) is not recognized, and its program groups on the +/// clock PID's random access points instead. +const VIDEO_STREAM_TYPES: &[u8] = &[ + 0x01, // MPEG-1 video + 0x02, // MPEG-2 video + 0x10, // MPEG-4 part 2 + 0x1B, // H.264 + 0x1F, // H.264 SVC sub-bitstream + 0x20, // H.264 MVC sub-bitstream + 0x24, // H.265 + 0x33, // H.266 + 0x42, // AVS + 0xD1, // Dirac + 0xEA, // VC-1 +]; + +/// The input is not a stream of 188-byte packets. +/// +/// A 192-byte source (BDAV M2TS, a four-byte timestamp before each packet) or a 204-byte one +/// (Reed-Solomon parity after each) is refused rather than reframed, since dropping those bytes +/// would break the byte identity passthrough exists for. +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +#[non_exhaustive] +#[error("transport stream uses {size}-byte packets; passthrough carries 188-byte packets only")] +pub struct PacketSizeError { + /// The packet size the sync bytes recur at. + pub size: usize, +} + +/// The clock PID's PCR went backwards without `discontinuity_indicator`. +/// +/// The source broke its own timebase without saying so, which leaves no timestamp that both +/// follows the PCR and stays on one timeline. +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +#[non_exhaustive] +#[error("PCR on PID {pid} went back from {from} to {to} without a discontinuity_indicator")] +pub struct PcrRewindError { + /// The clock PID. + pub pid: u16, + /// The previous PCR, raw 27 MHz ticks. + pub from: u64, + /// The PCR that stepped back, raw 27 MHz ticks. + pub to: u64, +} + +/// The clock PID's PCR went backwards with `discontinuity_indicator`: the source restarted. +/// +/// What follows is new content, which belongs in a new broadcast rather than on a timeline that +/// steps back, so the import ends here. +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +#[non_exhaustive] +#[error("PCR on PID {pid} restarted from {from} back to {to}; a restart needs a new broadcast")] +pub struct PcrRestartError { + /// The clock PID. + pub pid: u16, + /// The previous PCR, raw 27 MHz ticks. + pub from: u64, + /// The flagged PCR, raw 27 MHz ticks. + pub to: u64, +} + +/// The clock PID carried no PCR across more bytes than a group may hold. +/// +/// Those bytes cannot be timed, and holding them for a PCR that may never come is unbounded. +#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] +#[non_exhaustive] +#[error("no PCR on PID {pid} for {bytes} bytes")] +pub struct PcrGapError { + /// The clock PID. + pub pid: u16, + /// The bytes waiting on a PCR. + pub bytes: u64, +} + +/// Publishes a transport stream verbatim, as one track of whole 188-byte packets. +/// +/// See the [module documentation](self) for where objects and groups start and how they are +/// timestamped. +pub struct Passthrough { + catalog: crate::catalog::Producer, + /// Held until the first group is written, so the first catalog already names the track with + /// its final `randomAccess`. + reservation: Option>, + media: crate::container::Producer, + /// The section as last written to the catalog, `None` before the first group. + section: Option>, + + scratch: Vec, + framer: Framer, + /// Set once the packet size has been confirmed as 188. + sized: bool, + /// Bytes of whole packets seen so far: the position every PCR and object is placed at. + offset: u64, + + pat: PatReader, + crc_errors: u64, + /// The programs the PAT lists, in PAT order, without the network PID. + programs: Vec, + pmt_sections: HashMap, + /// Each listed program's clock and video PIDs, once its PMT has been read. + pmts: BTreeMap, + /// The PCR PID given by [`with_pcr_pid`](Self::with_pcr_pid), overriding the PMT's. + pcr_override: Option, + /// The PID pacing the multiplex. + pcr_pid: Option, + /// The PID whose `random_access_indicator` starts a group. + access_pid: Option, + /// Whether the layout allows random access at all: one program, at most one video PID, and + /// that program's own clock. + single: bool, + /// Whether the PMT of the clock's program has been read. + mapped: bool, + + timebase: Option, + /// A `discontinuity_indicator` on the clock PID, waiting for the PCR that it flags (normally + /// in the same packet). + pending_break: bool, + /// Before the first group: the PCR time the importer started waiting for a group start at. + waiting_since: Option, + /// The PCR time of the last object written, which no later one goes below. + written: u64, + + /// The object being filled, from its first packet. + open: Option, + /// Closed objects waiting on the next PCR for their timestamps. + queue: VecDeque, + group: Option, + /// Whether every group so far started at a random access point. + all_random_access: bool, + + mux_rate: Meter, +} + +/// Owns the `m2ts` catalog section once written, removing it however the track ends, as the +/// demultiplexed lane's catalog entries are. +struct Section { + catalog: crate::catalog::Producer, + written: hang::catalog::M2ts, +} + +impl Drop for Section { + fn drop(&mut self) { + // A closed catalog has nothing left to remove it from. + let _ = self.catalog.mutate(|catalog| catalog.m2ts = None); + } +} + +#[derive(Default)] +struct Program { + pcr_pid: Option, + video: Vec, +} + +/// The PCRs since the importer's first, or since the last flagged jump. +struct Timebase { + /// The last PCR, raw. + raw: u64, + /// The last few PCRs, newest last: enough to time every object since the one before last. + samples: VecDeque, +} + +#[derive(Clone, Copy)] +struct Sample { + /// Where the PCR's byte sits in the input. + at: u64, + /// The unwrapped PCR, in 27 MHz ticks. + ticks: u64, +} + +struct Object { + /// Where the object's first byte sits in the input. + at: u64, + payload: BytesMut, + /// Why this object starts a group, if it does. + start: Option, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Start { + RandomAccess, + Discontinuity, + Fallback, + /// The group before would have outgrown [`GROUP_BYTES_MAX`] or [`GROUP_OBJECTS_MAX`]. + Full, +} + +struct Group { + /// The PCR time of the group's first byte. + ticks: u64, + bytes: u64, + objects: usize, +} + +impl Passthrough { + /// A passthrough importer publishing one track on `broadcast`, listed in the catalog behind + /// `reserved`. + pub fn new(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> crate::Result { + let catalog = reserved.producer(); + let track = broadcast.unique_track(".m2ts", catalog.track_info(hang::catalog::PRIORITY.video))?; + let media = catalog.media_raw( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + )?; + Ok(Self { + catalog, + reservation: Some(reserved), + media, + section: None, + scratch: Vec::new(), + framer: Framer::default(), + sized: false, + offset: 0, + pat: PatReader::default(), + crc_errors: 0, + programs: Vec::new(), + pmt_sections: HashMap::new(), + pmts: BTreeMap::new(), + pcr_override: None, + pcr_pid: None, + access_pid: None, + single: false, + mapped: false, + timebase: None, + pending_break: false, + waiting_since: None, + written: 0, + open: None, + queue: VecDeque::new(), + group: None, + all_random_access: true, + mux_rate: Meter::default(), + }) + } + + /// Pace the multiplex on `pid` rather than on the `PCR_PID` of the PAT's first program. + /// + /// Groups then start at random access points of the video stream of the first program whose + /// PMT names `pid` as its clock. When no program does, they start at `pid`'s own random access + /// points, which mark no program's decoding boundary, so the track never claims random access. + pub fn with_pcr_pid(mut self, pid: u16) -> Self { + self.pcr_override = Some(pid); + self.pcr_pid = Some(pid); + self + } + + /// The name of the track carrying the packets. + pub fn track(&self) -> &str { + self.media.name() + } + + /// Append `data` and publish every whole packet it completes. A trailing partial packet is + /// kept for the next call. + pub fn decode(&mut self, data: &[u8]) -> anyhow::Result<()> { + self.scratch.extend_from_slice(data); + if !self.sized { + self.probe_packet_size(false)?; + } + self.packets() + } + + /// Publish what is left at the end of the input and close the track. + /// + /// No later PCR will arrive, so the objects since the last one are timed at the rate of the + /// interval before it. The catalog keeps the `m2ts` section until the importer drops. + pub fn finish(&mut self) -> anyhow::Result<()> { + if !self.sized { + self.probe_packet_size(true)?; + self.packets()?; + } + self.close_object(); + self.drain_queue()?; + self.reservation = None; + self.media.finish()?; + Ok(()) + } + + /// Abort the track with `err` instead of finishing, so subscribers see the real cause rather + /// than [`moq_net::Error::Dropped`]. Consumes the importer. + pub fn abort(self, err: moq_net::Error) { + let Self { media, section, .. } = self; + media.abort(err); + drop(section); + } + + /// Publish every whole packet in the scratch buffer, once the packet size is settled. + fn packets(&mut self) -> anyhow::Result<()> { + if !self.sized { + return Ok(()); + } + let mut off = 0; + while let Some(at) = self.framer.next(&self.scratch, &mut off) { + let pkt: [u8; TsPacket::SIZE] = self.scratch[at..at + TsPacket::SIZE].try_into().unwrap(); + self.packet(&pkt)?; + } + self.scratch.drain(..off); + Ok(()) + } + + /// Settle the packet size before the 188-byte framer can lock onto a false grid, refusing a + /// 192- or 204-byte source, and drop the bytes before the first sync byte of the grid. + /// + /// Each offset is tried, in order, once the bytes a run of every stride needs from it have + /// arrived, so where the grid is found does not depend on how the input was chunked. At the + /// end of the input a shorter run settles the offsets left, so a stream too short for a full + /// run still publishes. + fn probe_packet_size(&mut self, eof: bool) -> anyhow::Result<()> { + let scratch = &self.scratch; + let syncs = |k: usize, size: usize, run: usize| (0..run).all(|i| scratch.get(k + i * size) == Some(&0x47)); + let checkable = (scratch.len() + 1).saturating_sub(SYNC_WINDOW); + + let mut found = (0..checkable).find_map(|k| { + PACKET_SIZES + .iter() + .find(|&&size| syncs(k, size, SYNC_RUN)) + .map(|&size| (k, size)) + }); + if found.is_none() && eof { + // Every sync byte at the stride that still fits in what is left. + let fits = |k: usize, size: usize| (scratch.len() - k).div_ceil(size); + found = (checkable..scratch.len()) + .filter(|&k| scratch.len() - k >= TsPacket::SIZE) + .find_map(|k| { + PACKET_SIZES + .iter() + .find(|&&size| { + fits(k, size) >= 2 - usize::from(size == TsPacket::SIZE) && syncs(k, size, fits(k, size)) + }) + .map(|&size| (k, size)) + }); + } + + match found { + Some((k, TsPacket::SIZE)) => { + self.scratch.drain(..k); + self.sized = true; + } + Some((_, size)) => return Err(PacketSizeError { size }.into()), + // No grid starts at any offset tried: leading junk. + None => { + self.scratch.drain(..checkable); + } + } + Ok(()) + } + + fn packet(&mut self, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + let at = self.offset; + self.offset += TsPacket::SIZE as u64; + let pid = u16::from_be_bytes([pkt[1] & 0x1f, pkt[2]]); + + if pid == Pid::PAT { + self.pat_packet(pkt)?; + } else if let Ok(pmt_pid) = Pid::new(pid) + && self.pmt_sections.contains_key(&pmt_pid) + { + self.pmt_packet(pmt_pid, pkt)?; + } + self.mux_rate.packet(); + + // A packet the demodulator flagged corrupt, or whose adaptation field overruns it, is + // carried like any other, but nothing is read from it. + let readable = pkt[1] & 0x80 == 0 && adaptation_valid(pkt); + let mut start = None; + let mut clock = false; + if readable && self.pcr_pid == Some(pid) { + if discontinuity_indicator(pkt) { + self.pending_break = true; + } + if let Some(raw) = pcr(pkt) { + clock = true; + start = self.clock(pid, raw, at)?; + } + } + if readable && self.access_pid == Some(pid) && random_access_indicator(pkt) && self.can_start() { + start = Some(Start::RandomAccess); + } + + if start.is_some() || (clock && self.started()) { + if !clock { + self.close_object(); + } + let mut payload = BytesMut::with_capacity(TsPacket::SIZE * 8); + payload.extend_from_slice(pkt); + self.open = Some(Object { at, payload, start }); + } else if let Some(open) = self.open.as_mut() { + open.payload.extend_from_slice(pkt); + } + + let waiting = self.open.as_ref().map_or(0, |open| open.payload.len() as u64) + + self.queue.iter().map(|object| object.payload.len() as u64).sum::(); + if waiting > GROUP_BYTES_MAX { + return Err(PcrGapError { + pid: self.pcr_pid.unwrap_or_default(), + bytes: waiting, + } + .into()); + } + Ok(()) + } + + /// A PCR at `at` on the clock PID: close the object it ends, time everything waiting on it, + /// and say whether its packet starts a group. + fn clock(&mut self, pid: u16, raw: u64, at: u64) -> anyhow::Result> { + let at = at + PCR_OFFSET; + self.close_object(); + let flagged = std::mem::take(&mut self.pending_break); + + let Some(timebase) = self.timebase.as_mut() else { + // The first PCR starts the count at its own value. + self.timebase = Some(Timebase { + raw, + samples: VecDeque::from([Sample { at, ticks: raw }]), + }); + self.waiting_since = self.mapped.then_some(raw); + if self.mux_rate.pcr(raw) { + self.record()?; + } + return Ok(None); + }; + + // A retransmitted break (the same flagged PCR again) declares nothing new. + let flagged = flagged && !(timebase.samples.len() == 1 && timebase.raw == raw); + let step = (raw + PCR_WRAP - timebase.raw) % PCR_WRAP; + if step > PCR_WRAP / 2 { + let from = timebase.raw; + return Err(match flagged { + true => PcrRestartError { pid, from, to: raw }.into(), + false => PcrRewindError { pid, from, to: raw }.into(), + }); + } + let last = timebase.samples.back().expect("a timebase has a sample").ticks; + + if flagged { + // No PCR on the old side is coming: what waits on one is timed by the rate before it, + // and the new side carries on from there. + self.drain_queue()?; + let ticks = (last + step).max(self.written); + self.timebase = Some(Timebase { + raw, + samples: VecDeque::from([Sample { at, ticks }]), + }); + if self.mux_rate.discontinuity() { + self.record()?; + } + if self.mux_rate.pcr(raw) { + self.record()?; + } + // The jump is not elapsed time, so the wait for the first group starts over. + if self.waiting_since.is_some() { + self.waiting_since = Some(ticks); + } + return Ok(self.started().then_some(Start::Discontinuity)); + } + + let ticks = last + step; + timebase.raw = raw; + timebase.samples.push_back(Sample { at, ticks }); + if timebase.samples.len() > 3 { + timebase.samples.pop_front(); + } + self.drain_queue()?; + if self.mux_rate.pcr(raw) { + self.record()?; + } + + let since = match &self.group { + Some(group) => Some(group.ticks), + None if self.mapped => Some(*self.waiting_since.get_or_insert(ticks)), + None => None, + }; + Ok(since + .filter(|since| ticks.saturating_sub(*since) >= FALLBACK_TICKS && self.can_start()) + .map(|_| Start::Fallback)) + } + + /// Whether a group may start here: once the first one has, always; before it, only once the + /// clock's program is mapped and a PCR has been seen. + fn can_start(&self) -> bool { + self.started() || (self.mapped && self.timebase.is_some()) + } + + /// Whether the first group has started, written or still waiting on its timestamp. + fn started(&self) -> bool { + self.group.is_some() || self.open.is_some() || !self.queue.is_empty() + } + + fn close_object(&mut self) { + if let Some(object) = self.open.take() { + self.queue.push_back(object); + } + } + + /// Time and write every waiting object against the current timebase. + fn drain_queue(&mut self) -> anyhow::Result<()> { + while let Some(object) = self.queue.pop_front() { + let timebase = self.timebase.as_ref().expect("an object follows a PCR"); + let ticks = locate(object.at, &timebase.samples); + self.write(object, ticks)?; + } + Ok(()) + } + + fn write(&mut self, object: Object, ticks: u64) -> anyhow::Result<()> { + // Extrapolating past the last PCR before a small flagged jump can overshoot the jump. + let ticks = ticks.max(self.written); + self.written = ticks; + let timestamp = moq_net::Timestamp::from_micros(ticks / (PCR_HZ / 1_000_000))?; + let size = object.payload.len() as u64; + let start = object.start.or_else(|| { + self.group + .as_ref() + .filter(|group| group.bytes + size > GROUP_BYTES_MAX || group.objects >= GROUP_OBJECTS_MAX) + .map(|_| Start::Full) + }); + + if let Some(start) = start { + if start == Start::Discontinuity && self.group.is_some() { + self.media.discontinuity()?; + } + if start != Start::RandomAccess { + self.all_random_access = false; + } + if self.reservation.is_some() { + // The first object of the import is live on arrival. + self.catalog.anchor(timestamp)?; + } + self.group = Some(Group { + ticks, + bytes: 0, + objects: 0, + }); + } + + // Only an object that starts a group is ever opened before the first one. + let group = self.group.as_mut().expect("the first object starts a group"); + group.bytes += size; + group.objects += 1; + self.media.write(crate::container::Frame { + timestamp, + duration: None, + payload: object.payload.freeze(), + keyframe: start.is_some(), + })?; + if start.is_some() { + self.record()?; + self.reservation = None; + } + Ok(()) + } + + /// Write the section to the catalog if it changed. Nothing is written before the first group. + fn record(&mut self) -> anyhow::Result<()> { + if self.group.is_none() && self.section.is_none() { + return Ok(()); + } + let mut section = hang::catalog::M2ts::new(self.media.name()); + section.random_access = self.single && self.all_random_access; + section.mux_rate = self.mux_rate.published(); + if self.section.as_ref().is_some_and(|written| written.written == section) { + return Ok(()); + } + self.catalog.mutate(|catalog| catalog.m2ts = Some(section.clone()))?; + match &mut self.section { + Some(written) => written.written = section, + None => { + self.section = Some(Section { + catalog: self.catalog.clone(), + written: section, + }) + } + } + Ok(()) + } + + fn pat_packet(&mut self, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + let Some(pat) = self.pat.push(pkt, &mut self.crc_errors) else { + return Ok(()); + }; + let programs: Vec<_> = pat + .programs + .into_iter() + .filter(|program| program.program_number != 0) + .collect(); + if programs == self.programs { + return Ok(()); + } + self.pmt_sections + .retain(|pid, _| programs.iter().any(|program| program.pmt_pid == *pid)); + for program in &programs { + self.pmt_sections.entry(program.pmt_pid).or_default(); + } + self.pmts + .retain(|number, _| programs.iter().any(|program| program.program_number == *number)); + self.programs = programs; + self.remap() + } + + fn pmt_packet(&mut self, pid: Pid, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + let mut sections = Vec::new(); + self.pmt_sections.entry(pid).or_default().push(pkt, &mut sections); + let mut changed = false; + for section in sections { + if section.first() != Some(&psi::PMT_TABLE_ID) || !psi::crc_ok(§ion) { + continue; + } + let Some(pmt) = Pmt::parse(§ion) else { + continue; + }; + let listed = self + .programs + .iter() + .any(|program| program.program_number == pmt.program_number && program.pmt_pid == pid); + if !listed { + continue; + } + let program = Program { + pcr_pid: pmt.pcr_pid.map(|pid| pid.as_u16()), + video: pmt + .streams + .iter() + .filter(|es| VIDEO_STREAM_TYPES.contains(&es.stream_type)) + .map(|es| es.pid.as_u16()) + .collect(), + }; + let known = self.pmts.get(&pmt.program_number); + if known.is_none_or(|known| known.pcr_pid != program.pcr_pid || known.video != program.video) { + self.pmts.insert(pmt.program_number, program); + changed = true; + } + } + if changed { + self.remap()?; + } + Ok(()) + } + + /// The PAT or a PMT changed: find the clock, the PID that marks random access, and whether + /// the layout allows random access at all. + fn remap(&mut self) -> anyhow::Result<()> { + let clock = match self.pcr_override { + Some(pid) => self + .programs + .iter() + .find(|program| { + self.pmts + .get(&program.program_number) + .is_some_and(|pmt| pmt.pcr_pid == Some(pid)) + }) + .or(self.programs.first()), + None => self.programs.first(), + }; + let pmt = clock.and_then(|program| self.pmts.get(&program.program_number)); + + let pcr_pid = self.pcr_override.or(pmt.and_then(|pmt| pmt.pcr_pid)); + // Whether the clock is the mapped program's own. An overriding clock no program names + // marks its own random access points, which are no program's decoding boundary. + let owned = pmt.is_some_and(|pmt| pmt.pcr_pid == pcr_pid); + let access_pid = match pmt { + Some(pmt) if owned => pmt.video.first().copied().or(pcr_pid), + _ => pcr_pid, + }; + let mapped = pmt.is_some(); + let single = self.programs.len() == 1 && owned && pmt.is_some_and(|pmt| pmt.video.len() <= 1); + + if pcr_pid != self.pcr_pid { + // A new clock PID is a new clock: its next PCR is taken as flagged. + if self.pcr_pid.is_some() { + self.pending_break = true; + } + self.pcr_pid = pcr_pid; + } + self.access_pid = access_pid; + self.mapped = mapped; + self.single = single; + self.record() + } +} + +/// Whether a packet's adaptation field sets `random_access_indicator`. +fn random_access_indicator(pkt: &[u8; TsPacket::SIZE]) -> bool { + pkt[3] & 0x20 != 0 && pkt[4] > 0 && pkt[5] & 0x40 != 0 +} + +/// The PCR time of the byte at `at`, from the PCRs of its timebase. +/// +/// Interpolated between the two PCRs either side of it; extrapolated at the rate of the nearest +/// interval when it lies before the timebase's first PCR or after its last. With a single PCR +/// there is no rate, and the byte takes that PCR's time. +fn locate(at: u64, samples: &VecDeque) -> u64 { + let (first, last) = (samples[0], samples[samples.len() - 1]); + let (a, b) = match samples.iter().zip(samples.iter().skip(1)).find(|(_, b)| at < b.at) { + Some((a, b)) => (*a, *b), + None if samples.len() >= 2 => (samples[samples.len() - 2], last), + None => return last.ticks, + }; + let span = u128::from(b.at - a.at); + let ticks = u128::from(b.ticks - a.ticks); + if at >= a.at { + a.ticks + (u128::from(at - a.at) * ticks / span) as u64 + } else { + debug_assert_eq!(a.at, first.at); + a.ticks + .saturating_sub((u128::from(a.at - at) * ticks).div_ceil(span) as u64) + } +} diff --git a/rs/moq-mux/src/container/ts/passthrough_test.rs b/rs/moq-mux/src/container/ts/passthrough_test.rs new file mode 100644 index 0000000000..bb9fb73d18 --- /dev/null +++ b/rs/moq-mux/src/container/ts/passthrough_test.rs @@ -0,0 +1,905 @@ +//! Tests for [`Passthrough`](super::Passthrough). +//! +//! The Kyrion captures are a hardware encoder's broadcast output: a dedicated PCR PID (33) beside +//! the PMT (32) and video (256), so they exercise clock discovery on real bytes. Everything the +//! captures don't carry (scrambling, several programs, a wrap, a flagged jump, oversized groups) +//! is built by [`Mux`]. + +use std::collections::HashMap; +use std::time::Duration; + +use bytes::Bytes; + +use super::passthrough::{ + GROUP_BYTES_MAX, GROUP_OBJECTS_MAX, PacketSizeError, Passthrough, PcrGapError, PcrRestartError, PcrRewindError, +}; +use super::psi; + +/// A delay budget no test timeline comes close to, so the reader sees every group. +const RECORDING_MAX_DELAY: Duration = Duration::from_secs(30); + +const KYRION: &[u8] = include_bytes!("test_data/kyrion_mpeg2av_ac3.ts"); +const KYRION_H264: &[u8] = include_bytes!("test_data/scte35/kyrion_dirtystart.ts"); + +/// One object as a subscriber reads it. +#[derive(Clone, Debug, PartialEq)] +struct Object { + group: u64, + micros: u64, + payload: Bytes, +} + +struct Imported { + catalog: hang::catalog::Catalog, + objects: Vec, +} + +impl Imported { + /// Every object's packets, concatenated in order. + fn bytes(&self) -> Vec { + self.objects + .iter() + .flat_map(|object| object.payload.iter().copied()) + .collect() + } + + fn section(&self) -> &hang::catalog::M2ts { + self.catalog.m2ts.as_ref().expect("an m2ts section") + } + + /// Each group's objects. + fn groups(&self) -> Vec> { + let mut groups: Vec> = Vec::new(); + for object in &self.objects { + match groups.last_mut() { + Some(group) if group[0].group == object.group => group.push(object), + _ => groups.push(vec![object]), + } + } + groups + } +} + +/// Import `data` in chunks of `chunk` bytes and read back every object. +async fn import(data: &[u8], chunk: usize, pcr_pid: Option) -> anyhow::Result { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default())?; + let mut passthrough = Passthrough::new(broadcast.clone(), catalog.reserve())?; + if let Some(pid) = pcr_pid { + passthrough = passthrough.with_pcr_pid(pid); + } + let name = passthrough.track().to_string(); + for chunk in data.chunks(chunk) { + passthrough.decode(chunk)?; + } + passthrough.finish()?; + + let mut track = consumer + .track(&name)? + .subscribe(moq_net::track::Subscription::default().with_max_delay(RECORDING_MAX_DELAY)) + .await?; + let mut objects = Vec::new(); + while let Some(mut group) = tokio::time::timeout(Duration::from_secs(1), track.recv_group()).await?? { + while let Some(frame) = group.read_frame().await? { + let frame = hang::container::Frame::decode(frame.payload)?; + objects.push(Object { + group: group.sequence, + micros: frame.timestamp.as_micros() as u64, + payload: frame.payload, + }); + } + } + Ok(Imported { + catalog: catalog.snapshot(), + objects, + }) +} + +/// What a packet's header and adaptation field say. +#[derive(Debug, Default)] +struct Header { + pid: u16, + pcr: Option, + random_access: bool, + discontinuity: bool, +} + +fn header(pkt: &[u8]) -> Header { + let pid = u16::from_be_bytes([pkt[1] & 0x1f, pkt[2]]); + let mut header = Header { + pid, + ..Default::default() + }; + if pkt[3] & 0x20 != 0 && pkt[4] > 0 { + let flags = pkt[5]; + header.discontinuity = flags & 0x80 != 0; + header.random_access = flags & 0x40 != 0; + if flags & 0x10 != 0 && pkt[4] >= 7 { + let base = (u64::from(pkt[6]) << 25) + | (u64::from(pkt[7]) << 17) + | (u64::from(pkt[8]) << 9) + | (u64::from(pkt[9]) << 1) + | (u64::from(pkt[10]) >> 7); + header.pcr = Some(base * 300 + ((u64::from(pkt[10] & 1) << 8) | u64::from(pkt[11]))); + } + } + header +} + +/// Builds a transport stream packet by packet. +#[derive(Default)] +struct Mux { + out: Vec, + cc: HashMap, +} + +#[derive(Clone, Copy, Default)] +struct Flags { + pcr: Option, + random_access: bool, + discontinuity: bool, +} + +impl Flags { + fn pcr(ticks: u64) -> Self { + Self { + pcr: Some(ticks), + ..Default::default() + } + } + + fn random_access(mut self) -> Self { + self.random_access = true; + self + } + + fn discontinuity(mut self) -> Self { + self.discontinuity = true; + self + } +} + +const ACCESS: Flags = Flags { + pcr: None, + random_access: true, + discontinuity: false, +}; + +impl Mux { + fn cc(&mut self, pid: u16) -> u8 { + let cc = self.cc.entry(pid).or_default(); + let out = *cc; + *cc = (*cc + 1) & 0x0f; + out + } + + /// One packet on `pid`, its payload filled with the PID's low byte. + fn packet(&mut self, pid: u16, flags: Flags) -> &mut Self { + let cc = self.cc(pid); + let mut pkt = vec![0x47, (pid >> 8) as u8 & 0x1f, pid as u8]; + let marked = flags.pcr.is_some() || flags.random_access || flags.discontinuity; + if marked { + pkt.push(0x30 | cc); + let mut field = vec![ + u8::from(flags.discontinuity) << 7 + | u8::from(flags.random_access) << 6 + | u8::from(flags.pcr.is_some()) << 4, + ]; + if let Some(pcr) = flags.pcr { + let (base, ext) = (pcr / 300, pcr % 300); + field.extend_from_slice(&[ + (base >> 25) as u8, + (base >> 17) as u8, + (base >> 9) as u8, + (base >> 1) as u8, + ((base & 1) << 7) as u8 | 0x7e | (ext >> 8) as u8, + ext as u8, + ]); + } + pkt.push(field.len() as u8); + pkt.extend_from_slice(&field); + } else { + pkt.push(0x10 | cc); + } + pkt.resize(188, pid as u8); + self.out.extend_from_slice(&pkt); + self + } + + /// `n` plain packets on `pid`. + fn fill(&mut self, pid: u16, n: usize) -> &mut Self { + for _ in 0..n { + self.packet(pid, Flags::default()); + } + self + } + + /// A section on `pid`, split across as many packets as it needs. + fn section(&mut self, pid: u16, section: &[u8]) -> &mut Self { + let mut payload = vec![0u8]; + payload.extend_from_slice(section); + for (i, chunk) in payload.chunks(184).enumerate() { + let cc = self.cc(pid); + let pusi = if i == 0 { 0x40 } else { 0 }; + let mut pkt = vec![0x47, pusi | (pid >> 8) as u8 & 0x1f, pid as u8, 0x10 | cc]; + pkt.extend_from_slice(chunk); + pkt.resize(188, 0xff); + self.out.extend_from_slice(&pkt); + } + self + } + + fn pat(&mut self, programs: &[(u16, u16)]) -> &mut Self { + let body: Vec = programs + .iter() + .flat_map(|&(number, pmt)| [(number >> 8) as u8, number as u8, 0xe0 | (pmt >> 8) as u8, pmt as u8]) + .collect(); + self.section(0, &psi::section(psi::PAT_TABLE_ID, 1, 0, 0, 0, &body)) + } + + /// A PMT whose `program_info` holds `info` bytes of private descriptors, so a large one + /// spans packets. + fn pmt(&mut self, pid: u16, program: u16, pcr_pid: u16, streams: &[(u8, u16)], info: usize) -> &mut Self { + let mut descriptors = Vec::new(); + while descriptors.len() < info { + let len = (info - descriptors.len()).clamp(2, 200) - 2; + descriptors.push(0x80); + descriptors.push(len as u8); + descriptors.extend(std::iter::repeat_n(0xaa, len)); + } + let mut body = vec![ + 0xe0 | (pcr_pid >> 8) as u8, + pcr_pid as u8, + 0xf0 | (descriptors.len() >> 8) as u8, + descriptors.len() as u8, + ]; + body.extend_from_slice(&descriptors); + for &(stream_type, es) in streams { + body.extend_from_slice(&[stream_type, 0xe0 | (es >> 8) as u8, es as u8, 0xf0, 0x00]); + } + self.section(pid, &psi::section(psi::PMT_TABLE_ID, program, 0, 0, 0, &body)) + } + + fn bytes(&self) -> Vec { + self.out.clone() + } +} + +/// 27 MHz ticks per millisecond. +const MS: u64 = 27_000; + +/// A single program: PMT on 0x20, H.264 video on 0x100, audio on 0x101, the clock on `pcr_pid`. +fn program(mux: &mut Mux, pcr_pid: u16) { + mux.pat(&[(1, 0x20)]) + .pmt(0x20, 1, pcr_pid, &[(0x1b, 0x100), (0x0f, 0x101)], 0); +} + +/// `seconds` of a stream paced 40 ms per PCR on `pcr_pid`, starting at `start` ticks, with a +/// video random access point every `gop` PCRs (none when zero). Each PCR interval is the PCR +/// packet then `fill` packets alternating video and audio. +fn paced(mux: &mut Mux, pcr_pid: u16, start: u64, intervals: u64, gop: u64, fill: usize) { + for i in 0..intervals { + mux.packet(pcr_pid, Flags::pcr(start + i * 40 * MS)); + if gop != 0 && i % gop == 0 { + mux.packet(0x100, ACCESS); + } + for j in 0..fill { + mux.fill(if j % 2 == 0 { 0x100 } else { 0x101 }, 1); + } + } +} + +/// Where in `data` the packet starting the first object sits, checking every packet from there on +/// arrived verbatim. +fn assert_suffix(data: &[u8], imported: &Imported) -> usize { + let bytes = imported.bytes(); + assert!(!bytes.is_empty(), "nothing was published"); + assert_eq!(bytes.len() % 188, 0, "objects are whole packets"); + let start = data.len() - bytes.len(); + assert_eq!(start % 188, 0, "the first object starts on the packet grid"); + assert!( + data[start..] == bytes[..], + "the output is the input from the first group on" + ); + start +} + +#[tokio::test(start_paused = true)] +async fn a_broadcast_capture_is_byte_identical_from_the_first_group() { + for data in [KYRION, KYRION_H264] { + let imported = import(data, data.len(), None).await.unwrap(); + let start = assert_suffix(data, &imported); + + // Groups start at the video PID's random access points, not on the dedicated clock PID. + let groups = imported.groups(); + assert!(groups.len() >= 2, "{} groups", groups.len()); + let mut at = start; + for group in &groups { + let first = header(&group[0].payload); + assert_eq!(first.pid, 256, "a group starts on the video PID"); + assert!(first.random_access, "a group starts at a random access point"); + for object in &group[1..] { + let first = header(&object.payload); + assert_eq!(first.pid, 33, "an object inside a group starts at a PCR packet"); + assert!(first.pcr.is_some()); + } + at += group.iter().map(|object| object.payload.len()).sum::(); + } + assert_eq!(at, data.len()); + assert!( + imported.section().random_access, + "every group starts at a random access point" + ); + assert!(imported.section().track.ends_with(".m2ts")); + assert!(imported.catalog.video.renditions.is_empty(), "nothing is demultiplexed"); + + // The bytes before the first group are the ones dropped, and the first group is the first + // random access point once a PCR has been seen and a PMT has followed a PAT. Both captures + // open on a PMT the importer cannot place yet, so it waits for the next. + let packets: Vec<(usize, Header, bool)> = (0..data.len() / 188) + .map(|i| (i, header(&data[i * 188..]), data[i * 188 + 1] & 0x40 != 0)) + .collect(); + let pat = packets.iter().find(|(_, h, pusi)| h.pid == 0 && *pusi).unwrap().0; + let pmt = packets[pat..] + .iter() + .find(|(_, h, pusi)| h.pid == 32 && *pusi) + .unwrap() + .0; + let pcr = packets + .iter() + .find(|(_, h, _)| h.pid == 33 && h.pcr.is_some()) + .unwrap() + .0; + let first_rai = packets[pmt.max(pcr)..] + .iter() + .find(|(_, h, _)| h.pid == 256 && h.random_access) + .unwrap() + .0; + assert_eq!(start, first_rai * 188); + } +} + +#[tokio::test(start_paused = true)] +async fn an_object_at_a_pcr_is_stamped_just_before_it() { + let imported = import(KYRION, KYRION.len(), None).await.unwrap(); + let mut previous = 0; + for object in &imported.objects { + assert!(object.micros >= previous, "timestamps rise through one timebase"); + previous = object.micros; + if let Some(pcr) = header(&object.payload).pcr { + let pcr = pcr / 27; + // Ten bytes before the PCR's own byte: microseconds at broadcast rates. + assert!( + object.micros <= pcr && pcr - object.micros < 100, + "{} vs {pcr}", + object.micros + ); + } + } +} + +#[tokio::test(start_paused = true)] +async fn objects_do_not_depend_on_how_the_input_is_chunked() { + let whole = import(KYRION_H264, KYRION_H264.len(), None).await.unwrap(); + for chunk in [7, 188, 1316, 65_536] { + let chunked = import(KYRION_H264, chunk, None).await.unwrap(); + assert_eq!(chunked.objects, whole.objects, "chunks of {chunk}"); + assert_eq!(chunked.catalog.m2ts, whole.catalog.m2ts, "chunks of {chunk}"); + } +} + +/// Junk as long as a 204-byte packet, a lone sync byte in it, then a clean stream: the grid is +/// found behind it whether the input arrives at once or in pieces. +#[tokio::test(start_paused = true)] +async fn leading_junk_does_not_hide_the_grid() { + let mut data: Vec = (0..204).map(|i| if i == 60 { 0x47 } else { 0 }).collect(); + data.extend_from_slice(KYRION_H264); + let clean = import(KYRION_H264, KYRION_H264.len(), None).await.unwrap(); + for chunk in [7, 188, 1316, 65_536, data.len()] { + let imported = import(&data, chunk, None).await.unwrap(); + assert_eq!(imported.objects, clean.objects, "chunks of {chunk}"); + assert_eq!(imported.catalog.m2ts, clean.catalog.m2ts, "chunks of {chunk}"); + } +} + +/// Three packets, too few for a full run of sync bytes, still settle the grid at the end of the +/// input and publish. +#[tokio::test(start_paused = true)] +async fn a_stream_shorter_than_a_sync_run_still_publishes() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + mux.packet(0x100, Flags::pcr(0).random_access()); + let data = mux.bytes(); + for chunk in [7, data.len()] { + let imported = import(&data, chunk, None).await.unwrap(); + assert_eq!(assert_suffix(&data, &imported), 2 * 188, "chunks of {chunk}"); + } +} + +/// TS scrambling leaves the header and adaptation field in clear, which is all grouping reads. +#[tokio::test(start_paused = true)] +async fn a_scrambled_twin_groups_and_paces_the_same() { + let mut scrambled = KYRION.to_vec(); + for pkt in scrambled.chunks_mut(188) { + let pid = u16::from_be_bytes([pkt[1] & 0x1f, pkt[2]]); + if ![256, 257, 258].contains(&pid) || pkt[3] & 0x10 == 0 { + continue; + } + pkt[3] |= 0x80; + let start = if pkt[3] & 0x20 != 0 { 5 + usize::from(pkt[4]) } else { 4 }; + for byte in &mut pkt[start.min(188)..] { + *byte ^= 0x5a; + } + } + let clear = import(KYRION, KYRION.len(), None).await.unwrap(); + let twin = import(&scrambled, scrambled.len(), None).await.unwrap(); + assert_suffix(&scrambled, &twin); + + let shape = |imported: &Imported| { + imported + .objects + .iter() + .map(|object| (object.group, object.micros, object.payload.len())) + .collect::>() + }; + assert_eq!(shape(&twin), shape(&clear)); + assert_eq!(twin.catalog.m2ts, clear.catalog.m2ts); +} + +#[tokio::test(start_paused = true)] +async fn two_programs_on_one_clock_publish_no_random_access() { + let mut mux = Mux::default(); + mux.pat(&[(1, 0x20), (2, 0x30)]) + .pmt(0x20, 1, 0x100, &[(0x1b, 0x100)], 0) + .pmt(0x30, 2, 0x100, &[(0x1b, 0x200)], 0); + for i in 0..100 { + mux.packet(0x100, Flags::pcr(i * 40 * MS)); + // Staggered GOPs: program 2's random access points fall between program 1's. + if i % 10 == 0 { + mux.packet(0x100, ACCESS); + } + if i % 10 == 5 { + mux.packet(0x200, ACCESS); + } + mux.fill(0x200, 3); + } + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert!(!imported.section().random_access); + for group in imported.groups() { + let first = header(&group[0].payload); + assert_eq!( + (first.pid, first.random_access), + (0x100, true), + "groups follow program 1's video" + ); + } +} + +#[tokio::test(start_paused = true)] +async fn two_video_streams_in_one_program_publish_no_random_access() { + let mut mux = Mux::default(); + mux.pat(&[(1, 0x20)]) + .pmt(0x20, 1, 0x100, &[(0x1b, 0x100), (0x24, 0x102)], 0); + for i in 0..100 { + mux.packet(0x100, Flags::pcr(i * 40 * MS)); + if i % 10 == 0 { + mux.packet(0x100, ACCESS); + } + if i % 10 == 5 { + mux.packet(0x102, ACCESS); + } + mux.fill(0x102, 3); + } + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert!(!imported.section().random_access); + assert_eq!(imported.groups().len(), 10); +} + +#[tokio::test(start_paused = true)] +async fn a_clock_on_the_audio_pid_still_groups_at_video_random_access() { + let mut mux = Mux::default(); + program(&mut mux, 0x101); + for i in 0..100 { + // Audio marks every frame random access; only video's mark starts a group. + mux.packet(0x101, Flags::pcr(i * 40 * MS).random_access()); + if i % 12 == 0 { + mux.packet(0x100, ACCESS); + } + mux.fill(0x100, 4); + } + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert!(imported.section().random_access); + let groups = imported.groups(); + assert_eq!(groups.len(), 9); + for group in &groups { + let first = header(&group[0].payload); + assert_eq!((first.pid, first.random_access), (0x100, true)); + for object in &group[1..] { + assert_eq!(header(&object.payload).pid, 0x101, "objects start at the clock's PCRs"); + } + } +} + +#[tokio::test(start_paused = true)] +async fn the_clock_is_found_behind_a_pmt_split_across_packets() { + let mut mux = Mux::default(); + mux.pat(&[(1, 0x20)]) + .pmt(0x20, 1, 0x1ff, &[(0x02, 0x100), (0x81, 0x101)], 400); + assert!(mux.out.len() >= 188 * 4, "the PMT spans packets"); + paced(&mut mux, 0x1ff, 0, 50, 10, 6); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert!(imported.section().random_access); + assert_eq!(imported.groups().len(), 5); + assert_eq!(header(&imported.objects[1].payload).pid, 0x1ff); +} + +#[tokio::test(start_paused = true)] +async fn a_source_without_random_access_groups_every_second() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 0, 200, 0, 4); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert!(!imported.section().random_access); + let groups = imported.groups(); + // The first group waits a second for a random access point that never comes. + assert_eq!(groups[0][0].micros, 1_000_000 - micros_before_pcr(&data, 1_000_000)); + for group in &groups { + assert_eq!(group.len(), 25, "a second of 40 ms objects"); + } +} + +/// How far before its PCR an object starting at that PCR's packet is stamped, in microseconds. +fn micros_before_pcr(data: &[u8], pcr_micros: u64) -> u64 { + let at = (0..data.len() / 188) + .find(|i| header(&data[i * 188..]).pcr == Some(pcr_micros * 27)) + .unwrap(); + let prev = (0..at).rev().find(|i| header(&data[i * 188..]).pcr.is_some()).unwrap(); + // Ten bytes at the rate of the 40 ms interval ending at this PCR. + let span = ((at - prev) * 188) as u64; + (10 * 40 * MS).div_ceil(span).div_ceil(27) +} + +#[tokio::test(start_paused = true)] +async fn a_random_access_point_between_pcrs_is_interpolated() { + let mut mux = Mux::default(); + program(&mut mux, 0x101); + // One microsecond per byte: 1,880 bytes between PCRs, 1,880 * 27 ticks apart. + let step = 1_880 * 27; + for i in 0..4 { + mux.packet(0x101, Flags::pcr(1_000 * 27 + i * step)); + mux.fill(0x100, 4); + if i == 1 { + mux.packet(0x100, ACCESS); + mux.fill(0x100, 4); + } else { + mux.fill(0x100, 5); + } + } + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + let first = &imported.objects[0]; + assert!(header(&first.payload).random_access); + // The random access point is five packets after the second PCR's packet, which is ten bytes + // before that PCR. + assert_eq!(first.micros, 1_000 + 1_880 + 5 * 188 - 10); +} + +/// Objects after the last PCR are timed at the rate of the interval before it. +#[tokio::test(start_paused = true)] +async fn objects_after_the_last_pcr_extrapolate_its_rate() { + let mut mux = Mux::default(); + program(&mut mux, 0x101); + // One microsecond per byte, and no PCR after the random access point to bound it. + let step = 1_880 * 27; + for i in 0..4 { + mux.packet(0x101, Flags::pcr(1_000 * 27 + i * step)); + if i == 3 { + mux.fill(0x100, 4); + mux.packet(0x100, ACCESS); + mux.fill(0x100, 4); + } else { + mux.fill(0x100, 9); + } + } + let data = mux.bytes(); + for chunk in [7, 188, data.len()] { + let imported = import(&data, chunk, None).await.unwrap(); + assert_eq!( + assert_suffix(&data, &imported), + data.len() - 5 * 188, + "chunks of {chunk}" + ); + assert_eq!(imported.objects.len(), 1, "chunks of {chunk}"); + assert_eq!( + imported.objects[0].micros, + 1_000 + 3 * 1_880 + 5 * 188 - 10, + "chunks of {chunk}" + ); + } +} + +/// The section describes the track for as long as the importer lives, however the track ends. +#[tokio::test(start_paused = true)] +async fn the_section_goes_with_the_importer() { + for abort in [false, true] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut passthrough = Passthrough::new(broadcast.clone(), catalog.reserve()).unwrap(); + passthrough.decode(KYRION_H264).unwrap(); + assert!(catalog.snapshot().m2ts.is_some(), "abort {abort}"); + if abort { + passthrough.abort(moq_net::Error::Cancel); + } else { + passthrough.finish().unwrap(); + assert!(catalog.snapshot().m2ts.is_some(), "a finished track is still described"); + drop(passthrough); + } + assert!(catalog.snapshot().m2ts.is_none(), "abort {abort}"); + } +} + +#[tokio::test(start_paused = true)] +async fn the_pcr_wrap_continues_the_timeline() { + const WRAP: u64 = (1 << 33) * 300; + let mut mux = Mux::default(); + program(&mut mux, 0x100); + let start = WRAP - 2_000 * MS; + for i in 0..100 { + mux.packet(0x100, Flags::pcr((start + i * 40 * MS) % WRAP)); + if i % 10 == 0 { + mux.packet(0x100, ACCESS); + } + mux.fill(0x101, 3); + } + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + let micros: Vec = imported.objects.iter().map(|object| object.micros).collect(); + assert!(micros.windows(2).all(|pair| pair[0] < pair[1]), "{micros:?}"); + let last = *micros.last().unwrap(); + assert!(last > WRAP / 27, "unwrapped past 2^33 * 300 ticks: {last}"); + assert!(imported.section().random_access, "a wrap is not a discontinuity"); +} + +/// A flagged jump forward starts a group behind a break marker, and the count carries on by the +/// jump, so the timeline never steps back. +#[tokio::test(start_paused = true)] +async fn a_flagged_jump_forward_starts_a_group_behind_a_break() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 10_000 * MS, 50, 10, 4); + // A splice forward to thirty seconds, flagged, with no random access point at the break. + mux.packet(0x100, Flags::pcr(30_000 * MS).discontinuity()); + mux.fill(0x101, 4); + paced(&mut mux, 0x100, 30_040 * MS, 50, 10, 4); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert_suffix(&data, &imported); + + let micros: Vec = imported.objects.iter().map(|object| object.micros).collect(); + assert!(micros.windows(2).all(|pair| pair[0] <= pair[1]), "{micros:?}"); + let groups = imported.groups(); + let jump = groups + .iter() + .position(|group| header(&group[0].payload).discontinuity) + .expect("the break starts a group"); + assert!(jump > 0); + let before = groups[jump - 1].last().unwrap().micros; + let after = groups[jump][0].micros; + assert!((11_959_000..11_960_000).contains(&before), "{before}"); + assert!((29_999_000..30_000_000).contains(&after), "{after}"); + assert_eq!( + groups[jump][0].group, + groups[jump - 1][0].group + 2, + "a skipped sequence marks the break" + ); + assert!( + !imported.section().random_access, + "a group started by a break alone is not a random access point" + ); +} + +/// A flagged step back is a restart: new content, which a timeline cannot step back into. +#[tokio::test(start_paused = true)] +async fn a_flagged_step_back_is_a_restart() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 10_000 * MS, 30, 10, 2); + mux.packet(0x100, Flags::pcr(2_000 * MS).discontinuity()); + let data = mux.bytes(); + let err = match import(&data, data.len(), None).await { + Ok(_) => panic!("a restart was carried on one timeline"), + Err(err) => err, + }; + let restart = err.downcast_ref::().expect("a typed refusal"); + assert_eq!((restart.pid, restart.to), (0x100, 2_000 * MS)); +} + +/// A break before any group has started is no group start, so the first group still waits for a +/// random access point. +#[tokio::test(start_paused = true)] +async fn a_break_before_the_first_group_does_not_start_it() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 0, 5, 0, 2); + mux.packet(0x100, Flags::pcr(5_000 * MS).discontinuity()); + mux.fill(0x101, 2); + paced(&mut mux, 0x100, 5_040 * MS, 30, 10, 2); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert!(header(&imported.objects[0].payload).random_access); + assert!(imported.section().random_access); +} + +#[tokio::test(start_paused = true)] +async fn an_unflagged_rewind_is_refused() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 10_000 * MS, 30, 10, 2); + mux.packet(0x100, Flags::pcr(9_000 * MS)); + let data = mux.bytes(); + let err = match import(&data, data.len(), None).await { + Ok(_) => panic!("a rewind was accepted"), + Err(err) => err, + }; + let rewind = err.downcast_ref::().expect("a typed refusal"); + assert_eq!((rewind.pid, rewind.to), (0x100, 9_000 * MS)); +} + +#[tokio::test(start_paused = true)] +async fn a_retransmitted_break_declares_one_jump() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 10_000 * MS, 30, 10, 2); + let flagged = Flags::pcr(30_000 * MS).discontinuity(); + mux.packet(0x100, flagged); + mux.out.extend_from_within(mux.out.len() - 188..); + paced(&mut mux, 0x100, 30_040 * MS, 30, 10, 2); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + let breaks = imported + .groups() + .iter() + .filter(|group| header(&group[0].payload).discontinuity) + .count(); + assert_eq!(breaks, 1); + assert_suffix(&data, &imported); +} + +#[tokio::test(start_paused = true)] +async fn a_long_gop_splits_at_the_byte_ceiling_between_objects() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + mux.packet(0x100, Flags::pcr(0)); + mux.packet(0x100, ACCESS); + // 100 packets per PCR, a microsecond apart: far past the ceiling in well under a second. + let objects = GROUP_BYTES_MAX as usize * 5 / 2 / (100 * 188); + for i in 1..=objects as u64 { + mux.packet(0x100, Flags::pcr(i * 27)); + mux.fill(0x101, 99); + } + let data = mux.bytes(); + let imported = import(&data, 1 << 20, None).await.unwrap(); + assert_suffix(&data, &imported); + let groups = imported.groups(); + assert_eq!(groups.len(), 3); + for group in &groups { + let bytes: usize = group.iter().map(|object| object.payload.len()).sum(); + assert!(bytes as u64 <= GROUP_BYTES_MAX, "{bytes}"); + } + for group in &groups[1..] { + assert!(header(&group[0].payload).pcr.is_some(), "the split falls on an object"); + } + assert!(!imported.section().random_access); +} + +#[tokio::test(start_paused = true)] +async fn a_clock_in_every_packet_splits_at_the_object_ceiling() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + mux.packet(0x100, Flags::pcr(0).random_access()); + for i in 1..(GROUP_OBJECTS_MAX as u64 * 2 + 10) { + mux.packet(0x100, Flags::pcr(i * 27)); + } + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + let sizes: Vec = imported.groups().iter().map(Vec::len).collect(); + assert_eq!(sizes[..2], [GROUP_OBJECTS_MAX, GROUP_OBJECTS_MAX]); + assert!(!imported.section().random_access); +} + +#[tokio::test(start_paused = true)] +async fn a_clock_that_stops_is_refused_once_a_group_would_overflow() { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + mux.packet(0x100, Flags::pcr(0)); + mux.packet(0x100, ACCESS); + mux.fill(0x101, GROUP_BYTES_MAX as usize / 188 + 1); + let data = mux.bytes(); + let err = match import(&data, 1 << 20, None).await { + Ok(_) => panic!("an untimed run was accepted"), + Err(err) => err, + }; + assert_eq!(err.downcast_ref::().map(|gap| gap.pid), Some(0x100)); +} + +#[tokio::test(start_paused = true)] +async fn nothing_is_published_before_the_pmt_and_a_pcr() { + let mut mux = Mux::default(); + // A random access point before any PSI, then one before any PCR. + mux.packet(0x100, ACCESS); + program(&mut mux, 0x100); + mux.packet(0x100, ACCESS); + mux.fill(0x101, 3); + paced(&mut mux, 0x100, 0, 30, 10, 3); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + let start = assert_suffix(&data, &imported); + let first_pcr = (0..data.len() / 188) + .find(|i| header(&data[i * 188..]).pcr.is_some()) + .unwrap(); + assert!(start > first_pcr * 188); +} + +#[tokio::test(start_paused = true)] +async fn a_192_or_204_byte_source_is_refused() { + for size in [192, 204] { + let data: Vec = KYRION[..188 * 40] + .chunks(188) + .flat_map(|pkt| { + let mut wide = if size == 192 { vec![0, 1, 2, 3] } else { Vec::new() }; + wide.extend_from_slice(pkt); + wide.resize(size, 0); + wide + }) + .collect(); + let err = match import(&data, data.len(), None).await { + Ok(_) => panic!("{size}-byte packets were accepted"), + Err(err) => err, + }; + assert_eq!(err.downcast_ref::(), Some(&PacketSizeError { size })); + } +} + +#[tokio::test(start_paused = true)] +async fn a_cbr_source_records_its_mux_rate() { + let data = include_bytes!("test_data/bbb_cbr.ts"); + let imported = import(data, data.len(), None).await.unwrap(); + let rate = imported.section().mux_rate.expect("a CBR source records its mux rate"); + assert!(rate.abs_diff(400_000) * 1000 <= 400_000, "{rate}"); + assert_suffix(data, &imported); +} + +#[tokio::test(start_paused = true)] +async fn an_overriding_clock_paces_on_its_own_pcrs() { + let mut mux = Mux::default(); + // The PMT names the audio PID's clock; the override paces on a PID no PMT names. + program(&mut mux, 0x101); + for i in 0..60 { + mux.packet(0x101, Flags::pcr(i * 40 * MS)); + mux.fill(0x100, 2); + mux.packet(0x1ee, Flags::pcr(i * 40 * MS + 20 * MS)); + if i % 10 == 0 { + mux.packet(0x1ee, ACCESS); + } + mux.fill(0x100, 2); + } + let data = mux.bytes(); + let imported = import(&data, data.len(), Some(0x1ee)).await.unwrap(); + for group in imported.groups() { + assert_eq!(header(&group[0].payload).pid, 0x1ee, "its own random access points"); + for object in &group[1..] { + assert_eq!(header(&object.payload).pid, 0x1ee, "objects start at its PCRs"); + } + } + assert_eq!(imported.groups().len(), 6); + assert!( + !imported.section().random_access, + "a clock no program names marks no program's decoding boundary" + ); +} From 4ca4bebd6859296e68ba9ce638faada5e774f70c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 09:47:53 -0700 Subject: [PATCH 2/3] fix(moq-mux): mark the break when a splice is also a random access point Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/passthrough.rs | 17 ++++- .../src/container/ts/passthrough_test.rs | 68 ++++++++++--------- 2 files changed, 49 insertions(+), 36 deletions(-) diff --git a/rs/moq-mux/src/container/ts/passthrough.rs b/rs/moq-mux/src/container/ts/passthrough.rs index f5f6ed8008..51e50230c0 100644 --- a/rs/moq-mux/src/container/ts/passthrough.rs +++ b/rs/moq-mux/src/container/ts/passthrough.rs @@ -21,7 +21,8 @@ //! split to meet the bound. //! //! Only the first of these makes a group a random access point, so any other start publishes -//! [`random_access`](hang::catalog::M2ts::random_access) false from then on. +//! [`random_access`](hang::catalog::M2ts::random_access) false from then on. A group the PCR +//! discontinuity starts is behind a break even when it is also a random access point. //! //! Bytes before the first group are dropped: until the PMT, a PCR, and a group start have all //! been seen there is no group to put them in. The track is byte-identical to the input from its @@ -256,6 +257,9 @@ struct Object { payload: BytesMut, /// Why this object starts a group, if it does. start: Option, + /// Whether a flagged PCR jump comes before this object, which the track marks with a break + /// whatever else starts its group. + discontinuity: bool, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -445,6 +449,7 @@ impl Passthrough { let readable = pkt[1] & 0x80 == 0 && adaptation_valid(pkt); let mut start = None; let mut clock = false; + let mut discontinuity = false; if readable && self.pcr_pid == Some(pid) { if discontinuity_indicator(pkt) { self.pending_break = true; @@ -452,6 +457,7 @@ impl Passthrough { if let Some(raw) = pcr(pkt) { clock = true; start = self.clock(pid, raw, at)?; + discontinuity = start == Some(Start::Discontinuity); } } if readable && self.access_pid == Some(pid) && random_access_indicator(pkt) && self.can_start() { @@ -464,7 +470,12 @@ impl Passthrough { } let mut payload = BytesMut::with_capacity(TsPacket::SIZE * 8); payload.extend_from_slice(pkt); - self.open = Some(Object { at, payload, start }); + self.open = Some(Object { + at, + payload, + start, + discontinuity, + }); } else if let Some(open) = self.open.as_mut() { open.payload.extend_from_slice(pkt); } @@ -597,7 +608,7 @@ impl Passthrough { }); if let Some(start) = start { - if start == Start::Discontinuity && self.group.is_some() { + if object.discontinuity && self.group.is_some() { self.media.discontinuity()?; } if start != Start::RandomAccess { diff --git a/rs/moq-mux/src/container/ts/passthrough_test.rs b/rs/moq-mux/src/container/ts/passthrough_test.rs index bb9fb73d18..4bd5eae8c2 100644 --- a/rs/moq-mux/src/container/ts/passthrough_test.rs +++ b/rs/moq-mux/src/container/ts/passthrough_test.rs @@ -667,41 +667,43 @@ async fn the_pcr_wrap_continues_the_timeline() { } /// A flagged jump forward starts a group behind a break marker, and the count carries on by the -/// jump, so the timeline never steps back. +/// jump, so the timeline never steps back. The break is marked whether or not the splice is also a +/// random access point, and only a group started by the break alone gives up random access. #[tokio::test(start_paused = true)] async fn a_flagged_jump_forward_starts_a_group_behind_a_break() { - let mut mux = Mux::default(); - program(&mut mux, 0x100); - paced(&mut mux, 0x100, 10_000 * MS, 50, 10, 4); - // A splice forward to thirty seconds, flagged, with no random access point at the break. - mux.packet(0x100, Flags::pcr(30_000 * MS).discontinuity()); - mux.fill(0x101, 4); - paced(&mut mux, 0x100, 30_040 * MS, 50, 10, 4); - let data = mux.bytes(); - let imported = import(&data, data.len(), None).await.unwrap(); - assert_suffix(&data, &imported); - - let micros: Vec = imported.objects.iter().map(|object| object.micros).collect(); - assert!(micros.windows(2).all(|pair| pair[0] <= pair[1]), "{micros:?}"); - let groups = imported.groups(); - let jump = groups - .iter() - .position(|group| header(&group[0].payload).discontinuity) - .expect("the break starts a group"); - assert!(jump > 0); - let before = groups[jump - 1].last().unwrap().micros; - let after = groups[jump][0].micros; - assert!((11_959_000..11_960_000).contains(&before), "{before}"); - assert!((29_999_000..30_000_000).contains(&after), "{after}"); - assert_eq!( - groups[jump][0].group, - groups[jump - 1][0].group + 2, - "a skipped sequence marks the break" - ); - assert!( - !imported.section().random_access, - "a group started by a break alone is not a random access point" - ); + for random_access in [false, true] { + let mut mux = Mux::default(); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 10_000 * MS, 50, 10, 4); + // A splice forward to thirty seconds, flagged. + let mut splice = Flags::pcr(30_000 * MS).discontinuity(); + splice.random_access = random_access; + mux.packet(0x100, splice); + mux.fill(0x101, 4); + paced(&mut mux, 0x100, 30_040 * MS, 50, 10, 4); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert_suffix(&data, &imported); + + let micros: Vec = imported.objects.iter().map(|object| object.micros).collect(); + assert!(micros.windows(2).all(|pair| pair[0] <= pair[1]), "{micros:?}"); + let groups = imported.groups(); + let jump = groups + .iter() + .position(|group| header(&group[0].payload).discontinuity) + .expect("the break starts a group"); + assert!(jump > 0); + let before = groups[jump - 1].last().unwrap().micros; + let after = groups[jump][0].micros; + assert!((11_959_000..11_960_000).contains(&before), "{before}"); + assert!((29_999_000..30_000_000).contains(&after), "{after}"); + assert_eq!( + groups[jump][0].group, + groups[jump - 1][0].group + 2, + "a skipped sequence marks the break, random access {random_access}" + ); + assert_eq!(imported.section().random_access, random_access); + } } /// A flagged step back is a restart: new content, which a timeline cannot step back into. From dc1ca074cf59b98f73e2b5253dceaa644ca17b8e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 10:21:21 -0700 Subject: [PATCH 3/3] fix(moq-mux): keep randomAccess false after a multi-program layout Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-mpegts.md | 2 +- rs/moq-mux/src/container/ts/passthrough.rs | 15 +++++++++++---- rs/moq-mux/src/container/ts/passthrough_test.rs | 17 +++++++++++++++++ 3 files changed, 29 insertions(+), 5 deletions(-) diff --git a/drafts/draft-lcurley-moq-mpegts.md b/drafts/draft-lcurley-moq-mpegts.md index fd70e07fa0..2826578518 100644 --- a/drafts/draft-lcurley-moq-mpegts.md +++ b/drafts/draft-lcurley-moq-mpegts.md @@ -289,7 +289,7 @@ A group starts at: - the frame that would take a group past a size bound the publisher chooses below what a relay caches of one group; frames are never split to meet it. `randomAccess` is true only while the PAT lists one program, its PMT declares at most one video stream, and the clock PID is that PMT's `PCR_PID`, and only if every group so far has started at a `random_access_indicator`. -The first group that starts any other way republishes the section with `randomAccess` false for the rest of the track. +The first group that starts any other way, or that carries packets while that layout does not hold, republishes the section with `randomAccess` false for the rest of the track. With more programs or video streams the random access points can stagger, so a group start is not one for all of them. Both indicators sit in the adaptation field, which TS-level scrambling leaves in the clear, so a scrambled multiplex groups like its clear twin. diff --git a/rs/moq-mux/src/container/ts/passthrough.rs b/rs/moq-mux/src/container/ts/passthrough.rs index 51e50230c0..0e2ff52ee8 100644 --- a/rs/moq-mux/src/container/ts/passthrough.rs +++ b/rs/moq-mux/src/container/ts/passthrough.rs @@ -21,8 +21,10 @@ //! split to meet the bound. //! //! Only the first of these makes a group a random access point, so any other start publishes -//! [`random_access`](hang::catalog::M2ts::random_access) false from then on. A group the PCR -//! discontinuity starts is behind a break even when it is also a random access point. +//! [`random_access`](hang::catalog::M2ts::random_access) false from then on, as does any group +//! carried while the layout allows no random access (more than one program or video stream). A +//! group the PCR discontinuity starts is behind a break even when it is also a random access +//! point. //! //! Bytes before the first group are dropped: until the PMT, a PCR, and a group start have all //! been seen there is no group to put them in. The track is byte-identical to the input from its @@ -209,7 +211,8 @@ pub struct Passthrough { /// Closed objects waiting on the next PCR for their timestamps. queue: VecDeque, group: Option, - /// Whether every group so far started at a random access point. + /// Whether every group so far started at a random access point, and the layout allowed random + /// access throughout. all_random_access: bool, mux_rate: Meter, @@ -611,7 +614,7 @@ impl Passthrough { if object.discontinuity && self.group.is_some() { self.media.discontinuity()?; } - if start != Start::RandomAccess { + if start != Start::RandomAccess || !self.single { self.all_random_access = false; } if self.reservation.is_some() { @@ -766,6 +769,10 @@ impl Passthrough { self.access_pid = access_pid; self.mapped = mapped; self.single = single; + // The group in progress now carries what no random access point covers, and it stays cached. + if !single && self.started() { + self.all_random_access = false; + } self.record() } } diff --git a/rs/moq-mux/src/container/ts/passthrough_test.rs b/rs/moq-mux/src/container/ts/passthrough_test.rs index 4bd5eae8c2..7b2616d3a1 100644 --- a/rs/moq-mux/src/container/ts/passthrough_test.rs +++ b/rs/moq-mux/src/container/ts/passthrough_test.rs @@ -477,6 +477,23 @@ async fn two_programs_on_one_clock_publish_no_random_access() { } } +/// Groups started while a second program was on air stay cached, so dropping back to one program +/// does not win random access back. +#[tokio::test(start_paused = true)] +async fn random_access_lost_to_a_second_program_stays_lost() { + let mut mux = Mux::default(); + mux.pat(&[(1, 0x20), (2, 0x30)]) + .pmt(0x20, 1, 0x100, &[(0x1b, 0x100)], 0) + .pmt(0x30, 2, 0x100, &[(0x1b, 0x200)], 0); + paced(&mut mux, 0x100, 0, 30, 10, 4); + program(&mut mux, 0x100); + paced(&mut mux, 0x100, 1_200 * MS, 30, 10, 4); + let data = mux.bytes(); + let imported = import(&data, data.len(), None).await.unwrap(); + assert!(imported.groups().len() >= 6); + assert!(!imported.section().random_access); +} + #[tokio::test(start_paused = true)] async fn two_video_streams_in_one_program_publish_no_random_access() { let mut mux = Mux::default();