diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 4ce49cc7b16..ac6b3650772 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -154,12 +154,6 @@ moq --connect https://relay.example.com/anon --broadcast camera.hang \ import rtsp --connect rtsp://admin:secret@192.168.1.64/stream1 ``` -`--hls` publishes [HLS segment records](/bin/hls#how-export-works) beside each -video and audio rendition, which `export hls` needs to serve the broadcast. -Renditions' segments line up wherever the source's keyframes share timestamps. -The stdin formats, `hls`, `rtsp`, and `capture` take it; the other gateways build their -own catalog and refuse it. - ## Export The `fmp4`, `mkv`, `flv`, `h264`, and `h265` exports pick renditions with @@ -366,7 +360,7 @@ over one connection: ```bash moq --connect https://relay.example.com/anon \ - import --broadcast event.hang --hls rtsp --connect rtsp://camera.local/stream1 \ + import --broadcast event.hang srt --listen 0.0.0.0:9000 \ -- export --broadcast event.hang hls --listen 0.0.0.0:8080 \ -- export --broadcast event.hang archive file:///recordings ``` diff --git a/doc/bin/hls.md b/doc/bin/hls.md index b833fa8bc04..8e922ef324c 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -10,9 +10,9 @@ that can't speak MoQ, records it as HLS files, or both. `moq import hls` pulls a remote HLS master or media playlist into a broadcast. ```bash -# Publish with HLS records, then serve. Players open +# Publish, then serve. Players open # http://localhost:8089/my-stream.hang/master.m3u8 (or manifest.mpd for DASH) -ffmpeg -re -i input.mp4 -c copy -f mp4 -movflags cmaf+separate_moof+delay_moov+skip_trailer -frag_duration 1000 - | moq --connect https://relay.example.com/anon --broadcast my-stream.hang import --hls fmp4 +ffmpeg -re -i input.mp4 -c copy -f mp4 -movflags cmaf+separate_moof+delay_moov+skip_trailer -frag_duration 1000 - | moq --connect https://relay.example.com/anon --broadcast my-stream.hang import fmp4 moq --connect https://relay.example.com/anon --broadcast my-stream.hang export hls --listen '[::]:8089' # Record into ./recording, ending once the broadcast finishes @@ -49,7 +49,7 @@ Media is never transcoded, so the player must support the publisher's codecs. What that means for you: -- **Opt in at the publisher.** Only a rendition whose catalog entry carries `hls` gets a playlist; nothing publishes it by default. Pass `--hls` to `moq import` (the stdin formats, `hls`, `rtsp`, and `capture`), or enable it with `moq_mux::catalog::Config::with_hls` in Rust. Without a duration hint, a rendition declares its records after its first segment. +- **The publisher writes the records.** Only a rendition whose catalog entry carries `hls` gets a playlist. Rust publishers built on `moq-mux` (the CLI, the RTMP, SRT, and WebRTC gateways, and the native bindings) write them by default, and `moq_mux::catalog::Config::with_hls(None)` turns them off; `@moq/publish` does not write them yet. Without a duration hint, a rendition declares its records after its first segment. - **Each rendition cuts its own segments.** A group starts a segment when its pts falls in a later multiple of the segment duration than the previous group's, and each rendition numbers its own segments. Players match renditions by timestamp, so nothing coordinates the ladder: segments line up wherever keyframes share timestamps, a rendition that starts late lines up from its next multiple, and audio lines up with video whose keyframes sit on those multiples. The DASH manifest leaves `segmentAlignment` out. A break in one rendition's content starts a discontinuity on that rendition alone; `EXT-X-PROGRAM-DATE-TIME`, written after every discontinuity when the catalog has a clock, lets players resync. - **Idle renditions cost nothing**, and a segment is servable for as long as the relay's [cache](/bin/relay/config#cache) retains it. Keep `--window` (the playlist length, default 16 s) within that retention. - **Live joins cost the window.** The exporter reads the record checkpoint and lists at most 256 segments. diff --git a/doc/concept/hang.md b/doc/concept/hang.md index 53eb3b4b07c..41802ac4890 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -70,7 +70,7 @@ A few things the catalog can express beyond decoder config: - **Spatial media.** A video rendition may carry `projection` (flat `rectilinear`, or `equirectangular` with the degrees of longitude it spans, 180 for VR180) and `stereo` (`mono`, `left-right`, or `top-bottom` frame packing), and an audio rendition may carry `layout: "ambisonic"` for AmbiX channels. Readers treat a value they don't recognize as absent. Nothing in `@moq/watch` renders them yet. - **Disabled renditions.** An audio or video rendition with `enabled: false` has no frames coming, and a player must not select it. Pausing a rendition in `@moq/publish` keeps it in the catalog this way rather than removing it, so a mute is a one-field update; a disabled video rendition encodes one black keyframe first, so a player that predates the field shows black instead of a frozen picture. The field is written only when false. The older video `stalled` field is no longer written, and readers ignore it. - **Preference.** A video rendition may carry a signed `preference`, default `0`. A player keeps only the highest preference it can decode, then picks by size and bitrate within it, as do the single-rendition egresses (RTMP play, WHEP, FLV, the transcoder's source). A publisher that adds its own H.264 copy of an H.265 source marks the copy `-1`, so only a viewer that can't decode H.265 pulls it. Picking a rendition by name still overrides it. The field is written only when nonzero. -- **HLS records.** A video or audio rendition may opt in with an `hls` field naming a small track of HLS segment records, each segment's first group and pts, plus the nominal segment duration. Each rendition numbers its own segments, and an edge serves a rendition's segment `n` from its record `n`; segments line up across renditions wherever their keyframes do. The records are what let the [HLS gateway](/bin/hls) build playlists without subscribing to media; a rendition without them gets no playlist. Nothing publishes them unless the application enables them, and a reader that cannot decode the field ignores it. +- **HLS records.** A video or audio rendition may carry an `hls` field naming a small track of HLS segment records, each segment's first group and pts, plus the nominal segment duration. Each rendition numbers its own segments, and an edge serves a rendition's segment `n` from its record `n`; segments line up across renditions wherever their keyframes do. The records are what let the [HLS gateway](/bin/hls) build playlists without subscribing to media; a rendition without them gets no playlist. Rust publishers (`moq-mux` and everything built on it) publish them by default; `@moq/publish` does not yet. A reader that cannot decode the field ignores it. - **Clock.** The optional root `clock` maps PTS zero to wall time so every media track and every HLS record share one fixed epoch after timescale conversion. Publish independent sources as sibling broadcasts under one prefix, such as `robot/front.hang` and `robot/rear.hang`; each broadcast's audio and video entries are renditions of that source. In one Rust publisher, pass the same `moq_mux::Clock` to each `catalog::Config::with_clock`; JS publishers on one page already share the page clock. Consumers discover the siblings by prefix and align their timestamps through the advertised wall mapping; `@moq/watch` plays one broadcast at a time. - **Version.** The root `version` is the oldest moq-hang draft that reads the catalog correctly, written only when a field needs one newer than 03: a `track` that differs from its ID, or a video `warmup`. Readers ignore fields they don't know, but refuse a `version` above their own with a typed "unsupported catalog version" error, so a player can say to upgrade instead of misreading the catalog. An additive field never raises it. - **Extensions.** The root is a loose object. Applications add their own sections (`scte35`, for example) next to the ones hang defines, optionally naming a track that carries the data. Every library exposes a way to write your section without clobbering the built-in ones, and readers ignore what they don't know. diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index b90b5591a28..ee1ecd9583b 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -62,10 +62,10 @@ These land with the next breaking release. released verifiers reject, so upgrade every auth server and verifier before issuing capped tokens. - **HLS records replace archive timelines.** The hang catalog's root `archive` - section is gone. A video or audio rendition opts in to HLS with its own `hls` - field naming a `.hls` record track, and nothing publishes one by - default. Neither side reads the other's shape, so upgrade HLS exporters and - their publishers together, and opt the publishers in (`moq import --hls`). + section is gone. A video or audio rendition carries its own `hls` field + naming a `.hls` record track, which Rust publishers write by default. + Neither side reads the other's shape, so upgrade HLS exporters and their + publishers together. - **Announce restarts.** moq-lite 07 adds `ANNOUNCE_RESTART` (0x3): another publisher instance now serves the prefix. lite-06's `ANNOUNCE_RESTART` is named `ANNOUNCE_UPDATE`, as it always meant. moq-transport with the epoch diff --git a/drafts/draft-lcurley-moq-hang.md b/drafts/draft-lcurley-moq-hang.md index 4f47a853c4b..d8992f76fab 100644 --- a/drafts/draft-lcurley-moq-hang.md +++ b/drafts/draft-lcurley-moq-hang.md @@ -962,11 +962,11 @@ HLS records exist only so an edge can serve a broadcast as HLS or DASH without s MoQ groups carry only an opaque sequence number; the timestamps live inside the media frames. A record track republishes where each segment of one rendition starts, a few bytes per segment, so an edge builds playlists from the records alone and fetches a segment's groups only when a player requests it. -Records are optional and opt-in: a video or audio rendition carries them only when its publisher enables them, and a rendition without them is not HLS-renderable. +Records are optional: a video or audio rendition carries them only when its publisher publishes them, and a rendition without them is not HLS-renderable. Other tracks, including the catalog and data tracks ({{data}}), carry none. ## Catalog Field {#hls-catalog} -A video or audio rendition opts in with its `hls` field: +A video or audio rendition advertises its records with its `hls` field: ~~~ type HlsSchema = { @@ -1144,7 +1144,7 @@ This document has no IANA actions. ## moq-hang-04 {:numbered="false"} -- Replaced the root `archive` section and its per-track timelines with an opt-in, per-rendition `hls` field on video and audio ({{hls}}). Each record is a segment's first `group` and `pts` in milliseconds, with an optional `duration` where zero marks a stop; frame positions, `end`, `keyframe`, `timescale`, `replay`, `replayEpoch`, `store`, and `version` are gone. Each rendition numbers its own segments, and an edge maps a rendition's record `n` to its segment `n`. The record track is named `.hls` and is not compressed. +- Replaced the root `archive` section and its per-track timelines with an optional, per-rendition `hls` field on video and audio ({{hls}}). Each record is a segment's first `group` and `pts` in milliseconds, with an optional `duration` where zero marks a stop; frame positions, `end`, `keyframe`, `timescale`, `replay`, `replayEpoch`, `store`, and `version` are gone. Each rendition numbers its own segments, and an edge maps a rendition's record `n` to its segment `n`. The record track is named `.hls` and is not compressed. - A reader refuses a JSON frame in which an object repeats a member name. - Added the root `version` ({{catalog-version}}): the oldest draft that reads the catalog, which a consumer refuses when above its own. A `track` other than the entry's ID or a video `warmup` needs 4. Consumers MUST ignore unknown fields. - A data track marks a discontinuity with a group of one object with no bytes, since its empty payload is data. An object with no bytes on an audio or video track, or sharing its group, is malformed. diff --git a/js/hang/src/catalog/hls.ts b/js/hang/src/catalog/hls.ts index 775e82412e8..678a5361f96 100644 --- a/js/hang/src/catalog/hls.ts +++ b/js/hang/src/catalog/hls.ts @@ -4,7 +4,7 @@ import { u53Schema } from "./integers"; /** * A rendition's HLS segment records, which exist only for HLS compatibility. * - * Opt-in, and only on video and audio renditions. Each rendition numbers its own segments. A + * Optional, and only on video and audio renditions. Each rendition numbers its own segments. A * rendition without it is not HLS-renderable. See the `Hls` module for the records. */ export const HlsSchema = z.object({ diff --git a/js/hang/src/hls.ts b/js/hang/src/hls.ts index 9551f8fdc1a..4a3cc106d47 100644 --- a/js/hang/src/hls.ts +++ b/js/hang/src/hls.ts @@ -1,7 +1,7 @@ /** * HLS segment records: where each of a rendition's HLS segments starts. * - * A rendition opts in with its catalog `hls` field, which names a track carrying one + * A rendition advertises them with its catalog `hls` field, which names a track carrying one * {@link Record} per segment. A segment starts at a group boundary, so a record is the segment's * first group and pts. It runs to the next record: it holds every group up to the next record's * `group` and lasts until the next record's `pts`, unless it carries an explicit `duration`. A diff --git a/quest/m1/README.md b/quest/m1/README.md index b86dd5c48b9..110b304ffbf 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -155,7 +155,6 @@ surface and zero-copy texture import, land before the cut. - [Main is no slower than release](/quest/m1/release-perf/README.md) - every benchmark where main is slower than release is fixed or accepted with a reason, checked by a final paired run before the cut - [Interop coverage](/quest/m1/interop-coverage/README.md) - the riskiest moq-transport features moq-net exposes get cross-implementation tests over drafts 18 and 22 - [Calibrate the paired nightly benchmarks](/quest/m1/bench-ci.md) - A/A and deliberately slowed Nightly runs set the paired comparison's threshold, so its regression alerts can be trusted -- [HLS opt-in for gateways](/quest/m1/hls-opt-in-gateways.md) - SRT, RTMP, WebRTC ingest and the bindings can publish HLS records - [HLS and DASH segment conformance](/quest/m1/hls-conformance.md) - nightly ffprobe checks that advertised timing, sync flags, and bytes match the media served - [App-limited BBR3 estimate ages](/quest/m1/bbr-app-limited.md) - an app-limited BBR3 estimate falls within seconds after the path drops - [Aborted epoch reads keep their cause](/quest/m1/epoch-abort-cause.md) - an aborted read through an epoch route ends with its own error, not NotFound diff --git a/quest/m1/hls-opt-in-gateways.md b/quest/m1/hls-opt-in-gateways.md deleted file mode 100644 index af8051d214f..00000000000 --- a/quest/m1/hls-opt-in-gateways.md +++ /dev/null @@ -1,24 +0,0 @@ -# [S] HLS records opt-in for gateways and bindings - -## Goal - -Every publisher that builds its own catalog can opt in to HLS records, so -`export hls` can serve its broadcasts: the SRT, RTMP, and WebRTC ingest -gateways (`moq-srt`, `moq-rtmp`, `moq-rtc`) and the native bindings -(`moq-ffi` and its wrappers, `moq-c`, `moq-gst`). Today only the CLI's own -imports take `--hls`, and `import srt`, `import rtmp`, and `import rtc` refuse -it. - -## Plan - -- Each gateway's publish builder gains the opt-in beside `with_max_age`, and - `moq import --hls` passes it through instead of refusing. -- The bindings expose the opt-in where they build a catalog config, mirrored - across languages. -- Update `doc/bin/cli.md`, `doc/bin/hls.md`, and the binding docs. - -Public API: one opt-in per gateway builder and binding. Wire: none. - -## Related - -- [JS HLS field](/quest/m2/js-hls.md) - the `@moq/publish` opt-in diff --git a/quest/m1/segmenter-cuts.md b/quest/m1/segmenter-cuts.md index 49818433fd3..b463355e371 100644 --- a/quest/m1/segmenter-cuts.md +++ b/quest/m1/segmenter-cuts.md @@ -11,7 +11,7 @@ Found while landing [moq-dev/moq#6783](https://github.com/moq-dev/moq/pull/6783) `Segmenter.cuts` (`rs/moq-mux/src/hls.rs`) is a `VecDeque` that `cut()` appends to without bound, and only a new group start drains it. Cuts come from a declared fMP4 segment, including a `styp` box in network input -(for example `moq-cli` HLS ingest of a remote origin with `--hls`). An origin +(for example `moq-cli` HLS ingest of a remote origin). An origin that sends `styp` segments with rising pts and no sync sample never starts a group, so the queue grows by one entry per segment. diff --git a/quest/m1/time-addressing/consumers.md b/quest/m1/time-addressing/consumers.md index e66bbf9f482..a67af766638 100644 --- a/quest/m1/time-addressing/consumers.md +++ b/quest/m1/time-addressing/consumers.md @@ -4,7 +4,8 @@ Every first-party consumer that numbered groups works by start timestamp, in Rust and JS, and hang stops writing duration and discontinuity markers -that group ends and gaps now carry. +that group ends and gaps now carry. The HLS/DASH export lists segments +straight from group starts and ends, so the HLS records go away. ## Plan @@ -19,6 +20,11 @@ Decided 2026-10-25 (maintainer), from the spike: - Legacy egress synthesizes neither: old players lose end durations and pause markers, which they already tolerate being absent. +Decided 2026-10-27 (maintainer): moq-mux publishes HLS records by default +until this lands. Then the `.hls` record tracks, the catalog `hls` +field (hang draft and both catalog libraries), the `catalog::Config::with_hls` +opt-out, and any A/V segment alignment go away. + Convert each consumer that numbers groups: hang and moq-mux timelines, HLS/DASH export, `seek` for GOP-aligned failover in moq-mux, HLS import's discontinuity packing, transcode's 1:1 numbering, archive keys and resume, @@ -27,8 +33,10 @@ JS container consumers and `js/watch`. Update the hang draft and the docs each one makes stale; rename CLI flags per the repo's CLI rule. Public API: breaking where a consumer exposes group numbers (CLI -`--group`, archive keys). Wire: hang drops empty duration frames and -discontinuity marker groups. +`--group`, archive keys), and `moq_mux::hls`, `hang::hls`, and +`catalog::Config::with_hls` are deleted. Wire: hang drops empty duration +frames, discontinuity marker groups, the `hls` catalog field, and the record +tracks. ## Required diff --git a/quest/m2/js-hls.md b/quest/m2/js-hls.md index b0244dd3cdc..b9d7b25365b 100644 --- a/quest/m2/js-hls.md +++ b/quest/m2/js-hls.md @@ -2,22 +2,22 @@ ## Goal -`@moq/publish` publishes the opt-in per-rendition `hls` field and tracks Rust -does, with the same records and declared duration, when the application -enables it, so a browser-published broadcast exports as HLS like a native one. -Nothing publishes it by default. +`@moq/publish` publishes the per-rendition `hls` field and record tracks by +default, with the same records and declared duration Rust does, so a +browser-published broadcast exports as HLS like a native one. ## Plan -Retargeted 2026-10-17: the field exists only for HLS compatibility and is -opt-in (Catalog HLS field, landed). The archive -does not use it, so the old idle `flush()` and catalog and data track timelines -are dropped from this port. +Decided 2026-10-27 (maintainer): records are on by default in Rust +(`moq_mux::catalog::Config` publishes them unless `with_hls(None)`), and JS +matches, until [time addressing](/quest/m1/time-addressing/consumers.md) lets +the export list segments from group starts and deletes the records. The +archive does not use the field, so the old idle `flush()` and catalog and data +track timelines stay out of this port. Each rendition cuts its own segments on a pts grid and numbers its own records, sharing no state with the others, as `rs/moq-mux/src/hls.rs` does. The read side follows Catalog HLS field. Port the Rust producer -(`rs/moq-mux`) to `js/hang`, wire the opt-in through `@moq/publish`, and check -its records against Rust output in the interop suite. - +(`rs/moq-mux`) to `js/hang`, publish it from `@moq/publish` by default, and +check its records against Rust output in the interop suite. diff --git a/rs/hang/MIGRATE.md b/rs/hang/MIGRATE.md index fd5d9e0ee7d..5250ea618cf 100644 --- a/rs/hang/MIGRATE.md +++ b/rs/hang/MIGRATE.md @@ -16,7 +16,7 @@ Delete matches on `TooManyRenditions`. An application that wants a limit counts ### HLS records replace the root archive `catalog::Archive`, `catalog::Timeline`, `Catalog::archive`, and the `timeline` module are gone. -A video or audio rendition opts in to HLS with its own `hls: Option` field, naming its record track and nominal segment duration. +A video or audio rendition carries its own `hls: Option` field, naming its record track and nominal segment duration. Records are `hls::Record { sequence, group, pts, duration }` in milliseconds on a `.hls` track (`hls::default_name`, `hls::SUFFIX`). Build one with `Record::new(sequence, group, pts)`, adding `with_duration(0)` where content stops. A catalog with an old root `archive` still parses, with the entry ignored. diff --git a/rs/hang/README.md b/rs/hang/README.md index 0c55f7b98c4..be5e03cf58f 100644 --- a/rs/hang/README.md +++ b/rs/hang/README.md @@ -9,7 +9,7 @@ A media library built on top of `moq-lite` for streaming audio and video. - **Broadcast**: A discoverable collection of tracks, documented using a catalog. - **Catalog**: Metadata describing the available tracks, codec information, etc. This is a live track itself and is updated as tracks are added/removed/changed. -- **HLS records**: An opt-in per-rendition index of aligned HLS segments, so an edge can build playlists without downloading media. Advertised by a rendition's `hls` field; itself a live track. +- **HLS records**: An optional per-rendition index of aligned HLS segments, so an edge can build playlists without downloading media. Advertised by a rendition's `hls` field; itself a live track. - **Track**: Audio/video streams, as well as other types of data. - **Group**: A group of pictures (video) or collection of samples (audio). Each group is independently decodable. - **Frame**: A timestamp and a codec payload pair. diff --git a/rs/hang/src/catalog/hls.rs b/rs/hang/src/catalog/hls.rs index 6361d590e5f..b7d063f1f92 100644 --- a/rs/hang/src/catalog/hls.rs +++ b/rs/hang/src/catalog/hls.rs @@ -6,7 +6,7 @@ use super::millis::MillisCeil; /// A rendition's HLS segment records, which exist only for HLS compatibility. /// -/// Opt-in, and only on video and audio renditions. Each rendition numbers its own segments. A +/// Optional, and only on video and audio renditions. Each rendition numbers its own segments. A /// rendition without it is not HLS-renderable. See the [`hls`](crate::hls) module for the /// records. #[serde_with::serde_as] diff --git a/rs/hang/src/hls.rs b/rs/hang/src/hls.rs index 2012fa78c89..34c37f7dabe 100644 --- a/rs/hang/src/hls.rs +++ b/rs/hang/src/hls.rs @@ -1,6 +1,6 @@ //! HLS segment records: where each of a rendition's HLS segments starts. //! -//! A rendition opts in with its catalog [`hls`](crate::catalog::Hls) field, which names a track +//! A rendition advertises them with its catalog [`hls`](crate::catalog::Hls) field, which names a track //! carrying one [`Record`](crate::hls::Record) per segment. A segment starts at a group boundary, so a record is the //! segment's first group and pts. It runs to the next record: it holds every group up to the next //! record's `group` and lasts until the next record's `pts`, unless it carries an explicit diff --git a/rs/hang/src/lib.rs b/rs/hang/src/lib.rs index 4e534acc14c..87777a5985b 100644 --- a/rs/hang/src/lib.rs +++ b/rs/hang/src/lib.rs @@ -6,7 +6,7 @@ //! //! - **Catalog**: A JSON track containing codec info and track metadata, updated live as tracks change. //! - **Tracks**: Audio or video, supporting one or more renditions. -//! - **HLS records**: An opt-in JSON track per rendition marking each of its HLS segments' first group +//! - **HLS records**: An optional JSON track per rendition marking each of its HLS segments' first group //! and pts, so an edge can render playlists without downloading media. Advertised by the rendition's `hls` field. //! //! Each track specifies a container format: @@ -35,7 +35,7 @@ pub mod catalog; /// The container is the contents of each media track. pub mod container; -/// HLS segment records, opt-in per rendition. +/// HLS segment records, optional per rendition. pub mod hls; /// Export the moq-net version we use. diff --git a/rs/moq-audio/src/encode/capture.rs b/rs/moq-audio/src/encode/capture.rs index 6fdbfcf16e6..7059df00052 100644 --- a/rs/moq-audio/src/encode/capture.rs +++ b/rs/moq-audio/src/encode/capture.rs @@ -2109,8 +2109,7 @@ mod tests { let consumer = broadcast.consume(); let config = moq_mux::catalog::Config::default() .with_clock(clock) - .with_max_age(RETAIN) - .with_hls(moq_mux::hls::Config::default()); + .with_max_age(RETAIN); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap(); let mut options = Capture::default(); options.encode.track = Some("audio".into()); diff --git a/rs/moq-cli/MIGRATE.md b/rs/moq-cli/MIGRATE.md index 21c4b168e19..6571f46ad7f 100644 --- a/rs/moq-cli/MIGRATE.md +++ b/rs/moq-cli/MIGRATE.md @@ -4,11 +4,6 @@ How to update your invocation and config for each breaking change, newest first. ## Unreleased -### `import --hls` publishes the records `export hls` needs - -Imports no longer publish HLS records by default, so `export hls` lists nothing for a broadcast published without them. -Pass `--hls` to `import` (the stdin formats, `hls`, `rtsp`, and `capture`); renditions' segments line up wherever their keyframes share timestamps. - ### `--broadcast` refuses empty segments ([#6648](https://github.com/moq-dev/moq/pull/6648)) A broadcast name or prefix with a leading, trailing, or doubled `/` is an error instead of being collapsed: pass `--broadcast room/cam`, not `/room/cam/`. diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 349f8ace172..049e40ec9e8 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -357,12 +357,6 @@ impl Invocation { ); } } - if let Command::Import(import) = command - && import.hls - && let Some(reason) = import.source.hls_refusal() - { - anyhow::bail!("--hls records the segments of the catalog the import builds, but {reason}"); - } if let Command::Import(import) = command && import.linger.is_some() && let Some(reason) = import.source.linger_refusal() @@ -814,12 +808,6 @@ pub struct Import { #[usage(long = "latency-max", hide = true)] latency_max: Option, - /// Publish HLS segment records for every video and audio rendition, so `export hls` (or - /// any HLS/DASH edge) can serve the broadcast. The source must align keyframes across its - /// renditions. Not for `import rtmp`, `srt`, `rtc`, or `archive`. - #[usage(long)] - pub hls: bool, - /// Name a media report track in the catalog's `stats` section and fill it with what each /// track has written, about once a second while subscribed. The stdin imports only. #[usage(long)] @@ -924,19 +912,6 @@ impl ImportSource { self.catalog_refusal() } - /// Why `--hls` can't apply to this source, or `None` when it can: only a catalog the CLI - /// builds itself records segments. - fn hls_refusal(&self) -> Option<&'static str> { - match self { - Self::Rtmp(_) | Self::Srt(_) => Some("the gateway builds its own catalog"), - Self::Rtc(_) => Some("`import rtc` builds its own catalog"), - Self::Archive(_) => Some("`import archive` replays the catalog it recorded"), - #[cfg(feature = "capture")] - Self::Capture(_) => None, - Self::Avc3 | Self::Fmp4(_) | Self::Ts(_) | Self::Flv | Self::Hls(_) | Self::Rtsp(_) => None, - } - } - /// Why `--linger` can't apply to this source, or `None` when it can: only a source with a /// clean end has a finished broadcast to keep up. fn linger_refusal(&self) -> Option<&'static str> { diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs index b078c65e392..7711a85229e 100644 --- a/rs/moq-cli/src/hls.rs +++ b/rs/moq-cli/src/hls.rs @@ -168,15 +168,13 @@ pub async fn import(target: ImportTarget, args: ImportArgs, epoch: moq_net::Epoc name, max_age, linger, - hls, bandwidth, } = target; let producer = origin.create_broadcast(&name).context("failed to create broadcast")?; let config = moq_mux::catalog::Config::default() .with_max_age(max_age) - .with_bandwidth(bandwidth) - .with_hls(hls.then(moq_mux::hls::Config::default)); + .with_bandwidth(bandwidth); let catalog = moq_mux::catalog::Producer::new(&mut producer.broadcast().clone(), config)?; // Announced once the catalog first goes out, which waits for every rendition's init: a // relay queries a broadcast's tracks the moment it is announced, and keeps the refusals of a @@ -516,7 +514,6 @@ mod test { name: "hls".to_string(), max_age: None, linger: moq_mux::Linger::DEFAULT, - hls: false, bandwidth: Default::default(), }; let args = ImportArgs { diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 41508062042..eda6fb07fbb 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -733,7 +733,6 @@ fn spawn_import( name, max_age, linger, - hls: import.hls, bandwidth: bandwidth.clone(), }; @@ -743,8 +742,7 @@ fn spawn_import( warn_if_missing_format(&name); let config = moq_mux::catalog::Config::default() .with_max_age(max_age) - .with_bandwidth(bandwidth.clone()) - .with_hls(import.hls.then(moq_mux::hls::Config::default)); + .with_bandwidth(bandwidth.clone()); let mut publish = if let ImportSource::Ts(TsImport { program: Some(TsProgram::All), .. @@ -811,7 +809,7 @@ fn spawn_import( ImportSource::Capture(capture) => { warn_if_missing_format(&name); let broadcast = origin.create_broadcast(&name).context("failed to create broadcast")?; - let mut publish = Publish::capture(broadcast, &capture, bandwidth.clone(), max_age, import.hls)?; + let mut publish = Publish::capture(broadcast, &capture, bandwidth.clone(), max_age)?; publish.announce(epoch)?; local = Some(publish); } diff --git a/rs/moq-cli/src/moq.rs b/rs/moq-cli/src/moq.rs index e8321c7c39d..4d9ec745f2b 100644 --- a/rs/moq-cli/src/moq.rs +++ b/rs/moq-cli/src/moq.rs @@ -26,9 +26,6 @@ pub struct ImportTarget { /// How long a cleanly ended broadcast stays up (`--linger`). pub linger: std::time::Duration, - /// Whether the catalog records HLS segments (`--hls`). - pub hls: bool, - /// Connection allocator each passthrough track claims its peak-hold bitrate on, /// so a co-resident encoder targets what is left. Unlimited when there is no /// uplink estimate. diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index eda3a8a3fe3..068c293f6d1 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -478,12 +478,9 @@ impl Publish { args: &CaptureArgs, bandwidth: moq_net::bandwidth::Allocator, max_age: Option, - hls: bool, ) -> anyhow::Result { let mut broadcast = publication.broadcast().clone(); - let config = moq_mux::catalog::Config::default() - .with_max_age(max_age) - .with_hls(hls.then(moq_mux::hls::Config::default)); + let config = moq_mux::catalog::Config::default().with_max_age(max_age); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; let video = if args.no_video { diff --git a/rs/moq-cli/src/rtmp.rs b/rs/moq-cli/src/rtmp.rs index 6d9a1196f3f..8039444f956 100644 --- a/rs/moq-cli/src/rtmp.rs +++ b/rs/moq-cli/src/rtmp.rs @@ -71,7 +71,6 @@ pub async fn listen_import(target: ImportTarget, addr: SocketAddr) -> anyhow::Re name, max_age, linger, - hls: _, bandwidth, } = target; let name = moq_net::Path::new(&name)?; diff --git a/rs/moq-cli/src/rtsp.rs b/rs/moq-cli/src/rtsp.rs index 3e0e12290e2..2565277ca2f 100644 --- a/rs/moq-cli/src/rtsp.rs +++ b/rs/moq-cli/src/rtsp.rs @@ -75,8 +75,7 @@ async fn run(target: &ImportTarget, client: &moq_rtsp::Client, published: &mut b .context("failed to create broadcast")?; let config = moq_mux::catalog::Config::default() .with_max_age(target.max_age) - .with_bandwidth(target.bandwidth.clone()) - .with_hls(target.hls.then(moq_mux::hls::Config::default)); + .with_bandwidth(target.bandwidth.clone()); let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast.broadcast().clone(), config)?; // A relay asks for a broadcast's tracks the moment it is announced, so wait for the diff --git a/rs/moq-cli/src/srt.rs b/rs/moq-cli/src/srt.rs index 0454b462938..451a9bb3147 100644 --- a/rs/moq-cli/src/srt.rs +++ b/rs/moq-cli/src/srt.rs @@ -100,7 +100,6 @@ pub async fn listen_import( name, max_age, linger, - hls: _, bandwidth, } = target; let name = moq_net::Path::new(&name)?; diff --git a/rs/moq-gst/src/sink/pad.rs b/rs/moq-gst/src/sink/pad.rs index b817cd559d0..6bbae4ac086 100644 --- a/rs/moq-gst/src/sink/pad.rs +++ b/rs/moq-gst/src/sink/pad.rs @@ -88,6 +88,21 @@ impl Names { #[must_use] pub(super) struct Closed { track: Option, + /// The catalog `track` published in. + catalog: Option, +} + +impl Closed { + /// Drop the track and retire its HLS records, for a pad that finished mid-run: [`Names`] never + /// publishes its name again, so caps churn doesn't accumulate records. + fn retire(self) -> Result<()> { + let (Some(track), Some(catalog)) = (self.track, self.catalog) else { + return Ok(()); + }; + let name = track.name(); + drop(track); + Ok(catalog.retire(&name)?) + } } /// An audio or video pad, published through a codec importer. @@ -288,6 +303,8 @@ pub struct Pad { /// The run's, once reserved; until then this pad's own, which still covers renegotiation. names: Names, track: Option, + /// The catalog `track` publishes in, set with it. + catalog: Option, /// Whether `track` has a configured name, which may be requested again. A generated one never is. named: bool, caps: Option, @@ -312,6 +329,7 @@ impl Pad { wall: None, names: Names::default(), track: None, + catalog: None, named: false, caps: None, failed: false, @@ -394,7 +412,10 @@ impl Pad { } let reserved = self.reserved.take().unwrap_or_else(|| catalog.reserve()); match self.build(broadcast, reserved, options) { - Ok(name) => CapsOutcome::Active(name), + Ok(name) => { + self.catalog = Some(catalog.clone()); + CapsOutcome::Active(name) + } Err(err) => { gst::warning!(CAT, "invalidating pad: {err:?}"); self.fail(); @@ -899,10 +920,14 @@ impl Pad { /// Consumes the producer so a second call is a no-op (`Track::finish()` is not idempotent). Returns /// whether a producer was finalized. The importer accepts its track up front (in `Track::new`), so - /// `finish()` is safe even when no frame was ever decoded. + /// `finish()` is safe even when no frame was ever decoded. Unlike a run's end, it also retires the + /// track's HLS records. pub fn finalize(&mut self) -> Result { - let (_closed, result) = self.finish(); - result + let (closed, result) = self.finish(); + let retired = closed.retire(); + let finished = result?; + retired?; + Ok(finished) } /// Finish once, retaining rendition owners so closing a run cannot publish their removal. @@ -910,6 +935,7 @@ impl Pad { self.reserved.take(); let mut closed = Closed { track: self.track.take(), + catalog: self.catalog.take(), }; let Some(track) = closed.track.as_mut() else { return (closed, Ok(false)); @@ -1447,6 +1473,32 @@ mod tests { assert!(subscriber.recv_group().await.unwrap().is_none(), "the old track ended"); } + /// The old name never returns after a renegotiation, so its HLS records end with it instead of + /// accumulating with each caps change. + #[tokio::test(start_paused = true)] + async fn renegotiation_retires_the_old_records() { + gst::init().unwrap(); + let (broadcast, catalog) = producers(); + let mut pad = Pad::new(); + pad.observe_caps(&broadcast, &catalog, producer_options(&h264_caps(), Some("camera"))); + pad.observe_segment(time_segment()); + let pushed = pad.push_buffer(h264_keyframe_au(), Some(gst::ClockTime::ZERO), None, None); + assert_eq!(pushed, Ok(PushOutcome::Published)); + let mut records = broadcast + .consume() + .track(&hang::hls::default_name("camera")) + .unwrap() + .subscribe(None) + .await + .unwrap(); + + let vp8 = gst::Caps::builder("video/x-vp8").build(); + pad.observe_caps(&broadcast, &catalog, producer_options(&vp8, Some("camera"))); + let drained = async { while records.recv_group().await.unwrap().is_some() {} }; + let ended = tokio::time::timeout(Duration::from_secs(1), drained).await; + assert!(ended.is_ok(), "the old track's records end"); + } + /// A run never publishes a name it ended again, whoever asks for it next: each replacement mints /// the next free name. #[test] diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index 7163d60be3c..a211b17dba1 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -427,9 +427,9 @@ impl TrackState { /// Give up on this rendition's current importer generation after repeated failures. /// - /// Dropping the importer closes its recorders and retires its catalog entries, so players - /// stop selecting a rendition that no longer advances; a later successful step rebuilds it - /// from the init segment. + /// Dropping the importer ends its tracks and retires their catalog entries and HLS records, so + /// players stop selecting a rendition that no longer advances; a later successful step rebuilds + /// it from the init segment. fn evict(&mut self) { self.importer = None; self.map = None; @@ -2207,6 +2207,20 @@ mod tests { old_groups: Vec<(u64, u128)>, terminal: Option, new_groups: Vec<(u64, u128)>, + /// The first generation's HLS record track has ended. + old_records_ended: bool, + } + + /// Whether `subscriber` has ended, after draining the groups it already holds. + fn ended(subscriber: &mut moq_net::track::Ordered) -> bool { + loop { + match subscriber.poll_next_group(&kio::Waiter::noop()) { + Poll::Ready(Ok(Some(_))) => continue, + Poll::Ready(Ok(None)) => return true, + Poll::Pending => return false, + Poll::Ready(Err(error)) => panic!("the record track failed: {error:?}"), + } + } } /// A one-track fMP4 muxer for synthesized VP8 video or Opus audio. @@ -2287,6 +2301,13 @@ mod tests { .await .unwrap() .ordered(); + let mut old_records = consumer + .track(&hang::hls::default_name(&before_name)) + .unwrap() + .subscribe(replay.clone()) + .await + .unwrap() + .ordered(); let after = format!( "#EXTM3U\n#EXT-X-TARGETDURATION:2\n#EXT-X-MAP:URI=\"init.mp4\"\n#EXTINF:1,\n0.m4s\n{tags}#EXTINF:1,\n1.m4s\n" ); @@ -2294,6 +2315,7 @@ mod tests { let result = track.consume_segments(&local, &after, &base).await; let after_name = name(&catalog.snapshot()); let (old_groups, terminal) = group_stamps(&mut old); + let old_records_ended = ended(&mut old_records); let new_groups = if before_name != after_name { let mut new = consumer .track(&after_name) @@ -2316,6 +2338,7 @@ mod tests { old_groups, terminal, new_groups, + old_records_ended, } } @@ -2426,6 +2449,7 @@ mod tests { case.result.unwrap(); assert!(case.terminal.is_none(), "{terminal:?}", terminal = case.terminal); assert_ne!(case.before, case.after); + assert!(case.old_records_ended, "the replaced generation retires its records"); assert_eq!(case.old_groups, [(0, 5_000_000)]); assert_eq!(case.new_groups, [(moq_sequence(1, 1).unwrap(), u128::from(after_us))]); } diff --git a/rs/moq-mux/MIGRATE.md b/rs/moq-mux/MIGRATE.md index 45d078de028..14b133c68a8 100644 --- a/rs/moq-mux/MIGRATE.md +++ b/rs/moq-mux/MIGRATE.md @@ -26,12 +26,12 @@ Match `Target::Enabled(rate)` where you took the rate before. A policy without ` A TS PMT listing a PID twice routes it by the later entry instead of failing the import. Delete matches on `DuplicateTrackName`. -### HLS records are opt-in and replace timelines +### HLS records replace timelines -The `timeline` module is replaced by `hls`, and nothing publishes records by default. -Enable them with `catalog::Config::with_hls(hls::Config::default())`; `with_timeline` is gone. +The `timeline` module is replaced by `hls`, and records are still published by default. +`catalog::Config::with_timeline(config)` is `with_hls(config)`, and `with_hls(None)` publishes none. `catalog::Reserved::with_timeline_hint` is `with_hls_hint`, and `catalog::Producer::timeline()` is gone. -Each video and audio rendition then advertises its own `hls` field and `.hls` record track, cut at group boundaries; untimed and data tracks get none. +Each video and audio rendition advertises its own `hls` field and `.hls` record track, cut at group boundaries; untimed and data tracks get none. Read records with `hls::Consumer::subscribe(&broadcast, &hls)`, whose `Event::Push` carries an `hls::Entry`, or publish your own with `hls::Producer`. `Config::duration_min` is `hls::Config::duration`, `with_duration_min` is `with_duration`, and `DEFAULT_DURATION_MIN` is `DEFAULT_DURATION`. It is now the spacing of a pts grid rather than a floor: a group starts a segment when its pts falls in a later multiple of it than the previous group's, so a segment may be shorter. diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index b1f74ca9ac7..5df43ad42b5 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -166,7 +166,7 @@ struct Outputs { hang_track: moq_net::track::Producer, hangz: moq_json::snapshot::Producer>, msf_track: moq_net::track::Producer, - /// The broadcast's HLS record tracks, when [`Config::with_hls`] enabled them. + /// The broadcast's HLS record tracks, unless [`Config::with_hls`] disabled them. hls: Option, /// Keeps the broadcast open while the catalog is, so a publisher may hand its own handle to a /// media producer and drop it. @@ -275,7 +275,7 @@ pub struct Producer { current: Arc>>, - /// How each media track's HLS segments are cut, when enabled. See [`Config::with_hls`]. + /// How each media track's HLS segments are cut, unless disabled. See [`Config::with_hls`]. hls: Option, /// Retention override for the media tracks minted under this catalog, or `None` to keep /// hang's default. Fixed at construction, so every clone and every @@ -330,7 +330,7 @@ impl Default for Config<()> { max_age: None, bandwidth: moq_net::bandwidth::Allocator::unlimited(), clock: None, - hls: None, + hls: Some(crate::hls::Config::default()), } } } @@ -393,9 +393,8 @@ impl Config { self } - /// Publish HLS segment records for each video and audio rendition, cut by `hls`; off by default. - /// - /// See [`hls`](crate::hls) for what that costs and what it promises. + /// Publish HLS segment records for each video and audio rendition, cut by `hls`; `None` publishes + /// none. On by default, cut by [`hls::Config::default`](crate::hls::Config::default). pub fn with_hls(mut self, hls: impl Into>) -> Self { self.hls = hls.into(); self @@ -794,7 +793,7 @@ impl Producer { } /// Build the media [`container::Producer`](crate::container::Producer) for `track`, - /// enrolling a video or audio track for HLS records when [`Config::with_hls`] enabled them. + /// enrolling a video or audio track for HLS records unless [`Config::with_hls`] disabled them. pub(super) fn media( &self, track: moq_net::track::Producer, @@ -871,7 +870,7 @@ impl Producer { self.bandwidth.clone() } - /// Enroll `track` for HLS records when [`Config::with_hls`] enabled them, advertising them in its + /// Enroll `track` for HLS records unless [`Config::with_hls`] disabled them, advertising them in its /// rendition's `hls` field once the nominal duration is known. /// /// The video and audio track constructors call this for you. fMP4 passthrough calls it directly @@ -921,16 +920,21 @@ impl Producer { self.outputs.hls.as_ref().map_or(0, |tracks| tracks.len()) } - /// Retire `track`'s HLS records once its producer dropped, for an owner that never enrolls the - /// name again: the record track finishes and is forgotten. The catalog stops advertising it - /// with the rendition that carried it. - pub(crate) fn retire(&self, track: &str) -> crate::Result<()> { + /// Finish and forget the HLS records of a dropped `track` whose name is never published again. + pub fn retire(&self, track: &str) -> crate::Result<()> { match &self.outputs.hls { Some(tracks) => tracks.retire(track), None => Ok(()), } } + /// [`retire`](Self::retire) `track` from a teardown that cannot return the error. + pub(crate) fn retire_dropped(&self, track: &str) { + if let Err(err) = self.retire(track) { + tracing::warn!(%err, track, "failed to retire a dropped track's HLS records"); + } + } + /// Publish `track` as a latest-value JSON track, advertising it in the catalog. /// /// The caller creates the track on the broadcast, as it does for a media track; this writes its @@ -1503,7 +1507,7 @@ mod test { #[tokio::test(start_paused = true)] async fn abort_fails_the_hls_records() { let mut broadcast = moq_net::broadcast::Info::default().produce(); - let catalog = Producer::<()>::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::<()>::new(&mut broadcast, Config::default()).unwrap(); let _recorder = catalog .enroll("video0", Some(std::time::Duration::from_secs(4))) .unwrap(); @@ -1558,7 +1562,7 @@ mod test { use hang::catalog::PRIORITY; let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); let _recorder = catalog.enroll("video", None).unwrap(); let consumer = broadcast.consume(); @@ -1955,11 +1959,6 @@ mod test { ); } - /// A catalog publishing HLS records. - fn hls() -> Config { - Config::default().with_hls(crate::hls::Config::default()) - } - /// A video rendition `video0` on `catalog`, with a keyframe group at each of `starts` (ms). fn video0( broadcast: &mut moq_net::broadcast::Producer, @@ -1992,7 +1991,7 @@ mod test { #[test] fn hls_reports_a_track_collision() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); // Something else already took the name the record track wants. let _taken = broadcast.create_track("video0.hls", None).unwrap(); @@ -2000,9 +1999,9 @@ mod test { } #[test] - fn nothing_publishes_hls_unless_enabled() { + fn nothing_publishes_hls_when_disabled() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default().with_hls(None)).unwrap(); assert!( catalog .enroll("audio0", Some(std::time::Duration::from_secs(4))) @@ -2019,7 +2018,7 @@ mod test { #[tokio::test] async fn an_estimated_duration_is_advertised_after_its_first_segment() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); let mut consumer = catalog.consume().unwrap(); let mut video = video0(&mut broadcast, &catalog, &[0]); @@ -2044,7 +2043,7 @@ mod test { #[test] fn a_track_finished_before_its_first_segment_is_advertised() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); let mut video = video0(&mut broadcast, &catalog, &[0]); video.finish().unwrap(); @@ -2077,7 +2076,7 @@ mod test { #[test] fn only_the_enrolled_rendition_gets_its_records() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::<()>::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::<()>::new(&mut broadcast, Config::default()).unwrap(); let _video = catalog .enroll("video", Some(std::time::Duration::from_secs(2))) .unwrap(); @@ -2110,7 +2109,7 @@ mod test { #[test] fn retiring_a_track_ends_its_records() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); let recorder = catalog .enroll("video0", Some(std::time::Duration::from_secs(4))) .unwrap(); @@ -2135,7 +2134,7 @@ mod test { #[test] fn a_text_track_neither_records_nor_paces() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); let net = broadcast .create_track("text0", catalog.track_info(hang::catalog::PRIORITY.text)) .unwrap(); @@ -2158,7 +2157,7 @@ mod test { #[test] fn an_undeclared_enrollment_releases_the_broadcast() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, hls()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); let recorder = catalog.enroll("video0", None).unwrap(); let waiter = kio::Waiter::noop(); let mut subscribing = broadcast @@ -2179,7 +2178,7 @@ mod test { #[test] fn clock_is_advertised_without_hls() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + let catalog = Producer::new(&mut broadcast, Config::default().with_hls(None)).unwrap(); let snapshot = catalog.snapshot(); assert_eq!(snapshot.clock, catalog.clock().wall()); diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index 118700506b5..018ef2a12eb 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -393,7 +393,7 @@ impl Reserved { } /// Declare the track's regular group duration, which fixes its HLS segment duration up front - /// when [`Config::with_hls`](super::Config::with_hls) enabled HLS records. + /// unless [`Config::with_hls`](super::Config::with_hls) disabled HLS records. pub fn with_hls_hint(mut self, duration: impl Into>) -> Self { self.hls_hint = duration.into(); self diff --git a/rs/moq-mux/src/container/flv/import.rs b/rs/moq-mux/src/container/flv/import.rs index 4d17afcedfd..49780014e42 100644 --- a/rs/moq-mux/src/container/flv/import.rs +++ b/rs/moq-mux/src/container/flv/import.rs @@ -570,28 +570,40 @@ impl Import { Ok(()) } - /// Drop any existing video track `track_id` (finishing it and clearing its - /// catalog rendition) and allocate a fresh one. + /// Retire any existing video track `track_id` and allocate a fresh one. fn replace_video(&mut self, track_id: u8) -> anyhow::Result { - if let Some(mut old) = self.video.remove(&track_id) { - old.track.finish()?; + if let Some(old) = self.video.remove(&track_id) { + self.retire(old.track)?; } Ok(self .broadcast .unique_track(".flv-v", self.catalog.track_info(hang::catalog::PRIORITY.video))?) } - /// Drop any existing audio track `track_id` (finishing it and clearing its - /// catalog rendition) and allocate a fresh one. + /// Retire any existing audio track `track_id` and allocate a fresh one. fn replace_audio(&mut self, track_id: u8) -> anyhow::Result { - if let Some(mut old) = self.audio.remove(&track_id) { - old.track.finish()?; + if let Some(old) = self.audio.remove(&track_id) { + self.retire(old.track)?; } Ok(self .broadcast .unique_track(".flv-a", self.catalog.track_info(hang::catalog::PRIORITY.audio))?) } + /// End a track this importer is done with, clearing its catalog rendition and retiring its HLS + /// records. Every track takes a fresh name, so a publisher changing its config holds only its live + /// tracks. + fn retire( + &self, + mut track: crate::container::Producer, + ) -> anyhow::Result<()> { + let ended = track.end(); + let name = track.name().to_string(); + drop(track); + self.catalog.retire(&name)?; + Ok(ended?) + } + /// Close the current group on every track and reopen at `sequence`. pub fn seek(&mut self, sequence: u64) -> crate::Result<()> { for stream in self.video.values_mut() { @@ -622,6 +634,9 @@ impl Import { /// generic [`moq_net::Error::Dropped`] a bare drop surfaces. The counterpart to /// [`Self::finish`] for a failed teardown (e.g. the RTMP client disconnected). /// Consumes the importer. + /// + /// Unlike a drop, this leaves the tracks' HLS records to the catalog, whose abort fails them + /// rather than ending them as a complete recording. pub fn abort(mut self, err: moq_net::Error) { for stream in std::mem::take(&mut self.video).into_values() { stream.track.abort(err.clone()); @@ -632,6 +647,19 @@ impl Import { } } +impl Drop for Import { + fn drop(&mut self) { + let (video, audio) = (std::mem::take(&mut self.video), std::mem::take(&mut self.audio)); + let video = video.into_values().map(|stream| self.retire(stream.track)); + let audio = audio.into_values().map(|stream| self.retire(stream.track)); + for result in video.chain(audio) { + if let Err(err) = result { + tracing::warn!(%err, "failed to end a dropped FLV track"); + } + } + } +} + /// The multitrack framing common to every track in one tag: the layout and the /// real `VideoPacketType`/`AudioPacketType`, plus the shared FourCC (present for /// every layout except `ManyTracksManyCodecs`, where each track carries its own). diff --git a/rs/moq-mux/src/container/flv/import_test.rs b/rs/moq-mux/src/container/flv/import_test.rs index 8c9517491af..9c68a77a074 100644 --- a/rs/moq-mux/src/container/flv/import_test.rs +++ b/rs/moq-mux/src/container/flv/import_test.rs @@ -89,9 +89,8 @@ fn synth_flv() -> Vec { /// A rendition must never be advertised when its media producer could not be built. /// -/// Publishing the media producer is fallible (with HLS records enabled it mints the track's record -/// track, which can collide), so publishing the catalog entry first would -/// leave consumers a rendition that is announced but has no producer behind it and is therefore +/// Publishing the media producer is fallible (it mints the track's HLS record track, which can +/// collide), so publishing the catalog entry first would leave consumers a rendition that is announced but has no producer behind it and is therefore /// never served. #[tokio::test(start_paused = true)] async fn rendition_is_not_published_when_the_media_track_fails() { @@ -110,8 +109,7 @@ async fn rendition_is_not_published_when_the_media_track_fails() { }; let mut broadcast = moq_net::broadcast::Info::new().produce(); - let config = crate::catalog::Config::default().with_hls(crate::hls::Config::default()); - let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); // Squat the rendition's HLS record track, so enrolling it (and with it building its media // producer) fails. The handle must stay alive: the broadcast tracks names weakly, so @@ -465,6 +463,80 @@ async fn import_multitrack_video_many_codecs() { assert_eq!(catalog.snapshot().video.renditions.len(), 2); } +/// A publisher that keeps changing its audio and video config replaces each track under a fresh +/// name. A replaced track flushes its last frame and ends, and the importer retires it, so the HLS +/// records it holds stay those of the live renditions. Dropping the importer ends and retires those. +#[tokio::test(start_paused = true)] +async fn config_churn_keeps_flat_state() { + const NALU: [u8; 9] = [0, 0, 0, 5, 0x65, 0x88, 0x84, 0x21, 0x00]; + // AAC-LC stereo at 44.1 kHz, then at 48 kHz. + let ascs = [ASC, [0x11, 0x90]]; + let avccs = [avcc_level(0x1f), avcc_level(0x1e)]; + // One config change on both tracks, each followed by a frame. + let round = |round: u32| { + let mut out = Vec::new(); + let config = round as usize % 2; + let timestamp = round * 40; + let video = (super::FRAME_TYPE_KEY << 4) | super::VIDEO_CODEC_AVC; + + let mut vseq = vec![video, super::AVC_SEQUENCE_HEADER, 0, 0, 0]; + vseq.extend_from_slice(&avccs[config]); + write_tag(&mut out, super::TAG_VIDEO, timestamp, &vseq); + let mut aseq = vec![super::AAC_AUDIO_TAG_HEADER, super::AAC_SEQUENCE_HEADER]; + aseq.extend_from_slice(&ascs[config]); + write_tag(&mut out, super::TAG_AUDIO, timestamp, &aseq); + + let mut vframe = vec![video, super::AVC_NALU, 0, 0, 0]; + vframe.extend_from_slice(&NALU); + write_tag(&mut out, super::TAG_VIDEO, timestamp, &vframe); + let aframe = [super::AAC_AUDIO_TAG_HEADER, super::AAC_RAW, 0x21, 0x00]; + write_tag(&mut out, super::TAG_AUDIO, timestamp, &aframe); + out + }; + + let mut producer = moq_net::broadcast::Info::new().produce(); + let consumer = producer.consume(); + let catalog = crate::catalog::Producer::new(&mut producer, crate::catalog::Config::default()).unwrap(); + let mut importer = Import::new(producer, catalog.reserve()); + let mut first = flv_header(0x05); + first.extend_from_slice(&round(0)); + importer.decode(&first).unwrap(); + let replaced = catalog.snapshot().video.renditions.into_keys().next().unwrap(); + let replaced = consumer + .track(&replaced) + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(1))) + .await + .unwrap(); + + for n in 1..100 { + importer.decode(&round(n)).unwrap(); + } + let snap = catalog.snapshot(); + assert_eq!(snap.video.renditions.len(), 1); + assert_eq!(snap.audio.renditions.len(), 1); + assert_eq!(catalog.hls_tracks(), 2, "only the live renditions keep records"); + + let live = consumer + .track(snap.video.renditions.keys().next().unwrap()) + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(1))) + .await + .unwrap(); + drop(importer); + assert_eq!(catalog.hls_tracks(), 0, "a dropped importer retires its records"); + + for (track, label) in [(replaced, "replaced"), (live, "dropped")] { + let mut track = crate::container::Consumer::new( + track, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Data), + ); + let frame = track.read_frame().await.unwrap(); + assert_eq!(frame.expect(label).payload.as_ref(), &NALU, "{label}"); + assert!(track.read_frame().await.unwrap().is_none(), "the {label} track ends"); + } +} + #[tokio::test(start_paused = true)] async fn import_enhanced_av1() { let mut out = flv_header(0x01); diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index f74510af98d..e4d4f59cd0c 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -156,7 +156,7 @@ struct Passthrough { /// The latest group's start, which the next group may not start below. start: Option, - // Reports this track's groups into its HLS records, when enabled. Passthrough writes groups by + // Reports this track's groups into its HLS records, unless disabled. Passthrough writes groups by // hand (no `container::Producer`), so the recorder is fed directly rather than through // `with_recorder`. recorder: Option, @@ -193,6 +193,25 @@ impl Passthrough { } Ok(()) } + + /// Close the open group and finish the track. + fn finish(&mut self) -> crate::Result<()> { + self.close()?; + self.track.finish()?; + // The importer may be kept around, but its media stops holding the other tracks back. + self.estimator.discontinuity(); + // Dropping the recorder stops the open HLS segment, so the track is advertised. + self.recorder = None; + Ok(()) + } + + /// [`finish`](Self::finish) the track unless it already ended, for an importer about to drop it. + fn end(&mut self) -> crate::Result<()> { + if self.track.finished().is_some() || self.track.is_closed() { + return Ok(()); + } + self.finish() + } } /// Forward a call to whichever unwrapped producer a track writes. @@ -1362,20 +1381,16 @@ impl Import { // No frame follows to anchor the clock, so publish the declared track set now. self.initial_reservation = None; for track in self.tracks.values_mut() { - media!(&mut track.output, producer => producer.finish()?, passthrough => { - passthrough.close()?; - passthrough.track.finish()?; - // The importer may be kept around, but its media stops holding the other tracks back. - passthrough.estimator.discontinuity(); - // Dropping the recorder stops the open HLS segment, so the track is advertised. - passthrough.recorder = None; - }); + media!(&mut track.output, producer => producer.finish()?, passthrough => passthrough.finish()?); } Ok(()) } /// Abort all tracks with `err` instead of finishing, so subscribers see the real /// cause rather than [`moq_net::Error::Dropped`]. Consumes the importer. + /// + /// Unlike a drop, this leaves the tracks' HLS records to the catalog, whose abort fails them + /// rather than ending them as a complete recording. pub fn abort(mut self, err: moq_net::Error) { for track in std::mem::take(&mut self.text).into_values() { track.producer.abort(err.clone()); @@ -1433,3 +1448,25 @@ impl Import { Ok(()) } } + +impl Drop for Import { + /// End each track, then retire its HLS records: every track takes a fresh name, so nothing + /// publishes it again, as when an HLS import replaces this init segment's importer. + fn drop(&mut self) { + for track in self.text.values_mut() { + if let Err(err) = track.producer.end() { + tracing::warn!(%err, "failed to end a dropped fMP4 text track"); + } + } + for mut track in std::mem::take(&mut self.tracks).into_values() { + let (name, ended) = media!(&mut track.output, + producer => (producer.name().to_string(), producer.end()), + passthrough => (passthrough.track.name().to_string(), passthrough.end())); + drop(track); + if let Err(err) = ended { + tracing::warn!(%err, "failed to end a dropped fMP4 track"); + } + self.catalog.retire_dropped(&name); + } + } +} diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 52624e2df23..6f051634294 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1414,11 +1414,6 @@ async fn test_msf_catalog_roundtrip() { assert!(matches!(audio.container, Container::Cmaf { .. })); } -/// A catalog publishing HLS records. -fn hls() -> crate::catalog::Config { - crate::catalog::Config::default().with_hls(crate::hls::Config::default()) -} - /// Subscribe to `track`'s HLS records, before the catalog may advertise them. async fn subscribe(consumer: &moq_net::broadcast::Consumer, track: &str) -> crate::hls::Consumer { let hls = hang::catalog::Hls::new(hang::hls::default_name(track), std::time::Duration::ZERO); @@ -1451,7 +1446,7 @@ async fn groups(records: &mut crate::hls::Consumer) -> Vec<(u64, u64)> { async fn import_populates_the_hls_records() { let mut broadcast = moq_net::broadcast::Info::new().produce(); let consumer = broadcast.consume(); - let mut catalog = crate::catalog::Producer::new(&mut broadcast, hls()).unwrap(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()); let data = include_bytes!("test_data/bbb.mp4"); @@ -1812,7 +1807,7 @@ async fn segmented_source_indexes_one_group_per_record_in(passthrough: bool) { let mut broadcast = moq_net::broadcast::Info::new().produce(); let consumer = broadcast.consume(); - let mut catalog = crate::catalog::Producer::new(&mut broadcast, hls()).unwrap(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()).with_passthrough(passthrough); fmp4.decode(&init).unwrap(); @@ -1867,7 +1862,7 @@ async fn records_with_skew(video_offset_us: i64, passthrough: bool) -> (Vec<(u64 let mut broadcast = moq_net::broadcast::Info::new().produce(); let consumer = broadcast.consume(); - let mut catalog = crate::catalog::Producer::new(&mut broadcast, hls()).unwrap(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()).with_passthrough(passthrough); fmp4.decode(&init).unwrap(); @@ -2317,7 +2312,7 @@ async fn import_session(chunks: &[Vec]) -> Imported { let (init, _, _) = bbb_init(); let mut broadcast = moq_net::broadcast::Info::new().produce(); let consumer = broadcast.consume(); - let mut catalog = crate::catalog::Producer::new(&mut broadcast, hls()).unwrap(); + let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()); fmp4.decode(&init).unwrap(); diff --git a/rs/moq-mux/src/container/mkv/import.rs b/rs/moq-mux/src/container/mkv/import.rs index 35c882ab3e5..a31b9f13589 100644 --- a/rs/moq-mux/src/container/mkv/import.rs +++ b/rs/moq-mux/src/container/mkv/import.rs @@ -148,6 +148,14 @@ impl Media { } } + fn end(&mut self) -> crate::Result<()> { + match self { + Self::Video(track) => track.end(), + Self::Audio(track) => track.end(), + Self::Text(track) => track.end(), + } + } + fn abort(self, err: moq_net::Error) { match self { Self::Video(track) => track.abort(err), @@ -155,6 +163,14 @@ impl Media { Self::Text(track) => track.abort(err), } } + + fn name(&self) -> &str { + match self { + Self::Video(track) => track.name(), + Self::Audio(track) => track.name(), + Self::Text(track) => track.name(), + } + } } impl Import { @@ -616,6 +632,9 @@ impl Import { /// Abort all tracks with `err` instead of finishing, so subscribers see the real /// cause rather than [`moq_net::Error::Dropped`]. Consumes the importer. + /// + /// Unlike a drop, this leaves the tracks' HLS records to the catalog, whose abort fails them + /// rather than ending them as a complete recording. pub fn abort(mut self, err: moq_net::Error) { for mut track in std::mem::take(&mut self.tracks).into_values() { if let Some(g) = track.group.take() { @@ -626,6 +645,22 @@ impl Import { } } +impl Drop for Import { + /// End each track, then retire its HLS records: every track takes a fresh name, so nothing + /// publishes it again. + fn drop(&mut self) { + for mut track in std::mem::take(&mut self.tracks).into_values() { + let name = track.track.name().to_string(); + let ended = track.track.end(); + drop(track); + if let Err(err) = ended { + tracing::warn!(%err, "failed to end a dropped MKV track"); + } + self.catalog.retire_dropped(&name); + } + } +} + fn build_video_config( codec_id: &str, codec_private: Option<&Bytes>, diff --git a/rs/moq-mux/src/container/mkv/import_test.rs b/rs/moq-mux/src/container/mkv/import_test.rs index 01c3fe792b6..ebf4544f1b7 100644 --- a/rs/moq-mux/src/container/mkv/import_test.rs +++ b/rs/moq-mux/src/container/mkv/import_test.rs @@ -958,9 +958,8 @@ fn unsupported_audio_codec_is_refused() { /// A rendition must never be advertised when its media producer could not be built. /// -/// Publishing the media producer is fallible (with HLS records enabled it mints the track's record -/// track, which can collide), so publishing the catalog entry first would -/// leave consumers a rendition that is announced but has no producer behind it and is therefore +/// Publishing the media producer is fallible (it mints the track's HLS record track, which can +/// collide), so publishing the catalog entry first would leave consumers a rendition that is announced but has no producer behind it and is therefore /// never served. #[test] fn rendition_is_not_published_when_the_media_track_fails() { @@ -979,8 +978,7 @@ fn rendition_is_not_published_when_the_media_track_fails() { let name = control.video.renditions.keys().next().unwrap().clone(); let mut broadcast = moq_net::broadcast::Info::new().produce(); - let config = crate::catalog::Config::default().with_hls(crate::hls::Config::default()); - let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); // Squat the rendition's HLS record track, so enrolling it (and with it building its media // producer) fails. The handle must stay alive: the broadcast tracks names weakly, so diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index b53ba1729be..08d482f472f 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -58,7 +58,7 @@ pub struct Producer { /// Set by [`Self::seek`] and consumed on the next group creation. pending_sequence: Option, - /// Reports each written group into this rendition's HLS records, when the catalog enabled them. + /// Reports each written group into this rendition's HLS records, unless the catalog disabled them. recorder: Option, /// The furthest presentation point written, i.e. `max(timestamp + duration)`. Reported to @@ -748,6 +748,14 @@ where Ok(()) } + /// [`finish`](Self::finish) the track unless it already ended, for an owner about to drop it. + pub(crate) fn end(&mut self) -> crate::Result<()> { + if self.inner.finished().is_some() || self.inner.is_closed() { + return Ok(()); + } + self.finish() + } + /// Abort the track and any open group with the given error. /// /// The counterpart to [`Self::finish`] for a failed teardown: consumers observe diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index e115089876f..4b48737537c 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -232,6 +232,9 @@ pub struct Import { described: bool, /// Carried PIDs already warned about a TS-scrambled packet. scrambled: HashSet, + /// Whether [`finish`](Self::finish) or [`abort`](Self::abort) ended the input, even if finishing + /// failed part way. + finished: bool, } impl Import { @@ -303,6 +306,7 @@ impl Import { alone: false, described: false, scrambled: HashSet::new(), + finished: false, } } @@ -1766,6 +1770,7 @@ impl Import { /// Flush any buffered PES and finish every track. pub fn finish(&mut self) -> anyhow::Result<()> { + self.finished = true; self.demux.finish(); for pid in self.headers.keys().copied().collect::>() { self.damage(pid, &anyhow::anyhow!("PES header truncated at end of input"))?; @@ -1875,7 +1880,11 @@ impl Import { /// Abort every track with `err` instead of finishing, so subscribers see the /// real cause rather than [`moq_net::Error::Dropped`]. Buffered PES is discarded. /// Consumes the importer. + /// + /// Unlike a drop, this leaves the tracks' HLS records to the catalog, whose abort fails them + /// rather than ending them as a complete recording. pub fn abort(mut self, err: moq_net::Error) { + self.finished = true; for stream in std::mem::take(&mut self.streams).into_values() { stream.abort(err.clone()); } @@ -1890,6 +1899,25 @@ impl Import { } } +impl Drop for Import { + /// End the input unless it already ended, then retire every route's HLS records: each route + /// mints its own name, so nothing publishes it again. + fn drop(&mut self) { + if !self.finished + && let Err(err) = self.finish() + { + tracing::warn!(%err, "failed to finish a dropped TS import"); + } + for stream in std::mem::take(&mut self.streams).into_values() { + let name = stream.track_name(); + drop(stream); + if let Some(name) = name { + self.catalog.retire_dropped(&name); + } + } + } +} + /// A PAT listed more than one program, and the [`Import`] was not told which one to take. /// /// Importing them all onto one broadcast would put unrelated clocks on one timeline, so the @@ -10995,9 +11023,7 @@ pub(super) mod test { let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = crate::catalog::Producer::new( &mut broadcast, - crate::catalog::Config::default() - .with_catalog(crate::catalog::hang::Catalog::::default()) - .with_hls(crate::hls::Config::default()), + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), ) .unwrap(); let mut import = super::Import::new(broadcast, catalog.reserve()); diff --git a/rs/moq-mux/src/hls.rs b/rs/moq-mux/src/hls.rs index 970d8f80820..09c178688b0 100644 --- a/rs/moq-mux/src/hls.rs +++ b/rs/moq-mux/src/hls.rs @@ -1,11 +1,11 @@ -//! Opt-in HLS segment records: where each of a rendition's HLS segments starts. +//! HLS segment records: where each of a rendition's HLS segments starts. //! -//! Nothing here runs unless the application enables it with -//! [`catalog::Config::with_hls`](crate::catalog::Config::with_hls). Then every video and audio -//! rendition the catalog mints gets a track of [`hang::hls::Record`]s, one per segment, and its -//! catalog entry gets an [`hls`](hang::catalog::Hls) field naming that track and the nominal -//! segment duration. An edge renders HLS (or DASH) playlists from the records without subscribing -//! to media. +//! Every video and audio rendition a catalog mints gets a track of [`hang::hls::Record`]s, one per +//! segment, and its catalog entry gets an [`hls`](hang::catalog::Hls) field naming that track and +//! the nominal segment duration, unless the application disables them with +//! [`catalog::Config::with_hls`](crate::catalog::Config::with_hls). An edge renders HLS (or DASH) +//! playlists from the records without subscribing to media. A record track nobody subscribes to +//! costs only the in-process bookkeeping. //! //! ## Segments //! @@ -139,6 +139,8 @@ pub(crate) struct Segmenter { /// The newest boundary declared, so a repeat of it (several sources declaring the same ones) /// is ignored. cut: Option, + /// A boundary was queued since the newest group started. + queued: bool, /// Whether boundaries come only from cuts: the source declares them. manual: bool, ready: VecDeque, @@ -163,6 +165,7 @@ impl Segmenter { frontier: None, cuts: VecDeque::new(), cut: None, + queued: false, manual: false, ready: VecDeque::new(), closed: false, @@ -216,6 +219,9 @@ impl Segmenter { } else { self.begin(group, pts); } + if self.last != Some(group) { + self.queued = false; + } self.last = Some(group); self.advance(pts); Ok(()) @@ -250,6 +256,10 @@ impl Segmenter { /// Start a segment with group `group` at `pts`. fn begin(&mut self, group: u64, pts: Timestamp) { + // However the segment starts, a cut it reached is spent. + while self.cuts.front().is_some_and(|cut| cut.as_micros() <= pts.as_micros()) { + self.cuts.pop_front(); + } self.open = Some(pts); self.frontier = Some(pts); self.cell = self.cell(pts); @@ -260,6 +270,7 @@ impl Segmenter { /// Whether a group starting at `pts` in grid cell `cell` ends the segment that started at /// `start`. fn boundary(&mut self, start: Timestamp, pts: Timestamp, cell: u128) -> bool { + // A cut declared after its segment had already started is spent. while self .cuts .front() @@ -267,13 +278,10 @@ impl Segmenter { { self.cuts.pop_front(); } - if self.cuts.front().is_some_and(|cut| cut.as_micros() <= pts.as_micros()) { - while self.cuts.front().is_some_and(|cut| cut.as_micros() <= pts.as_micros()) { - self.cuts.pop_front(); - } - return true; + match self.cuts.front() { + Some(cut) if cut.as_micros() <= pts.as_micros() => true, + _ => !self.manual && cell > self.cell, } - !self.manual && cell > self.cell } /// Report that the content extends to `pts`, without a frame starting there. @@ -301,7 +309,9 @@ impl Segmenter { /// /// The first cut takes over from grid pacing for good, since the grid would otherwise end a /// segment just before the caller declares where it really ends. A cut at or before an earlier - /// one is ignored, so several sources declaring the same boundaries cost nothing. + /// one is ignored, so several sources declaring the same boundaries cost nothing. So is a cut + /// after one no group has started since: the next group ends the segment at the earlier one, and + /// a source declaring segments without sync samples would otherwise queue cuts without bound. pub fn cut(&mut self, pts: Timestamp) { if self.closed { return; @@ -311,7 +321,9 @@ impl Segmenter { return; } self.cut = Some(pts); - self.cuts.push_back(pts); + if !std::mem::replace(&mut self.queued, true) { + self.cuts.push_back(pts); + } } /// Content stopped before group `group`: publish a zero-duration record where it reached. @@ -354,7 +366,7 @@ impl Iterator for Segmenter { } /// Publishes one track's records onto its HLS record track, for a publisher that segments -/// itself; [`catalog::Config::with_hls`](crate::catalog::Config::with_hls) records them for you. +/// itself; a [`catalog::Producer`](crate::catalog::Producer) records them for you. /// /// Records must be pushed in sequence order, starting at the window's next index. pub struct Producer { @@ -958,15 +970,53 @@ mod test { #[test] fn explicit_cuts_override_the_grid() { let mut segmenter = Segmenter::new(Config::default()); - segmenter.cut(ms(2_000)); - segmenter.cut(ms(3_000)); - groups(&mut segmenter, 0..5, 1_000); + for group in 0..5u64 { + if matches!(group, 2 | 3) { + segmenter.cut(ms(group * 1_000)); + } + segmenter.frame(group, ms(group * 1_000)).unwrap(); + } assert_eq!( drain(&mut segmenter), [(0, 0, None), (2, 2_000, None), (3, 3_000, None)] ); } + /// A source declaring segments that start no group (no sync sample) queues one cut, which the + /// next group meets. + #[test] + fn cuts_without_a_group_queue_one() { + let mut segmenter = Segmenter::new(Config::default()); + segmenter.frame(0, ms(0)).unwrap(); + for step in 1..1_000u64 { + segmenter.cut(ms(step)); + segmenter.frame(0, ms(step)).unwrap(); + } + assert_eq!(segmenter.cuts.len(), 1); + segmenter.frame(1, ms(1_000)).unwrap(); + assert_eq!(drain(&mut segmenter), [(0, 0, None), (1, 1_000, None)]); + } + + /// A segment spends the cuts it reached however it starts: on a cut, at the ceiling, or after + /// a skipped group. + #[test] + fn every_segment_start_spends_its_cuts() { + let mut segmenter = Segmenter::new(Config::default()); + segmenter.frame(0, ms(0)).unwrap(); + segmenter.cut(ms(1_000)); + segmenter.frame(1, ms(1_000)).unwrap(); + // A one-second declaration puts the ceiling at three, so each later group ends a segment there. + for group in 2..100u64 { + let pts = ms(1_000 + (group - 1) * 3_000); + segmenter.cut(pts); + segmenter.frame(group, pts).unwrap(); + assert!(segmenter.cuts.is_empty(), "group {group}"); + } + segmenter.cut(ms(1_000_000)); + segmenter.frame(200, ms(1_000_000)).unwrap(); + assert!(segmenter.cuts.is_empty(), "after a skipped group"); + } + #[test] fn a_report_that_goes_backwards_is_refused() { let mut segmenter = Segmenter::new(hinted()); diff --git a/rs/moq-mux/src/lib.rs b/rs/moq-mux/src/lib.rs index 16c1a2a6291..518fce246b0 100644 --- a/rs/moq-mux/src/lib.rs +++ b/rs/moq-mux/src/lib.rs @@ -27,8 +27,8 @@ //! part of the format (MPEG-TS). //! - [`rate`] turns bandwidth grants into stable media bitrate targets shared //! by audio, video, and transcode senders. -//! - [`hls`](mod@hls) publishes opt-in HLS segment records per rendition, so an -//! edge can build HLS/DASH playlists without downloading media. +//! - [`hls`](mod@hls) publishes HLS segment records per rendition by default, so +//! an edge can build HLS/DASH playlists without downloading media. //! - [`follow`] follows a broadcast across the publisher instances that //! serve it, for a reader that restarts on each one. //! - [`Linger`] keeps a cleanly ended broadcast up for a fixed time, so a diff --git a/rs/moq-rtc/src/codec/vp8.rs b/rs/moq-rtc/src/codec/vp8.rs index 4ffae286a2c..0d8bf83fe48 100644 --- a/rs/moq-rtc/src/codec/vp8.rs +++ b/rs/moq-rtc/src/codec/vp8.rs @@ -65,8 +65,7 @@ mod tests { #[tokio::test] async fn importer_creation_failure_preserves_abort_error() { let mut broadcast = moq_net::broadcast::Info::new().produce(); - let config = moq_mux::catalog::Config::default().with_hls(moq_mux::hls::Config::default()); - let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); let _collision = broadcast.create_track(hang::hls::default_name("0.vp8"), None).unwrap(); let consumer = broadcast.consume(); let mut bridge = super::Bridge::new(broadcast, catalog).unwrap(); diff --git a/rs/moq-video/src/encode/producer.rs b/rs/moq-video/src/encode/producer.rs index be39b6aa4a8..3acba2f2f75 100644 --- a/rs/moq-video/src/encode/producer.rs +++ b/rs/moq-video/src/encode/producer.rs @@ -1194,8 +1194,7 @@ mod tests { let consumer = broadcast.consume(); let config = moq_mux::catalog::Config::default() .with_clock(clock) - .with_max_age(RETAIN) - .with_hls(moq_mux::hls::Config::default()); + .with_max_age(RETAIN); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap(); let track = broadcast .create_track(