diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4954667..5bea9f1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,3 +25,18 @@ jobs: - name: Run validation suite run: make ci + + unused-dependencies: + name: Unused dependencies + runs-on: ubuntu-latest + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Install cargo-machete + uses: taiki-e/install-action@v2 + with: + tool: cargo-machete + + - name: Fail on dependencies nothing uses + run: cargo machete diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 190e4b4..c9752fd 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -23,8 +23,11 @@ explains how the code is organised and [testing](docs/internals/testing.md) how - **Bounded by construction.** Anything read from a file or a mapper is untrusted. Check a length against the bytes available before allocating, use checked arithmetic, and keep the limits in `[limits]` meaningful. `unsafe` is forbidden. -- **Lint clean.** Clippy runs with `pedantic` and warnings are errors. Prefer fixing the code to - allowing the lint, and say why when you allow one. +- **Lint clean.** `make lint` runs Clippy with `pedantic` and the extra lints in `Cargo.toml`, and + warnings are errors. That includes code nothing calls (dead code), unused imports and variables, + and leftover `dbg!`, `todo!`, and `unimplemented!`. Delete unused code instead of silencing the + warning; when you must allow a lint, use `#[allow(..., reason = "...")]` and say why. CI also + runs `make unused-deps` (cargo-machete) to fail on dependencies nothing uses. - **Documented.** Behaviour that operators or mapper authors see belongs in the handbook. A change that alters an accepted design belongs in a design document first; see below. diff --git a/Cargo.toml b/Cargo.toml index aa2ba87..e348474 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,6 +52,15 @@ strip = "debuginfo" [lints.rust] missing_docs = "warn" unsafe_code = "forbid" +# Code that is never used is deleted, not left behind: dead_code is on by default, and CI turns +# every warning into an error. +unused_qualifications = "warn" +unused_import_braces = "warn" +unused_lifetimes = "warn" +trivial_numeric_casts = "warn" +redundant_lifetimes = "warn" +single_use_lifetimes = "warn" +unit_bindings = "warn" [lints.rustdoc] broken_intra_doc_links = "deny" @@ -59,3 +68,23 @@ broken_intra_doc_links = "deny" [lints.clippy] all = "warn" pedantic = "warn" +# Leftovers from debugging and unfinished work must not reach main. +dbg_macro = "warn" +todo = "warn" +unimplemented = "warn" +# Correctness and clarity picks from the restriction group. +allow_attributes_without_reason = "warn" +mem_forget = "warn" +rc_mutex = "warn" +str_to_string = "warn" +unused_result_ok = "warn" +verbose_file_reads = "warn" +empty_drop = "warn" +if_then_some_else_none = "warn" +needless_raw_strings = "warn" +redundant_type_annotations = "warn" +# Rust's own `Result<_, ()>` and friends aside, these catch code that compiles but is dead weight. +redundant_clone = "warn" +useless_let_if_seq = "warn" +unnecessary_self_imports = "warn" +unnecessary_wraps = "warn" diff --git a/Makefile b/Makefile index 9c059ab..c1ff98f 100644 --- a/Makefile +++ b/Makefile @@ -3,7 +3,7 @@ CARGO ?= cargo MDBOOK ?= mdbook -.PHONY: bench test-scripts bench-enforce conformance book book-serve book-test build check ci clean doc doc-open fixtures fmt fmt-check fuzz fuzz-check install-doc-tools lint serve demo site test deny +.PHONY: unused-deps bench test-scripts bench-enforce conformance book book-serve book-test build check ci clean doc doc-open fixtures fmt fmt-check fuzz fuzz-check install-doc-tools lint serve demo site test deny help: ## Show the available targets @printf '%s\n' \ @@ -29,6 +29,7 @@ help: ## Show the available targets 'lint Run Clippy with warnings denied' \ 'serve Run the example HLS service' \ 'demo Serve the web player demo on http://127.0.0.1:8080' \ + 'unused-deps Fail on dependencies that nothing uses (needs cargo-machete)' \ 'deny Check dependencies for advisories and licences (needs cargo-deny)' \ 'site Build mdBook with rustdoc under /api' \ 'test Run unit and integration tests' @@ -75,9 +76,12 @@ fuzz: ## Run media pipeline fuzzing with nightly cp tests/fixtures/*.mp4 fuzz/corpus/media-pipeline/ $(CARGO) +nightly fuzz run media-pipeline fuzz/corpus/media-pipeline -lint: ## Run Clippy with warnings denied +lint: ## Run Clippy with warnings denied (dead code, unused imports, and the lints in Cargo.toml) $(CARGO) clippy --all-features --all-targets -- -D warnings +unused-deps: ## Fail on dependencies nothing uses (needs `cargo install cargo-machete --locked`) + cargo machete + test: ## Run unit and integration tests $(CARGO) test --all-features --all-targets diff --git a/README.md b/README.md index 50cda78..97f558d 100644 --- a/README.md +++ b/README.md @@ -22,24 +22,23 @@ proportional to the segment asked for, not to the length of the video. - **Packages on demand.** Playlists and manifests are built when an asset loads and segments when they are requested, from a sample index kept in memory. There is no packaging step and no output to store. -- **Reads what encoders produce.** Progressive and fragmented MP4, M4A, and QuickTime files; H.264, - HEVC, VP9, and AV1; AAC, HE-AAC, AC-3, E-AC-3, Opus, and FLAC; edit lists, several audio tracks, - and audio-only files. See [Supported input](#supported-input). +- **Reads what encoders produce.** Progressive and fragmented MP4, M4A, and QuickTime files with + H.264, HEVC, VP9, or AV1 video and AAC, AC-3, E-AC-3, Opus, or FLAC audio. See + [supported input](docs/supported-input.md). - **Finds media through a mapper.** A mapper service answers "where is asset X?" with a file or an HTTP location, so the catalog lives wherever you already keep it. Remote origins are read with ranged requests and the server never downloads a whole file. - **Is bounded by construction.** Every length read from a file or a mapper is checked before it is allocated, and the limits are configuration. `unsafe` code is forbidden. The parser is run against corrupted input on every test run, and has a fuzz target. -- **Is built to be operated.** Prometheus metrics, request IDs, readiness and liveness endpoints, - load shedding, graceful shutdown, immutable content-versioned URLs for CDNs, and byte-range and - conditional requests. +- **Is built to be operated.** Prometheus metrics, health endpoints, load shedding, graceful + shutdown, and immutable content-versioned URLs for CDNs. ## What it does not do - **No transcoding.** The codecs must already suit the protocol, and there is no bitrate ladder - unless you supply the renditions. Adaptive renditions, WebVTT subtitles, and HLS I-frame - playlists are [planned](docs/technical-design/0006-trick-play-subtitles-and-renditions.md). + unless you supply the renditions. Adaptive renditions are + [planned](docs/technical-design/0006-trick-play-subtitles-and-renditions.md). - **No DRM, no live streaming, no MPEG-TS output.** Segments are fragmented MP4. DRM is planned after the features above. - **No TLS or authentication.** Run it behind a reverse proxy or CDN that provides both; see @@ -72,16 +71,15 @@ Then play it: ffplay http://127.0.0.1:3000/hls/sample/master.m3u8 ``` -or, in a second terminal, `make demo` and open for a player with live -server metrics next to it. `make help` lists everything else. +`make demo` starts a web player with live server metrics, and `make help` lists everything else; see +[Using segmentor](docs/usage.md). ## Install - **Release binaries.** Each [GitHub release](https://github.com/includeamin/segmentor/releases) attaches Linux binaries for x86-64 and arm64 with checksums and build attestations. [`install.sh`](install.sh) downloads one, verifies its checksum, and copies it into place - (`curl -fsSL https://raw.githubusercontent.com/includeamin/segmentor/main/install.sh | sh`; read - it first, it is short). + ). - **Container image.** `ghcr.io/includeamin/segmentor`, for the same architectures: ```sh @@ -93,135 +91,29 @@ server metrics next to it. `make help` lists everything else. - **From source.** `cargo install --git https://github.com/includeamin/segmentor`, which needs a recent stable Rust toolchain and a C compiler. -[Verifying a download](docs/releasing.md#verifying-a-release) explains how to check the -attestations and the image signature. [Deploying](docs/deployment.md) has a Docker Compose file, a -Kubernetes manifest, and a hardened systemd unit, and says what to put in front of it: segmentor has -no TLS or authentication of its own. - -## Supported input - -Progressive and fragmented MP4, M4A, and QuickTime `.mov` files, with `moov` first or last. Nothing is decoded or re-encoded, so the codecs must already suit the protocol: - -| | Supported | -| --- | --- | -| Video | H.264, HEVC (`hvc1`/`hev1`), VP9, AV1 | -| Audio | AAC-LC, HE-AAC and HE-AACv2 (explicit signaling), AC-3, E-AC-3, Opus, FLAC | -| Layout | one video track and any number of audio tracks, or audio only; edit lists of one edit, optionally after one empty edit | -| Skipped | tracks that are not audio or video (timecode, metadata, subtitles) | -| Rejected | encrypted media, samples in `moov` mixed with fragments, external data references, more than one sample description per track, other codecs (each error names what was found) | - -Whether a player can decode a codec is a separate question: HEVC and the Dolby codecs need Safari or a platform decoder, for instance. See [TDD 0004](docs/technical-design/0004-broader-mp4-input-support.md) for what was verified where. - -## Using it - -Start the example service with: - -```sh -make serve -``` - -The example asset is available at `http://127.0.0.1:3000/hls/sample/master.m3u8`. Configuration is loaded from `vod.example.toml`; asset IDs map to files beneath one canonical media root, and paths cannot escape that root. - -Logging is configured in the same file: - -```toml -[logging] -level = "info" # trace, debug, info, warn, or error -format = "json" # json or compact -buffer_capacity = 8192 -``` - -Logs are written by a dedicated worker thread through a bounded, lossy queue. If the logger cannot keep up, log lines are dropped instead of blocking media requests. Request and segment timing events use `debug`, so the default `info` level records lifecycle and asset-loading events without logging every media request. - -Each configured asset exposes: - -```text -/hls/{asset}/master.m3u8 -/hls/{asset}/video/index.m3u8 -/hls/{asset}/audio-{n}/index.m3u8 (one per audio track, numbered from 1) -/hls/{asset}/{track}/init.mp4 -/hls/{asset}/{track}/segments/{index}/media.m4s -/dash/{asset}/manifest.mpd -/dash/{asset}/{track}/init.mp4 -/dash/{asset}/{track}/segments/{index}/media.m4s -/health liveness -/ready readiness (503 once shutdown begins) -/metrics Prometheus text -``` - -Initialization and media responses support single and suffix byte ranges, `If-Range`, strong ETags, and immutable content-versioned URLs. Media URLs must carry the `v` query parameter the playlists emit; a missing or stale version is a `404`. Media payloads are read through a bounded backpressured stream instead of buffering the complete segment in each HTTP request. - -Assets can also be resolved on demand from an external mapper service, including media held on remote HTTP origins; see the [Mapper API reference](docs/mapper-api.md) and [docs/operations.md](docs/operations.md). - -CORS, shutdown behavior, concurrency limits, and metrics are configurable in the same file; see [docs/operations.md](docs/operations.md) for production guidance. - -## Web player demo - -`demo/index.html` is a single-file player (hls.js and dash.js, loaded from a CDN) that plays an asset over HLS or DASH and shows live server metrics parsed from `/metrics` next to it: request rate, throughput, per-route latency, errors, and resolver and cache events. It also shows player-side stats such as buffer, bandwidth, and dropped frames. - -```sh -make serve # terminal 1: the origin on :3000 -make demo # terminal 2: the player on http://127.0.0.1:8080 -``` - -The page reads `/metrics` cross-origin, so keep `[cors]` enabled in the config, as in `vod.example.toml`. - -## Packaging from the command line - -The packager parses a local MP4, creates keyframe-aligned segment plans, and writes separate fragmented MP4 audio and video tracks: - -```sh -cargo run -- package \ - --input tests/fixtures/h264-aac.mp4 \ - --output target/package-test -``` - -Regenerate the synthetic parser fixture and its FFprobe packet manifest with `make fixtures`. FFmpeg is used only for fixture generation and output validation, not by the application. - -## Releases - -Merges to `main` are tagged automatically with a semantic version derived from [Conventional Commits](https://www.conventionalcommits.org), so use `feat:`, `fix:`, or `type(scope)!:` in commit and pull request titles. Publish a release, with a generated changelog and a `latest` or `preview` flag, from the **Release** workflow. See [docs/releasing.md](docs/releasing.md). +See [Deploying](docs/deployment.md) for Compose, Kubernetes and systemd files, and +[Verifying a release](docs/releasing.md#verifying-a-release) for checking downloads. ## Documentation -The project handbook uses the same `mdBook` interface as the Rust Book. Install the pinned documentation tool and serve the book with live reload: - -```sh -make install-doc-tools -make book-serve -``` +The handbook, built with mdBook, covers everything beyond this page: -Build the complete static site, including the `rustdoc` API reference, with `make site`. The book starts at `target/book/index.html`, and its API Reference chapter links to the generated Rust documentation under `target/book/api/`. +- [Using segmentor](docs/usage.md): endpoints, the web player demo, the packaging command +- [Supported input](docs/supported-input.md), [Deploying](docs/deployment.md), and + [Operating the origin](docs/operations.md) +- [Mapper API](docs/mapper-api.md), [architecture](docs/architecture.md), and the + [technical designs](docs/technical-design/README.md) +- [Releasing](docs/releasing.md), including how versions and tags are made -See [docs/README.md](docs/README.md) for the documentation layout and authoring workflow. +Build it locally with `make book-serve` (see [docs/README.md](docs/README.md)). ## Contributing and security Contributions are welcome; start with [CONTRIBUTING.md](CONTRIBUTING.md), which covers building, -testing, design documents, and the sign-off every commit needs. Files that will not load or play -are the most useful reports there are: there is an issue form for them. Please report security -problems privately, as described in [SECURITY.md](SECURITY.md). Everyone taking part is expected to -follow the [Code of Conduct](CODE_OF_CONDUCT.md). - -## Development - -The repository uses stable Rust with `rustfmt` and Clippy. Run the complete local validation suite with: - -```sh -make ci -``` - -Run `make help` to list the individual build, check, format, lint, test, and documentation targets. - -`make conformance` runs the black-box HLS and DASH conformance suite, and `make bench` measures the performance budgets (see [docs/benchmarks.md](docs/benchmarks.md)). - -Compile the media-pipeline fuzz target on stable with `make fuzz-check`. To run a fuzz campaign, install the separate nightly tooling. `make fuzz` copies the generated MP4 fixtures into an ignored, writable corpus before starting libFuzzer: - -```sh -rustup toolchain install nightly -cargo install cargo-fuzz --locked -make fuzz -``` +testing, and the sign-off every commit needs. Files that will not load or play are the most useful +reports there are: there is an issue form for them. Report security problems privately, as +described in [SECURITY.md](SECURITY.md). Everyone taking part is expected to follow the +[Code of Conduct](CODE_OF_CONDUCT.md). ## License diff --git a/benches/budgets.rs b/benches/budgets.rs index 3a2d7c2..f2566c0 100644 --- a/benches/budgets.rs +++ b/benches/budgets.rs @@ -23,7 +23,8 @@ clippy::large_futures, clippy::redundant_closure_for_method_calls, clippy::too_many_lines, - clippy::trivially_copy_pass_by_ref + clippy::trivially_copy_pass_by_ref, + reason = "harness code, not the production crate" )] use std::path::{Path, PathBuf}; diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index b2c4299..263e83a 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -5,6 +5,8 @@ # Design - [Architecture](architecture.md) +- [Using segmentor](usage.md) +- [Supported input](supported-input.md) - [Deploying](deployment.md) - [Operating the origin](operations.md) - [Performance budgets](benchmarks.md) diff --git a/docs/mapper-api.md b/docs/mapper-api.md index 7dfaf06..30cbdeb 100644 --- a/docs/mapper-api.md +++ b/docs/mapper-api.md @@ -58,6 +58,7 @@ A file on an HTTP origin: | `location.type` | Yes | `file` or `http`. Anything else is rejected | | `location.path` | For `file` | Relative to `storage.media_root`; no leading `/`, no `.` or `..` components, no NUL, at most 4096 bytes | | `location.url` | For `http` | See [Remote locations](#remote-locations) | +| `subtitles` | No | Sidecar WebVTT files; see [Subtitles](#subtitles) | ### `304 Not Modified` @@ -105,6 +106,40 @@ Signed query parameters are treated as secrets: they are not logged at `info` an So: set `expires_at` on anything signed, keep the **same `version`** when you only re-sign, and answer unconditional requests (no `If-None-Match`) with a full body and a new signature. A `304` is only appropriate while the current signature is still valid. +## Subtitles + +An answer may attach WebVTT subtitle files to the asset: + +```json +{ + "asset_id": "movie", + "version": "2026-09-21-a", + "location": { "type": "file", "path": "movie.mp4" }, + "subtitles": [ + { "language": "en", "label": "English", "default": true, + "location": { "type": "file", "path": "subs/movie.en.vtt" } }, + { "language": "fr", "label": "Français", "forced": false, + "location": { "type": "http", "url": "https://origin.example.net/subs/movie.fr.vtt" } } + ] +} +``` + +| Field | Required | Rules | +| --- | --- | --- | +| `language` | Yes | A BCP 47 tag of letters, digits, and hyphens, starting with a letter, at most 35 characters. Unique within the asset, ignoring case. It appears in the URLs | +| `label` | No | What a player shows the viewer. Defaults to the language. At most 128 bytes, no control characters | +| `default` | No | The player selects this one unless the viewer chose otherwise. At most one entry may set it | +| `forced` | No | The track is meant to be shown even when the viewer has not asked for subtitles | +| `location` | Yes | A `file` or `http` location with the same rules as the media's, including the `[remote_media]` policy. An `http` origin must support ranged requests, as media origins do | + +The server fetches each file when the asset loads and keeps it in memory, so playback never touches the subtitle origin. A file must be UTF-8, must begin with `WEBVTT`, and must have readable cue timing lines. It is limited by `limits.max_subtitle_bytes` (2 MiB), `limits.max_subtitles_total_bytes` (8 MiB per asset), and `limits.max_subtitles` (16). **One bad file fails the whole asset** with the language named, so a viewer never gets a language that is silently missing. + +Cue times are read as times on the source file's own clock, the one its edit lists describe. Packaging can move a file onto a later timeline so that no timestamp is negative (this is what an edit list that trims encoder delay does, and it is typically a few tens of milliseconds), and the server adds that same offset to every cue so they stay in step with the picture. A video that simply starts late, through a leading empty edit, is not an offset: the cues were written against a clock that already includes that gap, so they are left alone. A fragmented file's timeline starts at zero and cues are not moved. Nothing else in the file changes. + +**Change `version` when a subtitle file changes.** The server reloads an asset only when its `version` or location changes, so an edited caption under an unchanged version is not picked up until the asset is evicted. + +The HLS master playlist gains an `#EXT-X-MEDIA:TYPE=SUBTITLES` entry per file, served from `/hls/{asset}/subtitles/{language}/index.m3u8`, and the DASH manifest gains a text adaptation set. Both point at `/{hls|dash}/{asset}/subtitles/{language}/sub.vtt?v={version}`. + ## What a `version` means to the server The server keeps one loaded copy per asset, keyed by `(asset_id, version)`. A different `version`, or the same version at a different location (a rotated signed URL), makes it reload from the new location. There is **no grace period**: players holding URLs from the old version get `404` and recover by fetching the playlist again. Change `version` only when the media actually changes. @@ -164,5 +199,6 @@ Then `curl http://127.0.0.1:3000/hls/movie/master.m3u8`. - Removed assets answer `404` or `410`, not `200`. - Answers are small (the server's default limit is 16 KiB). - The mapper answers quickly: the server's default per-request timeout is two seconds, with two retries. +- `version` also changes when a subtitle file changes. - `expires_at` is set for anything signed, and re-signing keeps the same `version`. - The mapper is reachable over `https` in production and requires the bearer token. diff --git a/docs/supported-input.md b/docs/supported-input.md new file mode 100644 index 0000000..a78930f --- /dev/null +++ b/docs/supported-input.md @@ -0,0 +1,13 @@ +# Supported input + +Progressive and fragmented MP4, M4A, and QuickTime `.mov` files, with `moov` first or last. Nothing is decoded or re-encoded, so the codecs must already suit the protocol: + +| | Supported | +| --- | --- | +| Video | H.264, HEVC (`hvc1`/`hev1`), VP9, AV1 | +| Audio | AAC-LC, HE-AAC and HE-AACv2 (explicit signaling), AC-3, E-AC-3, Opus, FLAC | +| Layout | one video track and any number of audio tracks, or audio only; edit lists of one edit, optionally after one empty edit | +| Skipped | tracks that are not audio or video (timecode, metadata, subtitles) | +| Rejected | encrypted media, samples in `moov` mixed with fragments, external data references, more than one sample description per track, other codecs (each error names what was found) | + +Whether a player can decode a codec is a separate question: HEVC and the Dolby codecs need Safari or a platform decoder, for instance. See [TDD 0004](technical-design/0004-broader-mp4-input-support.md) for what was verified where. diff --git a/docs/technical-design/0006-trick-play-subtitles-and-renditions.md b/docs/technical-design/0006-trick-play-subtitles-and-renditions.md index a3a1ae4..756eaf0 100644 --- a/docs/technical-design/0006-trick-play-subtitles-and-renditions.md +++ b/docs/technical-design/0006-trick-play-subtitles-and-renditions.md @@ -39,6 +39,8 @@ An asset is one MP4 today: one video track, some audio tracks, and a single vide ## 1. HLS I-frame playlists +> **Implemented** as designed. The conformance suite checks, for every fixture with video, that the playlist lists the source's keyframes and that each fragment decodes to one picture. The first version trusts `stss`, as described below; the open question about IDR pictures remains. + ### Output The master playlist gains an `#EXT-X-I-FRAME-STREAM-INF` line pointing at `video/iframes.m3u8`. That playlist has `#EXT-X-I-FRAMES-ONLY` and one entry per keyframe. Each entry is its own small resource, `/hls/{asset}/video/iframes/{n}/media.m4s`: a fragment holding that one sample, with the video track's existing init segment. @@ -60,6 +62,8 @@ Each I-frame resource, prefixed with the init segment, must decode to exactly on ## 2. Sidecar WebVTT subtitles +> **Implemented** as designed, with two refinements: an `http` subtitle origin must support ranged requests (it is opened like media, so it gets the same `[remote_media]` and redirect protection), and the size and count limits are `limits.max_subtitle_bytes`, `limits.max_subtitles_total_bytes`, and `limits.max_subtitles`. The version covers subtitle content; a mapper must still change its own `version` when a caption changes, because that is what triggers a reload. Checked in headless Chrome with hls.js and dash.js: cues appear at the right times, including on a file whose edit lists shift the timeline by 67 ms. The shift is the edit lists' shared offset `O`, not a track's own delay (a late-starting video is already part of the presentation the cues were written against), and a fragmented file is not shifted. Not checked in Safari. + ### Mapper answer An optional `subtitles` list, each entry: @@ -76,7 +80,7 @@ An optional `subtitles` list, each entry: The file is fetched when the asset loads and is held in memory, so requests never touch the origin: - **Validation.** UTF-8, at most `limits.max_subtitle_bytes` (default 2 MiB) each and a limit in total, and it must begin with `WEBVTT`. Anything else fails the asset load with a message naming the language. -- **Timeline correction.** An asset with an edit list or a fragmented start time is moved onto a shifted timeline (see [TDD 0004](0004-broader-mp4-input-support.md)), so a cue authored against the source would appear early or late by that offset. Cue timing lines are shifted by the asset's timeline offset when the file is served. Everything else in the file is passed through unchanged. +- **Timeline correction.** An asset whose edit lists trim encoder delay is served on a timeline `O` later than the source's clock (see [TDD 0004](0004-broader-mp4-input-support.md)), so a cue authored against the source would appear early by `O`. Cue timing lines are shifted by `O` when the asset loads. A track's own delay is not part of `O`, and a fragmented file, whose timeline starts at zero, is not shifted. Everything else in the file is passed through unchanged. - **Version.** The asset version covers the subtitle content, so changing a caption gives new URLs. ### Output diff --git a/docs/usage.md b/docs/usage.md new file mode 100644 index 0000000..f4495cb --- /dev/null +++ b/docs/usage.md @@ -0,0 +1,55 @@ +# Using segmentor + +## Run the example + +```sh +make serve +``` + +serves `tests/fixtures/h264-aac.mp4` as the asset `sample`, using `vod.example.toml`. Asset IDs map to files beneath one canonical media root, and paths cannot escape it. Assets can also be resolved on demand from a [mapper](mapper-api.md), including media on remote HTTP origins. + +## Endpoints + +Each asset exposes: + +```text +/hls/{asset}/master.m3u8 +/hls/{asset}/video/index.m3u8 +/hls/{asset}/audio-{n}/index.m3u8 (one per audio track, numbered from 1) +/hls/{asset}/video/iframes.m3u8 (I-frame playlist, when there is video) +/hls/{asset}/video/iframes/{n}/media.m4s (one keyframe as its own fragment) +/hls/{asset}/{track}/init.mp4 +/hls/{asset}/{track}/segments/{index}/media.m4s +/hls/{asset}/subtitles/{language}/index.m3u8 (when the mapper lists subtitles) +/hls/{asset}/subtitles/{language}/sub.vtt +/dash/{asset}/manifest.mpd +/dash/{asset}/subtitles/{language}/sub.vtt +/dash/{asset}/{track}/init.mp4 +/dash/{asset}/{track}/segments/{index}/media.m4s +/health liveness +/ready readiness (503 once shutdown begins) +/metrics Prometheus text +``` + +Initialization and media responses support single and suffix byte ranges, `If-Range`, strong ETags, and immutable content-versioned URLs. Media URLs must carry the `v` query parameter the playlists emit; a missing or stale version is a `404`. Media payloads are streamed through a bounded, backpressured reader instead of being buffered per request. + +Configuration, logging, limits, CORS, and shutdown are described in [Operating the origin](operations.md). + +## Web player demo + +`demo/index.html` is a single-file player (hls.js and dash.js, loaded from a CDN) that plays an asset over HLS or DASH and shows live server metrics parsed from `/metrics` next to it: request rate, throughput, per-route latency, errors, and resolver and cache events, plus player-side stats such as buffer, bandwidth, and dropped frames. + +```sh +make serve # terminal 1: the origin on :3000 +make demo # terminal 2: the player on http://127.0.0.1:8080 +``` + +The page reads `/metrics` cross-origin, so keep `[cors]` enabled, as in `vod.example.toml`. + +## Packaging from the command line + +`package` parses a local MP4, plans keyframe-aligned segments, and writes separate fragmented MP4 audio and video tracks: + +```sh +cargo run -- package --input tests/fixtures/h264-aac.mp4 --output target/package-test +``` diff --git a/src/asset.rs b/src/asset.rs index d3f8d02..20aff76 100644 --- a/src/asset.rs +++ b/src/asset.rs @@ -8,8 +8,9 @@ use crate::error::{Error, Result}; use crate::media::{MediaIndex, Sample, Track, TrackKey}; use crate::mp4::ParsedMedia; use crate::protocol::{Presentation, dash, hls}; -use crate::segment::SegmentPlan; +use crate::segment::{SegmentPlan, TrackSegment}; use crate::source::{ByteRange, LocalMediaSource, MediaSourceKind}; +use crate::subtitle::{self, Subtitle}; use crate::{fmp4, mp4, segment}; use std::sync::Arc; @@ -22,6 +23,7 @@ pub(crate) struct PackagedAsset { limits: LimitsConfig, version: String, rendered: RenderedManifests, + subtitles: Vec, } /// Playlists and manifests rendered once at load so requests never walk sample tables. @@ -29,6 +31,8 @@ pub(crate) struct PackagedAsset { struct RenderedManifests { hls_master: Bytes, hls_media: HashMap, + hls_iframes: Option, + hls_subtitle: Bytes, dash: Bytes, } @@ -51,11 +55,22 @@ impl PackagedAsset { source: MediaSourceKind, segment_duration_ms: u64, limits: &LimitsConfig, + ) -> Result { + Self::load_with_subtitles(source, Vec::new(), segment_duration_ms, limits).await + } + + /// Like [`Self::load`], with sidecar subtitle files already fetched. Each is validated and + /// moved onto the asset's timeline; one bad file fails the whole asset. + pub(crate) async fn load_with_subtitles( + source: MediaSourceKind, + subtitles: Vec, + segment_duration_ms: u64, + limits: &LimitsConfig, ) -> Result { let parsed = mp4::parse(&source, limits).await?; let limits = limits.clone(); tokio::task::spawn_blocking(move || { - Self::assemble(source, parsed, segment_duration_ms, &limits) + Self::assemble(source, parsed, subtitles, segment_duration_ms, &limits) }) .await .map_err(|error| Error::Io(std::io::Error::other(error)))? @@ -64,11 +79,13 @@ impl PackagedAsset { fn assemble( source: MediaSourceKind, parsed: ParsedMedia, + subtitles: Vec, segment_duration_ms: u64, limits: &LimitsConfig, ) -> Result { let ParsedMedia { index, metadata } = parsed; let plan = segment::plan(&index, segment_duration_ms, limits)?; + let subtitles = prepare_subtitles(&index, subtitles)?; let init_segments = index .tracks .iter() @@ -78,9 +95,10 @@ impl PackagedAsset { .map(|bytes| (track.key, bytes)) }) .collect::>>()?; - let version = version_of(&index); - let rendered = - RenderedManifests::render(Presentation::new(&index.tracks, &plan, &version))?; + let version = version_of(&index, &subtitles); + let rendered = RenderedManifests::render( + Presentation::new(&index.tracks, &plan, &version).with_subtitles(&subtitles), + )?; Ok(Self { source, index, @@ -89,6 +107,7 @@ impl PackagedAsset { limits: limits.clone(), version, rendered, + subtitles, }) } @@ -99,6 +118,7 @@ impl PackagedAsset { /// The read-only view renderers work from. pub(crate) fn presentation(&self) -> Presentation<'_> { Presentation::new(&self.index.tracks, &self.plan, &self.version) + .with_subtitles(&self.subtitles) } pub(crate) fn init_segment(&self, key: TrackKey) -> Result { @@ -120,6 +140,59 @@ impl PackagedAsset { .ok_or(Error::NotFound("track does not exist")) } + /// The video track's I-frame playlist, absent for an audio-only asset. + pub(crate) fn hls_iframe_playlist(&self) -> Result { + self.rendered + .hls_iframes + .clone() + .ok_or(Error::NotFound("asset has no video track")) + } + + /// The fragment holding only the `frame_index`th keyframe of the video track. + pub(crate) fn prepare_iframe(&self, frame_index: u32) -> Result { + let presentation = self.presentation(); + let track = presentation + .video() + .ok_or(Error::NotFound("asset has no video track"))?; + let position = track + .samples + .iter() + .enumerate() + .filter(|(_, sample)| sample.is_sync) + .nth(usize::try_from(frame_index).map_err(|_| { + Error::InvalidMedia("keyframe index does not fit in memory".to_owned()) + })?) + .map(|(position, _)| position) + .ok_or(Error::NotFound("keyframe does not exist"))?; + let sample = track.samples[position]; + let segment = TrackSegment { + track_id: track.id, + first_sample: position, + end_sample: position + 1, + decode_time: sample.decode_time, + duration: u64::from(sample.duration), + }; + let sequence_number = frame_index + .checked_add(1) + .ok_or_else(|| Error::InvalidMedia("sequence number overflow".to_owned()))?; + fmp4::prepare_media_segment(track, segment, sequence_number, &self.limits) + } + + /// The HLS playlist that lists one subtitle file. + pub(crate) fn hls_subtitle_playlist(&self, language: &str) -> Result { + self.subtitle(language)?; + Ok(self.rendered.hls_subtitle.clone()) + } + + /// A subtitle file, ready to serve. + pub(crate) fn subtitle(&self, language: &str) -> Result { + self.subtitles + .iter() + .find(|subtitle| subtitle.language.eq_ignore_ascii_case(language)) + .map(|subtitle| subtitle.data.clone()) + .ok_or(Error::NotFound("subtitle does not exist")) + } + pub(crate) fn dash_manifest(&self) -> Bytes { self.rendered.dash.clone() } @@ -225,9 +298,15 @@ impl PackagedAsset { .values() .map(Bytes::len) .sum::(); - (samples.saturating_mul(std::mem::size_of::()) as u64) + (samples.saturating_mul(size_of::()) as u64) .saturating_add(init as u64) .saturating_add(rendered as u64) + .saturating_add( + self.subtitles + .iter() + .map(|subtitle| subtitle.data.len() as u64) + .sum(), + ) } } @@ -244,6 +323,8 @@ impl RenderedManifests { Ok(Self { hls_master: Bytes::from(hls::master_playlist(presentation)?), hls_media, + hls_iframes: hls::iframe_playlist(presentation)?.map(Bytes::from), + hls_subtitle: Bytes::from(hls::subtitle_playlist(presentation)), dash: Bytes::from(dash::manifest(presentation)?), }) } @@ -255,11 +336,26 @@ impl RenderedManifests { /// Media URLs are cached as immutable by browsers and CDNs, so a new build that answers an old /// URL with different bytes would be served stale content until the cache expires. Mixing the /// revision into the version gives such a build new URLs instead. -const FORMAT_REVISION: u32 = 1; +const FORMAT_REVISION: u32 = 2; + +/// Validates each subtitle file and moves its cues onto the asset's timeline, by the offset the +/// edit lists were resolved with. A track's own delay is not part of it: that is already in the +/// presentation the cues were written against. +fn prepare_subtitles(index: &MediaIndex, subtitles: Vec) -> Result> { + let offset_ms = index.presentation_offset_ms; + subtitles + .into_iter() + .map(|mut subtitle| { + subtitle.data = subtitle::prepare(&subtitle.language, &subtitle.data, offset_ms)?; + Ok(subtitle) + }) + .collect() +} /// The `v` value in media URLs: a hash of everything the index was built from (`moov`, and every -/// `moof` of a fragmented file) and [`FORMAT_REVISION`]. -fn version_of(index: &MediaIndex) -> String { +/// `moof` of a fragmented file), [`FORMAT_REVISION`], and any subtitle files, so changing a +/// caption gives new URLs. +fn version_of(index: &MediaIndex, subtitles: &[Subtitle]) -> String { use std::fmt::Write; use sha2::{Digest, Sha256}; @@ -272,6 +368,18 @@ fn version_of(index: &MediaIndex) -> String { .expect("parsed assets always have a metadata hash"), ); hasher.update(FORMAT_REVISION.to_be_bytes()); + // Absent subtitles add nothing, so an asset without them keeps the version it always had. + for subtitle in subtitles { + for part in [ + subtitle.language.as_bytes(), + subtitle.label.as_bytes(), + &[u8::from(subtitle.default), u8::from(subtitle.forced)], + &subtitle.data, + ] { + hasher.update((part.len() as u64).to_be_bytes()); + hasher.update(part); + } + } hasher .finalize() .iter() @@ -309,7 +417,7 @@ mod tests { assert!(asset.version().bytes().all(|byte| byte.is_ascii_hexdigit())); assert_ne!(asset.version(), moov_only); assert_eq!( - version_of(&asset.index), + version_of(&asset.index, &[]), asset.version(), "and it is stable" ); diff --git a/src/config/limits.rs b/src/config/limits.rs index 1521a14..70f779d 100644 --- a/src/config/limits.rs +++ b/src/config/limits.rs @@ -30,6 +30,12 @@ pub(crate) struct LimitsConfig { pub(crate) max_index_bytes: u64, pub(crate) max_connections: usize, pub(crate) header_read_timeout_ms: u64, + /// Sidecar subtitle files per asset. + pub(crate) max_subtitles: usize, + /// One subtitle file. + pub(crate) max_subtitle_bytes: u64, + /// All of one asset's subtitle files together. + pub(crate) max_subtitles_total_bytes: u64, } impl LimitsConfig { @@ -54,6 +60,9 @@ impl LimitsConfig { || self.max_index_bytes == 0 || self.max_connections == 0 || self.header_read_timeout_ms == 0 + || self.max_subtitles == 0 + || self.max_subtitle_bytes == 0 + || self.max_subtitles_total_bytes == 0 { return Err(Error::Configuration( "all resource limits must be greater than zero".to_owned(), @@ -87,6 +96,9 @@ impl Default for LimitsConfig { max_index_bytes: 4 * 1024 * 1024 * 1024, max_connections: 10_000, header_read_timeout_ms: 10_000, + max_subtitles: 16, + max_subtitle_bytes: 2 * 1024 * 1024, + max_subtitles_total_bytes: 8 * 1024 * 1024, } } } diff --git a/src/fmp4/init.rs b/src/fmp4/init.rs index 53716fb..e547928 100644 --- a/src/fmp4/init.rs +++ b/src/fmp4/init.rs @@ -371,8 +371,8 @@ mod tests { assert!(!contains(&rewritten, b"wave") && !contains(&rewritten, b"chan")); // Same audio, described the standard way. assert_eq!( - crate::mp4::codec::parse_aac(&rewritten, 2).unwrap(), - crate::mp4::codec::parse_aac(&source, 2).unwrap() + mp4::codec::parse_aac(&rewritten, 2).unwrap(), + mp4::codec::parse_aac(&source, 2).unwrap() ); assert_eq!( &rewritten[..8], diff --git a/src/http/error.rs b/src/http/error.rs index 69dbd3d..1587871 100644 --- a/src/http/error.rs +++ b/src/http/error.rs @@ -9,7 +9,7 @@ use crate::registry::RegistryError; const NO_STORE: HeaderValue = HeaderValue::from_static("no-store"); -pub(crate) type HttpResult = std::result::Result; +pub(crate) type HttpResult = Result; #[derive(Debug)] pub(crate) struct HttpError { diff --git a/src/http/handlers/media.rs b/src/http/handlers/media.rs index bb42be2..19c3c73 100644 --- a/src/http/handlers/media.rs +++ b/src/http/handlers/media.rs @@ -14,6 +14,7 @@ use tokio_stream::wrappers::ReceiverStream; use super::parse_track; use crate::asset::PackagedAsset; +use crate::fmp4::PreparedSegment; use crate::http::error::{HttpError, HttpResult}; use crate::http::range::{ByteInterval, range_not_satisfiable, requested_range}; use crate::http::state::AppState; @@ -70,20 +71,100 @@ pub(crate) async fn media_segment( version.require(&asset)?; let key = parse_track(&track)?; let etag = entity_tag(&asset, &format!("{track}-segment-{segment_index}")); + serve_segment( + &state, + &method, + &headers, + asset, + etag, + key.kind, + move |asset| asset.prepare_media_segment(key, segment_index), + ) + .await +} + +/// A sidecar `WebVTT` file, served under both protocols. +pub(crate) async fn subtitle_file( + State(state): State, + Path((asset_id, language)): Path<(String, String)>, + Query(version): Query, + headers: HeaderMap, +) -> HttpResult { + let asset = state.asset(&asset_id).await?; + version.require(&asset)?; + let etag = entity_tag( + &asset, + &format!("subtitle-{}", language.to_ascii_lowercase()), + ); if not_modified(&headers, &etag) { return not_modified_response(etag, "public, max-age=31536000, immutable"); } + let bytes = asset.subtitle(&language)?; + let total = bytes.len() as u64; + let Ok(range) = requested_range(&headers, total, &etag) else { + return range_not_satisfiable(total); + }; + let selected = range.unwrap_or(ByteInterval { + start: 0, + end: total, + }); + let (start, end) = ( + usize::try_from(selected.start).map_err(|_| HttpError::internal("range".to_owned()))?, + usize::try_from(selected.end).map_err(|_| HttpError::internal("range".to_owned()))?, + ); + media_response_builder(total, selected, etag, "text/vtt; charset=utf-8") + .body(Body::from(bytes.slice(start..end))) + .map_err(|error| HttpError::internal(error.to_string())) +} + +/// One keyframe of the video track as a fragment of its own, for HLS I-frame playlists. +pub(crate) async fn iframe_segment( + State(state): State, + method: Method, + Path((asset_id, frame_index)): Path<(String, u32)>, + Query(version): Query, + headers: HeaderMap, +) -> HttpResult { + let asset = state.asset(&asset_id).await?; + version.require(&asset)?; + let etag = entity_tag(&asset, &format!("iframe-{frame_index}")); + serve_segment( + &state, + &method, + &headers, + asset, + etag, + TrackKind::Video, + move |asset| asset.prepare_iframe(frame_index), + ) + .await +} + +/// Answers a request for a generated fragment: conditional, ranged, and streamed from the source +/// through a bounded queue. +async fn serve_segment( + state: &AppState, + method: &Method, + headers: &HeaderMap, + asset: Arc, + etag: HeaderValue, + kind: TrackKind, + prepare: impl FnOnce(&PackagedAsset) -> crate::error::Result + Send + 'static, +) -> HttpResult { + if not_modified(headers, &etag) { + return not_modified_response(etag, "public, max-age=31536000, immutable"); + } let started = Instant::now(); // Header generation is CPU work proportional to the segment's sample count, so it runs on // the blocking pool rather than an async worker. let prepared = { let asset = Arc::clone(&asset); - tokio::task::spawn_blocking(move || asset.prepare_media_segment(key, segment_index)) + tokio::task::spawn_blocking(move || prepare(&asset)) .await .map_err(|error| HttpError::internal(error.to_string()))?? }; let total_length = prepared.content_length; - let requested_interval = match requested_range(&headers, total_length, &etag) { + let requested_interval = match requested_range(headers, total_length, &etag) { Ok(range) => range.unwrap_or(ByteInterval { start: 0, end: total_length, @@ -97,7 +178,7 @@ pub(crate) async fn media_segment( total_length, requested_interval, etag, - segment_content_type(key.kind), + segment_content_type(kind), ) .body(Body::empty()) .map_err(|error| HttpError::internal(error.to_string())); @@ -122,9 +203,7 @@ pub(crate) async fn media_segment( ); tracing::debug!( event = "media_segment_generated", - asset.id = %asset_id, - media.track = %track, - media.segment = segment_index, + media.kind = ?kind, response.bytes = content_length, elapsed_us = started.elapsed().as_micros(), ); @@ -132,7 +211,7 @@ pub(crate) async fn media_segment( total_length, requested_interval, etag, - segment_content_type(key.kind), + segment_content_type(kind), ) .body(Body::from_stream(ReceiverStream::new(receiver))) .map_err(|error| HttpError::internal(error.to_string())) diff --git a/src/http/handlers/mod.rs b/src/http/handlers/mod.rs index ea59c29..feea074 100644 --- a/src/http/handlers/mod.rs +++ b/src/http/handlers/mod.rs @@ -7,8 +7,10 @@ use crate::http::error::{HttpError, HttpResult}; use crate::media::TrackKey; pub(crate) use health::{health, metrics, ready}; -pub(crate) use media::{init_segment, media_segment}; -pub(crate) use playlist::{dash_manifest, master_playlist, media_playlist}; +pub(crate) use media::{iframe_segment, init_segment, media_segment, subtitle_file}; +pub(crate) use playlist::{ + dash_manifest, iframe_playlist, master_playlist, media_playlist, subtitle_playlist, +}; pub(crate) fn parse_track(track: &str) -> HttpResult { TrackKey::parse(track).ok_or_else(|| HttpError::not_found("track does not exist")) diff --git a/src/http/handlers/playlist.rs b/src/http/handlers/playlist.rs index f5b4263..fd4eb0f 100644 --- a/src/http/handlers/playlist.rs +++ b/src/http/handlers/playlist.rs @@ -38,6 +38,35 @@ pub(crate) async fn media_playlist( playlist_response(asset.hls_media_playlist(key)?, etag) } +pub(crate) async fn iframe_playlist( + State(state): State, + Path(asset_id): Path, + headers: HeaderMap, +) -> HttpResult { + let asset = state.asset(&asset_id).await?; + let etag = entity_tag(&asset, "hls-iframe-playlist"); + if not_modified(&headers, &etag) { + return not_modified_response(etag, "public, max-age=60"); + } + playlist_response(asset.hls_iframe_playlist()?, etag) +} + +pub(crate) async fn subtitle_playlist( + State(state): State, + Path((asset_id, language)): Path<(String, String)>, + headers: HeaderMap, +) -> HttpResult { + let asset = state.asset(&asset_id).await?; + let etag = entity_tag( + &asset, + &format!("hls-subtitle-{}-playlist", language.to_ascii_lowercase()), + ); + if not_modified(&headers, &etag) { + return not_modified_response(etag, "public, max-age=60"); + } + playlist_response(asset.hls_subtitle_playlist(&language)?, etag) +} + pub(crate) async fn dash_manifest( State(state): State, Path(asset_id): Path, diff --git a/src/http/range.rs b/src/http/range.rs index c755cb3..7ebc491 100644 --- a/src/http/range.rs +++ b/src/http/range.rs @@ -31,7 +31,7 @@ pub(crate) fn requested_range( headers: &HeaderMap, total: u64, etag: &HeaderValue, -) -> std::result::Result, ()> { +) -> Result, ()> { let Some(value) = headers.get(RANGE) else { return Ok(None); }; diff --git a/src/http/router.rs b/src/http/router.rs index b237f93..b8e5366 100644 --- a/src/http/router.rs +++ b/src/http/router.rs @@ -9,8 +9,8 @@ use tower_http::trace::{DefaultOnFailure, DefaultOnRequest, DefaultOnResponse, T use tracing::Level; use super::handlers::{ - dash_manifest, health, init_segment, master_playlist, media_playlist, media_segment, metrics, - ready, + dash_manifest, health, iframe_playlist, iframe_segment, init_segment, master_playlist, + media_playlist, media_segment, metrics, ready, subtitle_file, subtitle_playlist, }; use super::middleware::{ X_REQUEST_ID, enforce_header_limit, record_metrics, request_id, shed_load, @@ -22,6 +22,11 @@ const READY: &str = "/ready"; const METRICS: &str = "/metrics"; const HLS_MASTER: &str = "/hls/{asset_id}/master.m3u8"; const HLS_MEDIA_PLAYLIST: &str = "/hls/{asset_id}/{track}/index.m3u8"; +const HLS_IFRAME_PLAYLIST: &str = "/hls/{asset_id}/video/iframes.m3u8"; +const HLS_IFRAME_SEGMENT: &str = "/hls/{asset_id}/video/iframes/{frame_index}/media.m4s"; +const HLS_SUBTITLE_PLAYLIST: &str = "/hls/{asset_id}/subtitles/{language}/index.m3u8"; +const HLS_SUBTITLE_FILE: &str = "/hls/{asset_id}/subtitles/{language}/sub.vtt"; +const DASH_SUBTITLE_FILE: &str = "/dash/{asset_id}/subtitles/{language}/sub.vtt"; const HLS_INIT: &str = "/hls/{asset_id}/{track}/init.mp4"; const HLS_SEGMENT: &str = "/hls/{asset_id}/{track}/segments/{segment_index}/media.m4s"; const DASH_MANIFEST: &str = "/dash/{asset_id}/manifest.mpd"; @@ -30,12 +35,17 @@ const DASH_SEGMENT: &str = "/dash/{asset_id}/{track}/segments/{segment_index}/me /// Every route template, used both to register handlers and to label metrics, so a route /// cannot be added to one and forgotten in the other. -pub(crate) const ROUTES: [&str; 10] = [ +pub(crate) const ROUTES: [&str; 15] = [ HEALTH, READY, METRICS, HLS_MASTER, HLS_MEDIA_PLAYLIST, + HLS_IFRAME_PLAYLIST, + HLS_IFRAME_SEGMENT, + HLS_SUBTITLE_PLAYLIST, + HLS_SUBTITLE_FILE, + DASH_SUBTITLE_FILE, HLS_INIT, HLS_SEGMENT, DASH_MANIFEST, @@ -50,6 +60,11 @@ pub(crate) fn router(state: AppState) -> Router { .route(METRICS, get(metrics)) .route(HLS_MASTER, get(master_playlist)) .route(HLS_MEDIA_PLAYLIST, get(media_playlist)) + .route(HLS_IFRAME_PLAYLIST, get(iframe_playlist)) + .route(HLS_IFRAME_SEGMENT, get(iframe_segment)) + .route(HLS_SUBTITLE_PLAYLIST, get(subtitle_playlist)) + .route(HLS_SUBTITLE_FILE, get(subtitle_file)) + .route(DASH_SUBTITLE_FILE, get(subtitle_file)) .route(HLS_INIT, get(init_segment)) .route(HLS_SEGMENT, get(media_segment)) .route(DASH_MANIFEST, get(dash_manifest)) diff --git a/src/http/stream.rs b/src/http/stream.rs index 66583ad..fd2fb7a 100644 --- a/src/http/stream.rs +++ b/src/http/stream.rs @@ -37,7 +37,7 @@ impl StreamJob { } } - async fn stream(&mut self) -> std::result::Result<(), StreamAbort> { + async fn stream(&mut self) -> Result<(), StreamAbort> { let header_end = self.prepared.header.len() as u64; if let Some(overlap) = self.interval.overlap(0, header_end) { let (Ok(start), Ok(end)) = @@ -69,7 +69,7 @@ impl StreamJob { Ok(()) } - async fn read(&mut self, offset: u64, length: u64) -> std::result::Result { + async fn read(&mut self, offset: u64, length: u64) -> Result { let permit = match self.first_permit.take() { Some(permit) => permit, None => match acquire_segment_permit(&self.segment_jobs, self.queue_timeout).await { @@ -96,7 +96,7 @@ impl StreamJob { } } - async fn send(&self, item: std::io::Result) -> std::result::Result<(), StreamAbort> { + async fn send(&self, item: std::io::Result) -> Result<(), StreamAbort> { let length = item.as_ref().map_or(0, Bytes::len); match timeout(self.idle_timeout, self.sender.send(item)).await { Ok(Ok(())) => { @@ -112,7 +112,7 @@ impl StreamJob { } /// Surfaces a generic body error to the client, then reports the abort. - async fn fail(&self, message: &'static str) -> std::result::Result { + async fn fail(&self, message: &'static str) -> Result { let _ = self.send(Err(std::io::Error::other(message))).await; Err(StreamAbort::Error) } diff --git a/src/http/tests.rs b/src/http/tests.rs index aa91f22..f7a835c 100644 --- a/src/http/tests.rs +++ b/src/http/tests.rs @@ -680,7 +680,7 @@ async fn ffmpeg_decodes_hls_and_dash_presentations() { eprintln!("skipping media validation because ffmpeg is unavailable"); return; } - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let server = tokio::spawn(async move { axum::serve(listener, app()).await.unwrap(); diff --git a/src/lib.rs b/src/lib.rs index 26c904e..9a82ddb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -23,6 +23,7 @@ mod registry; mod resolver; mod segment; mod source; +mod subtitle; #[cfg(test)] mod testutil; diff --git a/src/media/index.rs b/src/media/index.rs index af0e703..33f039e 100644 --- a/src/media/index.rs +++ b/src/media/index.rs @@ -7,6 +7,10 @@ pub(crate) struct MediaIndex { pub(crate) source: SourceIdentity, pub(crate) movie_timescale: u32, pub(crate) duration: u64, + /// How much later than the source's own clock every served timestamp is, in milliseconds. It + /// is the shared offset the edit lists were resolved with; zero for a file without edits and + /// for a fragmented file, whose timeline simply starts at zero. + pub(crate) presentation_offset_ms: u64, pub(crate) tracks: Vec, /// Tracks in the file that are not packaged, with the reason, so the registry can log them. pub(crate) skipped_tracks: Vec, diff --git a/src/mp4/edit.rs b/src/mp4/edit.rs index f99c75d..220f74b 100644 --- a/src/mp4/edit.rs +++ b/src/mp4/edit.rs @@ -129,6 +129,31 @@ fn unsupported(track_id: u32, detail: impl std::fmt::Display) -> Error { Error::Unsupported(format!("track {track_id}: {detail}")) } +/// The shared offset `O` as a fraction of a second: `(numerator, denominator)`, denominator +/// positive. Served timestamps are presentation timestamps plus `O`. +fn shared_offset(edits: &[(TrackEdit, u32)], movie_timescale: u32) -> (i128, i128) { + let movie = i128::from(movie_timescale); + let mut offset = (0i128, 1i128); + for (edit, timescale) in edits { + let timescale = i128::from(*timescale); + let numerator = i128::from(edit.media_time) * movie - i128::from(edit.delay) * timescale; + let denominator = timescale * movie; + // numerator/denominator > offset.0/offset.1, without dividing. + if numerator * offset.1 > offset.0 * denominator { + offset = (numerator, denominator); + } + } + offset +} + +/// `O` in milliseconds, rounded to the nearest: how far later than the source's own clock every +/// timestamp in the served presentation is. Sidecar subtitles are authored against the source's +/// clock, so their cues move by this much. +pub(super) fn shared_offset_millis(edits: &[(TrackEdit, u32)], movie_timescale: u32) -> u64 { + let (numerator, denominator) = shared_offset(edits, movie_timescale); + u64::try_from((2 * numerator * 1000 + denominator) / (2 * denominator)).unwrap_or(0) +} + /// One shift per input, in that track's ticks, from the shared offset `O`: /// /// ```text @@ -142,17 +167,7 @@ pub(super) fn timeline_shifts( movie_timescale: u32, ) -> Result> { let movie = i128::from(movie_timescale); - // O as a fraction: (numerator, denominator), denominator positive. - let mut offset = (0i128, 1i128); - for (edit, timescale) in edits { - let timescale = i128::from(*timescale); - let numerator = i128::from(edit.media_time) * movie - i128::from(edit.delay) * timescale; - let denominator = timescale * movie; - // numerator/denominator > offset.0/offset.1, without dividing. - if numerator * offset.1 > offset.0 * denominator { - offset = (numerator, denominator); - } - } + let offset = shared_offset(edits, movie_timescale); edits .iter() .map(|(edit, timescale)| { diff --git a/src/mp4/parser.rs b/src/mp4/parser.rs index bfe9a9f..ca87d8a 100644 --- a/src/mp4/parser.rs +++ b/src/mp4/parser.rs @@ -117,6 +117,7 @@ fn parse_metadata( .map(|(edit, track)| (*edit, track.timescale)) .collect::>(); let shifts = edit::timeline_shifts(×cales, moov.movie_timescale)?; + let presentation_offset_ms = edit::shared_offset_millis(×cales, moov.movie_timescale); for ((track, edit), shift) in tracks.iter_mut().zip(edits).zip(shifts) { edit::apply(track, edit, shift)?; } @@ -136,6 +137,7 @@ fn parse_metadata( source: identity, movie_timescale: moov.movie_timescale, duration, + presentation_offset_ms, tracks, skipped_tracks, fragmentation: moov.fragmented.then(|| crate::media::Fragmentation { @@ -561,7 +563,7 @@ mod tests { use crate::media::CodecConfig; use crate::source::LocalMediaSource; - fn block_on(future: F) -> F::Output { + fn block_on(future: F) -> F::Output { tokio::runtime::Builder::new_current_thread() .enable_all() .build() @@ -1228,7 +1230,7 @@ mod tests { #[test] fn rejects_run_length_entries_that_claim_more_samples_than_stsz() { let original = std::fs::read(fixture("h264-aac.mp4")).expect("fixture should read"); - let mut mutated = original.clone(); + let mut mutated = original; // stts payload: version/flags (4), entry count (4), then (sample_count, delta) pairs. let stts = find_type(&mutated, *b"stts"); mutated[stts + 12..stts + 16].copy_from_slice(&i32::MAX.to_be_bytes()); diff --git a/src/observability/metrics.rs b/src/observability/metrics.rs index a5bc13a..bc952f9 100644 --- a/src/observability/metrics.rs +++ b/src/observability/metrics.rs @@ -262,7 +262,7 @@ impl Metrics { .fetch_add(1, Relaxed); } - #[allow(clippy::too_many_lines)] + #[allow(clippy::too_many_lines, reason = "one flat listing of every metric")] pub(crate) fn render(&self, dropped_log_lines: usize) -> String { let mut out = String::with_capacity(8 * 1024); let _ = writeln!( diff --git a/src/protocol/dash.rs b/src/protocol/dash.rs index 9b495d4..644dc32 100644 --- a/src/protocol/dash.rs +++ b/src/protocol/dash.rs @@ -4,6 +4,7 @@ use super::Presentation; use super::hls::track_language; use crate::error::{Error, Result}; use crate::media::Track; +use crate::subtitle::Subtitle; pub(crate) fn manifest(presentation: Presentation<'_>) -> Result { let duration = presentation_duration(presentation)?; @@ -17,6 +18,9 @@ pub(crate) fn manifest(presentation: Presentation<'_>) -> Result { for audio in presentation.audio_tracks() { write_audio_adaptation(&mut manifest, presentation, audio, version)?; } + for subtitle in presentation.subtitles() { + write_subtitle_adaptation(&mut manifest, subtitle, version); + } manifest.push_str(" \n\n"); Ok(manifest) } @@ -71,6 +75,29 @@ fn write_audio_adaptation( Ok(()) } +/// A sidecar `WebVTT` file as a text adaptation set that names the file directly. +fn write_subtitle_adaptation(manifest: &mut String, subtitle: &Subtitle, version: &str) { + let mut roles = String::new(); + if subtitle.forced { + roles.push_str( + " \n", + ); + } else { + roles.push_str( + " \n", + ); + } + if subtitle.default { + roles.push_str(" \n"); + } + writeln!( + manifest, + " \n{roles} \n subtitles/{language}/sub.vtt?v={version}\n \n ", + language = subtitle.language + ) + .expect("writing to a String cannot fail"); +} + fn write_segment_template( manifest: &mut String, presentation: Presentation<'_>, diff --git a/src/protocol/hls.rs b/src/protocol/hls.rs index aae8cc5..12834a1 100644 --- a/src/protocol/hls.rs +++ b/src/protocol/hls.rs @@ -3,6 +3,7 @@ use std::fmt::Write; use super::Presentation; use crate::error::{Error, Result}; use crate::media::{Track, TrackKey}; +use crate::subtitle::Subtitle; pub(crate) fn master_playlist(presentation: Presentation<'_>) -> Result { let video = presentation.video(); @@ -30,6 +31,10 @@ pub(crate) fn master_playlist(presentation: Presentation<'_>) -> Result } } + for subtitle in presentation.subtitles() { + write_subtitle_rendition(&mut playlist, subtitle, version); + } + let mut bandwidth = 0u64; let mut average_bandwidth = 0u64; for track in video.into_iter().chain(audio) { @@ -46,10 +51,18 @@ pub(crate) fn master_playlist(presentation: Presentation<'_>) -> Result .map_or_else(String::new, |(width, height)| { format!(",RESOLUTION={width}x{height}") }); + if let Some(iframes) = iframe_stream(presentation, video)? { + playlist.push_str(&iframes); + } let audio_attribute = if audio_group { ",AUDIO=\"audio\"" } else { "" }; + let subtitle_attribute = if presentation.subtitles().is_empty() { + "" + } else { + ",SUBTITLES=\"subs\"" + }; writeln!( playlist, - "#EXT-X-STREAM-INF:BANDWIDTH={bandwidth},AVERAGE-BANDWIDTH={average_bandwidth},CODECS=\"{}\"{resolution}{audio_attribute}", + "#EXT-X-STREAM-INF:BANDWIDTH={bandwidth},AVERAGE-BANDWIDTH={average_bandwidth},CODECS=\"{}\"{resolution}{audio_attribute}{subtitle_attribute}", codecs.join(",") ) .expect("writing to a String cannot fail"); @@ -89,6 +102,154 @@ pub(crate) fn media_playlist(presentation: Presentation<'_>, key: TrackKey) -> R Ok(playlist) } +/// The playlist that lists a subtitle file: the whole file as a single segment as long as the +/// presentation, which is valid for VOD. It is the same for every language, because it names the +/// file relative to its own location. +pub(crate) fn subtitle_playlist(presentation: Presentation<'_>) -> String { + let seconds = presentation.duration_seconds().max(1); + let version = presentation.version(); + format!( + "#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-TARGETDURATION:{seconds}\n#EXT-X-MEDIA-SEQUENCE:0\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXTINF:{seconds}.000,\nsub.vtt?v={version}\n#EXT-X-ENDLIST\n" + ) +} + +/// The I-frame playlist of the video track: one entry per keyframe, each its own one-sample +/// fragment. `None` for an asset without video. +pub(crate) fn iframe_playlist(presentation: Presentation<'_>) -> Result> { + let Some(track) = presentation.video() else { + return Ok(None); + }; + let version = presentation.version(); + let frames = keyframes(track)?; + let target_duration = frames + .entries + .iter() + .map(|frame| frame.interval.div_ceil(u64::from(track.timescale))) + .max() + .unwrap_or(1); + let mut playlist = format!( + "#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-TARGETDURATION:{target_duration}\n#EXT-X-MEDIA-SEQUENCE:0\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-I-FRAMES-ONLY\n#EXT-X-MAP:URI=\"init.mp4?v={version}\"\n" + ); + for (index, frame) in frames.entries.iter().enumerate() { + let milliseconds = frame + .interval + .checked_mul(1000) + .and_then(|interval| interval.checked_div(u64::from(track.timescale))) + .ok_or_else(|| Error::InvalidMedia("keyframe interval overflow".to_owned()))?; + writeln!( + playlist, + "#EXTINF:{}.{:03},\niframes/{index}/media.m4s?v={version}", + milliseconds / 1000, + milliseconds % 1000 + ) + .expect("writing to a String cannot fail"); + } + playlist.push_str("#EXT-X-ENDLIST\n"); + Ok(Some(playlist)) +} + +/// The `#EXT-X-I-FRAME-STREAM-INF` line for the video track, or `None` when there is no video. +fn iframe_stream(presentation: Presentation<'_>, video: Option<&Track>) -> Result> { + let Some(track) = video else { + return Ok(None); + }; + let frames = keyframes(track)?; + let (Some(peak), Some(average)) = ( + frames.peak_bandwidth(track.timescale), + frames.average_bandwidth(track.timescale), + ) else { + return Ok(None); + }; + let resolution = track + .codec + .dimensions() + .map_or_else(String::new, |(width, height)| { + format!(",RESOLUTION={width}x{height}") + }); + Ok(Some(format!( + "#EXT-X-I-FRAME-STREAM-INF:BANDWIDTH={peak},AVERAGE-BANDWIDTH={average},CODECS=\"{}\"{resolution},URI=\"video/iframes.m3u8?v={}\"\n", + track.codec.codecs(), + presentation.version() + ))) +} + +/// The sync samples of a track: where each is, how long it stays on screen (until the next one), +/// and how big it is. +struct Keyframes { + entries: Vec, +} + +struct Keyframe { + /// Ticks from this keyframe to the next, or to the end of the track. + interval: u64, + bytes: u64, +} + +fn keyframes(track: &Track) -> Result { + let overflow = || Error::InvalidMedia("keyframe interval overflow".to_owned()); + let sync = track + .samples + .iter() + .filter(|sample| sample.is_sync) + .collect::>(); + let end = track + .samples + .last() + .map(|last| last.decode_time.checked_add(u64::from(last.duration))) + .ok_or_else(overflow)? + .ok_or_else(overflow)?; + let mut entries = Vec::with_capacity(sync.len()); + for (position, sample) in sync.iter().enumerate() { + let next = sync.get(position + 1).map_or(end, |next| next.decode_time); + entries.push(Keyframe { + interval: next.checked_sub(sample.decode_time).ok_or_else(overflow)?, + bytes: u64::from(sample.size), + }); + } + Ok(Keyframes { entries }) +} + +impl Keyframes { + /// The largest keyframe, in bits per second over the interval it stands for. + fn peak_bandwidth(&self, timescale: u32) -> Option { + self.entries + .iter() + .filter_map(|frame| bits_per_second(frame.bytes, frame.interval, timescale)) + .max() + } + + /// All keyframes' bytes over the time they cover. + fn average_bandwidth(&self, timescale: u32) -> Option { + let bytes = self.entries.iter().map(|frame| frame.bytes).sum(); + let ticks = self.entries.iter().map(|frame| frame.interval).sum(); + bits_per_second(bytes, ticks, timescale) + } +} + +fn bits_per_second(bytes: u64, ticks: u64, timescale: u32) -> Option { + bytes + .checked_mul(8)? + .checked_mul(u64::from(timescale))? + .checked_div(ticks) +} + +/// One `#EXT-X-MEDIA:TYPE=SUBTITLES` line. `AUTOSELECT` follows `DEFAULT`, and `FORCED` marks +/// a track a player should show even when the viewer has not asked for subtitles. +fn write_subtitle_rendition(playlist: &mut String, subtitle: &Subtitle, version: &str) { + let flag = |value: bool| if value { "YES" } else { "NO" }; + let name = subtitle.label.replace('"', "'"); + writeln!( + playlist, + "#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID=\"subs\",NAME=\"{name}\",LANGUAGE=\"{}\",DEFAULT={},AUTOSELECT={},FORCED={},URI=\"subtitles/{}/index.m3u8?v={version}\"", + subtitle.language, + flag(subtitle.default), + flag(subtitle.default || subtitle.forced), + flag(subtitle.forced), + subtitle.language + ) + .expect("writing to a String cannot fail"); +} + /// One `#EXT-X-MEDIA` line. Renditions are named by position, because the handler names encoders /// write (`SoundHandler`) say nothing to a viewer; the language is added when the file has one. fn write_audio_rendition(playlist: &mut String, track: &Track, index: usize, version: &str) { @@ -141,6 +302,44 @@ mod tests { ); } + #[test] + fn the_iframe_playlist_lists_one_fragment_per_keyframe() { + let loaded = Loaded::h264_aac(); + let keyframes = loaded.index.tracks[0] + .samples + .iter() + .filter(|sample| sample.is_sync) + .count(); + + let playlist = iframe_playlist(loaded.presentation()) + .expect("playlist should render") + .expect("a video asset has one"); + + assert!(playlist.contains("#EXT-X-I-FRAMES-ONLY\n"), "{playlist}"); + assert!(playlist.contains(&format!("#EXT-X-MAP:URI=\"init.mp4?v={}\"", loaded.version))); + assert_eq!(playlist.matches("#EXTINF:").count(), keyframes); + assert!(playlist.contains(&format!("iframes/{}/media.m4s?v=", keyframes - 1))); + let master = master_playlist(loaded.presentation()).expect("master should render"); + assert!( + master.contains("#EXT-X-I-FRAME-STREAM-INF:BANDWIDTH="), + "{master}" + ); + assert!(master.contains(&format!("URI=\"video/iframes.m3u8?v={}\"", loaded.version))); + } + + #[test] + fn an_audio_only_asset_has_no_iframe_playlist() { + let loaded = Loaded::fixture("aac-only.m4a"); + + assert!( + iframe_playlist(loaded.presentation()) + .expect("rendering should succeed") + .is_none() + ); + let master = master_playlist(loaded.presentation()).expect("master should render"); + assert!(!master.contains("I-FRAME"), "{master}"); + } + #[test] fn master_lists_every_audio_track_and_defaults_the_first() { let loaded = Loaded::fixture("h264-aac-two-audio.mp4"); diff --git a/src/protocol/presentation.rs b/src/protocol/presentation.rs index 8388c80..7418db8 100644 --- a/src/protocol/presentation.rs +++ b/src/protocol/presentation.rs @@ -6,6 +6,7 @@ use crate::error::{Error, Result}; use crate::media::{Sample, Track, TrackKey, TrackKind}; use crate::segment::{SegmentPlan, TrackSegment}; +use crate::subtitle::Subtitle; /// Bits per second for one track, in the terms HLS and DASH declare them. #[derive(Debug, Clone, Copy)] @@ -20,6 +21,7 @@ pub(crate) struct Presentation<'a> { tracks: &'a [Track], plan: &'a SegmentPlan, version: &'a str, + subtitles: &'a [Subtitle], } impl<'a> Presentation<'a> { @@ -28,9 +30,28 @@ impl<'a> Presentation<'a> { tracks, plan, version, + subtitles: &[], } } + /// The same view with the asset's sidecar subtitles. + pub(crate) const fn with_subtitles(self, subtitles: &'a [Subtitle]) -> Self { + Self { subtitles, ..self } + } + + pub(crate) const fn subtitles(&self) -> &'a [Subtitle] { + self.subtitles + } + + /// The presentation's length in seconds, rounded up: the longest track. + pub(crate) fn duration_seconds(&self) -> u64 { + self.tracks + .iter() + .map(|track| track.duration.div_ceil(u64::from(track.timescale))) + .max() + .unwrap_or(0) + } + pub(crate) const fn tracks(&self) -> &'a [Track] { self.tracks } diff --git a/src/registry/mod.rs b/src/registry/mod.rs index ae32816..131682e 100644 --- a/src/registry/mod.rs +++ b/src/registry/mod.rs @@ -29,8 +29,11 @@ use crate::asset::PackagedAsset; use crate::config::{Config, LimitsConfig, is_valid_asset_id}; use crate::error::{Error, Result}; use crate::observability::metrics::{CacheEvent, Metrics, ResolverOutcome}; -use crate::resolver::{AssetLocation, AssetResolver, Resolution, ResolveError, ResolvedAsset}; +use crate::resolver::{ + AssetLocation, AssetResolver, Resolution, ResolveError, ResolvedAsset, SubtitleLocation, +}; use crate::source::LocationRefresher; +use crate::subtitle::Subtitle; use cache::LoadedCache; pub(crate) use opener::SourceOpener; @@ -546,6 +549,49 @@ impl AssetRegistry { } } + /// Fetches every sidecar subtitle file the mapper listed, within the configured limits. A file + /// that cannot be read fails the asset, so a viewer never gets a silently missing language. + async fn fetch_subtitles(&self, listed: &[SubtitleLocation]) -> Result> { + if listed.len() > self.limits.max_subtitles { + return Err(Error::InvalidMedia(format!( + "the mapper listed {} subtitles, more than limits.max_subtitles ({})", + listed.len(), + self.limits.max_subtitles + ))); + } + let mut total = 0u64; + let mut subtitles = Vec::with_capacity(listed.len()); + for entry in listed { + let data = self + .opener + .read_whole(&entry.location, self.limits.max_subtitle_bytes) + .await + .map_err(|error| match error { + // Only a bad file is named; an outage or a rejected location keeps its kind, so + // it is retried or reported as the mapper's fault and not as broken media. + Error::InvalidMedia(message) => { + Error::InvalidMedia(format!("subtitle `{}`: {message}", entry.language)) + } + other => other, + })?; + total = total.saturating_add(data.len() as u64); + if total > self.limits.max_subtitles_total_bytes { + return Err(Error::InvalidMedia(format!( + "subtitles exceed limits.max_subtitles_total_bytes ({})", + self.limits.max_subtitles_total_bytes + ))); + } + subtitles.push(Subtitle { + language: entry.language.clone(), + label: entry.label.clone(), + default: entry.default, + forced: entry.forced, + data, + }); + } + Ok(subtitles) + } + async fn load( self: &Arc, asset_id: &str, @@ -575,7 +621,14 @@ impl AssetRegistry { Arc::clone(&refresher) as Arc, ) .await?; - PackagedAsset::load(source, self.settings.segment_duration_ms, &self.limits).await + let subtitles = self.fetch_subtitles(&resolved.subtitles).await?; + PackagedAsset::load_with_subtitles( + source, + subtitles, + self.settings.segment_duration_ms, + &self.limits, + ) + .await } .await; if result.is_ok() { diff --git a/src/registry/opener.rs b/src/registry/opener.rs index c54d794..5cdde04 100644 --- a/src/registry/opener.rs +++ b/src/registry/opener.rs @@ -1,11 +1,17 @@ //! Turns a resolved location into an open media source. use std::path::{Path, PathBuf}; +use std::pin::Pin; use std::sync::Arc; +use bytes::Bytes; +use reqwest::Url; + use crate::error::{Error, Result}; use crate::resolver::AssetLocation; -use crate::source::{LocalMediaSource, LocationRefresher, MediaSourceKind, RemoteReader}; +use crate::source::{ + ByteRange, LocalMediaSource, LocationRefresher, MediaSourceKind, RemoteReader, +}; #[derive(Debug)] pub(crate) struct SourceOpener { @@ -45,6 +51,44 @@ impl SourceOpener { } } +impl SourceOpener { + /// Reads a whole small object, such as a subtitle file, refusing one over `max_bytes` before + /// any of it is read. It goes through the same confinement and remote-media rules as media. + pub(crate) async fn read_whole( + &self, + location: &AssetLocation, + max_bytes: u64, + ) -> Result { + let source = match location { + AssetLocation::File(_) => self.open(location, Arc::new(NoRefresh)).await?, + AssetLocation::Http(url) => { + MediaSourceKind::Http(Arc::new(self.remote.open(url.clone()).await?)) + } + }; + let length = source.len(); + if length > max_bytes { + return Err(Error::InvalidMedia(format!( + "file is {length} bytes, more than the limit of {max_bytes}" + ))); + } + source.read_range(ByteRange::new(0, length)).await + } +} + +/// For an object that has no signed URL to renew. +#[derive(Debug)] +struct NoRefresh; + +impl LocationRefresher for NoRefresh { + fn refresh(&self) -> Pin> + Send + '_>> { + Box::pin(async { + Err(Error::Upstream( + "this location cannot be refreshed".to_owned(), + )) + }) + } +} + /// Joins `path` to the media root, resolves symlinks, and requires a regular file that is still /// beneath the root, so no location can escape it. fn confine(root: &Path, path: &Path) -> Result { diff --git a/src/registry/tests.rs b/src/registry/tests.rs index 358e1c0..e1d3c06 100644 --- a/src/registry/tests.rs +++ b/src/registry/tests.rs @@ -1164,3 +1164,249 @@ fn locations_that_differ_only_in_their_query_are_the_same_object() { assert!(File("a.mp4".into()).same_object(&File("a.mp4".into()))); assert!(!File("a.mp4".into()).same_object(&url("https://o.example/a.mp4"))); } + +// --------------------------------------------------------------------------------------------- +// Sidecar subtitles +// --------------------------------------------------------------------------------------------- + +fn text(bytes: &Bytes) -> String { + String::from_utf8_lossy(bytes).into_owned() +} + +#[tokio::test] +async fn subtitles_from_the_mapper_appear_in_both_protocols_and_are_served() { + let h = harness().await; + h.mapper.state.set( + "movie", + Answer::file("v1", "h264-aac.mp4") + .with_subtitle("en", "subtitles-en.vtt") + .with_subtitle("fr", "subtitles-fr.vtt"), + ); + + let (status, _, master) = fetch(&h.app, "/hls/movie/master.m3u8").await; + let master_text = text(&master); + let version = version_in(&master); + + assert_eq!(status, StatusCode::OK); + assert!(master_text.contains("TYPE=SUBTITLES,GROUP-ID=\"subs\",NAME=\"EN\",LANGUAGE=\"en\"")); + assert!(master_text.contains(&format!("URI=\"subtitles/fr/index.m3u8?v={version}\""))); + assert!(master_text.contains(",SUBTITLES=\"subs\""), "{master_text}"); + + let (status, headers, playlist) = fetch(&h.app, "/hls/movie/subtitles/en/index.m3u8").await; + assert_eq!(status, StatusCode::OK); + assert_eq!(headers["content-type"], "application/vnd.apple.mpegurl"); + assert!(text(&playlist).contains(&format!("sub.vtt?v={version}"))); + + let uri = format!("/hls/movie/subtitles/en/sub.vtt?v={version}"); + let (status, headers, file) = fetch(&h.app, &uri).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(headers["content-type"], "text/vtt; charset=utf-8"); + assert!( + headers["cache-control"] + .to_str() + .unwrap() + .contains("immutable") + ); + assert!(text(&file).starts_with("WEBVTT")); + assert_eq!( + fetch( + &h.app, + &format!("/dash/movie/subtitles/en/sub.vtt?v={version}") + ) + .await + .2, + file + ); + + let (_, _, manifest) = fetch(&h.app, "/dash/movie/manifest.mpd").await; + let manifest = text(&manifest); + assert!( + manifest.contains("lang=\"fr\" mimeType=\"text/vtt\""), + "{manifest}" + ); + assert!(manifest.contains(&format!( + "subtitles/en/sub.vtt?v={version}" + ))); +} + +#[tokio::test] +async fn subtitle_urls_need_the_current_version_and_a_listed_language() { + let h = harness().await; + h.mapper.state.set( + "movie", + Answer::file("v1", "h264-aac.mp4").with_subtitle("en", "subtitles-en.vtt"), + ); + let version = version_in(&fetch(&h.app, "/hls/movie/master.m3u8").await.2); + + for uri in [ + "/hls/movie/subtitles/en/sub.vtt".to_owned(), + "/hls/movie/subtitles/en/sub.vtt?v=stale".to_owned(), + format!("/hls/movie/subtitles/de/sub.vtt?v={version}"), + "/hls/movie/subtitles/de/index.m3u8".to_owned(), + ] { + assert_eq!(status(&h.app, &uri).await, StatusCode::NOT_FOUND, "{uri}"); + } + let uri = format!("/hls/movie/subtitles/EN/sub.vtt?v={version}"); + assert_eq!( + status(&h.app, &uri).await, + StatusCode::OK, + "language matches without case" + ); +} + +#[tokio::test] +async fn an_asset_without_subtitles_is_unchanged() { + let h = harness().await; + h.mapper + .state + .set("movie", Answer::file("v1", "h264-aac.mp4")); + + let master = text(&fetch(&h.app, "/hls/movie/master.m3u8").await.2); + + assert!(!master.contains("SUBTITLES"), "{master}"); + assert!(!text(&fetch(&h.app, "/dash/movie/manifest.mpd").await.2).contains("text/vtt")); +} + +#[tokio::test] +async fn changing_a_caption_gives_the_asset_new_urls() { + let h = harness().await; + let mut first = Answer::file("v1", "h264-aac.mp4").with_subtitle("en", "subtitles-en.vtt"); + first.ttl_seconds = Some(0); + h.mapper.state.set("movie", first); + let old = version_in(&fetch(&h.app, "/hls/movie/master.m3u8").await.2); + + // The mapper changes its version when a caption changes, as its contract says it must. + let mut second = Answer::file("v2", "h264-aac.mp4").with_subtitle("en", "subtitles-fr.vtt"); + second.ttl_seconds = Some(0); + h.mapper.state.set("movie", second); + tokio::time::sleep(Duration::from_millis(40)).await; + let new = version_in(&fetch(&h.app, "/hls/movie/master.m3u8").await.2); + + assert_ne!(old, new); + assert_eq!( + status(&h.app, &format!("/hls/movie/subtitles/en/sub.vtt?v={old}")).await, + StatusCode::NOT_FOUND, + "the old URL stops resolving" + ); + let file = fetch(&h.app, &format!("/hls/movie/subtitles/en/sub.vtt?v={new}")) + .await + .2; + assert!(text(&file).contains("Bonjour")); +} + +#[tokio::test] +async fn a_bad_subtitle_fails_the_asset() { + for (path, expected) in [ + // Not WebVTT: the media is fine and the file is not. + ("subtitles-bad.srt", StatusCode::INTERNAL_SERVER_ERROR), + // Not there: the mapper pointed at something that does not exist. + ("missing.vtt", StatusCode::BAD_GATEWAY), + ] { + let h = harness().await; + h.mapper.state.set( + "movie", + Answer::file("v1", "h264-aac.mp4").with_subtitle("fr", path), + ); + + assert_eq!( + status(&h.app, "/hls/movie/master.m3u8").await, + expected, + "{path}" + ); + } +} + +#[tokio::test] +async fn subtitle_limits_are_enforced() { + let h = harness_with(|config| config.limits.max_subtitle_bytes = 20).await; + h.mapper.state.set( + "movie", + Answer::file("v1", "h264-aac.mp4").with_subtitle("en", "subtitles-en.vtt"), + ); + assert_ne!( + status(&h.app, "/hls/movie/master.m3u8").await, + StatusCode::OK, + "a file over max_subtitle_bytes fails the asset" + ); + + let h = harness_with(|config| config.limits.max_subtitles = 1).await; + h.mapper.state.set( + "movie", + Answer::file("v1", "h264-aac.mp4") + .with_subtitle("en", "subtitles-en.vtt") + .with_subtitle("fr", "subtitles-fr.vtt"), + ); + assert_ne!( + status(&h.app, "/hls/movie/master.m3u8").await, + StatusCode::OK, + "more files than max_subtitles fails the asset" + ); +} + +#[tokio::test] +async fn invalid_subtitle_entries_from_the_mapper_are_rejected() { + for entries in [ + // A language that is not a URL-safe tag. + r#"[{"language":"../x","location":{"type":"file","path":"subtitles-en.vtt"}}]"#, + // The same language twice, differing only in case. + r#"[{"language":"en","location":{"type":"file","path":"subtitles-en.vtt"}},{"language":"EN","location":{"type":"file","path":"subtitles-fr.vtt"}}]"#, + // Two defaults. + r#"[{"language":"en","default":true,"location":{"type":"file","path":"subtitles-en.vtt"}},{"language":"fr","default":true,"location":{"type":"file","path":"subtitles-fr.vtt"}}]"#, + // A path that escapes the media root. + r#"[{"language":"en","location":{"type":"file","path":"../secret.vtt"}}]"#, + // A remote host the policy does not allow. + r#"[{"language":"en","location":{"type":"http","url":"http://evil.example/x.vtt"}}]"#, + ] { + let h = harness().await; + *h.mapper.state.raw_body.lock().unwrap() = Some(format!( + r#"{{"asset_id":"movie","version":"v1","location":{{"type":"file","path":"h264-aac.mp4"}},"subtitles":{entries}}}"# + )); + assert_eq!( + status(&h.app, "/hls/movie/master.m3u8").await, + StatusCode::BAD_GATEWAY, + "{entries}" + ); + } +} + +#[tokio::test] +async fn cues_follow_the_shared_offset_of_the_edit_lists_and_nothing_else() { + let h = harness().await; + for (asset, file) in [ + // The edit lists trim encoder delay, so everything is served 66.7 ms later than the source. + ("edited", "h264-aac-default-edits.mp4"), + // The video starts 1.5 s in, which the presentation the cues were written against already + // contains: no shift, or the cues would be late by that much. + ("delayed", "h264-aac-video-delay.mp4"), + ("plain", "h264-aac.mp4"), + ] { + h.mapper.state.set( + asset, + Answer::file("v1", file).with_subtitle("en", "subtitles-en.vtt"), + ); + } + + let mut served = Vec::new(); + for asset in ["edited", "delayed", "plain"] { + let version = version_in(&fetch(&h.app, &format!("/hls/{asset}/master.m3u8")).await.2); + let uri = format!("/hls/{asset}/subtitles/en/sub.vtt?v={version}"); + served.push(text(&fetch(&h.app, &uri).await.2)); + } + + assert!( + served[0].contains("00:00:00.567 --> 00:00:01.567 line:90%"), + "{}", + served[0] + ); + assert!( + served[0].contains("00:00:01.567 --> 00:00:02.567\nWorld"), + "{}", + served[0] + ); + for unchanged in &served[1..] { + assert!( + unchanged.contains("00:00.500 --> 00:01.500 line:90%"), + "{unchanged}" + ); + } +} diff --git a/src/resolver/catalog.rs b/src/resolver/catalog.rs index 50cc0ac..0f52d35 100644 --- a/src/resolver/catalog.rs +++ b/src/resolver/catalog.rs @@ -29,6 +29,7 @@ impl StaticResolver { let path = self.assets.get(asset_id).ok_or(ResolveError::NotFound)?; Ok(Resolution::Resolved(ResolvedAsset { location: AssetLocation::File(path.clone()), + subtitles: Vec::new(), version: STATIC_VERSION.to_owned(), valid_until: Instant::now() + FOREVER, hard_expiry: None, diff --git a/src/resolver/mapper.rs b/src/resolver/mapper.rs index b7203ac..2eaae36 100644 --- a/src/resolver/mapper.rs +++ b/src/resolver/mapper.rs @@ -13,12 +13,13 @@ use time::format_description::well_known::Rfc3339; use tokio::time::sleep; use super::policy::{LocationPolicy, validate_relative_path}; -use super::{AssetLocation, Resolution, ResolveError, ResolvedAsset}; +use super::{AssetLocation, Resolution, ResolveError, ResolvedAsset, SubtitleLocation}; use crate::config::MapperConfig; use crate::config::Secret; use crate::observability::request_id; const MAX_VERSION_BYTES: usize = 256; +const MAX_LABEL_BYTES: usize = 128; const MAX_RETRY_AFTER: Duration = Duration::from_secs(1); #[derive(Debug)] @@ -38,6 +39,19 @@ struct Wire { ttl_seconds: Option, expires_at: Option, location: WireLocation, + #[serde(default)] + subtitles: Vec, +} + +#[derive(Debug, Deserialize)] +struct WireSubtitle { + language: String, + label: Option, + #[serde(default)] + default: bool, + #[serde(default)] + forced: bool, + location: WireLocation, } /// `serde` rejects an unknown `type`, which is how unknown location kinds are refused. @@ -222,6 +236,72 @@ impl HttpResolver { Duration::from_millis(requested.clamp(self.settings.min_ttl_ms, self.settings.max_ttl_ms)) } + /// A location, checked against the path rules or the remote-media policy. + fn interpret_location(&self, wire: &WireLocation) -> Result { + match wire { + WireLocation::File { path } => Ok(AssetLocation::File( + validate_relative_path(path).map_err(ResolveError::Rejected)?, + )), + WireLocation::Http { url } => { + let url = Url::parse(url) + .map_err(|_| ResolveError::Rejected("location URL is not valid".to_owned()))?; + self.policy + .check_url(&url) + .map_err(ResolveError::Rejected)?; + Ok(AssetLocation::Http(url)) + } + } + } + + fn interpret_subtitles( + &self, + wire: &[WireSubtitle], + ) -> Result, ResolveError> { + let reject = |message: String| Err(ResolveError::Rejected(message)); + let mut subtitles: Vec = Vec::with_capacity(wire.len()); + for entry in wire { + if !is_language_tag(&entry.language) { + return reject(format!( + "subtitle language `{}` is not a BCP 47 tag of letters, digits and hyphens", + entry.language.escape_default() + )); + } + if subtitles + .iter() + .any(|known| known.language.eq_ignore_ascii_case(&entry.language)) + { + return reject(format!( + "subtitle language `{}` is listed twice", + entry.language + )); + } + let label = entry + .label + .clone() + .unwrap_or_else(|| entry.language.clone()); + if label.is_empty() + || label.len() > MAX_LABEL_BYTES + || label.chars().any(char::is_control) + { + return reject(format!( + "subtitle `{}` has an empty, overlong, or control-character label", + entry.language + )); + } + subtitles.push(SubtitleLocation { + language: entry.language.clone(), + label, + default: entry.default, + forced: entry.forced, + location: self.interpret_location(&entry.location)?, + }); + } + if subtitles.iter().filter(|subtitle| subtitle.default).count() > 1 { + return reject("more than one subtitle is marked default".to_owned()); + } + Ok(subtitles) + } + /// Validates a `200` answer against the request and the location policy. fn interpret( &self, @@ -242,19 +322,8 @@ impl HttpResolver { { return reject("mapper version must be 1 to 256 visible ASCII characters"); } - let location = match &wire.location { - WireLocation::File { path } => { - AssetLocation::File(validate_relative_path(path).map_err(ResolveError::Rejected)?) - } - WireLocation::Http { url } => { - let url = Url::parse(url) - .map_err(|_| ResolveError::Rejected("location URL is not valid".to_owned()))?; - self.policy - .check_url(&url) - .map_err(ResolveError::Rejected)?; - AssetLocation::Http(url) - } - }; + let location = self.interpret_location(&wire.location)?; + let subtitles = self.interpret_subtitles(&wire.subtitles)?; let now = Instant::now(); let mut valid_until = now + self.ttl(wire.ttl_seconds, max_age_seconds); @@ -275,6 +344,7 @@ impl HttpResolver { } Ok(ResolvedAsset { location, + subtitles, version: wire.version, valid_until, hard_expiry, @@ -292,3 +362,13 @@ fn max_age(response: &Response) -> Option { .split(',') .find_map(|directive| directive.trim().strip_prefix("max-age=")?.parse().ok()) } + +/// A language tag as it can appear in a URL path: letters, digits, and single hyphens, starting +/// with a letter, at most 35 characters. +fn is_language_tag(tag: &str) -> bool { + tag.len() <= 35 + && tag + .split('-') + .all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_alphanumeric())) + && tag.as_bytes().first().is_some_and(u8::is_ascii_alphabetic) +} diff --git a/src/resolver/mod.rs b/src/resolver/mod.rs index e185945..6fe7f79 100644 --- a/src/resolver/mod.rs +++ b/src/resolver/mod.rs @@ -46,10 +46,23 @@ impl AssetLocation { } } +/// A sidecar `WebVTT` file the mapper attached to an asset. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct SubtitleLocation { + /// A BCP 47 tag, unique within the asset (case-insensitively). It appears in URLs. + pub(crate) language: String, + pub(crate) label: String, + pub(crate) default: bool, + pub(crate) forced: bool, + pub(crate) location: AssetLocation, +} + /// A resolver's answer for one asset. #[derive(Debug, Clone)] pub(crate) struct ResolvedAsset { pub(crate) location: AssetLocation, + /// Sidecar subtitles, in the order the mapper listed them. + pub(crate) subtitles: Vec, /// Opaque change token: equal versions mean identical media. pub(crate) version: String, /// After this instant the answer must be revalidated before use. diff --git a/src/source/metadata.rs b/src/source/metadata.rs index a406a28..3d58508 100644 --- a/src/source/metadata.rs +++ b/src/source/metadata.rs @@ -429,19 +429,17 @@ impl<'a> Walk<'a> { } // A cut `mdat` belongs to the `moof` just before it, which now describes samples that are // not all there. - let mut from = offset; - let mut dropped_fragments = 0; - if cut == Cut::Mdat + let (from, dropped_fragments) = if cut == Cut::Mdat && let Some(moof) = self.last_moof.take_if(|start| { self.fragments .last() .is_some_and(|fragment| fragment.offset == *start) - }) - { + }) { self.fragments.pop(); - from = moof; - dropped_fragments = 1; - } + (moof, 1) + } else { + (offset, 0) + }; if self.fragments.is_empty() { return Err(invalid( "the file ends before its first fragment is complete", @@ -615,11 +613,16 @@ mod tests { bytes } - /// Writes `bytes` under `target/` and opens it as a source. + /// Writes `bytes` under `target/` and opens it as a source. Tests run in parallel and several + /// use the same names, so each call gets a file of its own: one test rewriting a file while + /// another reads it made `every_kind_of_wrong_sidx_falls_back_to_the_sequential_walk` fail now + /// and then. fn source(name: &str, bytes: &[u8]) -> MediaSourceKind { + static NEXT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); let directory = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("target/metadata-tests"); std::fs::create_dir_all(&directory).unwrap(); - let path = directory.join(name); + let unique = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let path = directory.join(format!("{}-{unique}-{name}", std::process::id())); std::fs::write(&path, bytes).unwrap(); MediaSourceKind::Local(Arc::new(LocalMediaSource::open(path).unwrap())) } @@ -1092,7 +1095,7 @@ mod tests { ("zero.mp4", zero, 0), ("past.mp4", past_the_end, 0), // Media said to start in the middle of a box. - ("misaligned.mp4", honest.clone(), 5), + ("misaligned.mp4", honest, 5), ] { let file = with_sidx(&head, &parts, &sizes, first_offset); diff --git a/src/subtitle.rs b/src/subtitle.rs new file mode 100644 index 0000000..92653fb --- /dev/null +++ b/src/subtitle.rs @@ -0,0 +1,198 @@ +//! Sidecar `WebVTT` files: validation and timeline correction. +//! +//! A subtitle file is fetched once when its asset loads and held in memory. Packaging can move an +//! asset onto a shifted timeline (an edit list or a fragmented start time), so a cue authored +//! against the source would appear early or late; [`prepare`] adds that offset to every cue timing +//! line and leaves everything else in the file as it was. + +use std::fmt::Write; + +use bytes::Bytes; + +use crate::error::{Error, Result}; + +/// One subtitle file of an asset. Before [`prepare`] `data` is what was fetched; afterwards it is +/// what is served. +#[derive(Debug, Clone)] +pub(crate) struct Subtitle { + pub(crate) language: String, + pub(crate) label: String, + pub(crate) default: bool, + pub(crate) forced: bool, + pub(crate) data: Bytes, +} + +/// Validates `data` as `WebVTT` and moves every cue `offset_ms` later. +/// +/// The file must be UTF-8 and begin with `WEBVTT` (after an optional byte order mark). With no +/// offset the original bytes are returned untouched, but a cue timing line that cannot be read is +/// refused either way, so a bad file is caught when the asset loads and not when a viewer's +/// player meets it. +pub(crate) fn prepare(language: &str, data: &[u8], offset_ms: u64) -> Result { + let refuse = |reason: &str| Error::InvalidMedia(format!("subtitle `{language}`: {reason}")); + let text = std::str::from_utf8(data).map_err(|_| refuse("is not UTF-8"))?; + let body = text.strip_prefix('\u{feff}').unwrap_or(text); + let header_ok = body + .strip_prefix("WEBVTT") + .is_some_and(|rest| rest.is_empty() || rest.starts_with([' ', '\t', '\n', '\r'])); + if !header_ok { + return Err(refuse("does not begin with WEBVTT")); + } + + let mut shifted = String::with_capacity(body.len() + body.len() / 8); + for line in body.split_inclusive(['\n', '\r']) { + let content = line.trim_end_matches(['\n', '\r']); + let terminator = &line[content.len()..]; + if content.contains("-->") { + let timing = shift_timing_line(content, offset_ms) + .ok_or_else(|| refuse("has a cue timing line that cannot be read"))?; + shifted.push_str(&timing); + } else { + shifted.push_str(content); + } + shifted.push_str(terminator); + } + if offset_ms == 0 { + Ok(Bytes::copy_from_slice(data)) + } else { + Ok(Bytes::from(shifted)) + } +} + +/// `start --> end settings`, with both times moved. `None` when it is not a well-formed timing. +fn shift_timing_line(line: &str, offset_ms: u64) -> Option { + let (start, rest) = line.split_once("-->")?; + let start = start.trim(); + let rest = rest.trim_start(); + let (end, settings) = rest + .split_once([' ', '\t']) + .map_or((rest, ""), |(end, settings)| (end, settings)); + let start = parse_timestamp(start)?.checked_add(offset_ms)?; + let end = parse_timestamp(end)?.checked_add(offset_ms)?; + let mut out = String::new(); + write!( + out, + "{} --> {}", + format_timestamp(start), + format_timestamp(end) + ) + .ok()?; + if !settings.is_empty() { + out.push(' '); + out.push_str(settings.trim_start()); + } + Some(out) +} + +/// `[hh:]mm:ss.ttt` in milliseconds. Hours may have more than two digits. +fn parse_timestamp(text: &str) -> Option { + let (clock, millis) = text.split_once('.')?; + if millis.len() != 3 || !millis.bytes().all(|byte| byte.is_ascii_digit()) { + return None; + } + let fields = clock.split(':').collect::>(); + let (hours, minutes, seconds) = match fields.as_slice() { + [minutes, seconds] => ("0", *minutes, *seconds), + [hours, minutes, seconds] => (*hours, *minutes, *seconds), + _ => return None, + }; + let number = |field: &str, at_least: usize, at_most: usize| -> Option { + (field.bytes().all(|byte| byte.is_ascii_digit()) + && (at_least..=at_most).contains(&field.len())) + .then(|| field.parse().ok())? + }; + let hours = if fields.len() == 3 { + number(hours, 2, 10)? + } else { + 0 + }; + let minutes = number(minutes, 2, 2).filter(|minutes| *minutes < 60)?; + let seconds = number(seconds, 2, 2).filter(|seconds| *seconds < 60)?; + hours + .checked_mul(3_600_000)? + .checked_add(minutes * 60_000)? + .checked_add(seconds * 1000)? + .checked_add(millis.parse::().ok()?) +} + +fn format_timestamp(milliseconds: u64) -> String { + let hours = milliseconds / 3_600_000; + let minutes = milliseconds / 60_000 % 60; + let seconds = milliseconds / 1000 % 60; + format!( + "{hours:02}:{minutes:02}:{seconds:02}.{:03}", + milliseconds % 1000 + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn shifted(text: &str, offset_ms: u64) -> String { + String::from_utf8( + prepare("en", text.as_bytes(), offset_ms) + .expect("valid") + .to_vec(), + ) + .expect("UTF-8") + } + + #[test] + fn every_cue_moves_and_the_rest_of_the_file_does_not() { + let file = "WEBVTT - title\n\nNOTE a comment\n\nintro\n00:01.000 --> 00:02.500 line:90% align:start\nHello\n\n01:00:00.000 --> 01:00:01.000\nHour cue\n"; + + let out = shifted(file, 2500); + + assert_eq!( + out, + "WEBVTT - title\n\nNOTE a comment\n\nintro\n00:00:03.500 --> 00:00:05.000 line:90% align:start\nHello\n\n01:00:02.500 --> 01:00:03.500\nHour cue\n" + ); + } + + #[test] + fn the_shift_carries_across_seconds_minutes_and_hours() { + let out = shifted("WEBVTT\n\n59:59.900 --> 59:59.999\nx\n", 200); + + assert!(out.contains("01:00:00.100 --> 01:00:00.199"), "{out}"); + } + + #[test] + fn no_offset_returns_the_original_bytes() { + let file = "\u{feff}WEBVTT\r\n\r\n00:01.000 --> 00:02.000\r\nHi\r\n"; + + assert_eq!( + prepare("en", file.as_bytes(), 0).expect("valid"), + file.as_bytes() + ); + } + + #[test] + fn line_endings_survive_a_shift() { + let out = shifted("WEBVTT\r\n\r\n00:01.000 --> 00:02.000\r\nHi\r\n", 1000); + + assert_eq!(out, "WEBVTT\r\n\r\n00:00:02.000 --> 00:00:03.000\r\nHi\r\n"); + } + + #[test] + fn bad_files_are_refused_and_name_the_language() { + for (data, needle) in [ + (&b"\xff\xfe"[..], "not UTF-8"), + (b"", "WEBVTT"), + (b"WEBVTTX\n", "WEBVTT"), + (b"1\n00:01,000 --> 00:02,000\nsrt\n", "WEBVTT"), + (b"WEBVTT\n\n00:01.000 --> nonsense\nx\n", "timing"), + (b"WEBVTT\n\n00:61.000 --> 00:62.000\nx\n", "timing"), + (b"WEBVTT\n\n00:01.00 --> 00:02.000\nx\n", "timing"), + ( + b"WEBVTT\n\n99999999999999999999:00:00.000 --> 00:02.000\nx\n", + "timing", + ), + ] { + let error = prepare("fr", data, 0) + .expect_err("should be refused") + .to_string(); + assert!(error.contains("`fr`") && error.contains(needle), "{error}"); + } + } +} diff --git a/src/testutil.rs b/src/testutil.rs index 6dbfb6c..29212a7 100644 --- a/src/testutil.rs +++ b/src/testutil.rs @@ -53,15 +53,27 @@ pub(crate) struct Answer { pub(crate) location: Value, pub(crate) ttl_seconds: Option, pub(crate) expires_at: Option, + pub(crate) subtitles: Vec, } impl Answer { + /// Adds a `file` subtitle entry. + pub(crate) fn with_subtitle(mut self, language: &str, path: &str) -> Self { + self.subtitles.push(json!({ + "language": language, + "label": language.to_uppercase(), + "location": { "type": "file", "path": path }, + })); + self + } + pub(crate) fn file(version: &str, path: &str) -> Self { Self { version: version.to_owned(), location: json!({ "type": "file", "path": path }), ttl_seconds: Some(300), expires_at: None, + subtitles: Vec::new(), } } @@ -71,6 +83,7 @@ impl Answer { location: json!({ "type": "http", "url": url }), ttl_seconds: Some(300), expires_at: None, + subtitles: Vec::new(), } } } @@ -190,6 +203,9 @@ async fn mapper_asset( "version": answer.version, "location": answer.location, }); + if !answer.subtitles.is_empty() { + body["subtitles"] = json!(answer.subtitles); + } if let Some(ttl) = answer.ttl_seconds { body["ttl_seconds"] = json!(ttl); } diff --git a/tests/conformance.rs b/tests/conformance.rs index 6719ad9..d3da73b 100644 --- a/tests/conformance.rs +++ b/tests/conformance.rs @@ -22,7 +22,8 @@ clippy::large_futures, clippy::redundant_closure_for_method_calls, clippy::too_many_lines, - clippy::trivially_copy_pass_by_ref + clippy::trivially_copy_pass_by_ref, + reason = "harness code, not the production crate" )] use std::collections::HashMap; @@ -387,13 +388,11 @@ fn parse_fragment(data: &[u8]) -> Fragment { let flags = be32(trun) & 0x00ff_ffff; let count = be32(&trun[4..]) as usize; let mut cursor = 8; - let data_offset = if flags & 0x1 != 0 { + let data_offset = (flags & 0x1 != 0).then(|| { let value = i32::from_be_bytes(trun[cursor..cursor + 4].try_into().unwrap()); cursor += 4; - Some(value) - } else { - None - }; + value + }); if flags & 0x4 != 0 { cursor += 4; } @@ -1106,3 +1105,123 @@ fn audio_and_video_stay_in_sync_through_edit_lists() { } } } + +/// The I-frame playlist lists exactly the source's keyframes, and each listed fragment, with the +/// video init segment in front of it, decodes to one picture. +#[test] +fn every_iframe_fragment_decodes_to_exactly_one_frame() { + let ffprobe = Command::new("ffprobe").arg("-version").output().is_ok(); + let server = start_server(); + let directory = root().join("target/conformance/iframes"); + fs::create_dir_all(&directory).unwrap(); + + for (asset, file, _) in FIXTURES { + let master = get_ok(&server, &format!("/hls/{asset}/master.m3u8")).text(); + let has_video = expected_codecs(asset).0.is_some(); + let declared = master + .lines() + .find(|line| line.starts_with("#EXT-X-I-FRAME-STREAM-INF")); + assert_eq!( + declared.is_some(), + has_video, + "{asset}: only assets with video have an I-frame stream\n{master}" + ); + let Some(declared) = declared else { + assert_eq!( + get(&server, &format!("/hls/{asset}/video/iframes.m3u8")).status, + 404, + "{asset}: no video, no I-frame playlist" + ); + continue; + }; + assert!(declared.contains("CODECS=\""), "{asset}: {declared}"); + assert!(declared.contains("BANDWIDTH="), "{asset}: {declared}"); + assert!(!declared.contains("mp4a"), "{asset}: video codec only"); + + let playlist_url = format!("/hls/{asset}/video/iframes.m3u8"); + let playlist = get_ok(&server, &playlist_url); + assert_eq!( + playlist.headers["content-type"], + "application/vnd.apple.mpegurl" + ); + let text = playlist.text(); + assert!(text.contains("#EXT-X-I-FRAMES-ONLY"), "{asset}: {text}"); + assert!(text.ends_with("#EXT-X-ENDLIST\n"), "{asset}"); + let map = text + .lines() + .find_map(|line| line.strip_prefix("#EXT-X-MAP:URI=\"")) + .and_then(|rest| rest.strip_suffix('"')) + .expect("the playlist names the init segment"); + let init = get_ok(&server, &resolve(&playlist_url, map)).body; + let entries = text + .lines() + .filter(|line| line.starts_with("iframes/")) + .collect::>(); + + if ffprobe { + assert_eq!( + entries.len() as u64, + source_keyframes(&root().join("tests/fixtures").join(file)), + "{asset}: the playlist lists the source's keyframes" + ); + } + for (index, entry) in entries.iter().enumerate() { + assert!( + entry.starts_with(&format!("iframes/{index}/media.m4s?v=")), + "{asset}: {entry}" + ); + let fragment = get_ok(&server, &resolve(&playlist_url, entry)); + assert_eq!(fragment.headers["content-type"], "video/mp4"); + if !ffprobe || index % 4 != 0 { + continue; + } + let path = directory.join(format!("{asset}-{index}.mp4")); + let mut bytes = init.clone(); + bytes.extend_from_slice(&fragment.body); + fs::write(&path, bytes).unwrap(); + assert_eq!( + probe_frames(&path, "v:0"), + 1, + "{asset}: fragment {index} holds one picture" + ); + let errors = Command::new("ffmpeg") + .args(["-v", "error", "-i"]) + .arg(&path) + .args(["-f", "null", "-"]) + .output(); + if let Ok(output) = errors { + assert!( + output.stderr.is_empty(), + "{asset}: fragment {index} decode errors: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + } + let past_the_end = get( + &server, + &resolve( + &playlist_url, + &format!( + "iframes/{}/media.m4s?v={}", + entries.len(), + entries[0].rsplit('=').next().unwrap() + ), + ), + ); + assert_eq!(past_the_end.status, 404, "{asset}: no such keyframe"); + } +} + +/// The number of keyframes FFprobe finds in the first video stream. +fn source_keyframes(path: &Path) -> u64 { + let output = Command::new("ffprobe") + .args(["-v", "error", "-select_streams", "v:0"]) + .args(["-show_entries", "packet=flags", "-of", "csv=p=0"]) + .arg(path) + .output() + .expect("ffprobe should run"); + String::from_utf8_lossy(&output.stdout) + .lines() + .filter(|line| line.starts_with('K')) + .count() as u64 +} diff --git a/tests/fixtures/generate-variants.sh b/tests/fixtures/generate-variants.sh index 131ca24..9b0272b 100755 --- a/tests/fixtures/generate-variants.sh +++ b/tests/fixtures/generate-variants.sh @@ -30,6 +30,15 @@ ffmpeg_quiet \ -movflags +faststart \ "$fixture_dir/h264-aac-audio-delay.mp4" +# Video delayed by a second and a half: the video track gets a leading empty edit, so the whole +# presentation sits on a shifted timeline that sidecar subtitles must follow. +ffmpeg_quiet \ + -itsoffset 1.5 -f lavfi -i "$video" -f lavfi -i "$tone_a" \ + -c:v libx264 -pix_fmt yuv420p -preset medium -g 30 -keyint_min 30 -sc_threshold 0 -bf 2 \ + -c:a aac -profile:a aac_low -b:a 96k \ + -movflags +faststart \ + "$fixture_dir/h264-aac-video-delay.mp4" + # Two audio tracks with different languages. ffmpeg_quiet \ -f lavfi -i "$video" -f lavfi -i "$tone_a" -f lavfi -i "$tone_b" \ diff --git a/tests/fixtures/h264-aac-video-delay.mp4 b/tests/fixtures/h264-aac-video-delay.mp4 new file mode 100644 index 0000000..c387aac Binary files /dev/null and b/tests/fixtures/h264-aac-video-delay.mp4 differ diff --git a/tests/fixtures/subtitles-bad.srt b/tests/fixtures/subtitles-bad.srt new file mode 100644 index 0000000..882cb34 --- /dev/null +++ b/tests/fixtures/subtitles-bad.srt @@ -0,0 +1,3 @@ +1 +00:00:01,000 --> 00:00:02,000 +not webvtt diff --git a/tests/fixtures/subtitles-en.vtt b/tests/fixtures/subtitles-en.vtt new file mode 100644 index 0000000..1d9906a --- /dev/null +++ b/tests/fixtures/subtitles-en.vtt @@ -0,0 +1,7 @@ +WEBVTT + +00:00.500 --> 00:01.500 line:90% +Hello + +00:01.500 --> 00:02.500 +World diff --git a/tests/fixtures/subtitles-fr.vtt b/tests/fixtures/subtitles-fr.vtt new file mode 100644 index 0000000..2cb9f7f --- /dev/null +++ b/tests/fixtures/subtitles-fr.vtt @@ -0,0 +1,4 @@ +WEBVTT + +00:00.500 --> 00:01.500 +Bonjour